dajac commented on a change in pull request #11726:
URL: https://github.com/apache/kafka/pull/11726#discussion_r797572591
##########
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()
Review comment:
Could we import `TimeoutException` instead of specifying the full
qualified name every time?
##########
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:
I think that we could simplify this code and avoid using an executor by
doing as follow:
```
for (int i = 0; i < 10; i++) {
client.prepareResponse(
request -> {
time.sleep(defaultApiTimeoutMs / 10);
return request instanceof ListOffsetsRequest;
},
listOffsetsResponse(
Collections.emptyMap(),
Collections.singletonMap(tp0,
Errors.UNKNOWN_TOPIC_OR_PARTITION)
));
}
```
What do you think?
--
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
