You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "dajac (via GitHub)" <gi...@apache.org> on 2025/01/08 13:24:06 UTC
Re: [PR] KAFKA-17455: fix stuck producer when throttling or retrying [kafka]
dajac commented on code in PR #17527:
URL: https://github.com/apache/kafka/pull/17527#discussion_r1907171287
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -567,6 +567,44 @@ public void testMetadataTopicExpiry() throws Exception {
assertTrue(future.isDone(), "Request should be completed");
}
+ @Test
+ public void senderThreadShouldNotGetStuckWhenThrottledAndAddingPartitionsToTxn() {
+ // We want MockClient#poll() to advance time so that eventually the backoff expires.
+ client.advanceTimeDuringPoll(true);
+
+ 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);
Review Comment:
nit: Same question here. Could we assert that it equals to currentPollDelay?
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -567,6 +567,44 @@ public void testMetadataTopicExpiry() throws Exception {
assertTrue(future.isDone(), "Request should be completed");
}
+ @Test
+ public void senderThreadShouldNotGetStuckWhenThrottledAndAddingPartitionsToTxn() {
+ // We want MockClient#poll() to advance time so that eventually the backoff expires.
+ client.advanceTimeDuringPoll(true);
+
+ 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);
Review Comment:
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?
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -567,6 +567,44 @@ public void testMetadataTopicExpiry() throws Exception {
assertTrue(future.isDone(), "Request should be completed");
}
+ @Test
+ public void senderThreadShouldNotGetStuckWhenThrottledAndAddingPartitionsToTxn() {
+ // We want MockClient#poll() to advance time so that eventually the backoff expires.
+ client.advanceTimeDuringPoll(true);
+
+ 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);
+
+ client.advanceTimeDuringPoll(false);
Review Comment:
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.
##########
clients/src/test/java/org/apache/kafka/clients/MockClient.java:
##########
@@ -336,6 +336,12 @@ public List<ClientResponse> poll(long timeoutMs, long now) {
copy.add(response);
}
+ if (copy.isEmpty()) {
+ // Simulate time advancing. If no responses are received, then we know that
+ // we waited for the whole timeoutMs.
+ time.sleep(timeoutMs);
Review Comment:
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.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org