lucasbru commented on code in PR #23004:
URL: https://github.com/apache/kafka/pull/23004#discussion_r3924432945
##########
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
*/
- public void shutdown(final
org.apache.kafka.streams.CloseOptions.GroupMembershipOperation operation) {
+ public boolean shutdown(final
org.apache.kafka.streams.CloseOptions.GroupMembershipOperation operation) {
Review Comment:
Two of the four callers of shutdown() (close()'s shutdownHelper, and the
terminate-new-thread branch in addStreamThread) don't check this return value.
Is that intentional, or should they also handle the case where they lost the
race?
##########
streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java:
##########
@@ -804,6 +807,98 @@ public void shouldRemoveThread() throws Exception {
}
}
+ @Test
+ @Timeout(60)
+ public void shouldNotBlockOtherThreadChangesWhileRemovalWaitsForShutdown()
throws Exception {
+ // `removeStreamThread` used to wait for the removed thread to reach
DEAD while holding
Review Comment:
Nit: this comment reads as a description of the old locking bug rather than
what the test verifies now - could reword to describe current behavior instead.
##########
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:
If shutdown(LEAVE_GROUP) loses the race against the thread's own
uncaught-exception handler (which can call shutdown(DEFAULT) first on the same
thread), this returns false and we fall through to the "no threads eligible"
path, returning Optional.empty(). But the thread is actually being replaced at
that point, just not with LEAVE_GROUP semantics. Should we distinguish "nothing
to remove" from "lost the race, thread being replaced instead" in the return
value here?
--
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]
