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 9a239c6142a KAFKA-16955: fix synchronization of streams threadState
(#16337)
9a239c6142a is described below
commit 9a239c6142a8f2eb36f1600d1012224c31e58e71
Author: Rohan <desai.p.ro...@gmail.com>
AuthorDate: Fri Jun 14 10:44:36 2024 -0700
KAFKA-16955: fix synchronization of streams threadState (#16337)
Each KafkaStreams instance maintains a map from threadId to state
to use to aggregate to a KafkaStreams app state. The map is updated
on every state change, and when a new thread is created. State change
updates are done in a synchronized blocks, however the update that
happens on thread creation is not, which can raise
ConcurrentModificationException. This patch moves this update
into the listener object and protects it using the object's lock.
It also moves ownership of the state map into the listener so that
its less likely that future changes access it without locking
Reviewers: Matthias J. Sax <matth...@confluent.io>
---
.../org/apache/kafka/streams/KafkaStreams.java | 64 ++++++++++------------
1 file changed, 30 insertions(+), 34 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
index 9cf91f163e2..fcc999e92af 100644
--- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
+++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
@@ -176,7 +176,6 @@ public class KafkaStreams implements AutoCloseable {
private final long totalCacheSize;
private final StreamStateListener streamStateListener;
private final DelegatingStateRestoreListener
delegatingStateRestoreListener;
- private final Map<Long, StreamThread.State> threadState;
private final UUID processId;
private final KafkaClientSupplier clientSupplier;
protected final TopologyMetadata topologyMetadata;
@@ -633,17 +632,13 @@ public class KafkaStreams implements AutoCloseable {
/**
* Class that handles stream thread transitions
*/
- final class StreamStateListener implements StreamThread.StateListener {
+ private final class StreamStateListener implements
StreamThread.StateListener {
private final Map<Long, StreamThread.State> threadState;
private GlobalStreamThread.State globalThreadState;
- // this lock should always be held before the state lock
- private final Object threadStatesLock;
- StreamStateListener(final Map<Long, StreamThread.State> threadState,
- final GlobalStreamThread.State globalThreadState) {
- this.threadState = threadState;
+ StreamStateListener(final GlobalStreamThread.State globalThreadState) {
+ this.threadState = new HashMap<>();
this.globalThreadState = globalThreadState;
- this.threadStatesLock = new Object();
}
/**
@@ -675,33 +670,35 @@ public class KafkaStreams implements AutoCloseable {
public synchronized void onChange(final Thread thread,
final ThreadStateTransitionValidator
abstractNewState,
final ThreadStateTransitionValidator
abstractOldState) {
- synchronized (threadStatesLock) {
- // StreamThreads first
- if (thread instanceof StreamThread) {
- final StreamThread.State newState = (StreamThread.State)
abstractNewState;
- threadState.put(thread.getId(), newState);
-
- if (newState == StreamThread.State.PARTITIONS_REVOKED ||
newState == StreamThread.State.PARTITIONS_ASSIGNED) {
- setState(State.REBALANCING);
- } else if (newState == StreamThread.State.RUNNING) {
- maybeSetRunning();
- }
- } else if (thread instanceof GlobalStreamThread) {
- // global stream thread has different invariants
- final GlobalStreamThread.State newState =
(GlobalStreamThread.State) abstractNewState;
- globalThreadState = newState;
-
- if (newState == GlobalStreamThread.State.RUNNING) {
- maybeSetRunning();
- } else if (newState == GlobalStreamThread.State.DEAD) {
- if (state != State.PENDING_SHUTDOWN) {
- log.error("Global thread has died. The streams
application or client will now close to ERROR.");
- closeToError();
- }
+ // StreamThreads first
+ if (thread instanceof StreamThread) {
+ final StreamThread.State newState = (StreamThread.State)
abstractNewState;
+ threadState.put(thread.getId(), newState);
+
+ if (newState == StreamThread.State.PARTITIONS_REVOKED ||
newState == StreamThread.State.PARTITIONS_ASSIGNED) {
+ setState(State.REBALANCING);
+ } else if (newState == StreamThread.State.RUNNING) {
+ maybeSetRunning();
+ }
+ } else if (thread instanceof GlobalStreamThread) {
+ // global stream thread has different invariants
+ final GlobalStreamThread.State newState =
(GlobalStreamThread.State) abstractNewState;
+ globalThreadState = newState;
+
+ if (newState == GlobalStreamThread.State.RUNNING) {
+ maybeSetRunning();
+ } else if (newState == GlobalStreamThread.State.DEAD) {
+ if (state != State.PENDING_SHUTDOWN) {
+ log.error("Global thread has died. The streams
application or client will now close to ERROR.");
+ closeToError();
}
}
}
}
+
+ private synchronized void registerStreamThread(final StreamThread
streamThread) {
+ threadState.put(streamThread.getId(), streamThread.state());
+ }
}
static final class DelegatingStateRestoreListener implements
StateRestoreListener {
@@ -1047,8 +1044,7 @@ public class KafkaStreams implements AutoCloseable {
globalThreadState = globalStreamThread.state();
}
- threadState = new HashMap<>(numStreamThreads);
- streamStateListener = new StreamStateListener(threadState,
globalThreadState);
+ streamStateListener = new StreamStateListener(globalThreadState);
final GlobalStateStoreProvider globalStateStoreProvider = new
GlobalStateStoreProvider(this.topologyMetadata.globalStateStores());
@@ -1084,9 +1080,9 @@ public class KafkaStreams implements AutoCloseable {
KafkaStreams.this::closeToError,
streamsUncaughtExceptionHandler
);
+ streamStateListener.registerStreamThread(streamThread);
streamThread.setStateListener(streamStateListener);
threads.add(streamThread);
- threadState.put(streamThread.getId(), streamThread.state());
queryableStoreProvider.addStoreProviderForThread(streamThread.getName(), new
StreamThreadStateStoreProvider(streamThread));
return streamThread;
}
