Punctuation (from StreamThread to PunctuationQueue)

Author: jaceklaskowski

kafka-streams-AssignedStreamsTasks.adoc

@@ -89,16 +89,32 @@ In the end, `maybeCommit` gives the number of running stream tasks that <<kafka-
 
 NOTE: `maybeCommit` is used exclusively when `TaskManager` is requested to link:kafka-streams-internals-TaskManager.adoc#maybeCommitActiveTasks[maybeCommitActiveTasks].
 
-=== [[punctuate]] `punctuate` Method
+=== [[punctuate]] Punctuating Running Stream Tasks (by Stream and System Time) -- `punctuate` Method
 
 [source, java]
 ----
 int punctuate()
 ----
 
-`punctuate`...FIXME
+`punctuate` walks over the <<kafka-streams-internals-AssignedTasks.adoc#running, running stream tasks>> and requests them to <<kafka-streams-internals-StreamTask.adoc#maybePunctuateStreamTime, maybePunctuateStreamTime>> and <<kafka-streams-internals-StreamTask.adoc#maybePunctuateSystemTime, maybePunctuateSystemTime>>.
 
-NOTE: `punctuate` is used exclusively when `TaskManager` is requested to link:kafka-streams-internals-TaskManager.adoc#punctuate[punctuate].
+For every positive response (<<kafka-streams-internals-StreamTask.adoc#maybePunctuateStreamTime, maybePunctuateStreamTime>> or <<kafka-streams-internals-StreamTask.adoc#maybePunctuateSystemTime, maybePunctuateSystemTime>> returned `true`), an internal `punctuated` counter is incremented.
+
+In the end, `punctuate` returns the internal `punctuated` counter.
+
+In case of a `TaskMigratedException`, `punctuate` prints out the following INFO message to the logs, <<kafka-streams-internals-AssignedTasks.adoc#closeZombieTask, closes the zombie task>>, and possibly removes the task from the <<kafka-streams-internals-AssignedTasks.adoc#running, running stream tasks>>.
+
+```
+Failed to punctuate stream task [taskId] since it got migrated to another thread already. Closing it as zombie before triggering a new rebalance.
+```
+
+In case of a `KafkaException`, `punctuate` prints out the following ERROR message to the logs and re-throws the exception.
+
+```
+Failed to punctuate stream task [taskId] due to the following error:
+```
+
+NOTE: `punctuate` is used exclusively when `TaskManager` is requested to <<kafka-streams-internals-TaskManager.adoc#punctuate, punctuate stream tasks>>.
 
 === [[updateRestored]] `updateRestored` Method
 

kafka-streams-PunctuationQueue.adoc

@@ -1,17 +1,22 @@
 == [[PunctuationQueue]] PunctuationQueue
 
-`PunctuationQueue` is...FIXME
+[[pq]]
+`PunctuationQueue` manages a https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/PriorityQueue.html[java.util.PriorityQueue] of <<kafka-streams-PunctuationSchedule.adoc#, PunctuationSchedules>>.
 
-=== [[mayPunctuate]] `mayPunctuate` Method
+=== [[mayPunctuate]] Attempting to Punctuate -- `mayPunctuate` Method
 
 [source, java]
 ----
 boolean mayPunctuate(
-  final long timestamp,
-  final PunctuationType type,
-  final ProcessorNodePunctuator processorNodePunctuator)
+  long timestamp,
+  PunctuationType type,
+  ProcessorNodePunctuator processorNodePunctuator)
 ----
 
-`mayPunctuate`...FIXME
+`mayPunctuate` takes the <<kafka-streams-PunctuationSchedule.adoc#, PunctuationSchedules>> off the <<pq, PriorityQueue>> for which the <<kafka-streams-Stamped.adoc#timestamp, timestamp>> is older (_smaller_) than the given `timestamp`.
 
-NOTE: `mayPunctuate` is used when `StreamTask` is requested to link:kafka-streams-internals-StreamTask.adoc#maybePunctuateStreamTime[maybePunctuateStreamTime] and link:kafka-streams-internals-StreamTask.adoc#maybePunctuateSystemTime[maybePunctuateSystemTime].
+`mayPunctuate` then requests the given <<kafka-streams-ProcessorNodePunctuator.adoc#, ProcessorNodePunctuator>> to <<kafka-streams-ProcessorNodePunctuator.adoc#punctuate, punctuate>> (with the <<kafka-streams-PunctuationSchedule.adoc#node, node>> and the <<kafka-streams-PunctuationSchedule.adoc#punctuator, punctuator>> of every `PunctuationSchedule`, the given `timestamp` and the `PunctuationType`).
+
+In the end, `mayPunctuate` returns whether a `PunctuationSchedule` was punctuated (`true`) or not (`false`).
+
+NOTE: `mayPunctuate` is used when `StreamTask` is requested to attempt to punctuate by <<kafka-streams-internals-StreamTask.adoc#maybePunctuateStreamTime, stream>> and <<kafka-streams-internals-StreamTask.adoc#maybePunctuateSystemTime, system>> time.

kafka-streams-internals-PartitionGroup.adoc

@@ -67,7 +67,7 @@ int numBuffered(final TopicPartition partition)
 
 NOTE: `numBuffered` is used when...FIXME
 
-=== [[timestamp]] Minimum Partition Timestamp Across All TopicPartitions -- `timestamp` Method
+=== [[timestamp]] Minimum Partition Timestamp Across All Partitions -- `timestamp` Method
 
 [source, java]
 ----

kafka-streams-internals-StreamTask.adoc

@@ -290,18 +290,24 @@ NOTE: `punctuate` is part of link:kafka-streams-ProcessorNodePunctuator.adoc#pun
 
 `punctuate`...FIXME
 
-=== [[maybePunctuateStreamTime]] `maybePunctuateStreamTime` Method
+=== [[maybePunctuateStreamTime]] Attempting to Punctuate by Stream Time -- `maybePunctuateStreamTime` Method
 
 [source, java]
 ----
 boolean maybePunctuateStreamTime()
 ----
 
-`maybePunctuateStreamTime`...FIXME
+`maybePunctuateStreamTime` requests the <<partitionGroup, PartitionGroup>> for the <<kafka-streams-internals-PartitionGroup.adoc#timestamp, minimum partition timestamp across all partitions>>.
 
-NOTE: `maybePunctuateStreamTime` is used exclusively when `AssignedStreamsTasks` is requested to link:kafka-streams-AssignedStreamsTasks.adoc#punctuate[punctuate].
+`maybePunctuateStreamTime` requests the <<streamTimePunctuationQueue, stream-time PunctuationQueue>> to <<kafka-streams-PunctuationQueue.adoc#mayPunctuate, mayPunctuate>> with the minimum timestamp.
 
-=== [[maybePunctuateSystemTime]] `maybePunctuateSystemTime` Method
+In the end, `maybePunctuateStreamTime` returns whatever the stream-time `PunctuationQueue` returned.
+
+If the minimum timestamp is <<kafka-streams-internals-RecordQueue.adoc#UNKNOWN, UNKNOWN>>, `maybePunctuateStreamTime` returns `false`.
+
+NOTE: `maybePunctuateStreamTime` is used exclusively when `AssignedStreamsTasks` is requested to <<kafka-streams-AssignedStreamsTasks.adoc#punctuate, punctuate running stream tasks>>.
+
+=== [[maybePunctuateSystemTime]] Attempting to Punctuate by System Time -- `maybePunctuateSystemTime` Method
 
 [source, java]
 ----
@@ -310,7 +316,7 @@ boolean maybePunctuateSystemTime()
 
 `maybePunctuateSystemTime`...FIXME
 
-NOTE: `maybePunctuateSystemTime` is used exclusively when `AssignedStreamsTasks` is requested to link:kafka-streams-AssignedStreamsTasks.adoc#punctuate[punctuate].
+NOTE: `maybePunctuateSystemTime` is used exclusively when `AssignedStreamsTasks` is requested to <<kafka-streams-AssignedStreamsTasks.adoc#punctuate, punctuate running stream tasks>>.
 
 === [[schedule]] `schedule` Method
 

kafka-streams-internals-StreamThread.adoc

@@ -520,17 +520,6 @@ In the end, `create` creates a <<creating-instance, StreamThread>>.
 
 NOTE: `create` is used exclusively when `KafkaStreams` is <<kafka-streams-KafkaStreams.adoc#, created>>.
 
-=== [[punctuate]] `punctuate` Internal Method
-
-[source, java]
-----
-void punctuate()
-----
-
-`punctuate`...FIXME
-
-NOTE: `punctuate` is used when `StreamThread` is requested to <<run, start>> (and <<runOnce, polls records once>> and <<processAndMaybeCommit, processes records (with optional commit)>>).
-
 === [[enforceRebalance]] Enforcing Rebalance -- `enforceRebalance` Internal Method
 
 [source, java]
@@ -577,6 +566,21 @@ Committed all active tasks [activeTaskIds] and standby tasks [standbyTaskIds] in
 
 NOTE: `maybeCommit` is used when `StreamThread` is requested to <<runOnce, poll records once>> (directly and indirectly in <<processAndMaybeCommit, processAndMaybeCommit>>).
 
+=== [[maybePunctuate]] Attempting to Punctuate (Running Stream Tasks) -- `maybePunctuate` Internal Method
+
+[source, java]
+----
+boolean maybePunctuate()
+----
+
+`maybePunctuate` requests the <<taskManager, TaskManager>> to <<kafka-streams-internals-TaskManager.adoc#punctuate, punctuate>>.
+
+If the punctuate returned a positive number (greater than `0`), `maybePunctuate` <<advanceNowAndComputeLatency, advanceNowAndComputeLatency>> and requests the <<streamsMetrics, StreamsMetricsThreadImpl>> for the <<kafka-streams-StreamsMetricsThreadImpl.adoc#punctuateTimeSensor, punctuateTimeSensor>> to record the punctuate time.
+
+In the end, `maybePunctuate` returns whether the punctuate returned a positive number (`true`) or not (`false`).
+
+NOTE: `maybePunctuate` is used exclusively when `StreamThread` is requested to <<runOnce, poll records once and add them to active stream tasks>>.
+
 === [[internal-properties]] Internal Properties
 
 [cols="30m,70",options="header",width="100%"]
@@ -591,10 +595,14 @@ a| [[builder]] <<kafka-streams-internals-InternalTopologyBuilder.adoc#, Internal
 a| [[lastCommitMs]] Time of the last <<maybeCommit, commit>>
 
 | numIterations
-a| [[numIterations]]
+a| [[numIterations]] Number of iterations when the <<taskManager, TaskManager>> is requested to <<kafka-streams-internals-TaskManager.adoc#process, process records by running stream tasks (one record per task)>> while `StreamThread` is <<runOnce, polling records once and adding them to active stream tasks>>
 
 Default: `1`
 
+Incremented while <<runOnce, polling records once and adding them to active stream tasks>>
+
+Decremented by half while <<runOnce, polling records once and adding them to active stream tasks>>
+
 | processStandbyRecords
 a| [[processStandbyRecords]] Flag to control whether to <<maybeUpdateStandbyTasks, maybeUpdateStandbyTasks>> after <<maybeCommit, maybeCommit>> was executed
 

kafka-streams-internals-TaskManager.adoc

@@ -358,16 +358,16 @@ In the end, `maybeCommitActiveTasks` gives the number of running stream tasks th
 
 NOTE: `maybeCommitActiveTasks` is used exclusively when `StreamThread` is requested to link:kafka-streams-internals-StreamThread.adoc#processAndMaybeCommit[processAndMaybeCommit].
 
-=== [[punctuate]] `punctuate` Method
+=== [[punctuate]] Punctuating Stream Tasks -- `punctuate` Method
 
 [source, java]
 ----
 int punctuate()
 ----
 
-`punctuate` simply requests <<active, AssignedStreamsTasks>> to link:kafka-streams-AssignedStreamsTasks.adoc#punctuate[punctuate].
+`punctuate` simply requests the <<active, AssignedStreamsTasks>> to <<kafka-streams-AssignedStreamsTasks.adoc#punctuate, punctuate>>.
 
-NOTE: `punctuate` is used exclusively when `StreamThread` is requested to link:kafka-streams-internals-StreamThread.adoc#punctuate[punctuate].
+NOTE: `punctuate` is used exclusively when `StreamThread` is requested to <<kafka-streams-internals-StreamThread.adoc#maybePunctuate, attempt to punctuate>>.
 
 === [[commitAll]] Committing Active and Standby Tasks -- `commitAll` Method