From 4175df13f82d3972bf3d9fa46680b27e9218c53b Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Fri, 19 Nov 2021 09:59:30 -0500 Subject: [PATCH 1/5] KAFKA-12980: Return empty record batch from Consumer::poll when position advances --- .../kafka/clients/consumer/KafkaConsumer.java | 25 ++-- .../clients/consumer/internals/Fetch.java | 123 ++++++++++++++++++ .../clients/consumer/internals/Fetcher.java | 46 +++---- .../consumer/internals/FetcherTest.java | 113 +++++++++------- 4 files changed, 220 insertions(+), 87 deletions(-) create mode 100644 clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetch.java diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java index 2e2c09f57af3d..c4b0eda2266b5 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java @@ -26,6 +26,7 @@ import org.apache.kafka.clients.consumer.internals.ConsumerInterceptors; import org.apache.kafka.clients.consumer.internals.ConsumerMetadata; import org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient; +import org.apache.kafka.clients.consumer.internals.Fetch; import org.apache.kafka.clients.consumer.internals.Fetcher; import org.apache.kafka.clients.consumer.internals.FetcherMetricsRegistry; import org.apache.kafka.clients.consumer.internals.KafkaConsumerMetrics; @@ -1235,9 +1236,15 @@ private ConsumerRecords poll(final Timer timer, final boolean includeMetad } } - final Map>> records = pollForFetches(timer); - if (!records.isEmpty()) { - // before returning the fetched records, we can send off the next round of fetches + final Fetch fetch = pollForFetches(timer); + if (!fetch.isEmpty()) { + if (fetch.records().isEmpty()) { + log.debug( + "Returning empty records from poll since the consumer's position has advanced " + + "for at least one topic partition; this may happen in the case of aborted or empty transactions" + ); + return ConsumerRecords.empty(); + } // before returning the fetched records, we can send off the next round of fetches // and avoid block waiting for their responses to enable pipelining while the user // is handling the fetched records. // @@ -1247,7 +1254,7 @@ private ConsumerRecords poll(final Timer timer, final boolean includeMetad client.transmitSends(); } - return this.interceptors.onConsume(new ConsumerRecords<>(records)); + return this.interceptors.onConsume(new ConsumerRecords<>(fetch.records())); } } while (timer.notExpired()); @@ -1269,14 +1276,14 @@ boolean updateAssignmentMetadataIfNeeded(final Timer timer, final boolean waitFo /** * @throws KafkaException if the rebalance callback throws exception */ - private Map>> pollForFetches(Timer timer) { + private Fetch pollForFetches(Timer timer) { long pollTimeout = coordinator == null ? timer.remainingMs() : Math.min(coordinator.timeToNextPoll(timer.currentTimeMs()), timer.remainingMs()); // if data is available already, return it immediately - final Map>> records = fetcher.fetchedRecords(); - if (!records.isEmpty()) { - return records; + final Fetch fetch = fetcher.collectFetch(); + if (!fetch.isEmpty()) { + return fetch; } // send any new fetches (won't resend pending fetches) @@ -1301,7 +1308,7 @@ private Map>> pollForFetches(Timer tim }); timer.update(pollTimer.currentTimeMs()); - return fetcher.fetchedRecords(); + return fetcher.collectFetch(); } /** diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetch.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetch.java new file mode 100644 index 0000000000000..0f13b9020a823 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetch.java @@ -0,0 +1,123 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.clients.consumer.internals; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.TopicPartition; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +import static org.apache.kafka.common.utils.Utils.mkEntry; +import static org.apache.kafka.common.utils.Utils.mkMap; + +public class Fetch { + private final Map>> records; + private boolean positionAdvanced; + private int numRecords; + + public static Fetch empty() { + return new Fetch<>(new HashMap<>(), false, 0); + } + + public static Fetch forPartition( + TopicPartition partition, + List> records, + boolean positionAdvanced + ) { + Map>> recordsMap = records.isEmpty() + ? new HashMap<>() + : mkMap(mkEntry(partition, records)); + return new Fetch<>(recordsMap, positionAdvanced, records.size()); + } + + private Fetch( + Map>> records, + boolean positionAdvanced, + int numRecords + ) { + this.records = records; + this.positionAdvanced = positionAdvanced; + this.numRecords = numRecords; + } + + /** + * Add another {@link Fetch} to this one; all of its records will be added to this fetch's + * {@link #records()} records}, and if the other fetch + * {@link #positionAdvanced() advanced the consume position for any topic partition}, + * this fetch will be marked as having advanced the consume position as well. + * @param fetch the other fetch to add; may not be null + */ + public void add(Fetch fetch) { + Objects.requireNonNull(fetch); + addRecords(fetch.records); + this.positionAdvanced |= fetch.positionAdvanced; + } + + /** + * @return all of the non-control messages for this fetch, grouped by partition + */ + public Map>> records() { + return Collections.unmodifiableMap(records); + } + + /** + * @return whether the fetch caused the consumer's + * {@link org.apache.kafka.clients.consumer.KafkaConsumer#position(TopicPartition) position} to advance for at + * least one of the topic partitions in this fetch + */ + public boolean positionAdvanced() { + return positionAdvanced; + } + + /** + * @return the total number of non-control messages for this fetch, across all partitions + */ + public int numRecords() { + return numRecords; + } + + /** + * @return {@code true} if and only if this fetch did not return any user-visible (i.e., non-control) records, and + * did not cause the consumer position to advance for any topic partitions + */ + public boolean isEmpty() { + return numRecords == 0 && !positionAdvanced; + } + + private void addRecords(Map>> records) { + records.forEach((partition, partRecords) -> { + this.numRecords += partRecords.size(); + List> currentRecords = this.records.get(partition); + if (currentRecords == null) { + this.records.put(partition, partRecords); + } else { + // this case shouldn't usually happen because we only send one fetch at a time per partition, + // but it might conceivably happen in some rare cases (such as partition leader changes). + // we have to copy to a new list because the old one may be immutable + List> newRecords = new ArrayList<>(currentRecords.size() + partRecords.size()); + newRecords.addAll(currentRecords); + newRecords.addAll(partRecords); + this.records.put(partition, newRecords); + } + }); + } +} \ No newline at end of file diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index d567f5b5107c9..c6f29cc0eddce 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -109,8 +109,6 @@ import java.util.function.Function; import java.util.stream.Collectors; -import static java.util.Collections.emptyList; - /** * This class manages the fetching process with the brokers. *

@@ -637,15 +635,15 @@ private Map beginningOrEndOffset(Collection>> fetchedRecords() { - Map>> fetched = new HashMap<>(); + public Fetch collectFetch() { + Fetch fetch = Fetch.empty(); Queue pausedCompletedFetches = new ArrayDeque<>(); int recordsRemaining = maxPollRecords; @@ -665,7 +663,7 @@ public Map>> fetchedRecords() { // in cases such as the TopicAuthorizationException, and the second condition ensures that no // potential data loss due to an exception in a following record. FetchResponseData.PartitionData partition = records.partitionData; - if (fetched.isEmpty() && FetchResponse.recordsOrFail(partition).sizeInBytes() == 0) { + if (fetch.isEmpty() && FetchResponse.recordsOrFail(partition).sizeInBytes() == 0) { completedFetches.poll(); } throw e; @@ -681,28 +679,13 @@ public Map>> fetchedRecords() { pausedCompletedFetches.add(nextInLineFetch); nextInLineFetch = null; } else { - List> records = fetchRecords(nextInLineFetch, recordsRemaining); - - if (!records.isEmpty()) { - TopicPartition partition = nextInLineFetch.partition; - List> currentRecords = fetched.get(partition); - if (currentRecords == null) { - fetched.put(partition, records); - } else { - // this case shouldn't usually happen because we only send one fetch at a time per partition, - // but it might conceivably happen in some rare cases (such as partition leader changes). - // we have to copy to a new list because the old one may be immutable - List> newRecords = new ArrayList<>(records.size() + currentRecords.size()); - newRecords.addAll(currentRecords); - newRecords.addAll(records); - fetched.put(partition, newRecords); - } - recordsRemaining -= records.size(); - } + Fetch nextFetch = fetchRecords(nextInLineFetch, recordsRemaining); + recordsRemaining -= nextFetch.numRecords(); + fetch.add(nextFetch); } } } catch (KafkaException e) { - if (fetched.isEmpty()) + if (fetch.isEmpty()) throw e; } finally { // add any polled completed fetches for paused partitions back to the completed fetches queue to be @@ -710,10 +693,10 @@ public Map>> fetchedRecords() { completedFetches.addAll(pausedCompletedFetches); } - return fetched; + return fetch; } - private List> fetchRecords(CompletedFetch completedFetch, int maxRecords) { + private Fetch fetchRecords(CompletedFetch completedFetch, int maxRecords) { if (!subscriptions.isAssigned(completedFetch.partition)) { // this can happen when a rebalance happened before fetched records are returned to the consumer's poll call log.debug("Not returning fetched records for partition {} since it is no longer assigned", @@ -735,6 +718,8 @@ private List> fetchRecords(CompletedFetch completedFetch, i log.trace("Returning {} fetched records at offset {} for assigned partition {}", partRecords.size(), position, completedFetch.partition); + boolean positionAdvanced = false; + if (completedFetch.nextFetchOffset > position.offset) { FetchPosition nextPosition = new FetchPosition( completedFetch.nextFetchOffset, @@ -742,6 +727,7 @@ private List> fetchRecords(CompletedFetch completedFetch, i position.currentLeader); log.trace("Update fetching position to {} for partition {}", nextPosition, completedFetch.partition); subscriptions.position(completedFetch.partition, nextPosition); + positionAdvanced = true; } Long partitionLag = subscriptions.partitionLag(completedFetch.partition, isolationLevel); @@ -753,7 +739,7 @@ private List> fetchRecords(CompletedFetch completedFetch, i this.sensors.recordPartitionLead(completedFetch.partition, lead); } - return partRecords; + return Fetch.forPartition(completedFetch.partition, partRecords, positionAdvanced); } else { // these records aren't next in line based on the last consumed position, ignore them // they must be from an obsolete request @@ -765,7 +751,7 @@ private List> fetchRecords(CompletedFetch completedFetch, i log.trace("Draining fetched records for partition {}", completedFetch.partition); completedFetch.drain(); - return emptyList(); + return Fetch.empty(); } // Visible for testing diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java index b3dee9ecd6712..545bf24b9ec25 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java @@ -804,7 +804,7 @@ public byte[] deserialize(String topic, byte[] data) { // The fetcher should block on Deserialization error for (int i = 0; i < 2; i++) { try { - fetcher.fetchedRecords(); + fetcher.collectFetch(); fail("fetchedRecords should have raised"); } catch (SerializationException e) { // the position should not advance since no data has been returned @@ -865,7 +865,7 @@ public void testParseCorruptedRecord() throws Exception { consumerClient.poll(time.timer(0)); // the first fetchedRecords() should return the first valid message - assertEquals(1, fetcher.fetchedRecords().get(tp0).size()); + assertEquals(1, fetchedRecords().get(tp0).size()); assertEquals(1, subscriptions.position(tp0).offset); ensureBlockOnRecord(1L); @@ -885,7 +885,7 @@ private void ensureBlockOnRecord(long blockedOffset) { // the fetchedRecords() should always throw exception due to the invalid message at the starting offset. for (int i = 0; i < 2; i++) { try { - fetcher.fetchedRecords(); + fetcher.collectFetch(); fail("fetchedRecords should have raised KafkaException"); } catch (KafkaException e) { assertEquals(blockedOffset, subscriptions.position(tp0).offset); @@ -897,7 +897,7 @@ private void seekAndConsumeRecord(ByteBuffer responseBuffer, long toOffset) { // Seek to skip the bad record and fetch again. subscriptions.seekUnvalidated(tp0, new SubscriptionState.FetchPosition(toOffset, Optional.empty(), metadata.currentLeader(tp0))); // Should not throw exception after the seek. - fetcher.fetchedRecords(); + fetcher.collectFetch(); assertEquals(1, fetcher.sendFetches()); client.prepareResponse(fullFetchResponse(tidp0, MemoryRecords.readableRecords(responseBuffer), Errors.NONE, 100L, 0)); consumerClient.poll(time.timer(0)); @@ -941,7 +941,7 @@ public void testInvalidDefaultRecordBatch() { // the fetchedRecords() should always throw exception due to the bad batch. for (int i = 0; i < 2; i++) { try { - fetcher.fetchedRecords(); + fetcher.collectFetch(); fail("fetchedRecords should have raised KafkaException"); } catch (KafkaException e) { assertEquals(0, subscriptions.position(tp0).offset); @@ -970,7 +970,7 @@ public void testParseInvalidRecordBatch() { client.prepareResponse(fullFetchResponse(tidp0, MemoryRecords.readableRecords(buffer), Errors.NONE, 100L, 0)); consumerClient.poll(time.timer(0)); try { - fetcher.fetchedRecords(); + fetcher.collectFetch(); fail("fetchedRecords should have raised"); } catch (KafkaException e) { // the position should not advance since no data has been returned @@ -1144,7 +1144,7 @@ public void testFetchRequestWhenRecordTooLarge() { client.setNodeApiVersions(NodeApiVersions.create(ApiKeys.FETCH.id, (short) 2, (short) 2)); makeFetchRequestWithIncompleteRecord(); try { - fetcher.fetchedRecords(); + fetcher.collectFetch(); fail("RecordTooLargeException should have been raised"); } catch (RecordTooLargeException e) { assertTrue(e.getMessage().startsWith("There are some messages at [Partition=Offset]: ")); @@ -1167,7 +1167,7 @@ public void testFetchRequestInternalError() { buildFetcher(); makeFetchRequestWithIncompleteRecord(); try { - fetcher.fetchedRecords(); + fetcher.collectFetch(); fail("RecordTooLargeException should have been raised"); } catch (KafkaException e) { assertTrue(e.getMessage().startsWith("Failed to make progress reading messages")); @@ -1200,7 +1200,7 @@ public void testUnauthorizedTopic() { client.prepareResponse(fullFetchResponse(tidp0, this.records, Errors.TOPIC_AUTHORIZATION_FAILED, 100L, 0)); consumerClient.poll(time.timer(0)); try { - fetcher.fetchedRecords(); + fetcher.collectFetch(); fail("fetchedRecords should have thrown"); } catch (TopicAuthorizationException e) { assertEquals(singleton(topicName), e.unauthorizedTopics()); @@ -1228,7 +1228,7 @@ public void testFetchDuringEagerRebalance() { consumerClient.poll(time.timer(0)); // The active fetch should be ignored since its position is no longer valid - assertTrue(fetcher.fetchedRecords().isEmpty()); + assertTrue(fetchedRecords().isEmpty()); } @Test @@ -1269,7 +1269,7 @@ public void testInFlightFetchOnPausedPartition() { client.prepareResponse(fullFetchResponse(tidp0, this.records, Errors.NONE, 100L, 0)); consumerClient.poll(time.timer(0)); - assertNull(fetcher.fetchedRecords().get(tp0)); + assertNull(fetchedRecords().get(tp0)); } @Test @@ -1298,11 +1298,10 @@ public void testFetchOnCompletedFetchesForPausedAndResumedPartitions() { client.prepareResponse(fullFetchResponse(tidp0, this.records, Errors.NONE, 100L, 0)); consumerClient.poll(time.timer(0)); - Map>> fetchedRecords = fetchedRecords(); - assertEquals(0, fetchedRecords.size(), "Should not return any records when partition is paused"); + assertEmptyFetch("Should not return any records or advance position when partition is paused"); + assertTrue(fetcher.hasCompletedFetches(), "Should still contain completed fetches"); assertFalse(fetcher.hasAvailableFetches(), "Should not have any available (non-paused) completed fetches"); - assertNull(fetchedRecords.get(tp0)); assertEquals(0, fetcher.sendFetches()); subscriptions.resume(tp0); @@ -1310,14 +1309,13 @@ public void testFetchOnCompletedFetchesForPausedAndResumedPartitions() { assertTrue(fetcher.hasAvailableFetches(), "Should have available (non-paused) completed fetches"); consumerClient.poll(time.timer(0)); - fetchedRecords = fetchedRecords(); + Map>> fetchedRecords = fetchedRecords(); assertEquals(1, fetchedRecords.size(), "Should return records when partition is resumed"); assertNotNull(fetchedRecords.get(tp0)); assertEquals(3, fetchedRecords.get(tp0).size()); consumerClient.poll(time.timer(0)); - fetchedRecords = fetchedRecords(); - assertEquals(0, fetchedRecords.size(), "Should not return records after previously paused partitions are fetched"); + assertEmptyFetch("Should not return records or advance position after previously paused partitions are fetched"); assertFalse(fetcher.hasCompletedFetches(), "Should no longer contain completed fetches"); } @@ -1351,8 +1349,7 @@ public void testFetchOnCompletedFetchesForSomePausedPartitions() { assertNotNull(fetchedRecords.get(tp1)); assertNull(fetchedRecords.get(tp0)); - fetchedRecords = fetchedRecords(); - assertEquals(0, fetchedRecords.size(), "Should return no records for remaining paused partition"); + assertEmptyFetch("Should not return records or advance position for remaining paused partition"); assertTrue(fetcher.hasCompletedFetches(), "Should still contain completed fetches"); } @@ -1382,8 +1379,7 @@ public void testFetchOnCompletedFetchesForAllPausedPartitions() { consumerClient.poll(time.timer(0)); - fetchedRecords = fetchedRecords(); - assertEquals(0, fetchedRecords.size(), "Should return no records for all paused partitions"); + assertEmptyFetch("Should not return records or advance position for all paused partitions"); assertTrue(fetcher.hasCompletedFetches(), "Should still contain completed fetches"); assertFalse(fetcher.hasAvailableFetches(), "Should not have any available (non-paused) completed fetches"); } @@ -1414,7 +1410,7 @@ public void testPartialFetchWithPausedPartitions() { fetchedRecords = fetchedRecords(); - assertEquals(0, fetchedRecords.size(), "Should return no records for paused partitions"); + assertEmptyFetch("Should not return records or advance position for paused partitions"); assertTrue(fetcher.hasCompletedFetches(), "Should have 1 entry in completed fetches"); assertFalse(fetcher.hasAvailableFetches(), "Should not have any available (non-paused) completed fetches"); @@ -1443,9 +1439,9 @@ public void testFetchDiscardedAfterPausedPartitionResumedAndSeekedToNewOffset() consumerClient.poll(time.timer(0)); assertTrue(fetcher.hasCompletedFetches(), "Should have 1 entry in completed fetches"); - Map>> fetchedRecords = fetchedRecords(); - assertEquals(0, fetchedRecords.size(), "Should not return any records because we seeked to a new offset"); - assertNull(fetchedRecords.get(tp0)); + Fetch fetch = collectFetch(); + assertEquals(emptyMap(), fetch.records(), "Should not return any records because we seeked to a new offset"); + assertFalse(fetch.positionAdvanced()); assertFalse(fetcher.hasCompletedFetches(), "Should have no completed fetches"); } @@ -1458,7 +1454,7 @@ public void testFetchNotLeaderOrFollower() { assertEquals(1, fetcher.sendFetches()); client.prepareResponse(fullFetchResponse(tidp0, this.records, Errors.NOT_LEADER_OR_FOLLOWER, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEmptyFetch("Should not return records or advance position on fetch error"); assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds())); } @@ -1471,7 +1467,7 @@ public void testFetchUnknownTopicOrPartition() { assertEquals(1, fetcher.sendFetches()); client.prepareResponse(fullFetchResponse(tidp0, this.records, Errors.UNKNOWN_TOPIC_OR_PARTITION, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEmptyFetch("Should not return records or advance position on fetch error"); assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds())); } @@ -1484,7 +1480,7 @@ public void testFetchUnknownTopicId() { assertEquals(1, fetcher.sendFetches()); client.prepareResponse(fullFetchResponse(tidp0, this.records, Errors.UNKNOWN_TOPIC_ID, -1L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEmptyFetch("Should not return records or advance position on fetch error"); assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds())); } @@ -1497,7 +1493,7 @@ public void testFetchSessionIdError() { assertEquals(1, fetcher.sendFetches()); client.prepareResponse(fetchResponseWithTopLevelError(tidp0, Errors.FETCH_SESSION_TOPIC_ID_ERROR, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEmptyFetch("Should not return records or advance position on fetch error"); assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds())); } @@ -1510,7 +1506,7 @@ public void testFetchInconsistentTopicId() { assertEquals(1, fetcher.sendFetches()); client.prepareResponse(fullFetchResponse(tidp0, this.records, Errors.INCONSISTENT_TOPIC_ID, -1L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEmptyFetch("Should not return records or advance position on fetch error"); assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds())); } @@ -1524,7 +1520,7 @@ public void testFetchFencedLeaderEpoch() { client.prepareResponse(fullFetchResponse(tidp0, this.records, Errors.FENCED_LEADER_EPOCH, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size(), "Should not return any records"); + assertEmptyFetch("Should not return records or advance position on fetch error"); assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds()), "Should have requested metadata update"); } @@ -1538,7 +1534,7 @@ public void testFetchUnknownLeaderEpoch() { client.prepareResponse(fullFetchResponse(tidp0, this.records, Errors.UNKNOWN_LEADER_EPOCH, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size(), "Should not return any records"); + assertEmptyFetch("Should not return records or advance position on fetch error"); assertNotEquals(0L, metadata.timeToNextUpdate(time.milliseconds()), "Should not have requested metadata update"); } @@ -1580,7 +1576,7 @@ public void testFetchOffsetOutOfRange() { assertEquals(1, fetcher.sendFetches()); client.prepareResponse(fullFetchResponse(tidp0, this.records, Errors.OFFSET_OUT_OF_RANGE, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEmptyFetch("Should not return records or advance position on fetch error"); assertTrue(subscriptions.isOffsetResetNeeded(tp0)); assertNull(subscriptions.validPosition(tp0)); assertNull(subscriptions.position(tp0)); @@ -1598,7 +1594,7 @@ public void testStaleOutOfRangeError() { client.prepareResponse(fullFetchResponse(tidp0, this.records, Errors.OFFSET_OUT_OF_RANGE, 100L, 0)); subscriptions.seek(tp0, 1); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEmptyFetch("Should not return records or advance position on fetch error"); assertFalse(subscriptions.isOffsetResetNeeded(tp0)); assertEquals(1, subscriptions.position(tp0).offset); } @@ -1616,7 +1612,7 @@ public void testFetchedRecordsAfterSeek() { consumerClient.poll(time.timer(0)); assertFalse(subscriptions.isOffsetResetNeeded(tp0)); subscriptions.seek(tp0, 2); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEmptyFetch("Should not return records or advance position after seeking to end of topic partition"); } @Test @@ -1634,7 +1630,7 @@ public void testFetchOffsetOutOfRangeException() { assertFalse(subscriptions.isOffsetResetNeeded(tp0)); for (int i = 0; i < 2; i++) { OffsetOutOfRangeException e = assertThrows(OffsetOutOfRangeException.class, () -> - fetcher.fetchedRecords()); + fetcher.collectFetch()); assertEquals(singleton(tp0), e.offsetOutOfRangePartitions().keySet()); assertEquals(0L, e.offsetOutOfRangePartitions().get(tp0).longValue()); } @@ -1788,7 +1784,7 @@ public void testSeekBeforeException() { client.prepareResponse(fullFetchResponse(tidp0, this.records, Errors.NONE, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(2, fetcher.fetchedRecords().get(tp0).size()); + assertEquals(2, fetchedRecords().get(tp0).size()); subscriptions.assignFromUser(mkSet(tp0, tp1)); subscriptions.seekUnvalidated(tp1, new SubscriptionState.FetchPosition(1, Optional.empty(), metadata.currentLeader(tp1))); @@ -1801,11 +1797,11 @@ public void testSeekBeforeException() { .setHighWatermark(100)); client.prepareResponse(FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, new LinkedHashMap<>(partitions))); consumerClient.poll(time.timer(0)); - assertEquals(1, fetcher.fetchedRecords().get(tp0).size()); + assertEquals(1, fetchedRecords().get(tp0).size()); subscriptions.seek(tp1, 10); // Should not throw OffsetOutOfRangeException after the seek - assertEquals(0, fetcher.fetchedRecords().size()); + assertEmptyFetch("Should not return records or advance position after seeking to end of topic partitions"); } @Test @@ -1818,7 +1814,7 @@ public void testFetchDisconnected() { assertEquals(1, fetcher.sendFetches()); client.prepareResponse(fullFetchResponse(tidp0, this.records, Errors.NONE, 100L, 0), true); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEmptyFetch("Should not return records or advance position on disconnect"); // disconnects should have no affect on subscription state assertFalse(subscriptions.isOffsetResetNeeded(tp0)); @@ -2755,7 +2751,7 @@ public void testFetchResponseMetricsWithOnePartitionError() { assertEquals(1, fetcher.sendFetches()); client.prepareResponse(FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, new LinkedHashMap<>(partitions))); consumerClient.poll(time.timer(0)); - fetcher.fetchedRecords(); + fetcher.collectFetch(); int expectedBytes = 0; for (Record record : records.records()) @@ -2801,7 +2797,7 @@ public void testFetchResponseMetricsWithOnePartitionAtTheWrongOffset() { client.prepareResponse(FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, new LinkedHashMap<>(partitions))); consumerClient.poll(time.timer(0)); - fetcher.fetchedRecords(); + fetcher.collectFetch(); // we should have ignored the record at the wrong offset int expectedBytes = 0; @@ -3226,8 +3222,9 @@ public void testSkippingAbortedTransactions() { consumerClient.poll(time.timer(0)); assertTrue(fetcher.hasCompletedFetches()); - Map>> fetchedRecords = fetchedRecords(); - assertFalse(fetchedRecords.containsKey(tp0)); + Fetch fetch = collectFetch(); + assertEquals(emptyMap(), fetch.records()); + assertTrue(fetch.positionAdvanced()); } @Test @@ -3500,8 +3497,9 @@ public void testUpdatePositionOnEmptyBatch() { consumerClient.poll(time.timer(0)); assertTrue(fetcher.hasCompletedFetches()); - Map>> allFetchedRecords = fetchedRecords(); - assertTrue(allFetchedRecords.isEmpty()); + Fetch fetch = collectFetch(); + assertEquals(emptyMap(), fetch.records()); + assertTrue(fetch.positionAdvanced()); // The next offset should point to the next batch assertEquals(lastOffset + 1, subscriptions.position(tp0).offset); @@ -5015,9 +5013,28 @@ private MetadataResponse newMetadataResponse(String topic, Errors error) { initialUpdateResponse.controller().id(), Collections.singletonList(topicMetadata)); } - @SuppressWarnings("unchecked") + /** + * Assert that the {@link Fetcher#collectFetch() latest fetch} does not contain any + * {@link Fetch#records() user-visible records}, did not + * {@link Fetch#positionAdvanced() advance the consumer's position}, + * and is {@link Fetch#isEmpty() empty}. + * @param reason the reason to include for assertion methods such as {@link org.junit.jupiter.api.Assertions#assertTrue(boolean, String)} + */ + private void assertEmptyFetch(String reason) { + Fetch fetch = collectFetch(); + assertEquals(Collections.emptyMap(), fetch.records(), reason); + assertFalse(fetch.positionAdvanced(), reason); + assertTrue(fetch.isEmpty(), reason); + } + private Map>> fetchedRecords() { - return (Map) fetcher.fetchedRecords(); + Fetch fetch = collectFetch(); + return fetch.records(); + } + + @SuppressWarnings("unchecked") + private Fetch collectFetch() { + return (Fetch) fetcher.collectFetch(); } private void buildFetcher(int maxPollRecords) { From 62706bd5dca758ce9ab468d5c62c985908f760ad Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Mon, 22 Nov 2021 10:43:55 -0500 Subject: [PATCH 2/5] KAFKA-12980: Address review comments --- .../kafka/clients/consumer/KafkaConsumer.java | 15 +++++---------- .../kafka/clients/consumer/internals/Fetch.java | 2 +- .../kafka/clients/consumer/internals/Fetcher.java | 9 +++++++++ 3 files changed, 15 insertions(+), 11 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java index c4b0eda2266b5..0f0b3b6d0ac87 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java @@ -1176,9 +1176,10 @@ public ConsumerRecords poll(final long timeoutMs) { * offset for the subscribed list of partitions * *

- * This method returns immediately if there are records available. Otherwise, it will await the passed timeout. - * If the timeout expires, an empty record set will be returned. Note that this method may block beyond the - * timeout in order to execute custom {@link ConsumerRebalanceListener} callbacks. + * This method returns immediately if there are records available or if the position advances past control records. + * Otherwise, it will await the passed timeout. If the timeout expires, an empty record set will be returned. + * Note that this method may block beyond the timeout in order to execute custom + * {@link ConsumerRebalanceListener} callbacks. * * * @param timeout The maximum time to block (must not be greater than {@link Long#MAX_VALUE} milliseconds) @@ -1238,13 +1239,7 @@ private ConsumerRecords poll(final Timer timer, final boolean includeMetad final Fetch fetch = pollForFetches(timer); if (!fetch.isEmpty()) { - if (fetch.records().isEmpty()) { - log.debug( - "Returning empty records from poll since the consumer's position has advanced " - + "for at least one topic partition; this may happen in the case of aborted or empty transactions" - ); - return ConsumerRecords.empty(); - } // before returning the fetched records, we can send off the next round of fetches + // before returning the fetched records, we can send off the next round of fetches // and avoid block waiting for their responses to enable pipelining while the user // is handling the fetched records. // diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetch.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetch.java index 0f13b9020a823..54c8e6b4b9714 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetch.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetch.java @@ -120,4 +120,4 @@ private void addRecords(Map>> records) } }); } -} \ No newline at end of file +} diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index c6f29cc0eddce..62cae9d3da98e 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -728,6 +728,15 @@ private Fetch fetchRecords(CompletedFetch completedFetch, int maxRecords) log.trace("Update fetching position to {} for partition {}", nextPosition, completedFetch.partition); subscriptions.position(completedFetch.partition, nextPosition); positionAdvanced = true; + if (partRecords.isEmpty()) { + log.debug( + "Advanced position for partition {} without receiving any user-visible records. " + + "This is likely due to skipping over control records in the current fetch, " + + "and may result in the consumer returning an empty record batch when " + + "polled before its poll timeout has elapsed.", + completedFetch.partition + ); + } } Long partitionLag = subscriptions.partitionLag(completedFetch.partition, isolationLevel); From c90d85f49a709c0664289485f82be97bf99e21d8 Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Fri, 3 Dec 2021 14:14:17 -0500 Subject: [PATCH 3/5] KAFKA-12980: Simplify new logging messages --- .../apache/kafka/clients/consumer/KafkaConsumer.java | 3 ++- .../kafka/clients/consumer/internals/Fetcher.java | 12 ++++-------- 2 files changed, 6 insertions(+), 9 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java index 0f0b3b6d0ac87..62573f2b618d3 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java @@ -1176,7 +1176,8 @@ public ConsumerRecords poll(final long timeoutMs) { * offset for the subscribed list of partitions * *

- * This method returns immediately if there are records available or if the position advances past control records. + * This method returns immediately if there are records available or if the position advances past control records + * or aborted transactions when isolation.level=READ_COMMITTED. * Otherwise, it will await the passed timeout. If the timeout expires, an empty record set will be returned. * Note that this method may block beyond the timeout in order to execute custom * {@link ConsumerRebalanceListener} callbacks. diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 62cae9d3da98e..8e37b793d795e 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -725,17 +725,13 @@ private Fetch fetchRecords(CompletedFetch completedFetch, int maxRecords) completedFetch.nextFetchOffset, completedFetch.lastEpoch, position.currentLeader); - log.trace("Update fetching position to {} for partition {}", nextPosition, completedFetch.partition); + log.trace("Updating fetch position from {} to {} for partition {} and returning {} records from `poll()`", + position, nextPosition, completedFetch.partition, partRecords.size()); subscriptions.position(completedFetch.partition, nextPosition); positionAdvanced = true; if (partRecords.isEmpty()) { - log.debug( - "Advanced position for partition {} without receiving any user-visible records. " - + "This is likely due to skipping over control records in the current fetch, " - + "and may result in the consumer returning an empty record batch when " - + "polled before its poll timeout has elapsed.", - completedFetch.partition - ); + log.trace("Returning empty records from `poll()` " + + "since the consumer's position has advanced for at least one topic partition"); } } From e3fcdafe1999396bceb6a0beedff68b833ea063b Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Tue, 7 Dec 2021 11:24:35 -0500 Subject: [PATCH 4/5] Update clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java Co-authored-by: Jason Gustafson --- .../java/org/apache/kafka/clients/consumer/KafkaConsumer.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java index 62573f2b618d3..b77fd7c7de970 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java @@ -1177,7 +1177,7 @@ public ConsumerRecords poll(final long timeoutMs) { * *

* This method returns immediately if there are records available or if the position advances past control records - * or aborted transactions when isolation.level=READ_COMMITTED. + * or aborted transactions when isolation.level=read_committed. * Otherwise, it will await the passed timeout. If the timeout expires, an empty record set will be returned. * Note that this method may block beyond the timeout in order to execute custom * {@link ConsumerRebalanceListener} callbacks. From cb301155f3b63fb820f9962368c4efcdc1f13bcb Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Tue, 7 Dec 2021 16:38:00 -0500 Subject: [PATCH 5/5] KAFKA-12980: Move log message from Fetcher to KafkaConsumer --- .../org/apache/kafka/clients/consumer/KafkaConsumer.java | 5 +++++ .../org/apache/kafka/clients/consumer/internals/Fetcher.java | 4 ---- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java index b77fd7c7de970..4f481a5bfade0 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java @@ -1250,6 +1250,11 @@ private ConsumerRecords poll(final Timer timer, final boolean includeMetad client.transmitSends(); } + if (fetch.records().isEmpty()) { + log.trace("Returning empty records from `poll()` " + + "since the consumer's position has advanced for at least one topic partition"); + } + return this.interceptors.onConsume(new ConsumerRecords<>(fetch.records())); } } while (timer.notExpired()); diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 8e37b793d795e..54f70cafd72c8 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -729,10 +729,6 @@ private Fetch fetchRecords(CompletedFetch completedFetch, int maxRecords) position, nextPosition, completedFetch.partition, partRecords.size()); subscriptions.position(completedFetch.partition, nextPosition); positionAdvanced = true; - if (partRecords.isEmpty()) { - log.trace("Returning empty records from `poll()` " - + "since the consumer's position has advanced for at least one topic partition"); - } } Long partitionLag = subscriptions.partitionLag(completedFetch.partition, isolationLevel);