You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/02/02 06:50:00 UTC
[GitHub] [kafka] dengziming opened a new pull request #11726: KAFKA-13637: Use default.api.timeout.ms as default timeout value for KafkaConsumer.endOffsets
dengziming opened a new pull request #11726:
URL: https://github.com/apache/kafka/pull/11726
*More detailed description of your change*
In KafkaConsumer, we use `request.timeout.ms` in `endOffsets` and `default.api.timeout.ms` when in `beginningOffsets`, we should use `default.api.timeout.ms` for both.
*Summary of testing strategy (including rationale)*
QA
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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
[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
Posted by 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
[GitHub] [kafka] dengziming commented on a change in pull request #11726: KAFKA-13637: Use default.api.timeout.ms as default timeout value for KafkaConsumer.endOffsets
Posted by GitBox <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dengziming commented on pull request #11726: KAFKA-13637: Use default.api.timeout.ms as default timeout value for KafkaConsumer.endOffsets
Posted by GitBox <gi...@apache.org>.
dengziming commented on pull request #11726:
URL: https://github.com/apache/kafka/pull/11726#issuecomment-1027842174
@dajac , Yes, I added a unit test for this.
--
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
[GitHub] [kafka] dajac commented on pull request #11726: KAFKA-13637: Use default.api.timeout.ms as default timeout value for KafkaConsumer.endOffsets
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #11726:
URL: https://github.com/apache/kafka/pull/11726#issuecomment-1027677422
Good catch! It seems that we forgot it in https://github.com/apache/kafka/commit/53ca52f855e903907378188d29224b3f9cefa6cb. Have you tried to add a unit test for the bug?
--
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
[GitHub] [kafka] dengziming commented on a change in pull request #11726: KAFKA-13637: Use default.api.timeout.ms as default timeout value for KafkaConsumer.endOffsets
Posted by GitBox <gi...@apache.org>.
dengziming commented on a change in pull request #11726:
URL: https://github.com/apache/kafka/pull/11726#discussion_r798204734
##########
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:
we already imported java.util.concurrent.TimeoutException so can't import org.apache.kafka.common.errors.TimeoutException here.
--
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
[GitHub] [kafka] dajac merged pull request #11726: KAFKA-13637: Use default.api.timeout.ms as default timeout value for KafkaConsumer.endOffsets
Posted by GitBox <gi...@apache.org>.
dajac merged pull request #11726:
URL: https://github.com/apache/kafka/pull/11726
--
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