Copilot commented on code in PR #23004:
URL: https://github.com/apache/kafka/pull/23004#discussion_r3965678165
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1926,15 +1921,24 @@ private long advanceNowAndComputeLatency() {
* (e.g., in testing), hence the state is set only the first time
*
* @param operation the group membership operation to apply on shutdown.
Must be one of LEAVE_GROUP or REMAIN_IN_GROUP.
+ * @return true if this call initiated the shutdown, i.e., transitioned
the thread out of an
+ * alive state; false if the thread was already shutting down or
dead, in which case
+ * the group membership operation of the earlier shutdown request
is kept
Review Comment:
This return contract is inaccurate for a thread in `CREATED`:
`State.isAlive()` is false for that state, but `shutdown()` transitions it to
`PENDING_SHUTDOWN`, completes shutdown, and returns true. Describe success as
initiating the transition to `PENDING_SHUTDOWN` rather than transitioning out
of an alive state.
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -1216,48 +1231,106 @@ 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.
+ //
+ // Filtering on the Kafka Streams state rather than
`java.lang.Thread#isAlive`
+ // matters now that the lock is released during the wait below:
`shutdown()` moves
+ // a thread to PENDING_SHUTDOWN synchronously, while the
underlying Thread stays
+ // alive until `run()` returns, so two concurrent removals would
otherwise choose
+ // the same thread, both report it as removed, and leave the
thread count too high.
+ //
+ // Threads in CREATED are skipped deliberately: `addStreamThread`
publishes a thread
+ // to `threads` before starting it, and a thread that never ran
cannot reach DEAD.
+ for (final StreamThread streamThread : new ArrayList<>(threads)) {
+ final boolean isNotCurrentThread =
!streamThread.getName().equals(Thread.currentThread().getName());
+ if (streamThread.state().isAlive() && (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;
+ }
+
+ // 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);
+
queryableStoreProvider.removeStoreProviderForThread(threadToRemove.getName());
Review Comment:
An add can win this lock after the victim reaches `DEAD`.
`nextThreadIndex()` then removes the dead entry and reuses its name, while
`createAndAddStreamThread()` registers the new provider under that name. At
this point `threads.remove(threadToRemove)` is a no-op, but the unconditional
name-based removal deletes the new thread's provider and breaks interactive
queries. Make cleanup identity-aware (for example, compare-and-remove the
victim's provider) so a reused name cannot remove the replacement's
registration.
--
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]
