Hangleton commented on code in PR #13240:
URL: https://github.com/apache/kafka/pull/13240#discussion_r1116205901
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java:
##########
@@ -2604,11 +2650,95 @@ public void testCommitOffsetUnknownMemberId() {
client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE));
- prepareOffsetCommitRequest(singletonMap(t1p, 100L),
Errors.UNKNOWN_MEMBER_ID);
+ prepareOffsetCommitRequest(new OffsetCommitResponseSpec()
+ .expectedOffsets(singletonMap(t1p, 100L))
+ .error(Errors.UNKNOWN_MEMBER_ID));
+
assertThrows(CommitFailedException.class, () ->
coordinator.commitOffsetsSync(singletonMap(t1p,
new OffsetAndMetadata(100L, "metadata")),
time.timer(Long.MAX_VALUE)));
}
+ @Test
+ public void testCommitOffsetUnknownTopicId() {
+ client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
+ coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE));
+
+ // Prepare five OffsetCommit requests which return a retriable
UNKNOWN_TOPIC_ID error.
+ // Set the timeout accordingly so that commitOffsetsSync completes on
the fifth attempt.
+ // Note that the timer (MockTime) only ticks when its sleep(long)
method is invoked.
+ // Because the ConsumerNetworkClient does not call sleep() in its
network-level poll(.), this
+ // timer never moves forward once the network client is invoked. If
there is no available
+ // response to consume, its internal poll loop never completes. Hence,
the timeout needs to be
+ // enforced in the ConsumerCoordinator, and we need to make sure there
are enough responses
+ // queued in the MockClient to satisfy all invocations of the
ConsumerNetworkClient#poll(.).
+ int offsetCommitCalls = 5;
+ long timeoutMs = rebalanceConfig.retryBackoffMs * offsetCommitCalls;
+
+ IntStream.range(0, offsetCommitCalls).forEach(__ ->
+ prepareOffsetCommitRequest(new OffsetCommitResponseSpec()
+ .expectedOffsets(singletonMap(t1p, 100L))
+ .error(Errors.UNKNOWN_TOPIC_ID)));
+
+ // UnknownTopicIdException is retriable, hence will be retried by the
coordinator as long as
+ // the timeout allows. Note that since topic ids are not part of the
public API of the consumer,
+ // we cannot throw an UnknownTopicId to the user. By design, a false
boolean indicating the
+ // offset commit failed is returned.
+ assertFalse(coordinator.commitOffsetsSync(singletonMap(t1p,
+ new OffsetAndMetadata(100L, "metadata")),
time.timer(timeoutMs)));
+ }
+
+ @Test
+ public void testRetryCommitUnknownTopicId() {
+ client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
+ coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE));
+
+ client.prepareResponse(offsetCommitResponse(singletonMap(t1p,
Errors.UNKNOWN_TOPIC_ID)));
+ client.prepareResponse(offsetCommitResponse(singletonMap(t1p,
Errors.NONE)));
+
+ assertTrue(coordinator.commitOffsetsSync(singletonMap(t1p,
+ new OffsetAndMetadata(100L, "metadata")),
time.timer(Long.MAX_VALUE)));
+ }
+
+ @Test
+ public void testTopicIdsArePopulatedByTheConsumerCoordinator() {
Review Comment:
That is right, I added tests which invoke the sync and async offset commit
method.
--
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
