dengziming commented on a change in pull request #11726:
URL: https://github.com/apache/kafka/pull/11726#discussion_r798205426
##########
File path:
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
##########
@@ -2948,6 +2948,64 @@ public void testAssignorNameConflict() {
() -> new KafkaConsumer<>(configs, new StringDeserializer(), new
StringDeserializer()));
}
+ @Test
+ public void testOffsetsForTimesTimeout() {
+ final KafkaConsumer<String, String> consumer =
consumerForCheckingTimeoutException();
+ assertEquals(
+ "Failed to get offsets by times in 60000ms",
+
assertThrows(org.apache.kafka.common.errors.TimeoutException.class, () ->
consumer.offsetsForTimes(singletonMap(tp0, 0L))).getMessage()
+ );
+ }
+
+ @Test
+ public void testBeginningOffsetsTimeout() {
+ final KafkaConsumer<String, String> consumer =
consumerForCheckingTimeoutException();
+ assertEquals(
+ "Failed to get offsets by times in 60000ms",
+
assertThrows(org.apache.kafka.common.errors.TimeoutException.class, () ->
consumer.beginningOffsets(singletonList(tp0))).getMessage()
+ );
+ }
+
+ @Test
+ public void testEndOffsetsTimeout() {
+ final KafkaConsumer<String, String> consumer =
consumerForCheckingTimeoutException();
+ assertEquals(
+ "Failed to get offsets by times in 60000ms",
+
assertThrows(org.apache.kafka.common.errors.TimeoutException.class, () ->
consumer.endOffsets(singletonList(tp0))).getMessage()
+ );
+ }
+
+ private KafkaConsumer<String, String>
consumerForCheckingTimeoutException() {
+ final Time time = new MockTime();
+ SubscriptionState subscription = new SubscriptionState(new
LogContext(), OffsetResetStrategy.EARLIEST);
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+
+ initMetadata(client, singletonMap(topic, 1));
+ Node node = metadata.fetch().nodes().get(0);
+
+ ConsumerPartitionAssignor assignor = new RangeAssignor();
+
+ final KafkaConsumer<String, String> consumer = newConsumer(time,
client, subscription, metadata, assignor, false, groupInstanceId);
+
+ final ScheduledExecutorService exec =
Executors.newSingleThreadScheduledExecutor();
+ for (int i = 0; i < 10; i++) {
+ // Prepare a retriable error periodically for the client to retry
connection
+ exec.schedule(
+ () -> client.prepareResponseFrom(
+ listOffsetsResponse(
+ Collections.emptyMap(),
+ Collections.singletonMap(tp0,
Errors.UNKNOWN_TOPIC_OR_PARTITION)
+ ),
+ node), 50L, TimeUnit.MILLISECONDS);
+ // Sleep periodically to make loop retry timeout
+ exec.schedule(() -> time.sleep(defaultApiTimeoutMs / 10), 50L,
TimeUnit.MILLISECONDS);
+
+ }
Review comment:
Yes, this is a good idea, and this make the test more deterministic.
--
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
