mjsax commented on code in PR #23004:
URL: https://github.com/apache/kafka/pull/23004#discussion_r3985160722
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -1216,48 +1232,134 @@ 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 a thread that can still serve
queries counts as active
+ // name reuse: a replacement already in PENDING_SHUTDOWN
or DEAD serves none (its
+ // provider throws InvalidStateStoreException or returns
no stores), will be
+ // trimmed from `threads`, and would leave its provider
registered forever, so its
+ // registration is cleaned up here like the removed
thread's own.
+ final String removedThreadName = threadToRemove.getName();
+ final boolean nameReused = new
ArrayList<>(threads).stream()
+ .anyMatch(t -> {
+ if (!t.getName().equals(removedThreadName)) {
+ return false;
+ }
+ final StreamThread.State reusingThreadState =
t.state();
+ return reusingThreadState.isAlive() ||
reusingThreadState == StreamThread.State.CREATED;
+ });
+ if (nameReused) {
+ log.info("Skipping state-store provider cleanup for {}
since the name has been reused by a newly added thread",
+ removedThreadName);
+ } else {
+
queryableStoreProvider.removeStoreProviderForThread(removedThreadName);
+ }
+ } else {
+ log.warn("{} did not shutdown in the allotted time.",
threadToRemove.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
+ }
+ }
+ 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) {
Review Comment:
This might not be correct? We get the time already above and make this
check. -- It seems better to use a boolean (default `false`) and set it to
`true` if we did timeout and the thread did not reach DEAD state, and just
check the boolean.
Otherwise, we might throw this timeout even if the thread did exit on time,
because `time.milliseconds()` would have advance in-between.
--
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]
