This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new c31c9215e13 MINOR: Improve the Kafka Streams state listener JavaDocs
(#23011)
c31c9215e13 is described below
commit c31c9215e131f8c17e79f8901b48c13ee6aa8e7a
Author: Matthias J. Sax <[email protected]>
AuthorDate: Sun Aug 2 20:06:48 2026 -0700
MINOR: Improve the Kafka Streams state listener JavaDocs (#23011)
Updates outdate and incorrect JavaDocs, and aligns the information
provided across both listeners.
Reviewers: Andrew Schofield <[email protected]>
---
.../streams/processor/StandbyUpdateListener.java | 164 +++++++++++++++------
.../streams/processor/StateRestoreListener.java | 148 ++++++++++++-------
2 files changed, 208 insertions(+), 104 deletions(-)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/StandbyUpdateListener.java
b/streams/src/main/java/org/apache/kafka/streams/processor/StandbyUpdateListener.java
index 882747b47dc..0cc38136d8e 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/StandbyUpdateListener.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/StandbyUpdateListener.java
@@ -20,69 +20,139 @@ package org.apache.kafka.streams.processor;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.annotation.InterfaceAudience;
+/**
+ * Class for listening to the progress of the standby and warm-up tasks of a
Kafka Streams client.
+ *
+ * <p>When calling {@link
org.apache.kafka.streams.KafkaStreams#setStandbyUpdateListener(StandbyUpdateListener)}
+ * the passed instance is expected to be stateless since the {@code
StandbyUpdateListener} is shared across all state
+ * updater threads, of which there is one per stream thread ({@code
num.stream.threads}).
+ *
+ * <p>Users desiring stateful operations will need to provide synchronization
internally in
+ * the {@code StandbyUpdateListener} implementation.
+ *
+ * <p>Note that this listener is only registered at the per-client level and
users can use the {@code storeName}
+ * parameter to define specific monitoring for different {@link StateStore
StateStores}.
+ *
+ * <p>Warm-up tasks are reported through this interface as well as standby
tasks.
+ * A warm-up task is one that the group hands out so that this client can
build a task's state up before taking the task
+ * over, and the runtime runs it as a standby task, so the two are
indistinguishable to this listener.
+ * Relabelling a task between the two roles does not restart the task, and
does not trigger a callback; a task might be
+ * converted from a warm-up task to a standby task between two calls of {@link
#onBatchLoaded} transparently.
+ *
+ * <p>Also note that, unlike the restoration of an active task, the update of
a standby task never finishes, since it
+ * keeps applying the changelog records written by the active task.
+ * There is therefore no callback corresponding to
+ * {@link StateRestoreListener#onRestoreEnd(TopicPartition, String, long)}; a
standby task is reported only once it
+ * stops being updated, through {@link #onUpdateSuspended(TopicPartition,
String, long, long, SuspendReason)}.
+ *
+ * <p>Even if warm-up tasks are expected to "complete" restoration, the same
pattern as for standby tasks applies.
+ * Once a warm-up task is considered hot (based on {@code
acceptable.recovery.lag}), its restoration will be
+ * suspended when it is promoted to an active task, and the active task will
complete the restoration (if
+ * registered, the corresponding {@link StateRestoreListener} callback is
invoked on the active task for this
+ * case).
+ *
+ * <p>Incremental updates are exposed so users can estimate how much progress
has been made.
+ *
+ * @see StateRestoreListener
+ */
@InterfaceAudience.Public
public interface StandbyUpdateListener {
+ /**
+ * The reason why a standby or warm-up task stopped being updated.
+ */
enum SuspendReason {
+ /**
+ * The task stopped being updated as a standby or warm-up task, and
its state store was closed.
+ * This covers every reason other than a promotion: the task was
revoked, or it was assigned to another instance,
+ * or this instance is shutting down.
+ */
MIGRATED,
+
+ /**
+ * The task stopped being updated as a standby or warm-up task because
this instance is taking it over as an
+ * active task.
+ * Its state store stays open and keeps its contents, so the active
task resumes restoring where the update left
+ * off rather than from the beginning of the changelog.
+ * Restoring as an active task is reported through {@link
StateRestoreListener}.
+ */
PROMOTED
}
/**
- * A callback that will be invoked after registering the changelogs for
each state store in a standby
- * task. It is guaranteed to always be invoked before any records are
loaded into the standby store.
+ * Method called at the very beginning of {@link StateStore} updating.
*
- * @param topicPartition the changelog TopicPartition for this standby task
- * @param storeName the name of the store being loaded
- * @param startingOffset the offset from which the standby task begins
consuming from the changelog
+ * @param topicPartition
+ * the changelog {@link TopicPartition} for this task
+ * @param storeName
+ * the name of the store being loaded
+ * @param startingOffset
+ * the offset from which the task begins consuming from the
changelog
*/
- void onUpdateStart(final TopicPartition topicPartition,
- final String storeName,
- final long startingOffset);
+ void onUpdateStart(
+ final TopicPartition topicPartition,
+ final String storeName,
+ final long startingOffset
+ );
/**
- * Method called after loading a batch of records. In this case the
maximum size of the batch is whatever
- * the value of the MAX_POLL_RECORDS is set to.
- * <n>
- * This method is called after loading each batch and it is advised to
keep processing to a minimum.
- * Any heavy processing will block the state updater thread and slow down
the rate of standby task
- * loading. Therefore, if you need to do any extended processing or
connect to an external service,
- * consider doing so asynchronously.
+ * Method called after loading a batch of records.
+ * In this case the maximum size of the batch is whatever the value of
{@code max.poll.records} is set to.
+ *
+ * <p>This method is called after loading each batch and it is advised to
keep processing to a minimum.
+ * Any heavy processing will block the state updater thread and slow down
the rate of standby task loading.
+ * Therefore, if you need to do any extended processing or connect to an
external service, consider doing so
+ * asynchronously.
*
- * @param topicPartition the changelog TopicPartition for this standby task
- * @param storeName the name of the store being loaded
- * @param batchEndOffset batchEndOffset the changelog end offset
(inclusive) of the batch that was just loaded
- * @param batchSize the total number of records in the batch that was just
loaded
- * @param currentEndOffset the current end offset of the changelog topic
partition.
+ * @param topicPartition
+ * the changelog {@link TopicPartition} for this task
+ * @param storeName
+ * the name of the store being loaded
+ * @param taskId
+ * the {@link TaskId} of the task that owns the store being loaded
+ * @param batchEndOffset
+ * the changelog end offset (inclusive) of the batch that was just
loaded
+ * @param batchSize
+ * the total number of records in the batch that was just loaded
+ * @param currentEndOffset
+ * the current end offset of the changelog topic partition
*/
- void onBatchLoaded(final TopicPartition topicPartition,
- final String storeName,
- final TaskId taskId,
- final long batchEndOffset,
- final long batchSize,
- final long currentEndOffset);
+ void onBatchLoaded(
+ final TopicPartition topicPartition,
+ final String storeName,
+ final TaskId taskId,
+ final long batchEndOffset,
+ final long batchSize,
+ final long currentEndOffset
+ );
/**
- * This method is called when the corresponding standby task stops
updating, for the provided reason.
- * <p>
- * If the task was {@code MIGRATED} to another instance, this callback
will be invoked after this
- * state store (and the task itself) are closed (in which case the data
will be cleaned up after
- * state.cleanup.delay.ms).
- * If the task was {@code PROMOTED} to an active task, the state store
will not be closed, and the
- * callback will be invoked after unregistering it as a standby task but
before re-registering it as an active task
- * and beginning restoration. In other words, this will always called
before the corresponding
- * {@link StateRestoreListener#onRestoreStart} call is made.
+ * Method called when the corresponding standby or warm-up task stops
being updated, for the provided reason.
+ *
+ * <p>If the reason is {@link SuspendReason#MIGRATED}, this callback will
be invoked after this state store (and the
+ * task itself) are closed, in which case the data will be cleaned up
after {@code state.cleanup.delay.ms}.
+ * If the task was {@link SuspendReason#PROMOTED} to an active task, the
state store will not be closed, and the
+ * callback will be invoked after unregistering it as a standby task but
before re-registering it as an active task
+ * and beginning restoration.
+ * In other words, this will always be called before the corresponding
+ * {@link StateRestoreListener#onRestoreStart(TopicPartition, String,
long, long)} call is made.
*
- * @param topicPartition the changelog TopicPartition for this standby task
- * @param storeName the name of the store being loaded
- * @param storeOffset is the offset of the last changelog record that was
read and put into the store at the time
- * of suspension.
- * @param currentEndOffset the current end offset of the changelog topic
partition.
- * @param reason is the reason why the standby task was suspended.
+ * @param topicPartition
+ * the changelog {@link TopicPartition} for this task
+ * @param storeName
+ * the name of the store being loaded
+ * @param storeOffset
+ * the offset of the last changelog record that was read and put
into the store at the time of suspension
+ * @param currentEndOffset
+ * the current end offset of the changelog topic partition
+ * @param reason
+ * the reason why the task stopped being updated
*/
- void onUpdateSuspended(final TopicPartition topicPartition,
- final String storeName,
- final long storeOffset,
- final long currentEndOffset,
- final SuspendReason reason);
-}
\ No newline at end of file
+ void onUpdateSuspended(
+ final TopicPartition topicPartition,
+ final String storeName,
+ final long storeOffset,
+ final long currentEndOffset,
+ final SuspendReason reason
+ );
+}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/StateRestoreListener.java
b/streams/src/main/java/org/apache/kafka/streams/processor/StateRestoreListener.java
index adc0adc34e4..391d9631cbf 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/StateRestoreListener.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/StateRestoreListener.java
@@ -23,31 +23,30 @@ import org.apache.kafka.common.annotation.InterfaceAudience;
/**
* Class for listening to various states of the restoration process of a
StateStore.
*
- * <p>
- * When calling {@link
org.apache.kafka.streams.KafkaStreams#setGlobalStateRestoreListener(StateRestoreListener)}
- * the passed instance is expected to be stateless since the {@code
StateRestoreListener} is shared
- * across all {@link
org.apache.kafka.streams.processor.internals.StreamThread} instances.
+ * <p>When calling {@link
org.apache.kafka.streams.KafkaStreams#setGlobalStateRestoreListener(StateRestoreListener)}
+ * the passed instance is expected to be stateless since the {@code
StateRestoreListener} is shared across all state
+ * updater threads, of which there is one per stream thread ({@code
num.stream.threads}).
*
- * <p>
- * Users desiring stateful operations will need to provide synchronization
internally in
- * the {@code StateRestorerListener} implementation.
+ * <p>Users desiring stateful operations will need to provide synchronization
internally in
+ * the {@code StateRestoreListener} implementation.
*
- * <p>
- * Note that this listener is only registered at the per-client level and
users can base on the {@code storeName}
- * parameter to define specific monitoring for different {@link StateStore}s.
There is another
- * {@link StateRestoreCallback} interface which is registered via the
- * {@link StateStoreContext#register(StateStore, StateRestoreCallback,
CommitCallback)}
- * function per-store, and it is used to apply the fetched changelog records
into the local state store during restoration.
- * These two interfaces serve different restoration purposes and users should
not try to implement both of them in a single
- * class during state store registration.
+ * <p>Note that this listener is only registered at the per-client level and
users can use the {@code storeName}
+ * parameter to define specific monitoring for different {@link StateStore
StateStores}.
*
- * <p>
- * Also note that the update process of standby tasks is not monitored via
this interface, since a standby task does
- * note actually <it>restore</it> state, but keeps updating its state from the
changelogs written by the active task
+ * <p>{@code StateRestoreListener} must not be confused with the {@link
StateRestoreCallback} interface.
+ * The {@code StateRestoreListener} is a callback to observe restoration
progress.
+ * In contrast, the {@link StateRestoreCallback} interface is used by custom
state store implementations to provide
+ * the actual state restoration functionality.
+ *
+ * <p>Also note that the update process of standby tasks is not monitored via
this interface, since a standby task does
+ * not actually <em>restore</em> state, but keeps updating its state from the
changelogs written by the active task
* which does not ever finish.
+ * The same holds for warm-up tasks, which the runtime runs as standby tasks.
+ * Both are monitored via {@link StandbyUpdateListener} instead.
+ *
+ * <p>Incremental updates are exposed so users can estimate how much progress
has been made.
*
- * <p>
- * Incremental updates are exposed so users can estimate how much progress has
been made.
+ * @see StandbyUpdateListener
*/
@InterfaceAudience.Public
public interface StateRestoreListener {
@@ -55,58 +54,93 @@ public interface StateRestoreListener {
/**
* Method called at the very beginning of {@link StateStore} restoration.
*
- * @param topicPartition the TopicPartition containing the values to
restore
- * @param storeName the name of the store undergoing restoration
- * @param startingOffset the starting offset of the entire restoration
process for this TopicPartition
- * @param endingOffset the exclusive ending offset of the entire
restoration process for this TopicPartition
+ * <p>Called for every active task that begins restoring, including one
that has nothing to restore because its
+ * local state is already up to date -- a task whose state was just built
up as a standby or warm-up task, for
+ * instance.
+ * In that case {@code startingOffset} equals {@code endingOffset} and
+ * {@link #onRestoreEnd(TopicPartition, String, long)} follows with a
total of zero records restored, so an empty
+ * restoration is told apart by the offsets and the count rather than by
the callbacks being skipped.
+ *
+ * @param topicPartition
+ * the {@link TopicPartition} containing the values to restore
+ * @param storeName
+ * the name of the store undergoing restoration
+ * @param startingOffset
+ * the starting offset of the entire restoration process for this
{@link TopicPartition}
+ * @param endingOffset
+ * the exclusive ending offset of the entire restoration process
for this {@link TopicPartition}
*/
- void onRestoreStart(final TopicPartition topicPartition,
- final String storeName,
- final long startingOffset,
- final long endingOffset);
+ void onRestoreStart(
+ final TopicPartition topicPartition,
+ final String storeName,
+ final long startingOffset,
+ final long endingOffset
+ );
/**
- * Method called after restoring a batch of records. In this case the
maximum size of the batch is whatever
- * the value of the MAX_POLL_RECORDS is set to.
- *
- * This method is called after restoring each batch and it is advised to
keep processing to a minimum.
- * Any heavy processing will hold up recovering the next batch, hence
slowing down the restore process as a
- * whole.
+ * Method called after restoring a batch of records.
+ * In this case the maximum size of the batch is whatever the value of
{@code max.poll.records} is set to.
*
+ * <p>This method is called after restoring each batch and it is advised
to keep processing to a minimum.
+ * Any heavy processing will block the state updater thread and hold up
recovering the next batch, hence slowing
+ * down the restore process as a whole.
* If you need to do any extended processing or connecting to an external
service consider doing so asynchronously.
*
- * @param topicPartition the TopicPartition containing the values to
restore
- * @param storeName the name of the store undergoing restoration
- * @param batchEndOffset the inclusive ending offset for the current
restored batch for this TopicPartition
- * @param numRestored the total number of records restored in this batch
for this TopicPartition
+ * @param topicPartition
+ * the {@link TopicPartition} containing the values to restore
+ * @param storeName
+ * the name of the store undergoing restoration
+ * @param batchEndOffset
+ * the inclusive ending offset for the current restored batch for
this {@link TopicPartition}
+ * @param numRestored
+ * the total number of records restored in this batch for this
{@link TopicPartition}
*/
- void onBatchRestored(final TopicPartition topicPartition,
- final String storeName,
- final long batchEndOffset,
- final long numRestored);
+ void onBatchRestored(
+ final TopicPartition topicPartition,
+ final String storeName,
+ final long batchEndOffset,
+ final long numRestored
+ );
/**
* Method called when restoring the {@link StateStore} is complete.
*
- * @param topicPartition the TopicPartition containing the values to
restore
- * @param storeName the name of the store just restored
- * @param totalRestored the total number of records restored for this
TopicPartition
+ * @param topicPartition
+ * the {@link TopicPartition} containing the values to restore
+ * @param storeName
+ * the name of the store just restored
+ * @param totalRestored
+ * the total number of records restored for this {@link
TopicPartition}
*/
- void onRestoreEnd(final TopicPartition topicPartition,
- final String storeName,
- final long totalRestored);
+ void onRestoreEnd(
+ final TopicPartition topicPartition,
+ final String storeName,
+ final long totalRestored
+ );
/**
- * Method called when restoring the {@link StateStore} is suspended due to
the task being migrated out of the host.
- * If the migrated task is recycled or re-assigned back to the current
host, another
- * {@link #onRestoreStart(TopicPartition, String, long, long)} would be
called.
+ * Method called when restoring the {@link StateStore} is suspended before
it finished, because the task stopped
+ * being an active task on this host: it was assigned elsewhere, demoted
to a standby task, or this host is shutting
+ * down (the concrete reason is not reported).
+ * If the task becomes an active task on this host again -- assigned back
to it, or promoted from the standby task
+ * it was demoted to -- another {@link #onRestoreStart(TopicPartition,
String, long, long)} would be called.
+ *
+ * <p>A task whose restoration had already finished, and was reported
through
+ * {@link #onRestoreEnd(TopicPartition, String, long)}, is not reported
here when it is closed later on.
+ * In particular, an active task that had finished restoring and
transitioned into the {@code RUNNING} state and
+ * is then demoted to a standby task is not reported here at all; it
surfaces only as a
+ * {@link StandbyUpdateListener#onUpdateStart(TopicPartition, String,
long)} for the standby task that replaces it.
*
- * @param topicPartition the {@link TopicPartition} containing the values
to restore
- * @param storeName the name of the store just restored
- * @param totalRestored the total number of records restored for this
TopicPartition before being paused
+ * @param topicPartition
+ * the {@link TopicPartition} containing the values to restore
+ * @param storeName
+ * the name of the store just restored
+ * @param totalRestored
+ * the total number of records restored for this {@link
TopicPartition} before being suspended
*/
- default void onRestoreSuspended(final TopicPartition topicPartition,
- final String storeName,
- final long totalRestored) {
+ default void onRestoreSuspended(
+ final TopicPartition topicPartition,
+ final String storeName,
+ final long totalRestored) {
}
}
