frankvicky commented on code in PR #23004:
URL: https://github.com/apache/kafka/pull/23004#discussion_r3948261985
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -1216,48 +1226,97 @@ 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;
+ 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 the thread moved to
PENDING_SHUTDOWN between the
+ // isAlive() check above and this call, which means its
uncaught-exception
+ // handler won the race and will spawn a replacement: that
thread's death is
+ // already compensated and must not count as this removal,
so keep scanning.
+ if
(streamThread.shutdown(GroupMembershipOperation.LEAVE_GROUP)) {
Review Comment:
when `shutdown()` returns false we don't fall through immediately, the loop
keeps scanning and only returns empty if every alive thread lost the race. In
that case I think `Optional.empty()` is the honest answer: no thread was
removed and the thread count did not decrease, which is exactly what the
previous round of review asked us not to misreport. The caller can simply retry
and will then pick up the replacement thread.
I considered "claiming" the dying thread as this removal instead, but that
needs a per-thread claim flag checked by `replaceStreamThread` under the lock,
and the claimed thread would die with `DEFAULT` (i.e. `REMAIN_IN_GROUP` on the
classic protocol) rather than `LEAVE_GROUP`, so its tasks would only be
reassigned after the session timeout. That seemed worse than an empty result.
Distinguishing the two cases in the return value would need an API change.
What I did instead: updated the `@return` javadoc on both overloads to
document this case, and the log message now says explicitly that every alive
thread is already shutting down and that a retry will remove the replacement.
--
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]
