Skip to content

WIP- Move parseCompletedFetch to sendFetches callback #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
seglo wants to merge 5 commits into from
Closed

WIP- Move parseCompletedFetch to sendFetches callback #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
95d1405
KafkaConsumer should not throw away already fetched data for paused p…
seglo Jun 12, 2019
f042406
Exclude cached parsed records from fetchablePartitions, drain cached …
seglo Jul 4, 2019
ed60cfb
More tests
seglo Jul 4, 2019
906d160
WIP- Move parseCompletedFetch to sendFetches callback
seglo Jul 26, 2019
a4e2532
Initialize PartitionRecords in fetch callback
seglo Jul 28, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
119 changes: 87 additions & 32 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,7 +140,7 @@ public class Fetcher<K, V> implements Closeable {
private final ConsumerMetadata metadata;
private final FetchManagerMetrics sensors;
private final SubscriptionState subscriptions;
private final ConcurrentLinkedQueue<CompletedFetch> completedFetches;
private final ConcurrentLinkedQueue<PartitionRecords> completedFetches;
private final BufferSupplier decompressionBufferSupplier = BufferSupplier.create();
private final Deserializer<K> keyDeserializer;
private final Deserializer<V> valueDeserializer;
Expand Down Expand Up @@ -291,8 +291,11 @@ public void onSuccess(ClientResponse resp) {

log.debug("Fetch {} at offset {} for partition {} returned fetch data {}",
isolationLevel, fetchOffset, partition, fetchData);
completedFetches.add(new CompletedFetch(partition, fetchOffset, fetchData, metricAggregator,
resp.requestHeader().apiVersion()));

CompletedFetch completedFetch = new CompletedFetch(partition, fetchOffset,
fetchData, metricAggregator, resp.requestHeader().apiVersion());

completedFetches.add(parseCompletedFetch(completedFetch));
}
}

Expand Down Expand Up @@ -582,28 +585,34 @@ public Map<TopicPartition, List<ConsumerRecord<K, V>>> fetchedRecords() {
try {
while (recordsRemaining > 0) {
if (nextInLineRecords == null || nextInLineRecords.isFetched) {
CompletedFetch completedFetch = completedFetches.peek();
if (completedFetch == null) break;

try {
nextInLineRecords = parseCompletedFetch(completedFetch);
} catch (Exception e) {
// Remove a completedFetch upon a parse with exception if (1) it contains no records, and
// (2) there are no fetched records with actual content preceding this exception.
// The first condition ensures that the completedFetches is not stuck with the same completedFetch
// in cases such as the TopicAuthorizationException, and the second condition ensures that no
// potential data loss due to an exception in a following record.
FetchResponse.PartitionData partition = completedFetch.partitionData;
if (fetched.isEmpty() && (partition.records == null || partition.records.sizeInBytes() == 0)) {
completedFetches.poll();
PartitionRecords records = maybeGetPartitionRecords();

if (records == null) break;

if (records.notInitialized()) {
try {
nextInLineRecords = initializePartitionRecords(records);
} catch (Exception e) {
// Remove a completedFetch upon a parse with exception if (1) it contains no records, and
// (2) there are no fetched records with actual content preceding this exception.
// The first condition ensures that the completedFetches is not stuck with the same completedFetch
// in cases such as the TopicAuthorizationException, and the second condition ensures that no
// potential data loss due to an exception in a following record.
FetchResponse.PartitionData partition = records.completedFetch.partitionData;
if (fetched.isEmpty() && (partition.records == null || partition.records.sizeInBytes() == 0)) {
completedFetches.remove(records);
}
throw e;
}
throw e;
} else {
nextInLineRecords = records;
}
completedFetches.poll();
completedFetches.remove(records);
} else {
List<ConsumerRecord<K, V>> records = fetchRecords(nextInLineRecords, recordsRemaining);
TopicPartition partition = nextInLineRecords.partition;
if (!records.isEmpty()) {

if (!records.isEmpty() && nextInLineRecords != null) {
TopicPartition partition = nextInLineRecords.partition;
List<ConsumerRecord<K, V>> currentRecords = fetched.get(partition);
if (currentRecords == null) {
fetched.put(partition, records);
Expand All @@ -624,9 +633,23 @@ public Map<TopicPartition, List<ConsumerRecord<K, V>>> fetchedRecords() {
if (fetched.isEmpty())
throw e;
}

return fetched;
}

private PartitionRecords maybeGetPartitionRecords() {
PartitionRecords partRecords = null;
for (PartitionRecords records : completedFetches) {
if (subscriptions.isPaused(records.partition)) {
log.debug("Skipping the completed fetch for partition {} because its partition is paused", records.partition);
} else {
partRecords = records;
break;
}
}
return partRecords;
}

private List<ConsumerRecord<K, V>> fetchRecords(PartitionRecords partitionRecords, int maxRecords) {
if (!subscriptions.isAssigned(partitionRecords.partition)) {
// this can happen when a rebalance happened before fetched records are returned to the consumer's poll call
Expand All @@ -637,6 +660,15 @@ private List<ConsumerRecord<K, V>> fetchRecords(PartitionRecords partitionRecord
// poll call or if the offset is being reset
log.debug("Not returning fetched records for assigned partition {} since it is no longer fetchable",
partitionRecords.partition);

// when the partition is paused we add them back to completedFetches instead of draining them so that they
// can be returned on a subsequent poll if the partition is resumed
if (subscriptions.isPaused(partitionRecords.partition)) {
log.debug("Skipping fetching records for assigned partition {} because it is paused", partitionRecords.partition);
completedFetches.add(partitionRecords);
nextInLineRecords = null;
return emptyList();
}
} else {
SubscriptionState.FetchPosition position = subscriptions.position(partitionRecords.partition);
if (partitionRecords.nextFetchOffset == position.offset) {
Expand Down Expand Up @@ -670,7 +702,9 @@ private List<ConsumerRecord<K, V>> fetchRecords(PartitionRecords partitionRecord
}
}

log.trace("Draining fetched records for partition {}", partitionRecords.partition);
partitionRecords.drain();

return emptyList();
}

Expand Down Expand Up @@ -1025,7 +1059,7 @@ private List<TopicPartition> fetchablePartitions() {
if (nextInLineRecords != null && !nextInLineRecords.isFetched) {
exclude.add(nextInLineRecords.partition);
}
for (CompletedFetch completedFetch : completedFetches) {
for (PartitionRecords completedFetch : completedFetches) {
exclude.add(completedFetch.partition);
}
return subscriptions.fetchablePartitions(tp -> !exclude.contains(tp));
Expand Down Expand Up @@ -1125,20 +1159,31 @@ private <T> Map<Node, Map<TopicPartition, T>> regroupPartitionMapByNode(Map<Topi
}

/**
* The callback for fetch completion
* Parse a PartitionRecords object from a CompletedFetch
*/
private PartitionRecords parseCompletedFetch(CompletedFetch completedFetch) {
TopicPartition tp = completedFetch.partition;
FetchResponse.PartitionData<Records> partition = completedFetch.partitionData;

Iterator<? extends RecordBatch> batches = partition.records.batches().iterator();
return new PartitionRecords(tp, completedFetch, batches);
}

/**
* Initialize a PartitionRecords object.
*/
private PartitionRecords initializePartitionRecords(PartitionRecords partitionRecordsToInitialize) {
CompletedFetch completedFetch = partitionRecordsToInitialize.completedFetch;
TopicPartition tp = completedFetch.partition;
FetchResponse.PartitionData<Records> partition = completedFetch.partitionData;
long fetchOffset = completedFetch.fetchedOffset;
PartitionRecords partitionRecords = null;
Errors error = partition.error;

try {
if (!subscriptions.isFetchable(tp)) {
// this can happen when a rebalance happened or a partition consumption paused
// while fetch is still in-flight
log.debug("Ignoring fetched records for partition {} since it is no longer fetchable", tp);
if (!subscriptions.hasValidPosition(tp)) {
// this can happen when a rebalance happened while fetch is still in-flight
log.debug("Ignoring fetched records for partition {} since it no longer has valid position", tp);
} else if (error == Errors.NONE) {
// we are interested in this fetch only if the beginning offset matches the
// current consumed position
Expand All @@ -1152,7 +1197,7 @@ private PartitionRecords parseCompletedFetch(CompletedFetch completedFetch) {
log.trace("Preparing to read {} bytes of data for partition {} with offset {}",
partition.records.sizeInBytes(), tp, position);
Iterator<? extends RecordBatch> batches = partition.records.batches().iterator();
partitionRecords = new PartitionRecords(tp, completedFetch, batches);
partitionRecords = partitionRecordsToInitialize;

if (!batches.hasNext() && partition.records.sizeInBytes() > 0) {
if (completedFetch.responseVersion < 3) {
Expand Down Expand Up @@ -1196,6 +1241,8 @@ private PartitionRecords parseCompletedFetch(CompletedFetch completedFetch) {
});
}


partitionRecordsToInitialize.initialized = true;
} else if (error == Errors.NOT_LEADER_FOR_PARTITION ||
error == Errors.REPLICA_NOT_AVAILABLE ||
error == Errors.KAFKA_STORAGE_ERROR ||
Expand Down Expand Up @@ -1286,13 +1333,16 @@ private Optional<Integer> maybeLeaderEpoch(int leaderEpoch) {
* @param assignedPartitions newly assigned {@link TopicPartition}
*/
public void clearBufferedDataForUnassignedPartitions(Collection<TopicPartition> assignedPartitions) {
Iterator<CompletedFetch> itr = completedFetches.iterator();
while (itr.hasNext()) {
TopicPartition tp = itr.next().partition;
Iterator<PartitionRecords> completedFetchesItr = completedFetches.iterator();
while (completedFetchesItr.hasNext()) {
PartitionRecords records = completedFetchesItr.next();
TopicPartition tp = records.partition;
if (!assignedPartitions.contains(tp)) {
itr.remove();
records.drain();
completedFetchesItr.remove();
}
}

if (nextInLineRecords != null && !assignedPartitions.contains(nextInLineRecords.partition)) {
nextInLineRecords.drain();
nextInLineRecords = null;
Expand Down Expand Up @@ -1345,6 +1395,7 @@ private class PartitionRecords {
private boolean isFetched = false;
private Exception cachedRecordException = null;
private boolean corruptLastRecord = false;
private boolean initialized = false;

private PartitionRecords(TopicPartition partition,
CompletedFetch completedFetch,
Expand Down Expand Up @@ -1546,6 +1597,10 @@ private boolean containsAbortMarker(RecordBatch batch) {
Record firstRecord = batchIterator.next();
return ControlRecordType.ABORT == ControlRecordType.parse(firstRecord.key());
}

private boolean notInitialized() {
return !this.initialized;
}
}

private static class CompletedFetch {
Expand Down
139 changes: 139 additions & 0 deletions clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -897,6 +897,145 @@ public void testFetchOnPausedPartition() {
assertTrue(client.requests().isEmpty());
}

@Test
public void testFetchOnCompletedFetchesForPausedAndResumedPartitions() {
buildFetcher();

assignFromUser(singleton(tp0));
subscriptions.seek(tp0, 0);

assertEquals(1, fetcher.sendFetches());

subscriptions.pause(tp0);

client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NONE, 100L, 0));
consumerClient.poll(time.timer(0));

Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> fetchedRecords = fetchedRecords();
assertEquals("Should not return any records when partition is paused", 0, fetchedRecords.size());
assertTrue("Should still contain completed fetches", fetcher.hasCompletedFetches());
assertNull(fetchedRecords.get(tp0));
assertEquals(0, fetcher.sendFetches());

subscriptions.resume(tp0);

consumerClient.poll(time.timer(0));
fetchedRecords = fetchedRecords();
assertEquals("Should return records when partition is resumed", 1, fetchedRecords.size());
assertNotNull(fetchedRecords.get(tp0));
assertEquals(3, fetchedRecords.get(tp0).size());

consumerClient.poll(time.timer(0));
fetchedRecords = fetchedRecords();
assertEquals("Should not return records after previously paused partitions are fetched", 0, fetchedRecords.size());
assertFalse("Should no longer contain completed fetches", fetcher.hasCompletedFetches());
}

@Test
public void testFetchOnCompletedFetchesForSomePausedPartitions() {
buildFetcher();

Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> fetchedRecords;

assignFromUser(Utils.mkSet(tp0, tp1));

// seek to tp0 and tp1 in two polls to generate 2 complete requests and responses

// #1 seek, request, poll, response
subscriptions.seek(tp0, 1);
assertEquals(1, fetcher.sendFetches());
client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NONE, 100L, 0));
consumerClient.poll(time.timer(0));

// #2 seek, request, poll, response
subscriptions.seek(tp1, 1);
assertEquals(1, fetcher.sendFetches());
client.prepareResponse(fullFetchResponse(tp1, this.nextRecords, Errors.NONE, 100L, 0));

subscriptions.pause(tp0);
consumerClient.poll(time.timer(0));

fetchedRecords = fetchedRecords();
assertEquals("Should return completed fetch for unpaused partitions", 1, fetchedRecords.size());
assertTrue("Should still contain completed fetches", fetcher.hasCompletedFetches());
assertNotNull(fetchedRecords.get(tp1));
assertNull(fetchedRecords.get(tp0));

fetchedRecords = fetchedRecords();
assertEquals("Should return no records for remaining paused partition", 0, fetchedRecords.size());
assertTrue("Should still contain completed fetches", fetcher.hasCompletedFetches());
}

@Test
public void testFetchOnCompletedFetchesForAllPausedPartitions() {
buildFetcher();

Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> fetchedRecords;

assignFromUser(Utils.mkSet(tp0, tp1));

// seek to tp0 and tp1 in two polls to generate 2 complete requests and responses

// #1 seek, request, poll, response
subscriptions.seek(tp0, 1);
assertEquals(1, fetcher.sendFetches());
client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NONE, 100L, 0));
consumerClient.poll(time.timer(0));

// #2 seek, request, poll, response
subscriptions.seek(tp1, 1);
assertEquals(1, fetcher.sendFetches());
client.prepareResponse(fullFetchResponse(tp1, this.nextRecords, Errors.NONE, 100L, 0));

subscriptions.pause(tp0);
subscriptions.pause(tp1);

consumerClient.poll(time.timer(0));

fetchedRecords = fetchedRecords();
assertEquals("Should return no records for all paused partitions", 0, fetchedRecords.size());
assertTrue("Should still contain completed fetches", fetcher.hasCompletedFetches());
}

@Test
public void testPartialFetchWithPausedPartitions() {
// this test sends creates a completed fetch with 3 records and a max poll of 2 records to assert
// that a fetch that must be returned over at least 2 polls can be cached successfully when its partition is
// paused, then returned successfully after its been resumed again later
buildFetcher(2);

Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> fetchedRecords;

assignFromUser(Utils.mkSet(tp0, tp1));

subscriptions.seek(tp0, 1);
assertEquals(1, fetcher.sendFetches());
client.prepareResponse(fullFetchResponse(tp0, this.records, Errors.NONE, 100L, 0));
consumerClient.poll(time.timer(0));

fetchedRecords = fetchedRecords();

assertEquals("Should return 2 records from fetch with 3 records", 2, fetchedRecords.get(tp0).size());
assertFalse("Should have no completed fetches", fetcher.hasCompletedFetches());

subscriptions.pause(tp0);
consumerClient.poll(time.timer(0));

fetchedRecords = fetchedRecords();

assertEquals("Should return no records for paused partitions", 0, fetchedRecords.size());
assertTrue("Should have 1 entry in completed fetches", fetcher.hasCompletedFetches());

subscriptions.resume(tp0);

consumerClient.poll(time.timer(0));

fetchedRecords = fetchedRecords();

assertEquals("Should return last remaining record", 1, fetchedRecords.get(tp0).size());
assertFalse("Should have no completed fetches", fetcher.hasCompletedFetches());
}

@Test
public void testFetchNotLeaderForPartition() {
buildFetcher();
Expand Down