KAFKA-17455: fix stuck producer when throttling or retrying #17527

coltmcnealy-lh · 2024-10-17T15:53:22Z

fixes stuck producer by polling again after pollDelayMs in NetworkUtils#awaitReady()

coltmcnealy-lh · 2024-10-17T15:54:05Z

We are soaking this change right now. If it doesn't cause other problems, I will need help with unit tests. Addresses this ticket: https://issues.apache.org/jira/browse/KAFKA-17455

coltmcnealy-lh · 2024-10-17T23:17:57Z

@dajac any help would be appreciated with the tests. Not sure if what I attempted to do was clear or not, but (obviously) the changes I made to MockClient.java broke tons of stuff.

The reason for those changes was that since poll() didn't advance the time, it got stuck in an infinite loop before these changes. There must be a better way to test this.

coltmcnealy-lh · 2024-10-17T23:23:31Z

clients/src/test/java/org/apache/kafka/clients/MockClient.java

+        if (copy.isEmpty()) {
+            // Simulate time advancing. If no responses are received, then we know that
+            // we waited for the whole timeoutMs.
+            time.sleep(timeoutMs);


I noticed when developing my test that repeated calls to poll() with timeoutMs == 10 never resulted in time.milliseconds() advancing, so the fictitious Node in my test never became ready (time never advanced past the throttled time). That's why I made this change; however, it broke all sorts of things.

I guess most test don't expect that time advanced "automatically"... It think we need to remove this.

I think there two possibilities: start a background thread in your test before you call runOnce() and let the background thread advance mockTime (maybe too complex?), or, make the MockTime object more advanced and add some "auto-time-advance" feature, ie, each time milliseconds() is called, advance time by 1ms (or some other value passed into MockTime -- only your test would enable this feature.

Thanks Matthias. I did something somewhat inspired by your comment. Ready for review.

Should work, too, but why don't you just create new MockTime(1L) and use the already existing "auto-tick" feature?

@mjsax the problem is that the following test still passes on trunk using new MockTime(1L). However, the test that I have in the PR as it stands passes on my branch but fails on trunk:

@Test public void senderThreadShouldNotGetStuckWhenThrottledAndAddingPartitionsToTxn() { // We want MockClient#poll() to advance time so that eventually the backoff expires. // client.advanceTimeDuringPoll(true); time = new MockTime(1L); ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(123456L, (short) 0); apiVersions.update("0", NodeApiVersions.create(ApiKeys.INIT_PRODUCER_ID.id, (short) 0, (short) 3)); TransactionManager txnManager = new TransactionManager(logContext, "testUnresolvedSeq", 60000, 100, apiVersions); setupWithTransactionState(txnManager); doInitTransactions(txnManager, producerIdAndEpoch); int throttleTimeMs = 1000; long startTime = time.milliseconds(); Node nodeToThrottle = metadata.fetch().nodeById(0); client.throttle(nodeToThrottle, throttleTimeMs); // Verify node is throttled a little bit. In real-life Apache Kafka, we observe that this can happen // as done above by throttling or with a disconnect / backoff. long currentPollDelay = client.pollDelayMs(nodeToThrottle, startTime); assertTrue(currentPollDelay > 0); assertTrue(currentPollDelay <= throttleTimeMs); txnManager.beginTransaction(); txnManager.maybeAddPartition(tp0); assertFalse(txnManager.hasInFlightRequest()); sender.runOnce(); assertTrue(txnManager.hasInFlightRequest()); long totalTimeToRunOnce = time.milliseconds() - startTime; // It should have blocked roughly only the backoffTimeMs and some change. assertTrue(totalTimeToRunOnce < REQUEST_TIMEOUT); }

I suppose that the test passes with new MockTime(1L) because the new time is not used by the client. The client must be created after the time is set. Would it work? I would also prefer this over adding the sleep in the mock client.

mjsax · 2024-10-21T23:14:40Z

clients/src/test/java/org/apache/kafka/clients/MockClient.java

+        return connectionState(node.idString()).pollDelayMs(now);
+    }
+
+    public void setAdvanceTimeDuringPoll(boolean advanceTimeDuringPoll) {


Suggested change

public void setAdvanceTimeDuringPoll(boolean advanceTimeDuringPoll) {

public void advanceTimeDuringPoll(boolean advanceTimeDuringPoll) {

mjsax · 2024-10-21T23:16:48Z

clients/src/test/java/org/apache/kafka/clients/MockClient.java

+        if (copy.isEmpty()) {
+            // Simulate time advancing. If no responses are received, then we know that
+            // we waited for the whole timeoutMs.
+            time.sleep(timeoutMs);


Should work, too, but why don't you just create new MockTime(1L) and use the already existing "auto-tick" feature?

mjsax · 2024-10-21T23:17:54Z

clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java

+    @Test
+    public void senderThreadShouldNotGetStuckWhenThrottledAndAddingPartitionsToTxn() {
+        // We want MockClient#poll() to advance time so that eventually the backoff expires.
+        client.setAdvanceTimeDuringPoll(true);


Could be replace with this.time = new MockTime(1L); (or some other auto-tick value)?

I tried that. However, the problem is that the thing we want to test is not how many times we call time.milliseconds() but rather what values are passed into NetworkClient#poll().

I wasn't able to find a way to use this strategy to get a test that failed on trunk and passed on my branch.

…in NetworkUtils#awaitReady()

…advanceTimeDuringPoll()

mjsax

LGTM, but I am not an expert on producer code. \cc @dajac

dajac

@coltmcnealy-lh Thanks for the patch and sorry for the delay. I left a few minor comments for consideration.

dajac · 2025-01-08T13:18:19Z

clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java

+        assertTrue(currentPollDelay > 0);
+        assertTrue(currentPollDelay <= throttleTimeMs);


nit: I wonder if we could just assert that currentPollDelay equals to throttleTimeMs. The time does not seem to advance before pollDelayMs is called. Would it work?

dajac · 2025-01-08T13:19:11Z

clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java

+        long totalTimeToRunOnce = time.milliseconds() - startTime;
+
+        // It should have blocked roughly only the backoffTimeMs and some change.
+        assertTrue(totalTimeToRunOnce < REQUEST_TIMEOUT);


nit: Same question here. Could we assert that it equals to currentPollDelay?

dajac · 2025-01-08T13:20:30Z

clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java

+        // It should have blocked roughly only the backoffTimeMs and some change.
+        assertTrue(totalTimeToRunOnce < REQUEST_TIMEOUT);
+
+        client.advanceTimeDuringPoll(false);


Note that this line won't be called if any of the previous code throws (e.g. failed assertions). We should use a try..finally to ensure that we restore the state. An alternative would be to create the client in the setup in order to ensure a clean state for each test.

dajac · 2025-01-08T13:23:04Z

clients/src/test/java/org/apache/kafka/clients/MockClient.java

+        if (copy.isEmpty()) {
+            // Simulate time advancing. If no responses are received, then we know that
+            // we waited for the whole timeoutMs.
+            time.sleep(timeoutMs);


I suppose that the test passes with new MockTime(1L) because the new time is not used by the client. The client must be created after the time is set. Would it work? I would also prefer this over adding the sleep in the mock client.

coltmcnealy-lh · 2025-01-09T02:03:09Z

@dajac writes:

I suppose that the test passes with new MockTime(1L) because the new time is not used by the client. The client must be created after the time is set. Would it work? I would also prefer this over adding the sleep in the mock client.

It does pass; however, it also passes on trunk, which means the test has no signal. I am fine with making that change however @mjsax told me it was best to craft a test that fails before the patch (admittedly, this was tricky).

dajac · 2025-01-09T08:53:18Z

clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java

+            assertTrue(currentPollDelay > 0);
+            assertTrue(currentPollDelay == throttleTimeMs);


nit: We could use assertEquals(throttleTimeMs, currentPollDelay instead of those two lines.

dajac · 2025-01-09T08:59:11Z

@dajac writes:

I suppose that the test passes with new MockTime(1L) because the new time is not used by the client. The client must be created after the time is set. Would it work? I would also prefer this over adding the sleep in the mock client.

It does pass; however, it also passes on trunk, which means the test has no signal. I am fine with making that change however @mjsax told me it was best to craft a test that fails before the patch (admittedly, this was tricky).

@coltmcnealy-lh Thanks for trying. I also gave it a shot and I could not find a better way to do it. I think that we can keep your unit test as you proposed it. I left a minor comment.

dajac

lgtm

A producer might get stuck after it was throttled. This PR unblocks the producer by polling again after pollDelayMs in NetworkUtils#awaitReady(). Reviewers: Matthias J. Sax <[email protected]>, David Jacot <[email protected]>

mjsax · 2025-01-09T19:12:23Z

Thanks for the fix. Merged to trunk and cherry-picked to 4.0, 3.9, and 3.8 branches.

github-actions bot added clients small Small PRs labels Oct 17, 2024

github-actions bot added the producer label Oct 17, 2024

coltmcnealy-lh commented Oct 17, 2024

View reviewed changes

mjsax added the ci-approved label Oct 18, 2024

mjsax reviewed Oct 21, 2024

View reviewed changes

coltmcnealy-lh added 5 commits November 10, 2024 11:42

KAFKA-17455: fixes stuck producer by polling again after pollDelayMs …

d0956ba

…in NetworkUtils#awaitReady()

clarifies comments

d1d7d94

attempts to add test

54507da

Adds a test but my changes to MockClient.java broke all sorts of stuff

292d1ef

test that passes on my branch and fails on trunk

1bc3443

coltmcnealy-lh force-pushed the kafka-17455/fix-stuck-producer branch from 05200f7 to 1bc3443 Compare November 10, 2024 19:42

addresses PR feedback: rename MockClient#setAdvanceTimeDuringPoll to …

3da6309

…advanceTimeDuringPoll()

coltmcnealy-lh requested a review from mjsax November 12, 2024 18:12

mjsax reviewed Nov 14, 2024

View reviewed changes

dajac reviewed Jan 8, 2025

View reviewed changes

feedback from pull request

e56a563

dajac reviewed Jan 9, 2025

View reviewed changes

final PR feedback

344a519

dajac approved these changes Jan 9, 2025

View reviewed changes

mjsax merged commit bb22eec into apache:trunk Jan 9, 2025
9 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-17455: fix stuck producer when throttling or retrying #17527

KAFKA-17455: fix stuck producer when throttling or retrying #17527

coltmcnealy-lh commented Oct 17, 2024

coltmcnealy-lh commented Oct 17, 2024

coltmcnealy-lh commented Oct 17, 2024

coltmcnealy-lh Oct 17, 2024

mjsax Oct 18, 2024

coltmcnealy-lh Oct 19, 2024

mjsax Oct 21, 2024

coltmcnealy-lh Nov 12, 2024

dajac Jan 8, 2025

mjsax Oct 21, 2024

mjsax Oct 21, 2024

mjsax Oct 21, 2024

coltmcnealy-lh Nov 10, 2024

mjsax left a comment

dajac left a comment

dajac Jan 8, 2025

dajac Jan 8, 2025

dajac Jan 8, 2025

dajac Jan 8, 2025

coltmcnealy-lh commented Jan 9, 2025

dajac Jan 9, 2025

dajac commented Jan 9, 2025

dajac left a comment

mjsax commented Jan 9, 2025

	public void setAdvanceTimeDuringPoll(boolean advanceTimeDuringPoll) {
	public void advanceTimeDuringPoll(boolean advanceTimeDuringPoll) {

		assertTrue(currentPollDelay > 0);
		assertTrue(currentPollDelay <= throttleTimeMs);

		assertTrue(currentPollDelay > 0);
		assertTrue(currentPollDelay == throttleTimeMs);

KAFKA-17455: fix stuck producer when throttling or retrying #17527

KAFKA-17455: fix stuck producer when throttling or retrying #17527

Conversation

coltmcnealy-lh commented Oct 17, 2024

coltmcnealy-lh commented Oct 17, 2024

coltmcnealy-lh commented Oct 17, 2024

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

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

dajac 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

coltmcnealy-lh commented Jan 9, 2025

Choose a reason for hiding this comment

dajac commented Jan 9, 2025

dajac left a comment

Choose a reason for hiding this comment

mjsax commented Jan 9, 2025