lucasbru commented on code in PR #23004:
URL: https://github.com/apache/kafka/pull/23004#discussion_r3977294501
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -1216,48 +1232,119 @@ 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`
Review Comment:
This comment compares to the just-removed
`isThreadAlive()`/`Thread#isAlive()` rather than just stating the current
invariant - could read as narration of the diff rather than an explanation for
someone who never saw the old code. Maybe just: `shutdown()` moves a thread to
PENDING_SHUTDOWN synchronously while the underlying Thread stays alive until
`run()` returns, so filtering on Streams state (not Thread liveness) is what
keeps two concurrent removals from picking the same thread?
--
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]
