StreamTask.commit -- Committing Task

Author: jaceklaskowski

SUMMARY.adoc

@@ -162,6 +162,8 @@
 .. link:kafka-streams-StreamsMetricsImpl.adoc[StreamsMetricsImpl]
 .. link:kafka-streams-StreamsMetricsThreadImpl.adoc[StreamsMetricsThreadImpl]
 
+. link:kafka-streams-StreamTask-TaskMetrics.adoc[TaskMetrics]
+
 == Testing
 
 . link:kafka-streams-TopologyTestDriver.adoc[TopologyTestDriver]

kafka-streams-AbstractTask.adoc

@@ -198,3 +198,12 @@ void reinitializeStateStoresForPartitions(final Collection<TopicPartition> parti
 
 * `StreamThread` is requested to <<kafka-streams-StreamThread.adoc#maybeUpdateStandbyTasks, maybeUpdateStandbyTasks>>
 ====
+
+=== [[activeTaskCheckpointableOffsets]] Checkpointable Offsets -- `activeTaskCheckpointableOffsets` Method
+
+[source, java]
+----
+Map<TopicPartition, Long> activeTaskCheckpointableOffsets()
+----
+
+`activeTaskCheckpointableOffsets` simply returns an empty collection (of checkpointable offsets).

kafka-streams-StreamTask-TaskMetrics.adoc

@@ -0,0 +1,3 @@
+== [[TaskMetrics]] TaskMetrics
+
+`TaskMetrics` is...FIXME

kafka-streams-StreamTask.adoc

@@ -41,6 +41,9 @@ Used when `StreamTask` is requested for the following:
 
 | consumedOffsets
 | [[consumedOffsets]] Offsets by https://kafka.apache.org/20/javadoc/org/apache/kafka/common/TopicPartition.html[TopicPartitions] (`Map<TopicPartition, Long>`) that `StreamTask` has <<process, processed>> successfully
+
+| taskMetrics
+| [[taskMetrics]] <<kafka-streams-StreamTask-TaskMetrics.adoc#, TaskMetrics>>
 |===
 
 [[logging]]
@@ -85,19 +88,6 @@ NOTE: `suspend` is part of <<kafka-streams-Task.adoc#suspend, Task Contract>> to
 
 NOTE: The private `suspend` is used exclusively when `StreamTask` is requested to <<close, close>>.
 
-=== [[commit]] `commit` Method
-
-[source, java]
-----
-void commit() // <1>
-void commit(final boolean startNewTransaction)
-----
-<1> Calls `commit` with `startNewTransaction` flag on
-
-NOTE: `commit` is part of <<kafka-streams-Task.adoc#commit, Task Contract>> to...FIXME.
-
-`commit`...FIXME
-
 === [[close]] `close` Method
 
 [source, java]
@@ -362,3 +352,50 @@ void commitOffsets(final boolean startNewTransaction)
 `commitOffsets`...FIXME
 
 NOTE: `commitOffsets` is used exclusively when `StreamTask` is requested to <<commit, commit>>.
+
+=== [[commit]] Committing Task -- `commit` Method
+
+[source, java]
+----
+void commit()
+----
+
+NOTE: `commit` is part of <<kafka-streams-Task.adoc#commit, Task Contract>> to commit the task.
+
+`commit` simply <<commit-startNewTransaction, commits>> with the `startNewTransaction` flag on.
+
+=== [[commit-startNewTransaction]] `commit` Internal Method
+
+[source, java]
+----
+void commit(final boolean startNewTransaction)
+----
+
+`commit` prints out the following DEBUG message to the logs:
+
+```
+Committing
+```
+
+`commit` <<flushState, flushState>>.
+
+(only when <<kafka-streams-AbstractTask.adoc#eosEnabled, exactly-once support>> is disabled) `commit` requests the <<stateMgr, ProcessorStateManager>> to <<kafka-streams-ProcessorStateManager.adoc#checkpoint, checkpoint>> with the <<activeTaskCheckpointableOffsets, checkpointable offsets>>.
+
+`commit` <<commitOffsets, commitOffsets>> with the input `startNewTransaction` flag.
+
+`commit` turns the <<commitRequested, commitRequested>> internal flag off.
+
+In the end, `commit` requests the <<taskMetrics, TaskMetrics>> for the <<taskCommitTimeSensor, taskCommitTimeSensor>> and records the duration (i.e. the time since `commit` was executed).
+
+NOTE: `commit` is used when `StreamTask` is requested to <<commit, commit>> (that turns the input `startNewTransaction` flag on) and <<suspend, suspend>> (with the input `startNewTransaction` flag off).
+
+=== [[activeTaskCheckpointableOffsets]] `activeTaskCheckpointableOffsets` Method
+
+[source, java]
+----
+Map<TopicPartition, Long> activeTaskCheckpointableOffsets()
+----
+
+NOTE: `activeTaskCheckpointableOffsets` is part of the <<kafka-streams-AbstractTask.adoc#activeTaskCheckpointableOffsets, AbstractTask Contract>> to return the checkpointable offsets.
+
+`activeTaskCheckpointableOffsets`...FIXME

kafka-streams-StreamThread.adoc

@@ -605,7 +605,7 @@ Committing all active tasks [activeTaskIds] and standby tasks [standbyTaskIds] s
 
 Only if there are still running active and standby tasks, `maybeCommit` does the following:
 
-. Requests the <<streamsMetrics, StreamsMetricsThreadImpl>> for <<kafka-streams-StreamsMetricsThreadImpl.adoc#commitTimeSensor, commitTimeSensor>> and records the commit time (as the latency of committing all the tasks by their number)
+. Requests the <<streamsMetrics, StreamsMetricsThreadImpl>> for the <<kafka-streams-StreamsMetricsThreadImpl.adoc#commitTimeSensor, commitTimeSensor>> and records the commit time (as the latency of committing all the tasks by their number)
 
 . Requests the <<taskManager, TaskManager>> to <<kafka-streams-TaskManager.adoc#maybePurgeCommitedRecords, maybePurgeCommitedRecords>>
 

kafka-streams-Task.adoc

@@ -115,6 +115,8 @@ a| [[commit]]
 void commit()
 ----
 
+Commits the task
+
 Used when:
 
 * `AssignedStreamsTasks` is requested to execute the <<kafka-streams-AssignedStreamsTasks.adoc#maybeCommitAction, maybeCommitAction>> action