[GitHub] [kafka] dajac commented on a change in pull request #11726: KAFKA-13637: Use default.api.timeout.ms as default timeout value for KafkaConsumer.endOffsets

[GitBox <gi...@apache.org>]

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-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org