satishd commented on a change in pull request #11058:
URL: https://github.com/apache/kafka/pull/11058#discussion_r714973902
##########
File path:
storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java
##########
@@ -85,21 +90,68 @@
// Map of remote log metadata topic partition to consumed offsets.
private final Map<Integer, Long> partitionToConsumedOffsets = new
ConcurrentHashMap<>();
+ private Map<Integer, Long> committedPartitionToConsumedOffsets =
Collections.emptyMap();
+
+ private final long committedOffsetSyncIntervalMs;
+ private CommittedOffsetsFile committedOffsetsFile;
+ private long lastSyncedTimeMs;
+
public ConsumerTask(KafkaConsumer<byte[], byte[]> consumer,
RemotePartitionMetadataEventHandler
remotePartitionMetadataEventHandler,
- RemoteLogMetadataTopicPartitioner topicPartitioner) {
- Objects.requireNonNull(consumer);
- Objects.requireNonNull(remotePartitionMetadataEventHandler);
- Objects.requireNonNull(topicPartitioner);
-
- this.consumer = consumer;
- this.remotePartitionMetadataEventHandler =
remotePartitionMetadataEventHandler;
- this.topicPartitioner = topicPartitioner;
+ RemoteLogMetadataTopicPartitioner topicPartitioner,
+ Path committedOffsetsPath,
+ Time time,
+ long committedOffsetSyncIntervalMs) {
+ this.consumer = Objects.requireNonNull(consumer);
+ this.remotePartitionMetadataEventHandler =
Objects.requireNonNull(remotePartitionMetadataEventHandler);
+ this.topicPartitioner = Objects.requireNonNull(topicPartitioner);
+ this.time = Objects.requireNonNull(time);
+ this.committedOffsetSyncIntervalMs = committedOffsetSyncIntervalMs;
+
+ initializeConsumerAssignment(committedOffsetsPath);
+ }
+
+ private void initializeConsumerAssignment(Path committedOffsetsPath) {
+ try {
+ committedOffsetsFile = new
CommittedOffsetsFile(committedOffsetsPath.toFile());
+ } catch (IOException e) {
+ throw new KafkaException(e);
+ }
+
+ Map<Integer, Long> committedOffsets = Collections.emptyMap();
+ try {
+ // Load committed offset and assign them in the consumer.
+ committedOffsets = committedOffsetsFile.readEntries();
+ } catch (IOException e) {
+ // Ignore the error and consumer consumes from the earliest offset.
+ log.error("Encountered error while building committed offsets from
the file", e);
+ }
+
+ final Set<Map.Entry<Integer, Long>> entries =
committedOffsets.entrySet();
+
+ if (!entries.isEmpty()) {
+ // Assign topic partitions from the earlier committed offsets file.
+ Set<Integer> earlierAssignedPartitions = committedOffsets.keySet();
+ assignedMetaPartitions =
Collections.unmodifiableSet(earlierAssignedPartitions);
+ Set<TopicPartition> metadataTopicPartitions =
earlierAssignedPartitions.stream()
+
.map(x -> new TopicPartition(REMOTE_LOG_METADATA_TOPIC_NAME, x))
+
.collect(Collectors.toSet());
+ consumer.assign(metadataTopicPartitions);
+
+ // Seek to the committed offsets
+ for (Map.Entry<Integer, Long> entry : entries) {
+ partitionToConsumedOffsets.put(entry.getKey(),
entry.getValue());
+ consumer.seek(new
TopicPartition(REMOTE_LOG_METADATA_TOPIC_NAME, entry.getKey()),
entry.getValue());
+ }
+
+ committedPartitionToConsumedOffsets = committedOffsets;
+ }
}
@Override
public void run() {
log.info("Started Consumer task thread.");
+ lastSyncedTimeMs = time.milliseconds();
Review comment:
I guess the consumer would fetch from the earliest offset as there is no
state associated with in the consumer for those partitions as
`auto.offset.reset` is set as `earliest`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
