You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by lu...@apache.org on 2024/04/10 09:27:42 UTC
(kafka) branch trunk updated: KAFKA-16433: BeginningAndEndOffsets and OffsetsForTimes should send an event and return empty with zero timeout provided (#15688)
This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new e5a3f991005 KAFKA-16433: BeginningAndEndOffsets and OffsetsForTimes should send an event and return empty with zero timeout provided (#15688)
e5a3f991005 is described below
commit e5a3f991005dcc9c544918879bdd6127e1ff9aed
Author: Philip Nee <pn...@confluent.io>
AuthorDate: Wed Apr 10 02:27:35 2024 -0700
KAFKA-16433: BeginningAndEndOffsets and OffsetsForTimes should send an event and return empty with zero timeout provided (#15688)
beginningOrEndOffset and offsetforTimes should send a ListOffsetEvent and return empty results if the provided timeout is zero.
A minor change to the mock timer with zero timeout in AsyncConsumerTest.
Reviewers: Lucas Brutschy <lb...@confluent.io>
---
.../clients/consumer/internals/AsyncKafkaConsumer.java | 10 +++++++---
.../consumer/internals/AsyncKafkaConsumerTest.java | 15 ++++++++++-----
2 files changed, 17 insertions(+), 8 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
index f1ce5f46e57..429c0c1bb50 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
@@ -1100,8 +1100,10 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
// If timeout is set to zero return empty immediately; otherwise try to get the results
// and throw timeout exception if it cannot complete in time.
- if (timeout.toMillis() == 0L)
+ if (timeout.toMillis() == 0L) {
+ applicationEventHandler.add(listOffsetsEvent);
return listOffsetsEvent.emptyResults();
+ }
return applicationEventHandler.addAndGet(listOffsetsEvent, timer)
.entrySet()
@@ -1156,12 +1158,14 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
timer,
false);
- Map<TopicPartition, OffsetAndTimestampInternal> offsetAndTimestampMap;
+ // If timeout is set to zero return empty immediately; otherwise try to get the results
+ // and throw timeout exception if it cannot complete in time.
if (timeout.isZero()) {
- // Return an empty results but also send a request to update the highwatermark.
applicationEventHandler.add(listOffsetsEvent);
return listOffsetsEvent.emptyResults();
}
+
+ Map<TopicPartition, OffsetAndTimestampInternal> offsetAndTimestampMap;
offsetAndTimestampMap = applicationEventHandler.addAndGet(
listOffsetsEvent,
timer);
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
index 51fcd2e44eb..0c51a5dccfd 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
@@ -126,6 +126,7 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.atLeast;
@@ -147,7 +148,7 @@ public class AsyncKafkaConsumerTest {
private AsyncKafkaConsumer<String, String> consumer = null;
- private final Time time = new MockTime(1);
+ private Time time = new MockTime(0);
private final FetchCollector<String, String> fetchCollector = mock(FetchCollector.class);
private final ApplicationEventHandler applicationEventHandler = mock(ApplicationEventHandler.class);
private final ConsumerMetadata metadata = mock(ConsumerMetadata.class);
@@ -323,6 +324,7 @@ public class AsyncKafkaConsumerTest {
@Test
public void testCommitted() {
+ time = new MockTime(1);
consumer = newConsumer();
Map<TopicPartition, OffsetAndMetadata> topicPartitionOffsets = mockTopicPartitionOffset();
completeFetchedCommittedOffsetApplicationEventSuccessfully(topicPartitionOffsets);
@@ -842,7 +844,13 @@ public class AsyncKafkaConsumerTest {
consumer = newConsumer();
Map<TopicPartition, OffsetAndTimestampInternal> expectedOffsets = mockOffsetAndTimestamp();
- doReturn(expectedOffsets).when(applicationEventHandler).addAndGet(any(), any());
+ when(applicationEventHandler.addAndGet(any(ListOffsetsEvent.class), any())).thenAnswer(invocation -> {
+ Timer timer = invocation.getArgument(1);
+ if (timer.remainingMs() == 0) {
+ fail("Timer duration should not be zero.");
+ }
+ return expectedOffsets;
+ });
Map<TopicPartition, Long> result = assertDoesNotThrow(() -> consumer.beginningOffsets(expectedOffsets.keySet(), Duration.ofMillis(1)));
@@ -1581,7 +1589,6 @@ public class AsyncKafkaConsumerTest {
@Test
public void testProcessBackgroundEventsWithInitialDelay() throws Exception {
consumer = newConsumer();
- Time time = new MockTime();
Timer timer = time.timer(1000);
CompletableFuture<?> future = mock(CompletableFuture.class);
CountDownLatch latch = new CountDownLatch(3);
@@ -1617,7 +1624,6 @@ public class AsyncKafkaConsumerTest {
@Test
public void testProcessBackgroundEventsWithoutDelay() {
consumer = newConsumer();
- Time time = new MockTime();
Timer timer = time.timer(1000);
// Create a future that is already completed.
@@ -1639,7 +1645,6 @@ public class AsyncKafkaConsumerTest {
@Test
public void testProcessBackgroundEventsTimesOut() throws Exception {
consumer = newConsumer();
- Time time = new MockTime();
Timer timer = time.timer(1000);
CompletableFuture<?> future = mock(CompletableFuture.class);