KAFKA-15181: Improvements for TopicBaseRemoteLogMetadataManager

clients/src/main/java/org/apache/kafka/common/errors/ResourceNotReadyException.java

@@ -0,0 +1,14 @@
+package org.apache.kafka.common.errors;
+
+public class ResourceNotReadyException extends RetriableException {
+
+    public static final long serialVersionUID = 1L;
+
+    public ResourceNotReadyException() {
+        super();
+    }
+
+    public ResourceNotReadyException(String message) {
+        super(message);
+    }
+}

clients/src/main/java/org/apache/kafka/common/protocol/Errors.java

@@ -103,6 +103,7 @@
 import org.apache.kafka.common.errors.RecordTooLargeException;
 import org.apache.kafka.common.errors.ReplicaNotAvailableException;
 import org.apache.kafka.common.errors.ResourceNotFoundException;
+import org.apache.kafka.common.errors.ResourceNotReadyException;
 import org.apache.kafka.common.errors.RetriableException;
 import org.apache.kafka.common.errors.SaslAuthenticationException;
 import org.apache.kafka.common.errors.SecurityDisabledException;
@@ -380,7 +381,8 @@ public enum Errors {
     FENCED_MEMBER_EPOCH(110, "The member epoch is fenced by the group coordinator. The member must abandon all its partitions and rejoin.", FencedMemberEpochException::new),
     UNRELEASED_INSTANCE_ID(111, "The instance ID is still used by another member in the consumer group. That member must leave first.", UnreleasedInstanceIdException::new),
     UNSUPPORTED_ASSIGNOR(112, "The assignor or its version range is not supported by the consumer group.", UnsupportedAssignorException::new),
-    STALE_MEMBER_EPOCH(113, "The member epoch is stale. The member must retry after receiving its updated member epoch via the ConsumerGroupHeartbeat API.", StaleMemberEpochException::new);
+    STALE_MEMBER_EPOCH(113, "The member epoch is stale. The member must retry after receiving its updated member epoch via the ConsumerGroupHeartbeat API.", StaleMemberEpochException::new),
+    RESOURCE_NOT_READY(114, "The resource is not ready.", ResourceNotReadyException::new);
 
     private static final Logger log = LoggerFactory.getLogger(Errors.class);
 

storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerManager.java

@@ -27,9 +27,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.Closeable;
-import java.io.File;
 import java.io.IOException;
-import java.nio.file.Path;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.TimeoutException;
@@ -41,8 +39,6 @@
  */
 public class ConsumerManager implements Closeable {
 
-    public static final String COMMITTED_OFFSETS_FILE_NAME = "_rlmm_committed_offsets";
-
     private static final Logger log = LoggerFactory.getLogger(ConsumerManager.class);
     private static final long CONSUME_RECHECK_INTERVAL_MS = 50L;
 
@@ -60,15 +56,13 @@ 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());
-        Path committedOffsetsPath = new File(rlmmConfig.logDir(), COMMITTED_OFFSETS_FILE_NAME).toPath();
         consumerTask = new ConsumerTask(
-                consumer,
-                rlmmConfig.remoteLogMetadataTopicName(),
-                remotePartitionMetadataEventHandler,
-                topicPartitioner,
-                committedOffsetsPath,
-                time,
-                60_000L
+            remotePartitionMetadataEventHandler,
+            topicPartitioner,
+            consumer,
+            100L,
+            300_000L,
+            time
         );
         consumerTaskThread = KafkaThread.nonDaemon("RLMMConsumerTask", consumerTask);
     }
@@ -110,7 +104,7 @@ public void waitTillConsumptionCatchesUp(RecordMetadata recordMetadata,
         log.info("Waiting until consumer is caught up with the target partition: [{}]", partition);
 
         // If the current assignment does not have the subscription for this partition then return immediately.
-        if (!consumerTask.isPartitionAssigned(partition)) {
+        if (!consumerTask.isMetadataPartitionAssigned(partition)) {
             throw new KafkaException("This consumer is not assigned to the target partition " + partition + ". " +
                     "Partitions currently assigned: " + consumerTask.metadataPartitionsAssigned());
         }
@@ -119,17 +113,17 @@ public void waitTillConsumptionCatchesUp(RecordMetadata recordMetadata,
         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) {
+            long readOffset = consumerTask.readOffsetForMetadataPartition(partition).orElse(-1L);
+            if (readOffset >= offset) {
                 return;
             }
 
             log.debug("Expected offset [{}] for partition [{}], but the committed offset: [{}],  Sleeping for [{}] to retry again",
-                    offset, partition, receivedOffset, consumeCheckIntervalMs);
+                    offset, partition, readOffset, consumeCheckIntervalMs);
 
             if (time.milliseconds() - startTimeMs > timeoutMs) {
                 log.warn("Expected offset for partition:[{}] is : [{}], but the committed offset: [{}] ",
-                        partition, receivedOffset, offset);
+                        partition, readOffset, offset);
                 throw new TimeoutException("Timed out in catching up with the expected offset by consumer.");
             }
 
@@ -158,7 +152,7 @@ public void removeAssignmentsForPartitions(Set<TopicIdPartition> partitions) {
         consumerTask.removeAssignmentsForPartitions(partitions);
     }
 
-    public Optional<Long> receivedOffsetForPartition(int metadataPartition) {
-        return consumerTask.receivedOffsetForPartition(metadataPartition);
+    public Optional<Long> readOffsetForPartition(int metadataPartition) {
+        return consumerTask.readOffsetForMetadataPartition(metadataPartition);
     }
 }