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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
