mjsax commented on code in PR #23004:
URL: https://github.com/apache/kafka/pull/23004#discussion_r3985123443
##########
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");
Review Comment:
This is the same sentence as the other warn log, but w/o any long
explanation. When would we this case?
But making a step back, is it actually important to distinguish both cases?
-- I am also wondering why we log a WARN? Is this something that should go into
the logs, and if yes, would INFO not be sufficient? In the end, we return an
`Optional.empty` anyway and the application can react to the failed removal.
--
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]
