Skip to content

KAFKA-15181: fix(storage): wait for consumer to be synced after assigning partitions #14012

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
jeqo wants to merge 6 commits into from
Closed

KAFKA-15181: fix(storage): wait for consumer to be synced after assigning partitions #14012

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
5f92706
chore(storage): nit fixes on TBRLMM tests
jeqo Jul 13, 2023
bf65b07
feat(storage): add common set of properties
jeqo Jul 14, 2023
a19a210
test(storage): add test case to validate cache is in-sync on assignments
jeqo Jul 14, 2023
4c09893
fix(storage): wait for consumer to be synced after assigning partitions
jeqo Jul 14, 2023
6c9adaf
refactor: info log when waiting for consumer to catch up
jeqo Jul 17, 2023
6855c6b
fixup! test(storage): add test case to validate cache is in-sync on a…
jeqo Jul 19, 2023
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
2 changes: 1 addition & 1 deletion checkstyle/suppressions.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -337,7 +337,7 @@
<suppress checks="CyclomaticComplexity"
files="(LogValidator|RemoteLogManagerConfig).java"/>
<suppress checks="NPathComplexity"
files="(LogValidator|RemoteIndexCache).java"/>
files="(LogValidator|RemoteIndexCache|TopicBasedRemoteLogMetadataManager).java"/>
<suppress checks="ParameterNumber"
files="(LogAppendInfo|RemoteLogManagerConfig).java"/>

Expand Down
91 changes: 72 additions & 19 deletions ...ge/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,10 +16,13 @@
*/
package org.apache.kafka.server.log.remote.metadata.storage;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListOffsetsResult;
import org.apache.kafka.clients.admin.OffsetSpec;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.utils.KafkaThread;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
Expand All @@ -30,9 +33,14 @@
import java.io.File;
import java.io.IOException;
import java.nio.file.Path;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
* This class manages the consumer thread viz {@link ConsumerTask} that polls messages from the assigned metadata topic partitions.
Expand All @@ -46,6 +54,7 @@ public class ConsumerManager implements Closeable {
private static final Logger log = LoggerFactory.getLogger(ConsumerManager.class);
private static final long CONSUME_RECHECK_INTERVAL_MS = 50L;

private final AdminClient adminClient;
private final TopicBasedRemoteLogMetadataManagerConfig rlmmConfig;
private final Time time;
private final ConsumerTask consumerTask;
Expand All @@ -60,8 +69,17 @@ public ConsumerManager(TopicBasedRemoteLogMetadataManagerConfig rlmmConfig,

//Create a task to consume messages and submit the respective events to RemotePartitionMetadataEventHandler.
KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(rlmmConfig.consumerProperties());
adminClient = AdminClient.create(rlmmConfig.commonProperties());
Path committedOffsetsPath = new File(rlmmConfig.logDir(), COMMITTED_OFFSETS_FILE_NAME).toPath();
consumerTask = new ConsumerTask(consumer, remotePartitionMetadataEventHandler, topicPartitioner, committedOffsetsPath, time, 60_000L);
consumerTask = new ConsumerTask(
consumer,
rlmmConfig.remoteLogMetadataTopicName(),
remotePartitionMetadataEventHandler,
topicPartitioner,
committedOffsetsPath,
time,
60_000L
);
consumerTaskThread = KafkaThread.nonDaemon("RLMMConsumerTask", consumerTask);
}

Expand All @@ -76,47 +94,52 @@ public void startConsumerThread() {
}

/**
* Waits if necessary for the consumption to reach the offset of the given {@code recordMetadata}.
* Waits if necessary for the consumption to reach the {@code offset} of the given record
* at a certain {@code partition} of the metadata topic.
*
* @param recordMetadata record metadata to be checked for consumption.
Comment thread
jeqo marked this conversation as resolved.
Outdated
* @param partition partition of metadata topic to be checked for consumption
* @param offset record offset of metadata topic {@code partition} to be checked for consumption
* @throws TimeoutException if this method execution did not complete with in the wait time configured with
* property {@code TopicBasedRemoteLogMetadataManagerConfig#REMOTE_LOG_METADATA_CONSUME_WAIT_MS_PROP}.
*/
public void waitTillConsumptionCatchesUp(RecordMetadata recordMetadata) throws TimeoutException {
waitTillConsumptionCatchesUp(recordMetadata, rlmmConfig.consumeWaitMs());
public void waitTillConsumptionCatchesUp(int partition, long offset) throws TimeoutException {
waitTillConsumptionCatchesUp(partition, offset, rlmmConfig.consumeWaitMs());
}

/**
* Waits if necessary for the consumption to reach the offset of the given {@code recordMetadata}.
* Waits if necessary for the consumption to reach the {@code offset} of the given record
* at a certain {@code partition} of the metadata topic.
*
* @param recordMetadata record metadata to be checked for consumption.
* @param timeoutMs wait timeout in milli seconds
* @param partition partition of metadata topic to be checked for consumption
* @param offset record offset of metadata topic {@code partition} to be checked for consumption
* @param timeoutMs wait timeout in milliseconds
* @throws TimeoutException if this method execution did not complete with in the given {@code timeoutMs}.
*/
public void waitTillConsumptionCatchesUp(RecordMetadata recordMetadata,
long timeoutMs) throws TimeoutException {
final int partition = recordMetadata.partition();
public void waitTillConsumptionCatchesUp(int partition, long offset, long timeoutMs) throws TimeoutException {
log.info("Waiting until consumer is caught up with the target partition [{}]", partition);

final long consumeCheckIntervalMs = Math.min(CONSUME_RECHECK_INTERVAL_MS, timeoutMs);
Comment thread
jeqo marked this conversation as resolved.
Outdated

// If the current assignment does not have the subscription for this partition then return immediately.
if (!consumerTask.isPartitionAssigned(partition)) {
throw new KafkaException("This consumer is not subscribed to the target partition " + partition + " on which message is produced.");
throw new KafkaException("This consumer is not assigned to the target partition " + partition + ". " +
"Partitions currently assigned: " + consumerTask.metadataPartitionsAssigned());
}

final long offset = recordMetadata.offset();
long startTimeMs = time.milliseconds();
while (true) {
log.debug("Checking if partition [{}] is up to date with offset [{}]", partition, offset);
long receivedOffset = consumerTask.receivedOffsetForPartition(partition).orElse(-1L);
if (receivedOffset >= offset) {
return;
}

log.debug("Committed offset [{}] for partition [{}], but the target offset: [{}], Sleeping for [{}] to retry again",
offset, partition, receivedOffset, consumeCheckIntervalMs);
log.debug("Expected offset [{}] for partition [{}], but the committed offset: [{}], Sleeping for [{}] to retry again",
offset, partition, receivedOffset, consumeCheckIntervalMs);

if (time.milliseconds() - startTimeMs > timeoutMs) {
log.warn("Committed offset for partition:[{}] is : [{}], but the target offset: [{}] ",
partition, receivedOffset, offset);
log.warn("Expected offset for partition:[{}] is : [{}], but the committed offset: [{}] ",
partition, receivedOffset, offset);
throw new TimeoutException("Timed out in catching up with the expected offset by consumer.");
}

Expand All @@ -126,8 +149,9 @@ public void waitTillConsumptionCatchesUp(RecordMetadata recordMetadata,

@Override
public void close() throws IOException {
// Consumer task will close the task and it internally closes all the resources including the consumer.
// Consumer task will close the task, and it internally closes all the resources including the consumer.
Utils.closeQuietly(consumerTask, "ConsumerTask");
Utils.closeQuietly(adminClient, "AdminClient");

// Wait until the consumer thread finishes.
try {
Expand All @@ -139,6 +163,35 @@ public void close() throws IOException {

public void addAssignmentsForPartitions(Set<TopicIdPartition> partitions) {
consumerTask.addAssignmentsForPartitions(partitions);
waitTillMetaPartitionConsumptionCatchesUp();

@kamalcph kamalcph Jul 17, 2023 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a blocking call. It won't return until the partition is ready/consumption-catches-up. In high throughput cluster, this can anywhere take order of seconds to complete.

Moreover, when you reassign the partitions, the controller will send the LISR request for one partition at a time and the LISR request handling path shouldn't have any blocking call so that the controller can send the next LISR request on time. Otherwise, LISR request gets timed out.

@showuon showuon Jul 17, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point @kamalcph ! Do you have any suggestion here?

@mdedetrich mdedetrich Jul 18, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am also unclear as to what we should do here, I mean we could return a CompletableFuture<Void> but this would involve changing the interface?

@kamalcph kamalcph Jul 18, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One approach we can take is to wait for metadata partition to catch up asynchronously. All the calls to read the remote log will hit the RemotePartitionMetadataStore#getRemoteLogMetadataCache, there we can ensure that the partition is initialised via a latch.

The ConsumerTask can keep a marker of the latest offset of metadata-partition when user-partitions are assigned to them and can mark that user-partition as initialised in the cache once it read/re-read messages up-to that point.

We can have a latch with timeout of 2 mins to avoid holding the caller thread for a long amount of time and throw a retry-able error.

@showuon showuon Jul 18, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sounds great! I like the idea that we lazily wait for the consumer to catch up. One comment is why don't we return retriable error immediately, and ask the client to try it again, like when the leader is under election, the client will get NOT_LEADER_OR_FOLLOWER error immediately and retry, instead of waiting in the broker side until leader election completed. WDYT?

@kamalcph kamalcph Jul 18, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That also works! We have to look around for all the callers:

  1. Replica fetcher thread will retry the request after the replica.fetch.backoff.ms once the partition is marked as error/failed.
  2. Consumers may resend the FETCH request as soon as it fails without the request quota.

We can also think about adding one more API in RemoteLogMetadataManager to ensure that the partition is initialised before copying the segments (or) when the replica thread is being initialised.

boolean isInitialized(TopicIdPartition topicIdPartition) throws RemoteStorageException;

@satishd satishd Jul 18, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The latching mechanism as mentioned by @kamalcph is what we are currently using internally. @abhijeetk88 is planning to pull those changes to trunk and raise PR this week to handle the scenario.

}

/**
* Whenever there is a change on the remote log metadata topic partitions assigned to the {@link TopicBasedRemoteLogMetadataManager},
* wait for consumer (therefore the cache as well) to be in-sync before continuing
*/
private void waitTillMetaPartitionConsumptionCatchesUp() {
try {
final Set<TopicPartition> topicPartitions = consumerTask.metadataPartitionsAssigned();
if (!topicPartitions.isEmpty()) {
final Map<TopicPartition, OffsetSpec> listOffsetsRequest = topicPartitions.stream()
.collect(Collectors.toMap(Function.identity(), tp -> OffsetSpec.latest()));
final Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> latestOffsets = adminClient.listOffsets(listOffsetsRequest)
.all()
.get(rlmmConfig.consumeWaitMs(), TimeUnit.MILLISECONDS); // piggybacking on existing timeout

for (TopicPartition tp : topicPartitions) {
waitTillConsumptionCatchesUp(
tp.partition(),
latestOffsets.get(tp).offset() - 1 // as latest offset is the next latest written record
);
}
}
} catch (ExecutionException | InterruptedException | TimeoutException e) {
log.error("Error encountered while consuming from remote log metadata partitions: {}",
consumerTask.metadataPartitionsAssigned(), e);
throw new KafkaException("Error encountered while consuming from remote log metadata partitions", e);
}
}

public void removeAssignmentsForPartitions(Set<TopicIdPartition> partitions) {
Expand Down
17 changes: 14 additions & 3 deletions storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,12 +68,13 @@ class ConsumerTask implements Runnable, Closeable {

private final RemoteLogMetadataSerde serde = new RemoteLogMetadataSerde();
private final KafkaConsumer<byte[], byte[]> consumer;
private final String metadataTopicName;
private final RemotePartitionMetadataEventHandler remotePartitionMetadataEventHandler;
private final RemoteLogMetadataTopicPartitioner topicPartitioner;
private final Time time;

// It indicates whether the closing process has been started or not. If it is set as true,
// consumer will stop consuming messages and it will not allow partition assignments to be updated.
// consumer will stop consuming messages, and it will not allow partition assignments to be updated.
private volatile boolean closing = false;

// It indicates whether the consumer needs to assign the partitions or not. This is set when it is
Expand Down Expand Up @@ -101,12 +102,14 @@ class ConsumerTask implements Runnable, Closeable {
private long lastSyncedTimeMs;

public ConsumerTask(KafkaConsumer<byte[], byte[]> consumer,
String metadataTopicName,
RemotePartitionMetadataEventHandler remotePartitionMetadataEventHandler,
RemoteLogMetadataTopicPartitioner topicPartitioner,
Path committedOffsetsPath,
Time time,
long committedOffsetSyncIntervalMs) {
this.consumer = Objects.requireNonNull(consumer);
this.metadataTopicName = Objects.requireNonNull(metadataTopicName);
this.remotePartitionMetadataEventHandler = Objects.requireNonNull(remotePartitionMetadataEventHandler);
this.topicPartitioner = Objects.requireNonNull(topicPartitioner);
this.time = Objects.requireNonNull(time);
Expand Down Expand Up @@ -143,6 +146,7 @@ private void initializeConsumerAssignment(Path committedOffsetsPath) {

// Seek to the committed offsets
for (Map.Entry<Integer, Long> entry : committedOffsets.entrySet()) {
log.debug("Updating consumed offset: [{}] for partition [{}]", entry.getValue(), entry.getKey());
partitionToConsumedOffsets.put(entry.getKey(), entry.getValue());
consumer.seek(new TopicPartition(REMOTE_LOG_METADATA_TOPIC_NAME, entry.getKey()), entry.getValue());
}
Expand Down Expand Up @@ -187,6 +191,7 @@ private void processConsumerRecord(ConsumerRecord<byte[], byte[]> record) {
} else {
log.debug("This event {} is skipped as the topic partition is not assigned for this instance.", remoteLogMetadata);
}
log.debug("Updating consumed offset: [{}] for partition [{}]", record.offset(), record.partition());
partitionToConsumedOffsets.put(record.partition(), record.offset());
}
}
Expand All @@ -209,7 +214,7 @@ private void maybeSyncCommittedDataAndOffsets(boolean forceSync) {
if (offset != null) {
remotePartitionMetadataEventHandler.syncLogMetadataSnapshot(topicIdPartition, metadataPartition, offset);
} else {
log.debug("Skipping syncup of the remote-log-metadata-file for partition:{} , with remote log metadata partition{}, and no offset",
log.debug("Skipping sync-up of the remote-log-metadata-file for partition:{} , with remote log metadata partition{}, and no offset",

@kamalcph kamalcph Jul 17, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

partition{} -> partition {}

topicIdPartition, metadataPartition);
}
}
Expand Down Expand Up @@ -313,7 +318,7 @@ private void updateAssignmentsForPartitions(Set<TopicIdPartition> addedPartition
updatedAssignedMetaPartitions.add(topicPartitioner.metadataPartition(tp));
}

// Clear removed topic partitions from inmemory cache.
// Clear removed topic partitions from in-memory cache.
for (TopicIdPartition removedPartition : removedPartitions) {
remotePartitionMetadataEventHandler.clearTopicPartition(removedPartition);
}
Expand Down Expand Up @@ -353,4 +358,10 @@ public void close() {
}
}
}

public Set<TopicPartition> metadataPartitionsAssigned() {
return assignedMetaPartitions.stream()
.map(partitionNum -> new TopicPartition(metadataTopicName, partitionNum))
.collect(Collectors.toSet());
}
}
6 changes: 2 additions & 4 deletions ...ava/org/apache/kafka/server/log/remote/metadata/storage/RemotePartitionMetadataStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,11 +47,9 @@ public class RemotePartitionMetadataStore extends RemotePartitionMetadataEventHa

private final Path logDir;

private Map<TopicIdPartition, RemotePartitionDeleteMetadata> idToPartitionDeleteMetadata =
new ConcurrentHashMap<>();
private Map<TopicIdPartition, RemotePartitionDeleteMetadata> idToPartitionDeleteMetadata = new ConcurrentHashMap<>();

private Map<TopicIdPartition, FileBasedRemoteLogMetadataCache> idToRemoteLogMetadataCache =
new ConcurrentHashMap<>();
private Map<TopicIdPartition, FileBasedRemoteLogMetadataCache> idToRemoteLogMetadataCache = new ConcurrentHashMap<>();

public RemotePartitionMetadataStore(Path logDir) {
this.logDir = logDir;
Expand Down
Loading