StreamTask and the current record (PartitionGroup, ProcessorNode, and Processor)

Author: jaceklaskowski

kafka-streams-Processor.adoc

@@ -35,9 +35,9 @@ a| [[init]]
 void init(ProcessorContext context)
 ----
 
-Initializes the processor (with a <<kafka-streams-ProcessorContext.adoc#, ProcessorContext>>)
+Initializes the processor with a <<kafka-streams-ProcessorContext.adoc#, ProcessorContext>> (that can be used to decide whether a record should be <<kafka-streams-ProcessorContext.adoc#forward, forwarded downstream>> to child processors if there are any)
 
-Used exclusively when `ProcessorNode` is requested to <<kafka-streams-internals-ProcessorNode.adoc#init, init>>.
+Used exclusively when `ProcessorNode` is requested to <<kafka-streams-internals-ProcessorNode.adoc#init, init>>
 
 | process
 a| [[process]]
@@ -47,9 +47,9 @@ a| [[process]]
 void process(K key, V value)
 ----
 
-Processes a single record
+Processes a single record (as a pair of a `K` key and a `V` value)
 
-Used exclusively when `ProcessorNode` is requested to <<kafka-streams-internals-ProcessorNode.adoc#process, process>>.
+Used exclusively when `ProcessorNode` is requested to <<kafka-streams-internals-ProcessorNode.adoc#process, process a single record (as a key and value pair)>>.
 
 |===
 

kafka-streams-ProcessorContext.adoc

@@ -60,7 +60,7 @@ void forward(final K key, final V value)
 void forward(final K key, final V value, final To to)
 ----
 
-Forwards a record to downstream processors
+Forwards a record downstream (to child processors if there are any)
 
 | getStateStore
 a| [[getStateStore]]

kafka-streams-internals-PartitionGroup.adoc

@@ -1,6 +1,6 @@
 == [[PartitionGroup]] PartitionGroup
 
-`PartitionGroup` manages <<partitionQueues, RecordQueues per partition>> assigned to a <<kafka-streams-internals-StreamTask.adoc#, StreamTask>>.
+`PartitionGroup` is a collection of <<partitionQueues, RecordQueues>> (one per every <<kafka-streams-internals-StreamTask.adoc#partitions, partition>> assigned to a <<kafka-streams-internals-StreamTask.adoc#, StreamTask>>).
 
 `PartitionGroup` is <<creating-instance, created>> exclusively for a <<kafka-streams-internals-StreamTask.adoc#partitionGroup, StreamTask>>.
 

kafka-streams-internals-ProcessorNode.adoc

@@ -1,6 +1,6 @@
 == [[ProcessorNode]] ProcessorNode
 
-`ProcessorNode` is a <<process, processing node>> in a topology that is identified by a <<name, name>> and have zero or more <<children, child processor nodes>>.
+`ProcessorNode` is a <<process, processing node>> in a processor topology that is identified by a <<name, name>> and have zero or more <<children, child processor nodes>>.
 
 NOTE: link:kafka-streams-internals-SourceNode.adoc[SourceNode] and link:kafka-streams-internals-SinkNode.adoc[SinkNode] are specialized `ProcessorNodes`.
 
@@ -12,29 +12,6 @@ NOTE: link:kafka-streams-internals-SourceNode.adoc[SourceNode] and link:kafka-st
 
 `ProcessorNode` has a <<toString, human-friendly / textual representation>> that is particularly helpful for debugging.
 
-[[internal-registries]]
-.ProcessorNode's Internal Properties (e.g. Registries, Counters and Flags)
-[cols="1,2",options="header",width="100%"]
-|===
-| Name
-| Description
-
-| `children`
-| [[children]] Child `ProcessorNodes`
-
-Used when...FIXME
-
-| `nodeMetrics`
-| [[nodeMetrics]] link:kafka-streams-NodeMetrics.adoc[NodeMetrics]
-
-Used when...FIXME
-
-| `time`
-| [[time]] `Time`
-
-Used when...FIXME
-|===
-
 === [[init]] `init` Method
 
 [source, java]
@@ -53,14 +30,16 @@ void init(ProcessorContext context)
 * `StreamTask` is requested to <<kafka-streams-internals-StreamTask.adoc#initTopology, initTopology>>
 ====
 
-=== [[process]] Processing Record -- `process` Method
+=== [[process]] Processing Single Record (As Key and Value Pair) -- `process` Method
 
 [source, java]
 ----
 void process(final K key, final V value)
 ----
 
-`process`...FIXME
+`process` requests the <<processor, Processor>> to <<kafka-streams-Processor.adoc#process, process the given key and value pair>>.
+
+In the end, `process` requests the <<nodeMetrics, NodeMetrics>> for the <<kafka-streams-NodeMetrics.adoc#nodeProcessTimeSensor, nodeProcessTimeSensor>> to record the processing time.
 
 [NOTE]
 ====
@@ -70,18 +49,18 @@ void process(final K key, final V value)
 
 * `GlobalStateUpdateTask` is requested to <<kafka-streams-internals-GlobalStateUpdateTask.adoc#update, update>>
 
-* `StreamTask` is requested to <<kafka-streams-internals-StreamTask.adoc#process, process>>
+* `StreamTask` is requested to <<kafka-streams-internals-StreamTask.adoc#process, process a single record>>
 ====
 
 === [[creating-instance]] Creating ProcessorNode Instance
 
-`ProcessorNode` takes the following when created:
+`ProcessorNode` takes the following to be created:
 
 * [[name]] Name
-* [[processor]] link:kafka-streams-Processor.adoc[Processor] (of `K` keys and `V` values)
+* [[processor]] <<kafka-streams-Processor.adoc#, Processor>> (`Processor<K, V>` of `K` keys and `V` values)
 * [[stateStores]] Names of the associated state stores
 
-`ProcessorNode` initializes the <<internal-registries, internal registries and counters>>.
+`ProcessorNode` initializes the <<internal-properties, internal properties>>.
 
 === [[addChild]] Adding Child Processor Node -- `addChild` Method
 
@@ -130,3 +109,26 @@ void punctuate(final long timestamp, final Punctuator punctuator)
 `punctuate` requests the <<nodeMetrics, NodeMetrics>> for the link:kafka-streams-NodeMetrics.adoc#metrics[StreamsMetricsImpl] that is requested to link:kafka-streams-internals-StreamsMetricsImpl.adoc#measureLatencyNs[measureLatencyNs] (that records it with the link:kafka-streams-NodeMetrics.adoc#nodePunctuateTimeSensor[nodePunctuateTimeSensor] of the <<nodeMetrics, NodeMetrics>>).
 
 NOTE: `punctuate` is used exclusively when `StreamTask` is requested to link:kafka-streams-internals-StreamTask.adoc#punctuate[execute a scheduled periodic action].
+
+=== [[internal-properties]] Internal Properties
+
+[cols="30m,70",options="header",width="100%"]
+|===
+| Name
+| Description
+
+| children
+| [[children]] Child `ProcessorNodes`
+
+Used when...FIXME
+
+| nodeMetrics
+| [[nodeMetrics]] <<kafka-streams-NodeMetrics.adoc#, NodeMetrics>>
+
+Used when...FIXME
+
+| time
+| [[time]] `Time`
+
+Used when...FIXME
+|===

kafka-streams-internals-StreamTask.adoc

@@ -1,6 +1,10 @@
 == [[StreamTask]] StreamTask
 
-`StreamTask` is a concrete <<kafka-streams-internals-AbstractTask.adoc#, stream processor task>> that uses a <<partitionGroup, PartitionGroup>> (with the <<partitions, partitions assigned>>) to <<process, process records>> (one at a time) ordered by partition timestamp.
+`StreamTask` is a concrete <<kafka-streams-internals-AbstractTask.adoc#, stream processor task>> that uses a <<partitionGroup, PartitionGroup>> (with the <<partitions, partitions assigned>>) to determine which record should be <<process, processed>> (as ordered by partition timestamp).
+
+When requested to <<process, process a single record>>, `StreamTask` requests the <<partitionGroup, PartitionGroup>> for the <<kafka-streams-internals-PartitionGroup.adoc#nextRecord, next stamped record (record with timestamp) and the RecordQueue>>. `StreamTask` uses a <<recordInfo, RecordInfo>> to hold the <<kafka-streams-internals-RecordQueue.adoc#, RecordQueue>> (with the source <<kafka-streams-internals-ProcessorNode.adoc#, processor node>> and the partition) of the currently-processed stamped record. Eventually, `StreamTask` requests the source processor node (of the `RecordQueue` and the partition) to <<kafka-streams-internals-ProcessorNode.adoc#process, process the record>>.
+
+NOTE: It is at the discretion of a processor node (incl. a source processor node) to <<kafka-streams-ProcessorContext.adoc#forward, forward the record downstream>> (to child processors if there are any).
 
 `StreamTask` is <<creating-instance, created>> exclusively when `TaskCreator` is requested to <<kafka-streams-internals-TaskCreator.adoc#createTask, create one>>.
 
@@ -100,7 +104,7 @@ NOTE: `close` is part of link:kafka-streams-internals-Task.adoc#close[Task Contr
 `StreamTask` takes the following to be created:
 
 * [[id]] <<kafka-streams-TaskId.adoc#, TaskId>>
-* [[partitions]] Topic partitions (Kafka `TopicPartition`)
+* [[partitions]] Partitions (Kafka https://kafka.apache.org/22/javadoc/org/apache/kafka/common/TopicPartition.html[TopicPartition])
 * [[topology]] <<kafka-streams-internals-ProcessorTopology.adoc#, ProcessorTopology>>
 * [[consumer]] Kafka https://kafka.apache.org/22/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html[Consumer] (`Consumer<byte[], byte[]>`)
 * [[changelogReader]] <<kafka-streams-ChangelogReader.adoc#, ChangelogReader>>
@@ -493,7 +497,6 @@ NOTE: `producerMetrics` is used when...FIXME
 
 === [[internal-properties]] Internal Properties
 
-.StreamTask's Internal Properties (e.g. Registries, Counters and Flags)
 [cols="30m,70",options="header",width="100%"]
 |===
 | Name
@@ -506,10 +509,12 @@ NOTE: `producerMetrics` is used when...FIXME
 a| [[idleStartTime]]
 
 | partitionGroup
-a| [[partitionGroup]] <<kafka-streams-internals-PartitionGroup.adoc#, PartitionGroup>> (with <<kafka-streams-internals-RecordQueue.adoc#, RecordQueues>> per <<partitions, partition assigned>>)
+a| [[partitionGroup]] <<kafka-streams-internals-PartitionGroup.adoc#, PartitionGroup>> (with <<kafka-streams-internals-RecordQueue.adoc#, RecordQueues>> per every <<partitions, partition assigned>>)
 
 Used when `StreamTask` is requested for the following:
 
+* <<addRecords, Buffer new records (from a partition)>>
+
 * <<isProcessable, isProcessable>>
 
 * <<process, Process a single record>>
@@ -518,8 +523,6 @@ Used when `StreamTask` is requested for the following:
 
 * <<closeSuspended, closeSuspended>>
 
-* <<addRecords, Buffer new records (from a partition)>>
-
 * <<numBuffered, numBuffered>>
 
 * <<maybePunctuateStreamTime, maybePunctuateStreamTime>>
@@ -543,7 +546,9 @@ Used for the following:
 * <<producerMetrics, producerMetrics>>
 
 | recordInfo
-a| [[recordInfo]] <<kafka-streams-internals-RecordInfo.adoc#, RecordInfo>>
+a| [[recordInfo]] <<kafka-streams-internals-RecordInfo.adoc#, RecordInfo>> (that holds a <<kafka-streams-internals-RecordQueue.adoc#, RecordQueue>> with the source <<kafka-streams-internals-ProcessorNode.adoc#, processor node>> and the partition the <<process, currently-processed stamped record>> came from)
+
+Created empty alongside the <<StreamTask, StreamTask>> and _"fill up"_ with the <<kafka-streams-internals-RecordQueue.adoc#, RecordQueue>> when requested to <<process, process a single record>>
 
 | streamTimePunctuationQueue
 a| [[streamTimePunctuationQueue]] <<kafka-streams-PunctuationQueue.adoc#, PunctuationQueue>>