cadonna commented on code in PR #15669:
URL: https://github.com/apache/kafka/pull/15669#discussion_r1557825632
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -498,6 +538,40 @@ public void
testAutoCommitEmptyDoesNotLeaveInflightRequestFlagOn() {
verify(commitRequestManger, times(2)).resetAutoCommitTimer();
}
+ @Test
+ public void testAutoCommitOnIntervalSkippedIfPreviousOneInFlight() {
+ TopicPartition t1p = new TopicPartition("topic1", 0);
+ subscriptionState.assignFromUser(singleton(t1p));
+ subscriptionState.seek(t1p, 100);
+
+ CommitRequestManager commitRequestManager = create(true, 100);
+
+ // Send auto-commit request that will remain in-flight without a
response
+ time.sleep(100);
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
+ commitRequestManger.maybeAutoCommitAsync();
+ List<NetworkClientDelegate.FutureCompletionHandler> futures =
assertPoll(1, commitRequestManger);
+ assertEquals(1, futures.size());
+ NetworkClientDelegate.FutureCompletionHandler inflightCommitResult =
futures.get(0);
+ verify(commitRequestManger, times(1)).resetAutoCommitTimer();
+ clearInvocations(commitRequestManger);
+
+ // After next interval expires, no new auto-commit request should be
sent. The interval
+ // should not be reset either, to ensure that the next auto-commit is
sent out as soon as
+ // the inflight receives a response.
+ time.sleep(100);
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
+ commitRequestManger.maybeAutoCommitAsync();
+ assertPoll(0, commitRequestManger);
+ verify(commitRequestManger, never()).resetAutoCommitTimer();
+
+ // When a response for the inflight is received, a next auto-commit
should be sent when
+ // polling the manager.
+ inflightCommitResult.onComplete(
+ mockOffsetCommitResponse(t1p.topic(), t1p.partition(), (short) 1,
Errors.NONE));
+ assertPoll(1, commitRequestManger);
+ }
Review Comment:
```suggestion
// Send auto-commit request that will remain in-flight without a
response
time.sleep(100);
commitRequestManager.updateAutoCommitTimer(time.milliseconds());
commitRequestManager.maybeAutoCommitAsync();
List<NetworkClientDelegate.FutureCompletionHandler> futures =
assertPoll(1, commitRequestManager);
assertEquals(1, futures.size());
NetworkClientDelegate.FutureCompletionHandler inflightCommitResult =
futures.get(0);
verify(commitRequestManager, times(1)).resetAutoCommitTimer();
clearInvocations(commitRequestManager);
// After next interval expires, no new auto-commit request should be
sent. The interval
// should not be reset either, to ensure that the next auto-commit
is sent out as soon as
// the inflight receives a response.
time.sleep(100);
commitRequestManager.updateAutoCommitTimer(time.milliseconds());
commitRequestManager.maybeAutoCommitAsync();
assertPoll(0, commitRequestManager);
verify(commitRequestManager, never()).resetAutoCommitTimer();
// When a response for the inflight is received, a next auto-commit
should be sent when
// polling the manager.
inflightCommitResult.onComplete(
mockOffsetCommitResponse(t1p.topic(), t1p.partition(), (short)
1, Errors.NONE));
assertPoll(1, commitRequestManager);
}
```
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
