feat: trim stream prior to deleting objects (#578)

Signed-off-by: Li Zhanhui <lizhanhui@gmail.com> Signed-off-by: SSpirits <admin@lv5.moe> Co-authored-by: SSpirits <admin@lv5.moe>
AutoMQ · Nov 7, 2023 · 1f0f26c · 1f0f26c
1 parent 0674d5f
commit 1f0f26c
Show file tree

Hide file tree

Showing 7 changed files with 23 additions and 27 deletions.
diff --git a/broker/src/main/java/com/automq/rocketmq/broker/BrokerController.java b/broker/src/main/java/com/automq/rocketmq/broker/BrokerController.java
@@ -20,9 +20,9 @@
 import com.automq.rocketmq.common.api.DataStore;
 import com.automq.rocketmq.common.config.BrokerConfig;
 import com.automq.rocketmq.common.util.Lifecycle;
+import com.automq.rocketmq.controller.MetadataStore;
 import com.automq.rocketmq.controller.server.ControllerServiceImpl;
 import com.automq.rocketmq.controller.server.MetadataStoreBuilder;
-import com.automq.rocketmq.controller.MetadataStore;
 import com.automq.rocketmq.metadata.DefaultProxyMetadataService;
 import com.automq.rocketmq.metadata.DefaultStoreMetadataService;
 import com.automq.rocketmq.metadata.api.ProxyMetadataService;
@@ -75,7 +75,7 @@ public BrokerController(BrokerConfig brokerConfig) throws Exception {
         MessageStoreImpl messageStore = MessageStoreBuilder.build(brokerConfig.store(), brokerConfig.s3Stream(), storeMetadataService, dlqService);
         this.messageStore = messageStore;
 
-        DataStore dataStore = new DataStoreFacade(messageStore.getS3ObjectOperator(), messageStore.getTopicQueueManager());
+        DataStore dataStore = new DataStoreFacade(messageStore.streamStore(), messageStore.s3ObjectOperator(), messageStore.topicQueueManager());
         metadataStore.setDataStore(dataStore);
 
 

diff --git a/common/src/main/java/com/automq/rocketmq/common/api/DataStore.java b/common/src/main/java/com/automq/rocketmq/common/api/DataStore.java
@@ -24,6 +24,14 @@ public interface DataStore {
 
     CompletableFuture<Void> closeQueue(long topicId, int queueId);
 
+    /**
+     * Trim stream such that records with offsets prior to <code>offset</code> will be inaccessible.
+     *
+     * @param streamId ID of the stream to trim
+     * @param offset   Minimum offset of the stream after trim operation
+     */
+    CompletableFuture<Void> trimStream(long streamId, long offset);
+
     /**
      * Delete a list of S3 objects by object id.
      * <p>

diff --git a/proxy/src/test/java/com/automq/rocketmq/proxy/mock/MockMessageStore.java b/proxy/src/test/java/com/automq/rocketmq/proxy/mock/MockMessageStore.java
@@ -20,9 +20,7 @@
 import com.automq.rocketmq.common.model.FlatMessageExt;
 import com.automq.rocketmq.common.model.generated.FlatMessage;
 import com.automq.rocketmq.store.api.LogicQueue;
-import com.automq.rocketmq.store.api.LogicQueueManager;
 import com.automq.rocketmq.store.api.MessageStore;
-import com.automq.rocketmq.store.api.S3ObjectOperator;
 import com.automq.rocketmq.store.model.message.AckResult;
 import com.automq.rocketmq.store.model.message.ChangeInvisibleDurationResult;
 import com.automq.rocketmq.store.model.message.ClearRetryMessagesResult;
@@ -40,7 +38,6 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicLong;
-import org.apache.commons.lang3.NotImplementedException;
 import org.apache.commons.lang3.tuple.Pair;
 
 public class MockMessageStore implements MessageStore {
@@ -65,16 +62,6 @@ public void shutdown() {
 
     }
 
-    @Override
-    public LogicQueueManager getTopicQueueManager() {
-        throw new NotImplementedException();
-    }
-
-    @Override
-    public S3ObjectOperator getS3ObjectOperator() {
-        throw new NotImplementedException();
-    }
-
     @Override
     public CompletableFuture<PopResult> pop(long consumerGroupId, long topicId, int queueId, Filter filter,
         int batchSize, boolean fifo, boolean retry, long invisibleDuration) {

diff --git a/store/src/main/java/com/automq/rocketmq/store/DataStoreFacade.java b/store/src/main/java/com/automq/rocketmq/store/DataStoreFacade.java
@@ -20,16 +20,19 @@
 import com.automq.rocketmq.common.api.DataStore;
 import com.automq.rocketmq.store.api.LogicQueueManager;
 import com.automq.rocketmq.store.api.S3ObjectOperator;
+import com.automq.rocketmq.store.api.StreamStore;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 
 public class DataStoreFacade implements DataStore {
 
+    private final StreamStore streamStore;
     private final S3ObjectOperator s3ObjectOperator;
-
     private final LogicQueueManager logicQueueManager;
 
-    public DataStoreFacade(S3ObjectOperator s3ObjectOperator, LogicQueueManager logicQueueManager) {
+    public DataStoreFacade(StreamStore streamStore, S3ObjectOperator s3ObjectOperator,
+        LogicQueueManager logicQueueManager) {
+        this.streamStore = streamStore;
         this.s3ObjectOperator = s3ObjectOperator;
         this.logicQueueManager = logicQueueManager;
     }
@@ -39,6 +42,11 @@ public CompletableFuture<Void> closeQueue(long topicId, int queueId) {
         return logicQueueManager.close(topicId, queueId);
     }
 
+    @Override
+    public CompletableFuture<Void> trimStream(long streamId, long offset) {
+        return streamStore.trim(streamId, offset);
+    }
+
     @Override
     public CompletableFuture<List<Long>> batchDeleteS3Objects(List<Long> objectIds) {
         return s3ObjectOperator.delete(objectIds);

diff --git a/store/src/main/java/com/automq/rocketmq/store/MessageStoreImpl.java b/store/src/main/java/com/automq/rocketmq/store/MessageStoreImpl.java
@@ -81,8 +81,7 @@ public MessageStoreImpl(StoreConfig config, StreamStore streamStore,
         this.s3ObjectOperator = s3ObjectOperator;
     }
 
-    @Override
-    public LogicQueueManager getTopicQueueManager() {
+    public LogicQueueManager topicQueueManager() {
         return logicQueueManager;
     }
 
@@ -93,8 +92,7 @@ public StreamStore streamStore() {
     /**
      * @return {@link S3ObjectOperator} instance
      */
-    @Override
-    public S3ObjectOperator getS3ObjectOperator() {
+    public S3ObjectOperator s3ObjectOperator() {
         return s3ObjectOperator;
     }
 

diff --git a/store/src/main/java/com/automq/rocketmq/store/api/MessageStore.java b/store/src/main/java/com/automq/rocketmq/store/api/MessageStore.java
@@ -31,10 +31,6 @@
 
 public interface MessageStore extends Lifecycle {
 
-    LogicQueueManager getTopicQueueManager();
-
-    S3ObjectOperator getS3ObjectOperator();
-
     /**
      * Pop message from specified topic and queue.
      *

diff --git a/store/src/main/java/com/automq/rocketmq/store/metrics/StoreMetricsManager.java b/store/src/main/java/com/automq/rocketmq/store/metrics/StoreMetricsManager.java
@@ -98,7 +98,7 @@ public void run() {
             waitForRunning(config.periodicExporterIntervalInMills());
 
             StreamStore streamStore = messageStore.streamStore();
-            DefaultLogicQueueManager manager = (DefaultLogicQueueManager) messageStore.getTopicQueueManager();
+            DefaultLogicQueueManager manager = (DefaultLogicQueueManager) messageStore.topicQueueManager();
             Set<LagRecord> newLagRecordSet = Sets.newConcurrentHashSet();
             manager.logicQueueMap().forEach((topicQueueId, logicQueueFuture) -> {
                 if (!logicQueueFuture.isDone() || logicQueueFuture.isCompletedExceptionally()) {
@@ -148,7 +148,6 @@ public void initStaticMetrics(Meter meter) {
 
     @Override
     public void initDynamicMetrics(Meter meter) {
-        StoreMetricsManager.attributesBuilderSupplier = attributesBuilderSupplier;
         consumerLagMessages = meter.gaugeBuilder(GAUGE_CONSUMER_LAG_MESSAGES)
             .setDescription("Consumer lag messages")
             .ofLongs()