diff --git a/build.gradle b/build.gradle index 9c23d6f9d6ee4..d47e9233e39f5 100644 --- a/build.gradle +++ b/build.gradle @@ -1118,6 +1118,7 @@ project(':clients') { testCompile libs.bcpkix testCompile libs.junitJupiter testCompile libs.mockitoCore + testCompile libs.hamcrest testRuntime libs.slf4jlog4j testRuntime libs.jacksonDatabind diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml index ef02e05c516e6..a71c593f811b2 100644 --- a/checkstyle/suppressions.xml +++ b/checkstyle/suppressions.xml @@ -100,7 +100,10 @@ files="RequestResponseTest.java|FetcherTest.java|KafkaAdminClientTest.java"/> + files="MemoryRecordsTest|MetricsTest|RequestResponseTest|TestSslUtils|AclAuthorizerBenchmark|MockConsumer"/> + + diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecords.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecords.java index 4d0f62c3a1d15..d34bf55711bf0 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecords.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecords.java @@ -16,11 +16,13 @@ */ package org.apache.kafka.clients.consumer; +import org.apache.kafka.clients.consumer.internals.FetchedRecords; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.utils.AbstractIterator; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; @@ -32,14 +34,99 @@ * partition returned by a {@link Consumer#poll(java.time.Duration)} operation. */ public class ConsumerRecords implements Iterable> { - - @SuppressWarnings("unchecked") - public static final ConsumerRecords EMPTY = new ConsumerRecords<>(Collections.EMPTY_MAP); + public static final ConsumerRecords EMPTY = new ConsumerRecords<>( + Collections.emptyMap(), + Collections.emptyMap() + ); private final Map>> records; + private final Map metadata; + + public static final class Metadata { + + private final long receivedTimestamp; + private final Long position; + private final Long endOffset; + + public Metadata(final long receivedTimestamp, + final Long position, + final Long endOffset) { + this.receivedTimestamp = receivedTimestamp; + this.position = position; + this.endOffset = endOffset; + } + + /** + * @return The timestamp of the broker response that contained this metadata + */ + public long receivedTimestamp() { + return receivedTimestamp; + } + + /** + * @return The next position the consumer will fetch, or null if the consumer has no position. + */ + public Long position() { + return position; + } + + /** + * @return The lag between the next position to fetch and the current end of the partition, or + * null if the end offset is not known or there is no position. + */ + public Long lag() { + return endOffset == null || position == null ? null : endOffset - position; + } + + /** + * @return The current last offset in the partition. The determination of the "last" offset + * depends on the Consumer's isolation level. Under "read_uncommitted," this is the last successfully + * replicated offset plus one. Under "read_committed," this is the minimum of the last successfully + * replicated offset plus one or the smallest offset of any open transaction. Null if the end offset + * is not known. + */ + public Long endOffset() { + return endOffset; + } + + @Override + public String toString() { + return "Metadata{" + + "receivedTimestamp=" + receivedTimestamp + + ", position=" + position + + ", endOffset=" + endOffset + + '}'; + } + } + + private static Map extractMetadata(final FetchedRecords fetchedRecords) { + final Map metadata = new HashMap<>(); + for (final Map.Entry entry : fetchedRecords.metadata().entrySet()) { + metadata.put( + entry.getKey(), + new Metadata( + entry.getValue().receivedTimestamp(), + entry.getValue().position() == null ? null : entry.getValue().position().offset, + entry.getValue().endOffset() + ) + ); + } + return metadata; + } - public ConsumerRecords(Map>> records) { + public ConsumerRecords(final Map>> records) { this.records = records; + this.metadata = new HashMap<>(); + } + + public ConsumerRecords(final Map>> records, + final Map metadata) { + this.records = records; + this.metadata = metadata; + } + + public ConsumerRecords(final FetchedRecords fetchedRecords) { + this(fetchedRecords.records(), extractMetadata(fetchedRecords)); } /** @@ -55,6 +142,16 @@ public List> records(TopicPartition partition) { return Collections.unmodifiableList(recs); } + /** + * Get the updated metadata returned by the brokers along with this record set. + * May be empty or partial depending on the responses from the broker during this particular poll. + * May also include metadata for additional partitions than the ones for which there are records + * in this {@code ConsumerRecords} object. + */ + public Map metadata() { + return Collections.unmodifiableMap(metadata); + } + /** * Get just the records for the given topic */ 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 b6bebc1717ae0..e60eebe239c47 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 @@ -27,6 +27,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.FetchedRecords; import org.apache.kafka.clients.consumer.internals.Fetcher; import org.apache.kafka.clients.consumer.internals.FetcherMetricsRegistry; import org.apache.kafka.clients.consumer.internals.KafkaConsumerMetrics; @@ -1234,7 +1235,7 @@ private ConsumerRecords poll(final Timer timer, final boolean includeMetad } } - final Map>> records = pollForFetches(timer); + final FetchedRecords records = pollForFetches(timer); if (!records.isEmpty()) { // 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 @@ -1268,12 +1269,12 @@ boolean updateAssignmentMetadataIfNeeded(final Timer timer, final boolean waitFo /** * @throws KafkaException if the rebalance callback throws exception */ - private Map>> pollForFetches(Timer timer) { + private FetchedRecords 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(); + final FetchedRecords records = fetcher.fetchedRecords(); if (!records.isEmpty()) { return records; } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java index 7bf4c3f16dc94..7ddda76e416b0 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java @@ -218,7 +218,21 @@ public synchronized ConsumerRecords poll(final Duration timeout) { } toClear.forEach(p -> this.records.remove(p)); - return new ConsumerRecords<>(results); + + final Map metadata = new HashMap<>(); + for (final TopicPartition partition : subscriptions.assignedPartitions()) { + if (subscriptions.hasValidPosition(partition) && endOffsets.containsKey(partition)) { + final SubscriptionState.FetchPosition position = subscriptions.position(partition); + final long offset = position.offset; + final long endOffset = endOffsets.get(partition); + metadata.put( + partition, + new ConsumerRecords.Metadata(System.currentTimeMillis(), offset, endOffset) + ); + } + } + + return new ConsumerRecords<>(results, metadata); } public synchronized void addRecord(ConsumerRecord record) { @@ -229,6 +243,7 @@ public synchronized void addRecord(ConsumerRecord record) { throw new IllegalStateException("Cannot add records for a partition that is not assigned to the consumer"); List> recs = this.records.computeIfAbsent(tp, k -> new ArrayList<>()); recs.add(record); + endOffsets.compute(tp, (ignore, offset) -> offset == null ? record.offset() : Math.max(offset, record.offset())); } /** diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchedRecords.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchedRecords.java new file mode 100644 index 0000000000000..d8ef92bd6e48a --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchedRecords.java @@ -0,0 +1,102 @@ +/* + * 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.HashMap; +import java.util.List; +import java.util.Map; + +public class FetchedRecords { + private final Map>> records; + private final Map metadata; + + public static final class FetchMetadata { + + private final long receivedTimestamp; + private final SubscriptionState.FetchPosition position; + private final Long endOffset; + + public FetchMetadata(final long receivedTimestamp, + final SubscriptionState.FetchPosition position, + final Long endOffset) { + this.receivedTimestamp = receivedTimestamp; + this.position = position; + this.endOffset = endOffset; + } + + public long receivedTimestamp() { + return receivedTimestamp; + } + + public SubscriptionState.FetchPosition position() { + return position; + } + + public Long endOffset() { + return endOffset; + } + + @Override + public String toString() { + return "FetchMetadata{" + + "receivedTimestamp=" + receivedTimestamp + + ", position=" + position + + ", endOffset=" + endOffset + + '}'; + } + } + + public FetchedRecords() { + records = new HashMap<>(); + metadata = new HashMap<>(); + } + + public void addRecords(final TopicPartition topicPartition, final List> records) { + if (this.records.containsKey(topicPartition)) { + // 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 + final List> currentRecords = this.records.get(topicPartition); + final List> newRecords = new ArrayList<>(records.size() + currentRecords.size()); + newRecords.addAll(currentRecords); + newRecords.addAll(records); + this.records.put(topicPartition, newRecords); + } else { + this.records.put(topicPartition, records); + } + } + + public Map>> records() { + return records; + } + + public void addMetadata(final TopicPartition partition, final FetchMetadata fetchMetadata) { + metadata.put(partition, fetchMetadata); + } + + public Map metadata() { + return metadata; + } + + public boolean isEmpty() { + return records.isEmpty() && metadata.isEmpty(); + } +} 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 6488294d4c541..01efa762fc9bd 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 @@ -262,9 +262,8 @@ public synchronized int sendFetches() { .toForget(data.toForget()) .rackId(clientRackId); - if (log.isDebugEnabled()) { - log.debug("Sending {} {} to broker {}", isolationLevel, data.toString(), fetchTarget); - } + log.debug("Sending {} {} to broker {}", isolationLevel, data, fetchTarget); + RequestFuture future = client.send(fetchTarget, request); // We add the node to the set of nodes with pending fetch requests before adding the // listener because the future may have been fulfilled on another thread (e.g. during a @@ -319,7 +318,7 @@ public void onSuccess(ClientResponse resp) { short responseVersion = resp.requestHeader().apiVersion(); completedFetches.add(new CompletedFetch(partition, partitionData, - metricAggregator, batches, fetchOffset, responseVersion)); + metricAggregator, batches, fetchOffset, responseVersion, resp.receivedTimeMs())); } } @@ -598,8 +597,8 @@ private Map beginningOrEndOffset(Collection>> fetchedRecords() { - Map>> fetched = new HashMap<>(); + public FetchedRecords fetchedRecords() { + FetchedRecords fetched = new FetchedRecords<>(); Queue pausedCompletedFetches = new ArrayDeque<>(); int recordsRemaining = maxPollRecords; @@ -637,20 +636,28 @@ public Map>> fetchedRecords() { } else { List> records = fetchRecords(nextInLineFetch, recordsRemaining); + TopicPartition partition = nextInLineFetch.partition; + + // This can be false when a rebalance happened before fetched records + // are returned to the consumer's poll call + if (subscriptions.isAssigned(partition)) { + + // initializeCompletedFetch, above, has already persisted the metadata from the fetch in the + // SubscriptionState, so we can just read it out, which in particular lets us re-use the logic + // for determining the end offset + final long receivedTimestamp = nextInLineFetch.receivedTimestamp; + final Long endOffset = subscriptions.logEndOffset(partition, isolationLevel); + final FetchPosition fetchPosition = subscriptions.position(partition); + + final FetchedRecords.FetchMetadata metadata = + new FetchedRecords.FetchMetadata(receivedTimestamp, fetchPosition, endOffset); + + fetched.addMetadata(partition, metadata); + + } + 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); - } + fetched.addRecords(partition, records); recordsRemaining -= records.size(); } } @@ -1459,6 +1466,7 @@ private class CompletedFetch { private final FetchResponse.PartitionData partitionData; private final FetchResponseMetricAggregator metricAggregator; private final short responseVersion; + private final long receivedTimestamp; private int recordsRead; private int bytesRead; @@ -1477,13 +1485,15 @@ private CompletedFetch(TopicPartition partition, FetchResponseMetricAggregator metricAggregator, Iterator batches, Long fetchOffset, - short responseVersion) { + short responseVersion, + final long receivedTimestamp) { this.partition = partition; this.partitionData = partitionData; this.metricAggregator = metricAggregator; this.batches = batches; this.nextFetchOffset = fetchOffset; this.responseVersion = responseVersion; + this.receivedTimestamp = receivedTimestamp; this.lastEpoch = Optional.empty(); this.abortedProducerIds = new HashSet<>(); this.abortedTransactions = abortedTransactions(partitionData); diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java index 30491110a3d78..b2a5e51aaae8a 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java @@ -562,6 +562,18 @@ synchronized void updateLastStableOffset(TopicPartition tp, long lastStableOffse assignedState(tp).lastStableOffset(lastStableOffset); } + synchronized Long logStartOffset(TopicPartition tp) { + return assignedState(tp).logStartOffset; + } + + synchronized Long logEndOffset(TopicPartition tp, IsolationLevel isolationLevel) { + TopicPartitionState topicPartitionState = assignedState(tp); + if (isolationLevel == IsolationLevel.READ_COMMITTED) + return topicPartitionState.lastStableOffset == null ? null : topicPartitionState.lastStableOffset; + else + return topicPartitionState.highWatermark == null ? null : topicPartitionState.highWatermark; + } + /** * Set the preferred read replica with a lease timeout. After this time, the replica will no longer be valid and * {@link #preferredReadReplica(TopicPartition, long)} will return an empty result. diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java index 19999534f79d5..e92e684af0c1c 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java @@ -45,18 +45,19 @@ import org.apache.kafka.common.errors.InvalidTopicException; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.errors.WakeupException; +import org.apache.kafka.common.internals.ClusterResourceListeners; import org.apache.kafka.common.message.HeartbeatResponseData; import org.apache.kafka.common.message.JoinGroupRequestData; import org.apache.kafka.common.message.JoinGroupResponseData; import org.apache.kafka.common.message.LeaveGroupResponseData; +import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition; import org.apache.kafka.common.message.ListOffsetsResponseData; -import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse; import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse; -import org.apache.kafka.common.internals.ClusterResourceListeners; +import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse; import org.apache.kafka.common.message.SyncGroupResponseData; -import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.metrics.Sensor; +import org.apache.kafka.common.metrics.stats.Avg; import org.apache.kafka.common.network.Selectable; import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.protocol.Errors; @@ -91,8 +92,6 @@ import org.apache.kafka.test.MockConsumerInterceptor; import org.apache.kafka.test.MockMetricsReporter; import org.apache.kafka.test.TestUtils; -import org.apache.kafka.common.metrics.stats.Avg; - import org.junit.jupiter.api.Test; import javax.management.MBeanServer; @@ -101,7 +100,6 @@ import java.nio.ByteBuffer; import java.time.Duration; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.ConcurrentModificationException; @@ -128,10 +126,18 @@ import java.util.stream.Collectors; import java.util.stream.Stream; +import static java.util.Arrays.asList; +import static java.util.Collections.emptyMap; import static java.util.Collections.singleton; import static java.util.Collections.singletonList; import static java.util.Collections.singletonMap; import static org.apache.kafka.common.requests.FetchMetadata.INVALID_SESSION_ID; +import static org.apache.kafka.common.utils.Utils.mkEntry; +import static org.apache.kafka.common.utils.Utils.mkMap; +import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.greaterThanOrEqualTo; +import static org.hamcrest.Matchers.lessThanOrEqualTo; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; @@ -601,7 +607,7 @@ public void testFetchProgressWithMissingPartitionPosition() { initMetadata(client, Collections.singletonMap(topic, 2)); KafkaConsumer consumer = newConsumerNoAutoCommit(time, client, subscription, metadata); - consumer.assign(Arrays.asList(tp0, tp1)); + consumer.assign(asList(tp0, tp1)); consumer.seekToEnd(singleton(tp0)); consumer.seekToBeginning(singleton(tp1)); @@ -761,7 +767,7 @@ public void testCommitsFetchedDuringAssign() { client.prepareResponseFrom(offsetResponse(Collections.singletonMap(tp0, offset1), Errors.NONE), coordinator); assertEquals(offset1, consumer.committed(Collections.singleton(tp0)).get(tp0).offset()); - consumer.assign(Arrays.asList(tp0, tp1)); + consumer.assign(asList(tp0, tp1)); // fetch offset for two topics Map offsets = new HashMap<>(); @@ -833,14 +839,14 @@ public void testNoCommittedOffsets() { ConsumerPartitionAssignor assignor = new RoundRobinAssignor(); KafkaConsumer consumer = newConsumer(time, client, subscription, metadata, assignor, true, groupInstanceId); - consumer.assign(Arrays.asList(tp0, tp1)); + consumer.assign(asList(tp0, tp1)); // lookup coordinator client.prepareResponseFrom(FindCoordinatorResponse.prepareResponse(Errors.NONE, node), node); Node coordinator = new Node(Integer.MAX_VALUE - node.id(), node.host(), node.port()); // fetch offset for one topic - client.prepareResponseFrom(offsetResponse(Utils.mkMap(Utils.mkEntry(tp0, offset1), Utils.mkEntry(tp1, -1L)), Errors.NONE), coordinator); + client.prepareResponseFrom(offsetResponse(mkMap(mkEntry(tp0, offset1), mkEntry(tp1, -1L)), Errors.NONE), coordinator); final Map committed = consumer.committed(Utils.mkSet(tp0, tp1)); assertEquals(2, committed.size()); assertEquals(offset1, committed.get(tp0).offset()); @@ -1050,6 +1056,7 @@ public void fetchResponseWithUnexpectedPartitionIsIgnored() { ConsumerRecords records = consumer.poll(Duration.ZERO); assertEquals(0, records.count()); + assertThat(records.metadata(), equalTo(emptyMap())); consumer.close(Duration.ofMillis(0)); } @@ -1079,7 +1086,7 @@ public void testSubscriptionChangesWithAutoCommitEnabled() { KafkaConsumer consumer = newConsumer(time, client, subscription, metadata, assignor, true, groupInstanceId); // initial subscription - consumer.subscribe(Arrays.asList(topic, topic2), getConsumerRebalanceListener(consumer)); + consumer.subscribe(asList(topic, topic2), getConsumerRebalanceListener(consumer)); // verify that subscription has changed but assignment is still unchanged assertEquals(2, consumer.subscription().size()); @@ -1087,7 +1094,7 @@ public void testSubscriptionChangesWithAutoCommitEnabled() { assertTrue(consumer.assignment().isEmpty()); // mock rebalance responses - Node coordinator = prepareRebalance(client, node, assignor, Arrays.asList(tp0, t2p0), null); + Node coordinator = prepareRebalance(client, node, assignor, asList(tp0, t2p0), null); consumer.updateAssignmentMetadataIfNeeded(time.timer(Long.MAX_VALUE)); consumer.poll(Duration.ZERO); @@ -1116,10 +1123,12 @@ public void testSubscriptionChangesWithAutoCommitEnabled() { // verify that the fetch occurred as expected assertEquals(11, records.count()); assertEquals(1L, consumer.position(tp0)); + assertEquals(1L, (long) records.metadata().get(tp0).position()); assertEquals(10L, consumer.position(t2p0)); + assertEquals(10L, (long) records.metadata().get(t2p0).position()); // subscription change - consumer.subscribe(Arrays.asList(topic, topic3), getConsumerRebalanceListener(consumer)); + consumer.subscribe(asList(topic, topic3), getConsumerRebalanceListener(consumer)); // verify that subscription has changed but assignment is still unchanged assertEquals(2, consumer.subscription().size()); @@ -1134,7 +1143,7 @@ public void testSubscriptionChangesWithAutoCommitEnabled() { AtomicBoolean commitReceived = prepareOffsetCommitResponse(client, coordinator, partitionOffsets1); // mock rebalance responses - prepareRebalance(client, node, assignor, Arrays.asList(tp0, t3p0), coordinator); + prepareRebalance(client, node, assignor, asList(tp0, t3p0), coordinator); // mock a response to the next fetch from the new assignment Map fetches2 = new HashMap<>(); @@ -1147,7 +1156,9 @@ public void testSubscriptionChangesWithAutoCommitEnabled() { // verify that the fetch occurred as expected assertEquals(101, records.count()); assertEquals(2L, consumer.position(tp0)); + assertEquals(2L, (long) records.metadata().get(tp0).position()); assertEquals(100L, consumer.position(t3p0)); + assertEquals(100L, (long) records.metadata().get(t3p0).position()); // verify that the offset commits occurred as expected assertTrue(commitReceived.get()); @@ -1490,7 +1501,7 @@ public void testGracefulClose() throws Exception { response.put(tp0, Errors.NONE); OffsetCommitResponse commitResponse = offsetCommitResponse(response); LeaveGroupResponse leaveGroupResponse = new LeaveGroupResponse(new LeaveGroupResponseData().setErrorCode(Errors.NONE.code())); - consumerCloseTest(5000, Arrays.asList(commitResponse, leaveGroupResponse), 0, false); + consumerCloseTest(5000, asList(commitResponse, leaveGroupResponse), 0, false); } @Test @@ -1861,12 +1872,12 @@ public void testReturnRecordsDuringRebalance() throws InterruptedException { ConsumerPartitionAssignor assignor = new CooperativeStickyAssignor(); KafkaConsumer consumer = newConsumer(time, client, subscription, metadata, assignor, true, groupInstanceId); - initMetadata(client, Utils.mkMap(Utils.mkEntry(topic, 1), Utils.mkEntry(topic2, 1), Utils.mkEntry(topic3, 1))); + initMetadata(client, mkMap(mkEntry(topic, 1), mkEntry(topic2, 1), mkEntry(topic3, 1))); - consumer.subscribe(Arrays.asList(topic, topic2), getConsumerRebalanceListener(consumer)); + consumer.subscribe(asList(topic, topic2), getConsumerRebalanceListener(consumer)); Node node = metadata.fetch().nodes().get(0); - Node coordinator = prepareRebalance(client, node, assignor, Arrays.asList(tp0, t2p0), null); + Node coordinator = prepareRebalance(client, node, assignor, asList(tp0, t2p0), null); // a poll with non-zero milliseconds would complete three round-trips (discover, join, sync) TestUtils.waitForCondition(() -> { @@ -1897,7 +1908,7 @@ public void testReturnRecordsDuringRebalance() throws InterruptedException { client.respondFrom(fetchResponse(fetches1), node); // subscription change - consumer.subscribe(Arrays.asList(topic, topic3), getConsumerRebalanceListener(consumer)); + consumer.subscribe(asList(topic, topic3), getConsumerRebalanceListener(consumer)); // verify that subscription has changed but assignment is still unchanged assertEquals(Utils.mkSet(topic, topic3), consumer.subscription()); @@ -1943,7 +1954,7 @@ public void testReturnRecordsDuringRebalance() throws InterruptedException { client.respondFrom(fetchResponse(fetches1), node); // now complete the rebalance - client.respondFrom(syncGroupResponse(Arrays.asList(tp0, t3p0), Errors.NONE), coordinator); + client.respondFrom(syncGroupResponse(asList(tp0, t3p0), Errors.NONE), coordinator); AtomicInteger count = new AtomicInteger(0); TestUtils.waitForCondition(() -> { @@ -2041,6 +2052,118 @@ public void testInvalidGroupMetadata() throws InterruptedException { assertThrows(IllegalStateException.class, consumer::groupMetadata); } + @Test + public void testPollMetadata() { + final Time time = new MockTime(); + final SubscriptionState subscription = new SubscriptionState(new LogContext(), OffsetResetStrategy.EARLIEST); + final ConsumerMetadata metadata = createMetadata(subscription); + final MockClient client = new MockClient(time, metadata); + + initMetadata(client, singletonMap(topic, 1)); + final ConsumerPartitionAssignor assignor = new RoundRobinAssignor(); + + final KafkaConsumer consumer = + newConsumer(time, client, subscription, metadata, assignor, true, groupInstanceId); + + consumer.assign(singleton(tp0)); + consumer.seek(tp0, 50L); + + final FetchInfo fetchInfo = new FetchInfo(1L, 99L, 50L, 5); + client.prepareResponse(fetchResponse(singletonMap(tp0, fetchInfo))); + + final ConsumerRecords records = consumer.poll(Duration.ofMillis(1)); + assertEquals(5, records.count()); + assertEquals(55L, consumer.position(tp0)); + + // verify that the consumer computes the correct metadata based on the fetch response + final ConsumerRecords.Metadata actualMetadata = records.metadata().get(tp0); + assertEquals(100L, (long) actualMetadata.endOffset()); + assertEquals(55L, (long) actualMetadata.position()); + assertEquals(45L, (long) actualMetadata.lag()); + consumer.close(Duration.ZERO); + } + + + @Test + public void testPollMetadataWithExtraPartitions() { + final Time time = new MockTime(); + final SubscriptionState subscription = new SubscriptionState(new LogContext(), OffsetResetStrategy.EARLIEST); + final ConsumerMetadata metadata = createMetadata(subscription); + final MockClient client = new MockClient(time, metadata); + + initMetadata(client, singletonMap(topic, 2)); + final ConsumerPartitionAssignor assignor = new RoundRobinAssignor(); + + final KafkaConsumer consumer = + newConsumer(time, client, subscription, metadata, assignor, true, groupInstanceId); + + consumer.assign(asList(tp0, tp1)); + consumer.seek(tp0, 50L); + consumer.seek(tp1, 10L); + + client.prepareResponse( + fetchResponse( + mkMap( + mkEntry(tp0, new FetchInfo(1L, 99L, 50L, 5)), + mkEntry(tp1, new FetchInfo(0L, 29L, 10L, 0)) + ) + ) + ); + + final ConsumerRecords records = consumer.poll(Duration.ofMillis(1)); + assertEquals(5, records.count()); + assertEquals(55L, consumer.position(tp0)); + + assertEquals(5, records.records(tp0).size()); + final ConsumerRecords.Metadata tp0Metadata = records.metadata().get(tp0); + assertEquals(100L, (long) tp0Metadata.endOffset()); + assertEquals(55L, (long) tp0Metadata.position()); + assertEquals(45L, (long) tp0Metadata.lag()); + + // we may get back metadata for other assigned partitions even if we don't get records for them + assertEquals(0, records.records(tp1).size()); + final ConsumerRecords.Metadata tp1Metadata = records.metadata().get(tp1); + assertEquals(30L, (long) tp1Metadata.endOffset()); + assertEquals(10L, (long) tp1Metadata.position()); + assertEquals(20L, (long) tp1Metadata.lag()); + + consumer.close(Duration.ZERO); + } + + @Test + public void testPollMetadataWithNoRecords() { + final Time time = new MockTime(); + final SubscriptionState subscription = new SubscriptionState(new LogContext(), OffsetResetStrategy.EARLIEST); + final ConsumerMetadata metadata = createMetadata(subscription); + final MockClient client = new MockClient(time, metadata); + + initMetadata(client, singletonMap(topic, 1)); + final ConsumerPartitionAssignor assignor = new RoundRobinAssignor(); + + final KafkaConsumer consumer = + newConsumer(time, client, subscription, metadata, assignor, true, groupInstanceId); + + consumer.assign(singleton(tp0)); + consumer.seek(tp0, 50L); + + final FetchInfo fetchInfo = new FetchInfo(1L, 99L, 50L, 0); + client.prepareResponse(fetchResponse(singletonMap(tp0, fetchInfo))); + + final ConsumerRecords records = consumer.poll(Duration.ofMillis(1)); + + // we got no records back ... + assertEquals(0, records.count()); + assertEquals(50L, consumer.position(tp0)); + + // ... but we can still get metadata that was in the fetch response + final ConsumerRecords.Metadata actualMetadata = records.metadata().get(tp0); + assertEquals(100L, (long) actualMetadata.endOffset()); + assertEquals(50L, (long) actualMetadata.position()); + assertEquals(50L, (long) actualMetadata.lag()); + + consumer.close(Duration.ZERO); + } + private KafkaConsumer consumerWithPendingAuthenticationError() { Time time = new MockTime(); SubscriptionState subscription = new SubscriptionState(new LogContext(), OffsetResetStrategy.EARLIEST); @@ -2206,7 +2329,7 @@ private OffsetFetchResponse offsetResponse(Map offsets, Er } private ListOffsetsResponse listOffsetsResponse(Map offsets) { - return listOffsetsResponse(offsets, Collections.emptyMap()); + return listOffsetsResponse(offsets, emptyMap()); } private ListOffsetsResponse listOffsetsResponse(Map partitionOffsets, @@ -2242,6 +2365,8 @@ private FetchResponse fetchResponse(Map fetchResponse(Map( - Errors.NONE, 0, FetchResponse.INVALID_LAST_STABLE_OFFSET, - 0L, null, records)); + Errors.NONE, highWatermark, FetchResponse.INVALID_LAST_STABLE_OFFSET, + logStartOffset, null, records)); } return new FetchResponse<>(Errors.NONE, tpResponses, 0, INVALID_SESSION_ID); } @@ -2379,10 +2504,20 @@ private KafkaConsumer newConsumer(Time time, } private static class FetchInfo { + long logFirstOffset; + long logLastOffset; long offset; int count; FetchInfo(long offset, int count) { + this(0L, offset + count, offset, count); + } + + FetchInfo(long logFirstOffset, long logLastOffset, long offset, int count) { + assertThat(logFirstOffset, lessThanOrEqualTo(offset)); + assertThat(logLastOffset, greaterThanOrEqualTo(offset + count)); + this.logFirstOffset = logFirstOffset; + this.logLastOffset = logLastOffset; this.offset = offset; this.count = count; } @@ -2521,7 +2656,7 @@ public void testPollIdleRatio() { } private static boolean consumerMetricPresent(KafkaConsumer consumer, String name) { - MetricName metricName = new MetricName(name, "consumer-metrics", "", Collections.emptyMap()); + MetricName metricName = new MetricName(name, "consumer-metrics", "", emptyMap()); return consumer.metrics.metrics().containsKey(metricName); } @@ -2561,11 +2696,11 @@ public void testEnforceRebalanceTriggersRebalanceOnNextPoll() { ConsumerPartitionAssignor assignor = new RoundRobinAssignor(); KafkaConsumer consumer = newConsumer(time, client, subscription, metadata, assignor, true, groupInstanceId); MockRebalanceListener countingRebalanceListener = new MockRebalanceListener(); - initMetadata(client, Utils.mkMap(Utils.mkEntry(topic, 1), Utils.mkEntry(topic2, 1), Utils.mkEntry(topic3, 1))); + initMetadata(client, mkMap(mkEntry(topic, 1), mkEntry(topic2, 1), mkEntry(topic3, 1))); - consumer.subscribe(Arrays.asList(topic, topic2), countingRebalanceListener); + consumer.subscribe(asList(topic, topic2), countingRebalanceListener); Node node = metadata.fetch().nodes().get(0); - prepareRebalance(client, node, assignor, Arrays.asList(tp0, t2p0), null); + prepareRebalance(client, node, assignor, asList(tp0, t2p0), null); // a first rebalance to get the assignment, we need two poll calls since we need two round trips to finish join / sync-group consumer.poll(Duration.ZERO); 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 8dab0f03dbf83..6cf6eb4a2ea96 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 @@ -524,7 +524,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, fetcher.fetchedRecords().records().get(tp0).size()); assertEquals(1, subscriptions.position(tp0).offset); ensureBlockOnRecord(1L); @@ -928,7 +928,7 @@ public void testInFlightFetchOnPausedPartition() { client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NONE, 100L, 0)); consumerClient.poll(time.timer(0)); - assertNull(fetcher.fetchedRecords().get(tp0)); + assertNull(fetcher.fetchedRecords().records().get(tp0)); } @Test @@ -1117,7 +1117,7 @@ public void testFetchNotLeaderOrFollower() { assertEquals(1, fetcher.sendFetches()); client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NOT_LEADER_OR_FOLLOWER, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEquals(0, fetcher.fetchedRecords().records().size()); assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds())); } @@ -1130,7 +1130,7 @@ public void testFetchUnknownTopicOrPartition() { assertEquals(1, fetcher.sendFetches()); client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.UNKNOWN_TOPIC_OR_PARTITION, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEquals(0, fetcher.fetchedRecords().records().size()); assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds())); } @@ -1144,7 +1144,7 @@ public void testFetchFencedLeaderEpoch() { client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.FENCED_LEADER_EPOCH, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size(), "Should not return any records"); + assertEquals(0, fetcher.fetchedRecords().records().size(), "Should not return any records"); assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds()), "Should have requested metadata update"); } @@ -1158,7 +1158,7 @@ public void testFetchUnknownLeaderEpoch() { client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.UNKNOWN_LEADER_EPOCH, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size(), "Should not return any records"); + assertEquals(0, fetcher.fetchedRecords().records().size(), "Should not return any records"); assertNotEquals(0L, metadata.timeToNextUpdate(time.milliseconds()), "Should not have requested metadata update"); } @@ -1200,7 +1200,7 @@ public void testFetchOffsetOutOfRange() { assertEquals(1, fetcher.sendFetches()); client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.OFFSET_OUT_OF_RANGE, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEquals(0, fetcher.fetchedRecords().records().size()); assertTrue(subscriptions.isOffsetResetNeeded(tp0)); assertNull(subscriptions.validPosition(tp0)); assertNull(subscriptions.position(tp0)); @@ -1218,7 +1218,7 @@ public void testStaleOutOfRangeError() { client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.OFFSET_OUT_OF_RANGE, 100L, 0)); subscriptions.seek(tp0, 1); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEquals(0, fetcher.fetchedRecords().records().size()); assertFalse(subscriptions.isOffsetResetNeeded(tp0)); assertEquals(1, subscriptions.position(tp0).offset); } @@ -1236,7 +1236,7 @@ public void testFetchedRecordsAfterSeek() { consumerClient.poll(time.timer(0)); assertFalse(subscriptions.isOffsetResetNeeded(tp0)); subscriptions.seek(tp0, 2); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEquals(0, fetcher.fetchedRecords().records().size()); } @Test @@ -1392,7 +1392,7 @@ public void testSeekBeforeException() { client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NONE, 100L, 0)); consumerClient.poll(time.timer(0)); - assertEquals(2, fetcher.fetchedRecords().get(tp0).size()); + assertEquals(2, fetcher.fetchedRecords().records().get(tp0).size()); subscriptions.assignFromUser(Utils.mkSet(tp0, tp1)); subscriptions.seekUnvalidated(tp1, new SubscriptionState.FetchPosition(1, Optional.empty(), metadata.currentLeader(tp1))); @@ -1403,11 +1403,11 @@ public void testSeekBeforeException() { FetchResponse.INVALID_LAST_STABLE_OFFSET, FetchResponse.INVALID_LOG_START_OFFSET, Optional.empty(), null, MemoryRecords.EMPTY)); client.prepareResponse(new FetchResponse<>(Errors.NONE, new LinkedHashMap<>(partitions), 0, INVALID_SESSION_ID)); consumerClient.poll(time.timer(0)); - assertEquals(1, fetcher.fetchedRecords().get(tp0).size()); + assertEquals(1, fetcher.fetchedRecords().records().get(tp0).size()); subscriptions.seek(tp1, 10); // Should not throw OffsetOutOfRangeException after the seek - assertEquals(0, fetcher.fetchedRecords().size()); + assertEquals(0, fetcher.fetchedRecords().records().size()); } @Test @@ -1420,7 +1420,7 @@ public void testFetchDisconnected() { assertEquals(1, fetcher.sendFetches()); client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NONE, 100L, 0), true); consumerClient.poll(time.timer(0)); - assertEquals(0, fetcher.fetchedRecords().size()); + assertEquals(0, fetcher.fetchedRecords().records().size()); // disconnects should have no affect on subscription state assertFalse(subscriptions.isOffsetResetNeeded(tp0)); @@ -4511,7 +4511,7 @@ private MetadataResponse newMetadataResponse(String topic, Errors error) { @SuppressWarnings("unchecked") private Map>> fetchedRecords() { - return (Map) fetcher.fetchedRecords(); + return (Map) fetcher.fetchedRecords().records(); } private void buildFetcher(int maxPollRecords) { diff --git a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala index 97f4407e1bb39..5b09b288d7a06 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala +++ b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala @@ -581,7 +581,7 @@ class PlaintextConsumerTest extends BaseConsumerTest { consumer.seekToEnd(List(tp).asJava) assertEquals(totalRecords, consumer.position(tp)) - assertTrue(consumer.poll(Duration.ofMillis(50)).isEmpty) + assertTrue(pollForRecord(consumer, Duration.ofMillis(50)).isEmpty) consumer.seekToBeginning(List(tp).asJava) assertEquals(0L, consumer.position(tp)) @@ -599,7 +599,7 @@ class PlaintextConsumerTest extends BaseConsumerTest { consumer.seekToEnd(List(tp2).asJava) assertEquals(totalRecords, consumer.position(tp2)) - assertTrue(consumer.poll(Duration.ofMillis(50)).isEmpty) + assertTrue(pollForRecord(consumer, Duration.ofMillis(50)).isEmpty) consumer.seekToBeginning(List(tp2).asJava) assertEquals(0L, consumer.position(tp2)) @@ -668,7 +668,7 @@ class PlaintextConsumerTest extends BaseConsumerTest { consumer.pause(partitions) startingTimestamp = System.currentTimeMillis() sendRecords(producer, numRecords = 5, tp, startingTimestamp = startingTimestamp) - assertTrue(consumer.poll(Duration.ofMillis(100)).isEmpty) + assertTrue(pollForRecord(consumer, Duration.ofMillis(100)).isEmpty) consumer.resume(partitions) consumeAndVerifyRecords(consumer = consumer, numRecords = 5, startingOffset = 5, startingTimestamp = startingTimestamp) } @@ -716,7 +716,8 @@ class PlaintextConsumerTest extends BaseConsumerTest { // consuming a record that is too large should succeed since KIP-74 consumer.assign(List(tp).asJava) - val records = consumer.poll(Duration.ofMillis(20000)) + val duration = Duration.ofMillis(20000) + val records = pollForRecord(consumer, duration) assertEquals(1, records.count) val consumerRecord = records.iterator().next() assertEquals(0L, consumerRecord.offset) @@ -748,7 +749,7 @@ class PlaintextConsumerTest extends BaseConsumerTest { // we should only get the small record in the first `poll` consumer.assign(List(tp).asJava) - val records = consumer.poll(Duration.ofMillis(20000)) + val records = pollForRecord(consumer, Duration.ofMillis(20000)) assertEquals(1, records.count) val consumerRecord = records.iterator().next() assertEquals(0L, consumerRecord.offset) @@ -1802,12 +1803,12 @@ class PlaintextConsumerTest extends BaseConsumerTest { consumer3.assign(asList(tp)) consumer3.seek(tp, 1) - val numRecords1 = consumer1.poll(Duration.ofMillis(5000)).count() + val numRecords1 = pollForRecord(consumer1, Duration.ofMillis(5000)).count() assertThrows(classOf[InvalidGroupIdException], () => consumer1.commitSync()) assertThrows(classOf[InvalidGroupIdException], () => consumer2.committed(Set(tp).asJava)) - val numRecords2 = consumer2.poll(Duration.ofMillis(5000)).count() - val numRecords3 = consumer3.poll(Duration.ofMillis(5000)).count() + val numRecords2 = pollForRecord(consumer2, Duration.ofMillis(5000)).count() + val numRecords3 = pollForRecord(consumer3, Duration.ofMillis(5000)).count() consumer1.unsubscribe() consumer2.unsubscribe() @@ -1856,10 +1857,10 @@ class PlaintextConsumerTest extends BaseConsumerTest { consumer1.assign(asList(tp)) consumer2.assign(asList(tp)) - val records1 = consumer1.poll(Duration.ofMillis(5000)) + val records1 = pollForRecord(consumer1, Duration.ofMillis(5000)) consumer1.commitSync() - val records2 = consumer2.poll(Duration.ofMillis(5000)) + val records2 = pollForRecord(consumer2, Duration.ofMillis(5000)) consumer2.commitSync() consumer1.close() @@ -1870,4 +1871,19 @@ class PlaintextConsumerTest extends BaseConsumerTest { assertTrue(records2.count() == 1 && records2.records(tp).asScala.head.offset == 1, "Expected consumer2 to consume one message from offset 1, which is the committed offset of consumer1") } + + /** + * Consumer#poll returns early if there is metadata to return even if there are no records, + * so when we intend to wait for records, we can't just rely on long polling in the Consumer. + */ + private def pollForRecord(consumer: KafkaConsumer[Array[Byte], Array[Byte]], duration: Duration) = { + val deadline = System.currentTimeMillis() + duration.toMillis + var durationRemaining = deadline - System.currentTimeMillis() + var result = consumer.poll(Duration.ofMillis(durationRemaining)) + while (result.count() == 0 && durationRemaining > 0) { + result = consumer.poll(Duration.ofMillis(durationRemaining)) + durationRemaining = deadline - System.currentTimeMillis() + } + result + } } diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java index 637a96e13b610..56022b286f7cc 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java @@ -1662,7 +1662,7 @@ public void shouldCheckpointForSuspendedTask() { task.postCommit(true); EasyMock.verify(stateManager); } - + @Test public void shouldNotCheckpointForSuspendedRunningTaskWithSmallProgress() { EasyMock.expect(stateManager.changelogOffsets())