Skip to content

Commit b7b1c0a

Browse files
minjchominjcho
authored
KAFKA-16263: Add listeners and callbacks section to Streams developer guide (#22078)
Add a new **Listeners and callbacks** section to the *Running Streams Applications* developer guide to document the following four `KafkaStreams` listener/callback setters: - `setStateListener` - `setUncaughtExceptionHandler` - `setGlobalStateRestoreListener` - `setStandbyUpdateListener` Reviewers: Evan Zhou <ezhou@confluent.io>, Matthias J. Sax <matthias@confluent.io>
1 parent cb0a0db commit b7b1c0a

1 file changed

Lines changed: 112 additions & 0 deletions

File tree

‎docs/streams/developer-guide/running-app.md‎

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,118 @@ When you start your application you are launching a Kafka Streams instance of yo
4444

4545
When the application instance starts running, the defined processor topology will be initialized as one or more stream tasks. If the processor topology defines any state stores, these are also constructed during the initialization period. For more information, see the State restoration during workload rebalance section).
4646

47+
# Listeners and callbacks
48+
49+
`KafkaStreams` provides several listeners and callbacks that let you observe the internal state and behavior of your application. All listeners must be set **before** calling `start()`. Attempting to set a listener after `start()` has been called will throw an `IllegalStateException`. A setter may be called multiple times before `start()`. Only the most recent listener takes effect.
50+
51+
## State listener
52+
53+
You can set a `KafkaStreams.StateListener` to be notified whenever the `KafkaStreams` instance transitions between states. The possible states are: `CREATED`, `REBALANCING`, `RUNNING`, `PENDING_SHUTDOWN`, `NOT_RUNNING`, `PENDING_ERROR`, and `ERROR`. See the [`KafkaStreams.State`](/{version}/javadoc/org/apache/kafka/streams/KafkaStreams.State.html) javadocs for the meaning of each state and the allowed transitions.
54+
55+
KafkaStreams streams = new KafkaStreams(topology, props);
56+
57+
streams.setStateListener((newState, oldState) -> {
58+
if (newState == KafkaStreams.State.RUNNING) {
59+
// application is now ready to process records
60+
} else if (newState == KafkaStreams.State.ERROR) {
61+
// application has encountered a fatal error
62+
}
63+
});
64+
65+
streams.start();
66+
67+
## Uncaught exception handler
68+
69+
You can set a `StreamsUncaughtExceptionHandler` to handle unexpected exceptions thrown by internal stream threads. The handler receives the exception and must return a `StreamThreadExceptionResponse` indicating how to proceed:
70+
71+
* `REPLACE_THREAD` -- Replace the failed thread with a new one.
72+
* `SHUTDOWN_CLIENT` -- Shut down this `KafkaStreams` client.
73+
* `SHUTDOWN_APPLICATION` -- Request all instances of the application to shut down. This is best-effort: the signal is propagated via the rebalance protocol, so there is no guarantee that other instances will receive or act on it (for example, if they are unreachable).
74+
75+
Example:
76+
77+
import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
78+
79+
streams.setUncaughtExceptionHandler(exception -> {
80+
if (exception instanceof RetriableException) {
81+
return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD;
82+
}
83+
return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
84+
});
85+
86+
The handler executes on the thread that produced the exception. Because the handler is shared across all stream threads, the implementation **must be thread-safe**. To get the thread that threw the exception, call `Thread.currentThread()` from within the handler.
87+
88+
## State restore listener
89+
90+
You can set a `StateRestoreListener` to be notified about the progress of state store restoration. This is useful for monitoring and logging the restoration process. The listener provides callbacks for the following lifecycle events:
91+
92+
* `onRestoreStart` -- Called when restoration begins for a state store partition, providing the starting and ending offsets.
93+
* `onBatchRestored` -- Called after each batch of records is restored, providing the batch end offset and the number of records restored in the batch.
94+
* `onRestoreEnd` -- Called when restoration completes for a state store partition, providing the total number of records restored.
95+
* `onRestoreSuspended` -- Called when restoration is suspended because the task was migrated to another instance.
96+
97+
Example:
98+
99+
import org.apache.kafka.streams.processor.StateRestoreListener;
100+
101+
streams.setGlobalStateRestoreListener(new StateRestoreListener() {
102+
@Override
103+
public void onRestoreStart(TopicPartition topicPartition, String storeName,
104+
long startingOffset, long endingOffset) {
105+
// log that restoration has started
106+
}
107+
108+
@Override
109+
public void onBatchRestored(TopicPartition topicPartition, String storeName,
110+
long batchEndOffset, long numRestored) {
111+
// track progress
112+
}
113+
114+
@Override
115+
public void onRestoreEnd(TopicPartition topicPartition, String storeName,
116+
long totalRestored) {
117+
// log that restoration is complete
118+
}
119+
});
120+
121+
Because the listener is shared across all `StreamThread` instances, the implementation **must be thread-safe**. Note that this listener does **not** monitor standby task updates. To monitor standby tasks, use the standby update listener described below.
122+
123+
## Standby update listener
124+
125+
You can set a `StandbyUpdateListener` to be notified about updates to standby state store replicas. Standby replicas keep a copy of the state store on a different instance for faster failover. The listener provides callbacks for the following lifecycle events:
126+
127+
* `onUpdateStart` -- Called when a standby task begins consuming from the changelog, providing the starting offset.
128+
* `onBatchLoaded` -- Called after each batch of records is loaded into the standby store, providing the batch end offset, batch size, and the current end offset of the changelog partition.
129+
* `onUpdateSuspended` -- Called when the standby task stops updating. The `SuspendReason` parameter indicates why: `MIGRATED` means the task was moved to another instance, while `PROMOTED` means the standby was promoted to an active task (in which case the corresponding `StateRestoreListener.onRestoreStart` will be called next).
130+
131+
Example:
132+
133+
import org.apache.kafka.streams.processor.StandbyUpdateListener;
134+
135+
streams.setStandbyUpdateListener(new StandbyUpdateListener() {
136+
@Override
137+
public void onUpdateStart(TopicPartition topicPartition, String storeName,
138+
long startingOffset) {
139+
// log that standby update has started
140+
}
141+
142+
@Override
143+
public void onBatchLoaded(TopicPartition topicPartition, String storeName,
144+
TaskId taskId, long batchEndOffset,
145+
long batchSize, long currentEndOffset) {
146+
// track standby replication progress
147+
}
148+
149+
@Override
150+
public void onUpdateSuspended(TopicPartition topicPartition, String storeName,
151+
long storeOffset, long currentEndOffset,
152+
StandbyUpdateListener.SuspendReason reason) {
153+
// log reason for suspension
154+
}
155+
});
156+
157+
For more information about standby replicas, see [Standby Replicas](config-streams.md#num-standby-replicas).
158+
47159
# Elastic scaling of your application
48160

49161
Kafka Streams makes your stream processing applications elastic and scalable. You can add and remove processing capacity dynamically during application runtime without any downtime or data loss. This makes your applications resilient in the face of failures and for allows you to perform maintenance as needed (e.g. rolling upgrades).

0 commit comments

Comments
 (0)