KAFKA-17100: GlobalStreamThread#start should not busy-wait #16914

raminqaf · 2024-08-19T08:21:40Z

More detailed description
I have replaced the sleep method with a CountDownLatch. It will await until the latch count reaches zero. The countdown happens in the initialize method in a finally block.

Summary of testing strategy
All the tests in GlobalStreamThreadTest pass, indicating that the class's behavior has remained unchanged.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

raminqaf · 2024-08-19T08:23:59Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java

+            // Question: Is this a good idea? Or should we spread the latch countdown to finally blocks
+            // Count down the latch if transitioning from CREATED to any other state
+            if (oldState == State.CREATED && newState != State.CREATED) {
+                initializationLatch.countDown();
+            }


@gharris1727 This works (all the tests pass), but I want to double-check it with you. The other alternative would be to count down in finally blocks across the class.

The other alternative would be to count down in finally blocks across the class.

It could make sense to have a finally block in the initialize method and call countDown in shutdown, but the current implementation appears to have reasonable semantics.

I have a weak feeling that counting down this latch is too much for the setState method to be doing, but at the same time pushing the latch management out into the calling code risks missing a (possibly future) call-site which transitions away from CREATED.

I think i would leave this to see what the Streams folks think about it.

I agree with @gharris1727 -- this is not the right place for this code. I am not 100% sure right now what the best place would be, but both run() or initialize() look like good candidates (maybe initialize() is better).

I moved the countDown logic to the finally block of initialize. It sounds reasonable to me.

gharris1727

This seems reasonable to me, and captures the intent of my ticket.

@mjsax @ableegoldman Could one of you PTAL, thanks!

gharris1727 · 2024-08-19T21:42:57Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java

+            // Question: Is this a good idea? Or should we spread the latch countdown to finally blocks
+            // Count down the latch if transitioning from CREATED to any other state
+            if (oldState == State.CREATED && newState != State.CREATED) {
+                initializationLatch.countDown();
+            }


The other alternative would be to count down in finally blocks across the class.

It could make sense to have a finally block in the initialize method and call countDown in shutdown, but the current implementation appears to have reasonable semantics.

I have a weak feeling that counting down this latch is too much for the setState method to be doing, but at the same time pushing the latch management out into the calling code risks missing a (possibly future) call-site which transitions away from CREATED.

I think i would leave this to see what the Streams folks think about it.

gharris1727 · 2024-08-19T21:45:33Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java

+            initializationLatch.await();
+        } catch (final InterruptedException e) {
+            currentThread().interrupt();
+            throw new IllegalStateException("Thread was interrupted during initialization", e);


This is a "new" exception, as the previous code would have just spun calling Utils.sleep and continuously throwing InterruptedExceptions. It's more informative than the other IllegalStateException thrown here, so this seems reasonable.

mjsax · 2024-08-22T19:34:52Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java

+            initializationLatch.await();
+        } catch (final InterruptedException e) {
+            currentThread().interrupt();
+            throw new IllegalStateException("Thread was interrupted during initialization", e);


Suggested change

throw new IllegalStateException("Thread was interrupted during initialization", e);

throw new IllegalStateException("GlobalStreamThread was interrupted during initialization", e);

gharris1727 · 2024-08-23T16:03:26Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java

@@ -436,6 +430,8 @@ private StateConsumer initialize() {
        } catch (final Exception fatalException) {
            closeStateConsumer(stateConsumer, false);
            startupException = new StreamsException("Exception caught during initialization of GlobalStreamThread", fatalException);
+        } finally {
+            initializationLatch.countDown();


I think to keep the existing semantics, another one of these is needed in shutdown.

I don't think we need to right now, given that start() blocks and thus shutdown() cannot be triggered as long as init() did not finish (cf https://issues.apache.org/jira/browse/KAFKA-7380)

But maybe good to add to be future proof...

Interesting... 💭 I have added the countdown to the shutdown().

mjsax

@gharris1727 are you happy with the PR? Can we merge it?

gharris1727

LGTM, thanks @raminqaf!

mjsax · 2024-08-25T21:53:30Z

Thanks for the PR @raminqaf -- merged to trunk.

) This PR replaces a busy-wait sleep with a CountDownLatch. Reviewers: Greg Harris <[email protected]>, Matthias J. Sax <[email protected]>

raminqaf added 2 commits August 19, 2024 09:59

GlobalStreamThread#start should not busy-wait

758a1b4

move latch count down

49930b3

raminqaf commented Aug 19, 2024

View reviewed changes

raminqaf changed the title ~~GlobalStreamThread#start should not busy-wait~~ [KAFKA-17100] GlobalStreamThread#start should not busy-wait Aug 19, 2024

raminqaf changed the title ~~[KAFKA-17100] GlobalStreamThread#start should not busy-wait~~ KAFKA-17100: GlobalStreamThread#start should not busy-wait Aug 19, 2024

gharris1727 reviewed Aug 19, 2024

View reviewed changes

mjsax added the streams label Aug 22, 2024

mjsax reviewed Aug 22, 2024

View reviewed changes

raminqaf added 2 commits August 23, 2024 09:51

Change exception message and move countDown to initialize

143a185

remove line

7a9a300

raminqaf requested review from gharris1727 and mjsax August 23, 2024 07:55

gharris1727 reviewed Aug 23, 2024

View reviewed changes

add countdown latch to shutdown()

5d45fdd

mjsax approved these changes Aug 25, 2024

View reviewed changes

gharris1727 approved these changes Aug 25, 2024

View reviewed changes

mjsax merged commit 9eb7e1a into apache:trunk Aug 25, 2024
4 of 7 checks passed

raminqaf deleted the KAFKA-17100 branch August 28, 2024 07:37

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-17100: GlobalStreamThread#start should not busy-wait #16914

KAFKA-17100: GlobalStreamThread#start should not busy-wait #16914

raminqaf commented Aug 19, 2024 •

edited

Loading

raminqaf Aug 19, 2024

gharris1727 Aug 19, 2024

mjsax Aug 22, 2024

raminqaf Aug 23, 2024

gharris1727 left a comment

gharris1727 Aug 19, 2024

gharris1727 Aug 19, 2024

mjsax Aug 22, 2024

gharris1727 Aug 23, 2024

mjsax Aug 23, 2024

raminqaf Aug 24, 2024

mjsax left a comment

gharris1727 left a comment

mjsax commented Aug 25, 2024

	throw new IllegalStateException("Thread was interrupted during initialization", e);
	throw new IllegalStateException("GlobalStreamThread was interrupted during initialization", e);

KAFKA-17100: GlobalStreamThread#start should not busy-wait #16914

KAFKA-17100: GlobalStreamThread#start should not busy-wait #16914

Conversation

raminqaf commented Aug 19, 2024 • edited Loading

Committer Checklist (excluded from commit message)

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gharris1727 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax left a comment

Choose a reason for hiding this comment

gharris1727 left a comment

Choose a reason for hiding this comment

mjsax commented Aug 25, 2024

raminqaf commented Aug 19, 2024 •

edited

Loading