Copilot commented on code in PR #23004:
URL: https://github.com/apache/kafka/pull/23004#discussion_r3982941099
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -1216,48 +1232,124 @@ public Optional<String> removeStreamThread(final
Duration timeout) {
private Optional<String> removeStreamThread(final long timeoutMs) throws
TimeoutException {
final long startMs = time.milliseconds();
- if (isRunningOrRebalancing()) {
- synchronized (changeThreadCount) {
- // make a copy of threads to avoid holding lock
- for (final StreamThread streamThread : new
ArrayList<>(threads)) {
- final boolean callingThreadIsNotCurrentStreamThread =
!streamThread.getName().equals(Thread.currentThread().getName());
- if (streamThread.isThreadAlive() &&
(callingThreadIsNotCurrentStreamThread || numLiveStreamThreads() == 1)) {
- log.info("Removing StreamThread {}",
streamThread.getName());
-
streamThread.shutdown(org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.LEAVE_GROUP);
- if (callingThreadIsNotCurrentStreamThread) {
- final long remainingTimeMs = timeoutMs -
(time.milliseconds() - startMs);
- if (remainingTimeMs <= 0 ||
!streamThread.waitOnThreadState(StreamThread.State.DEAD, remainingTimeMs)) {
- log.warn("{} did not shutdown in the allotted
time.", streamThread.getName());
- // Don't remove from threads until shutdown is
complete. We will trim it from the
- // list once it reaches DEAD, and if for some
reason it's hanging indefinitely in the
- // shutdown then we should just consider this
thread.id to be burned
- } else {
- log.info("Successfully removed {} in {}ms",
streamThread.getName(), time.milliseconds() - startMs);
- threads.remove(streamThread);
-
queryableStoreProvider.removeStoreProviderForThread(streamThread.getName());
- }
- } else {
- log.info("{} is the last remaining thread and must
remove itself, therefore we cannot wait "
- + "for it to complete shutdown as this will
result in deadlock.", streamThread.getName());
- }
+ if (!isRunningOrRebalancing()) {
+ log.warn("Cannot remove a stream thread when Kafka Streams client
is in state {}", state());
+ return Optional.empty();
+ }
- final long cacheSizePerThread =
cacheSizePerThread(numLiveStreamThreads());
- log.info("Resizing thread cache due to thread removal,
new cache size per thread is {}", cacheSizePerThread);
- resizeThreadCache(cacheSizePerThread);
-
resizeMaxUncommittedBytes(maxUncommittedBytesPerThread(numLiveStreamThreads()));
- final long remainingTimeMs = timeoutMs -
(time.milliseconds() - startMs);
- if (remainingTimeMs <= 0) {
- throw new TimeoutException("Thread " +
streamThread.getName() + " did not stop in the allotted time");
- }
- return Optional.of(streamThread.getName());
+ // Phase 1: choose a thread and signal its shutdown under the lock. We
must not block
+ // on that thread's terminal state while holding `changeThreadCount`:
if the thread
+ // being removed is the one that concurrently entered the
REPLACE_THREAD
+ // uncaught-exception handler, its `replaceStreamThread ->
addStreamThread` path is
+ // already blocked on this same lock, so it can never reach DEAD (only
+ // `completeShutdown` sets that state) and the wait below would never
return.
+ final StreamThread threadToRemove;
+ boolean skippedThreadAlreadyShuttingDown = false;
+ synchronized (changeThreadCount) {
+ StreamThread candidate = null;
+ // Copy the threads list to avoid holding its intrinsic lock
during iteration.
+ //
+ // `shutdown()` moves a thread to PENDING_SHUTDOWN synchronously
while the underlying
+ // Thread stays alive until `run()` returns, so filtering on the
Streams state (not
+ // Thread liveness alone) is what keeps two concurrent removals
from picking the same
+ // thread.
+ //
+ // A thread in CREATED is removable only once started:
`addStreamThread` publishes the
+ // thread to `threads` before starting it, and a thread that never
ran cannot reach
+ // DEAD. A started thread stays in CREATED until `run()` begins
executing, so Thread
+ // liveness covers that scheduling window; shutting such a thread
down completes
+ // inline within shutdown().
+ for (final StreamThread streamThread : new ArrayList<>(threads)) {
+ final boolean isNotCurrentThread =
!streamThread.getName().equals(Thread.currentThread().getName());
+ final StreamThread.State threadState = streamThread.state();
+ final boolean removable = threadState.isAlive()
+ || (threadState == StreamThread.State.CREATED &&
streamThread.isThreadAlive());
+ if (removable && (isNotCurrentThread || numLiveStreamThreads()
== 1)) {
+ // shutdown() returns false if another caller requested
this thread's shutdown
+ // between the isAlive() check above and this call: either
its uncaught-exception
+ // handler, which will spawn a replacement, or a
concurrent client close. In both
+ // cases that caller owns the thread's death, so it must
not count as this
+ // removal; keep scanning for another candidate.
+ if
(streamThread.shutdown(GroupMembershipOperation.LEAVE_GROUP)) {
+ log.info("Removing StreamThread {}",
streamThread.getName());
+ candidate = streamThread;
+ break;
}
+ skippedThreadAlreadyShuttingDown = true;
}
}
- log.warn("There are no threads eligible for removal");
+ threadToRemove = candidate;
+ }
+
+ if (threadToRemove == null) {
+ if (skippedThreadAlreadyShuttingDown) {
+ log.warn("There are no threads eligible for removal: every
alive thread is already shutting down, "
+ + "either because it is being replaced after an uncaught
exception or because the client is closing. "
+ + "Retry to remove the replacement thread once it is
running.");
+ } else {
+ log.warn("There are no threads eligible for removal");
+ }
+ return Optional.empty();
+ }
+
+ final boolean callingThreadIsNotCurrentStreamThread =
+ !threadToRemove.getName().equals(Thread.currentThread().getName());
+
+ // Phase 2: wait for the thread to reach DEAD without holding
`changeThreadCount`, so
+ // that a concurrent add or thread replacement can make progress in
the meantime.
+ final boolean reachedDead;
+ if (callingThreadIsNotCurrentStreamThread) {
+ final long remainingTimeMs = timeoutMs - (time.milliseconds() -
startMs);
+ reachedDead = remainingTimeMs > 0
+ && threadToRemove.waitOnThreadState(StreamThread.State.DEAD,
remainingTimeMs);
} else {
- log.warn("Cannot remove a stream thread when Kafka Streams client
is in state {}", state());
+ log.info("{} is the last remaining thread and must remove itself,
therefore we cannot wait "
+ + "for it to complete shutdown as this will result in
deadlock.", threadToRemove.getName());
+ reachedDead = false;
}
- return Optional.empty();
+
+ // Phase 3: bookkeeping under the lock, so that the threads-list
update is serialized
+ // against concurrent add and remove callers. The cache sizes are
recomputed from
+ // `numLiveStreamThreads()`, which already excludes the
PENDING_SHUTDOWN thread we
+ // signalled in phase 1; that is also why releasing the lock during
the wait cannot make
+ // a concurrent `addStreamThread` size the caches against a stale
thread count.
+ synchronized (changeThreadCount) {
+ if (callingThreadIsNotCurrentStreamThread) {
+ if (reachedDead) {
+ log.info("Successfully removed {} in {}ms",
threadToRemove.getName(), time.milliseconds() - startMs);
+ threads.remove(threadToRemove);
+ // While the wait above did not hold the lock, a
concurrent addStreamThread may have
+ // trimmed the DEAD thread via nextThreadIndex and reused
its name, overwriting the
+ // store-provider registration with the new thread's. The
registrations are keyed by
+ // name, so removing by name would delete the
replacement's provider and break
+ // interactive queries; only clean up while no live thread
carries the name.
+ final String removedThreadName = threadToRemove.getName();
+ final boolean nameReused = new
ArrayList<>(threads).stream()
+ .anyMatch(t -> t.getName().equals(removedThreadName));
Review Comment:
The reuse check also matches a replacement that has already reached `DEAD`.
In that race this branch preserves its now-useless provider, and the subsequent
`numLiveStreamThreads()` call trims the dead replacement from `threads`,
leaving the provider holding that thread indefinitely. Treat only non-DEAD
threads as active name reuse; dead providers return no stores
(`StreamThreadStateStoreProvider.java:48-50`) and are safe to remove.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
