StreamThread et al.

Author: jaceklaskowski

SUMMARY.adoc

@@ -8,7 +8,7 @@
 . link:kafka-streams-multi-instance-kafka-streams-application.adoc[Multi-Instance Kafka Streams Applications]
 
 . link:kafka-streams-exactly-once-support-eos.adoc[Exactly-Once Support (EOS)]
-. link:kafka-streams-StreamThreads-StreamTasks-and-StandbyTasks.adoc[StreamThreads, StreamTasks and StandbyTasks]
+. link:kafka-streams-StreamThreads-StreamTasks-and-StandbyTasks.adoc[KafkaStreams, StreamThreads, StreamTasks and StandbyTasks]
 
 == Demos
 

graffles/kafka-streams-StreamThread-processAndMaybeCommit.graffle

@@ -52,7 +52,8 @@ Refer to link:kafka-logging.adoc#log4j.properties[Application Logging Using log4
 
 [source, java]
 ----
-void onPartitionsAssigned(final Collection<TopicPartition> assignment)
+void onPartitionsAssigned(
+  Collection<TopicPartition> assignment)
 ----
 
 NOTE: `onPartitionsAssigned` is part of `ConsumerRebalanceListener` Contract in Apache Kafka to...FIXME.
@@ -82,7 +83,8 @@ partition assignment took [duration] ms.
 
 [source, java]
 ----
-void onPartitionsRevoked(final Collection<TopicPartition> assignment)
+void onPartitionsRevoked(
+  Collection<TopicPartition> assignment)
 ----
 
 NOTE: `onPartitionsRevoked` is part of `ConsumerRebalanceListener` Contract in Apache Kafka to...FIXME.

images/kafka-streams-StreamThread-processAndMaybeCommit.png

@@ -1,6 +1,10 @@
-== StreamThreads, StreamTasks and StandbyTasks
+== KafkaStreams, StreamThreads, StreamTasks and StandbyTasks
 
-<<kafka-streams-StreamsConfig.adoc#NUM_STREAM_THREADS_CONFIG, StreamsConfig.NUM_STREAM_THREADS_CONFIG>> (`num.stream.threads`) is the number of <<kafka-streams-internals-StreamThread.adoc#, StreamThreads>> to execute stream processing.
+<<kafka-streams-StreamsConfig.adoc#NUM_STREAM_THREADS_CONFIG, StreamsConfig.NUM_STREAM_THREADS_CONFIG>> (`num.stream.threads`) is the number of <<kafka-streams-internals-StreamThread.adoc#, StreamThreads>> that a single <<kafka-streams-KafkaStreams.adoc#, KafkaStreams>> instance creates for stream processing.
+
+With that, there will always be at least two threads running for a Kafka Streams application - the main thread of the application and one or many `StreamThreads`.
+
+Every `StreamThread` manages its own <<kafka-streams-internals-TaskManager.adoc#, TaskManager>> (with the factories of <<kafka-streams-internals-TaskCreator.adoc#, StreamTasks>> and <<kafka-streams-internals-StandbyTaskCreator.adoc#, StandbyTasks>>).
 
 * Number of StreamTasks = # TaskCreator.createTask = # AbstractTaskCreator.createTasks = TaskManager.addStreamTasks
 
@@ -21,3 +25,5 @@ DEBUG Assigning tasks {} to clients {} with number of replicas {}
 Step 1. StreamsPartitionAssignor.onAssignment(final Assignment assignment) —> TaskManager.setAssignmentMetadata(Map<TaskId, Set<TopicPartition>> activeTasks, Map<TaskId, Set<TopicPartition>> standbyTasks)
 
 Step 2. RebalanceListener.onPartitionsAssigned(Collection<TopicPartition> assignment) —> TaskManager.createTasks(assignment)
+
+`StreamsPartitionAssignor` makes sure that the <<kafka-streams-internals-StreamsPartitionAssignor.adoc#processVersionOneAssignment, number of assigned partitions to a Kafka Streams application instance is exactly the same as number of active tasks>>.

kafka-streams-StreamThread-RebalanceListener.adoc

@@ -1,11 +1,11 @@
 == [[InternalTopologyBuilder]] InternalTopologyBuilder
 
-`InternalTopologyBuilder` is used to <<build, build a topology of processor nodes>> (for <<kafka-streams-internals-StandbyTaskCreator.adoc#, StreamThread.StandbyTaskCreators>> and <<kafka-streams-internals-TaskCreator.adoc#, StreamThread.TaskCreator>>).
+`InternalTopologyBuilder` is used to <<build, build a topology of processor nodes>> (for <<kafka-streams-internals-StandbyTaskCreator.adoc#, StreamThread.StandbyTaskCreator>> and <<kafka-streams-internals-TaskCreator.adoc#, StreamThread.TaskCreator>> task factories and hence <<kafka-streams-internals-Task.adoc#, stream processor tasks>> themselves).
 
 .InternalTopologyBuilder and ProcessorTopology
 image::images/kafka-streams-InternalTopologyBuilder-build.png[align="center"]
 
-`InternalTopologyBuilder` is <<creating-instance, created>> exclusively for a <<kafka-streams-Topology.adoc#internalTopologyBuilder, Topology>>.
+`InternalTopologyBuilder` is <<creating-instance, created>> exclusively for a <<kafka-streams-Topology.adoc#internalTopologyBuilder, Topology>> (which is simply the frontend to the internals of topology development).
 
 .InternalTopologyBuilder and Topology
 image::images/kafka-streams-InternalTopologyBuilder-Topology.png[align="center"]

kafka-streams-StreamThreads-StreamTasks-and-StandbyTasks.adoc

@@ -274,6 +274,35 @@ Map<String, Assignment> computeNewAssignment(
 
 NOTE: `computeNewAssignment` is used exclusively when `StreamsPartitionAssignor` is requested to <<assign, perform partition assignment>>.
 
+=== [[processVersionOneAssignment]] `processVersionOneAssignment` Internal Method
+
+[source, java]
+----
+void processVersionOneAssignment(
+  AssignmentInfo info,
+  List<TopicPartition> partitions,
+  Map<TaskId, Set<TopicPartition>> activeTasks)
+----
+
+`processVersionOneAssignment`...FIXME
+
+NOTE: `processVersionOneAssignment` is used when `StreamsPartitionAssignor` is requested to <<onAssignment, handle partition assignment from a group leader>> (for version 1) and <<processVersionTwoAssignment, process partition assignment (version 2)>>.
+
+=== [[processVersionTwoAssignment]] `processVersionTwoAssignment` Internal Method
+
+[source, java]
+----
+void processVersionTwoAssignment(
+  AssignmentInfo info,
+  List<TopicPartition> partitions,
+  Map<TaskId, Set<TopicPartition>> activeTasks,
+  Map<TopicPartition, PartitionInfo> topicToPartitionInfo)
+----
+
+`processVersionTwoAssignment`...FIXME
+
+NOTE: `processVersionTwoAssignment` is used when...FIXME
+
 === [[internal-properties]] Internal Properties
 
 [cols="30m,70",options="header",width="100%"]

kafka-streams-internals-InternalTopologyBuilder.adoc

@@ -398,16 +398,23 @@ int commitAll()
 
 NOTE: `commitAll` is used exclusively when `StreamThread` is requested to <<kafka-streams-internals-StreamThread.adoc#maybeCommit, commit all tasks (when commit interval elapsed)>>.
 
-=== [[activeTaskIds]] `activeTaskIds` Method
+=== [[activeTaskIds]] All Active Tasks -- `activeTaskIds` Method
 
 [source, java]
 ----
 Set<TaskId> activeTaskIds()
 ----
 
-`activeTaskIds`...FIXME
+`activeTaskIds` simply requests the <<active, AssignedStreamsTasks>> for the <<kafka-streams-internals-AssignedStreamsTasks.adoc#allAssignedTaskIds, assigned task IDs>>.
 
-NOTE: `activeTaskIds` is used when...FIXME
+[NOTE]
+====
+`activeTaskIds` is used when:
+
+* `StreamThread` is requested to <<kafka-streams-internals-StreamThread.adoc#maybeCommit, commit all active tasks (when commit interval elapsed)>>
+
+* `RebalanceListener` is requested to handle <<kafka-streams-StreamThread-RebalanceListener.adoc#onPartitionsAssigned, onPartitionsAssigned>> and <<kafka-streams-StreamThread-RebalanceListener.adoc#onPartitionsRevoked, onPartitionsRevoked>> events
+====
 
 === [[standbyTaskIds]] `standbyTaskIds` Method