[RIP-81] Master-slave metadata incremental synchronization

broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBSubscriptionGroupManager.java

@@ -128,8 +128,7 @@ public boolean stop() {
     @Override
     public SubscriptionGroupConfig putSubscriptionGroupConfig(SubscriptionGroupConfig subscriptionGroupConfig) {
         String groupName = subscriptionGroupConfig.getGroupName();
-        SubscriptionGroupConfig oldConfig = this.subscriptionGroupTable.put(groupName, subscriptionGroupConfig);
-
+        SubscriptionGroupConfig oldConfig = super.putSubscriptionGroupConfig(subscriptionGroupConfig);
         try {
             byte[] keyBytes = groupName.getBytes(DataConverter.CHARSET_UTF8);
             byte[] valueBytes = JSON.toJSONBytes(subscriptionGroupConfig, SerializerFeature.BrowserCompatible);
@@ -143,7 +142,7 @@ public SubscriptionGroupConfig putSubscriptionGroupConfig(SubscriptionGroupConfi
     @Override
     protected SubscriptionGroupConfig putSubscriptionGroupConfigIfAbsent(SubscriptionGroupConfig subscriptionGroupConfig) {
         String groupName = subscriptionGroupConfig.getGroupName();
-        SubscriptionGroupConfig oldConfig = this.subscriptionGroupTable.putIfAbsent(groupName, subscriptionGroupConfig);
+        SubscriptionGroupConfig oldConfig = super.putSubscriptionGroupConfigIfAbsent(subscriptionGroupConfig);
         if (oldConfig == null) {
             try {
                 byte[] keyBytes = groupName.getBytes(DataConverter.CHARSET_UTF8);
@@ -158,7 +157,7 @@ protected SubscriptionGroupConfig putSubscriptionGroupConfigIfAbsent(Subscriptio
 
     @Override
     protected SubscriptionGroupConfig removeSubscriptionGroupConfig(String groupName) {
-        SubscriptionGroupConfig subscriptionGroupConfig = this.subscriptionGroupTable.remove(groupName);
+        SubscriptionGroupConfig subscriptionGroupConfig = super.removeSubscriptionGroupConfig(groupName);
         try {
             this.rocksDBConfigManager.delete(groupName.getBytes(DataConverter.CHARSET_UTF8));
         } catch (Exception e) {

broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBTopicConfigManager.java

@@ -110,7 +110,8 @@ protected void decodeTopicConfig(byte[] key, byte[] body) {
     @Override
     public TopicConfig putTopicConfig(TopicConfig topicConfig) {
         String topicName = topicConfig.getTopicName();
-        TopicConfig oldTopicConfig = this.topicConfigTable.put(topicName, topicConfig);
+        TopicConfig oldTopicConfig = super.putTopicConfig(topicConfig);
+
         try {
             byte[] keyBytes = topicName.getBytes(DataConverter.CHARSET_UTF8);
             byte[] valueBytes = JSON.toJSONBytes(topicConfig, SerializerFeature.BrowserCompatible);
@@ -123,7 +124,7 @@ public TopicConfig putTopicConfig(TopicConfig topicConfig) {
 
     @Override
     protected TopicConfig removeTopicConfig(String topicName) {
-        TopicConfig topicConfig = this.topicConfigTable.remove(topicName);
+        TopicConfig topicConfig = super.removeTopicConfig(topicName);
         try {
             this.rocksDBConfigManager.delete(topicName.getBytes(DataConverter.CHARSET_UTF8));
         } catch (Exception e) {

broker/src/main/java/org/apache/rocketmq/broker/controller/ReplicasManager.java

@@ -354,7 +354,11 @@ private void handleSlaveSynchronize(final BrokerRole role) {
             slaveSyncFuture = this.brokerController.getScheduledExecutorService().scheduleAtFixedRate(() -> {
                 try {
                     if (System.currentTimeMillis() - lastSyncTimeMs > 10 * 1000) {
-                        brokerController.getSlaveSynchronize().syncAll();
+                        if (!this.brokerController.getBrokerConfig().isAllowMetadataIncrementalSync()) {
+                            this.brokerController.getSlaveSynchronize().syncAll();
+                        } else {
+                            this.brokerController.getSlaveSynchronize().start();
+                        }
                         lastSyncTimeMs = System.currentTimeMillis();
                     }
                     //timer checkpoint, latency-sensitive, so sync it more frequently

broker/src/main/java/org/apache/rocketmq/broker/dledger/DLedgerRoleChangeHandler.java

@@ -114,7 +114,11 @@ private void handleSlaveSynchronize(BrokerRole role) {
                 public void run() {
                     try {
                         if (System.currentTimeMillis() - lastSyncTimeMs > 10 * 1000) {
-                            brokerController.getSlaveSynchronize().syncAll();
+                            if (!brokerController.getBrokerConfig().isAllowMetadataIncrementalSync()) {
+                                brokerController.getSlaveSynchronize().syncAll();
+                            } else {
+                                brokerController.getSlaveSynchronize().start();
+                            }
                             lastSyncTimeMs = System.currentTimeMillis();
                         }
                         //timer checkpoint, latency-sensitive, so sync it more frequently

broker/src/main/java/org/apache/rocketmq/broker/loadbalance/MessageRequestModeManager.java

@@ -16,10 +16,12 @@
  */
 package org.apache.rocketmq.broker.loadbalance;
 
+import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.BrokerPathConfigHelper;
 import org.apache.rocketmq.common.ConfigManager;
+import org.apache.rocketmq.common.topic.TopicValidator;
 import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
 import org.apache.rocketmq.remoting.protocol.body.SetMessageRequestModeRequestBody;
 
@@ -42,15 +44,34 @@ public void setMessageRequestMode(String topic, String consumerGroup, SetMessage
         ConcurrentHashMap<String, SetMessageRequestModeRequestBody> consumerGroup2ModeMap = messageRequestModeMap.get(topic);
         if (consumerGroup2ModeMap == null) {
             consumerGroup2ModeMap = new ConcurrentHashMap<>();
-            ConcurrentHashMap<String, SetMessageRequestModeRequestBody> pre =
-                messageRequestModeMap.putIfAbsent(topic, consumerGroup2ModeMap);
-            if (pre != null) {
-                consumerGroup2ModeMap = pre;
+            ConcurrentHashMap<String, SetMessageRequestModeRequestBody>[] pre = new ConcurrentHashMap[1];
+            ConcurrentHashMap<String, SetMessageRequestModeRequestBody> finalConsumerGroup2ModeMap = consumerGroup2ModeMap;
+            messageRequestModeMap.compute(topic, (key, existingValue) -> {
+                if (existingValue == null) {
+                    notifyMessageRequestModeCreated(requestBody);
+                    pre[0] = null;
+                    return finalConsumerGroup2ModeMap;
+                } else {
+                    notifyMessageRequestModeUpdated(requestBody);
+                    pre[0] = existingValue;
+                    return existingValue;
+                }
+            });
+            if (pre[0] != null) {
+                consumerGroup2ModeMap = pre[0];
             }
         }
         consumerGroup2ModeMap.put(consumerGroup, requestBody);
     }
 
+    private void notifyMessageRequestModeCreated(SetMessageRequestModeRequestBody requestBody) {
+        brokerController.getMetadataChangeObserver().onCreated(TopicValidator.RMQ_SYS_MESSAGE_MODE_SYNC, requestBody.getTopic(), requestBody);
+    }
+
+    private void notifyMessageRequestModeUpdated(SetMessageRequestModeRequestBody requestBody) {
+        brokerController.getMetadataChangeObserver().onUpdated(TopicValidator.RMQ_SYS_MESSAGE_MODE_SYNC, requestBody.getTopic(), requestBody);
+    }
+
     public SetMessageRequestModeRequestBody getMessageRequestMode(String topic, String consumerGroup) {
         ConcurrentHashMap<String, SetMessageRequestModeRequestBody> consumerGroup2ModeMap = messageRequestModeMap.get(topic);
         if (consumerGroup2ModeMap != null) {
@@ -92,4 +113,31 @@ public void decode(String jsonString) {
     public String encode(boolean prettyFormat) {
         return RemotingSerializable.toJson(this, prettyFormat);
     }
+
+    public ConcurrentHashMap<String, ConcurrentHashMap<String, SetMessageRequestModeRequestBody>> deepCopyMessageRequestModeMapSnapshot() {
+        ConcurrentHashMap<String, ConcurrentHashMap<String, SetMessageRequestModeRequestBody>> newOuterMap = new ConcurrentHashMap<>(this.messageRequestModeMap.size());
+        for (Map.Entry<String, ConcurrentHashMap<String, SetMessageRequestModeRequestBody>> topicEntry : this.messageRequestModeMap.entrySet()) {
+            String topic = topicEntry.getKey();
+            ConcurrentHashMap<String, SetMessageRequestModeRequestBody> originalInnerMap = topicEntry.getValue();
+
+            if (originalInnerMap != null) {
+                ConcurrentHashMap<String, SetMessageRequestModeRequestBody> newInnerMap = new ConcurrentHashMap<>(originalInnerMap.size());
+                for (Map.Entry<String, SetMessageRequestModeRequestBody> groupEntry : originalInnerMap.entrySet()) {
+                    String consumerGroup = groupEntry.getKey();
+                    SetMessageRequestModeRequestBody originalBody = groupEntry.getValue();
+
+                    if (originalBody != null) {
+                        try {
+                            SetMessageRequestModeRequestBody clonedBody = originalBody.clone();
+                            newInnerMap.put(consumerGroup, clonedBody);
+                        } catch (CloneNotSupportedException e) {
+                            newInnerMap.put(consumerGroup, originalBody);
+                        }
+                    }
+                }
+                newOuterMap.put(topic, newInnerMap);
+            }
+        }
+        return newOuterMap;
+    }
 }

broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java

@@ -17,6 +17,7 @@
 package org.apache.rocketmq.broker.offset;
 
 import com.google.common.collect.Maps;
+
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -25,6 +26,7 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicLong;
 
 import com.google.common.base.Strings;
@@ -36,6 +38,7 @@
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.topic.TopicValidator;
 import org.apache.rocketmq.logging.org.slf4j.Logger;
 import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
 import org.apache.rocketmq.remoting.protocol.DataVersion;
@@ -54,7 +57,19 @@ public class ConsumerOffsetManager extends ConfigManager {
         new ConcurrentHashMap<>(512);
 
     private final ConcurrentMap<String/* topic@group */, ConcurrentMap<Integer, Long>> pullOffsetTable =
-        new ConcurrentHashMap<>(512);
+            new ConcurrentHashMap<>(512);
+
+    private volatile ConcurrentMap<String/* topic@group */, ConcurrentMap<Integer, Long>> workingBuffer =
+            new ConcurrentHashMap<>(512);
+
+    private volatile ConcurrentMap<String/* topic@group */, ConcurrentMap<Integer, Long>> syncBuffer =
+            new ConcurrentHashMap<>(512);
+
+    private final Object bufferSwapLock = new Object();
+
+    private transient ScheduledExecutorService scheduledExecutorService;
+
+    private final transient AtomicLong lastSyncTimestamp = new AtomicLong(System.currentTimeMillis());
 
     protected transient BrokerController brokerController;
 
@@ -70,6 +85,36 @@ public ConsumerOffsetManager(BrokerController brokerController) {
     protected void removeConsumerOffset(String topicAtGroup) {
 
     }
+    private void triggerBufferSwapAndSync() {
+        synchronized (bufferSwapLock) {
+            ConcurrentMap<String, ConcurrentMap<Integer, Long>> oldSyncBuffer = this.syncBuffer;
+            this.syncBuffer = this.workingBuffer;
+            this.workingBuffer = new ConcurrentHashMap<>(512);
+            if (oldSyncBuffer != null && !oldSyncBuffer.isEmpty()) {
+                oldSyncBuffer.forEach((key, map) ->
+                        this.syncBuffer.computeIfAbsent(key, k -> new ConcurrentHashMap<>()).putAll(map));
+            }
+        }
+        if (!this.syncBuffer.isEmpty()) {
+            try {
+                this.brokerController.getMetadataChangeObserver().onUpdated(TopicValidator.RMQ_SYS_CONSUMER_OFFSET_SYNC, String.valueOf(dataVersion), this.syncBuffer);
+            } catch (Exception e) {
+                LOG.error("Failed to sync consumer offsets.", e);
+            } finally {
+                this.lastSyncTimestamp.set(System.currentTimeMillis());
+                this.syncBuffer.clear();
+            }
+        }
+    }
+
+    public void checkAndSync() {
+        boolean timeCondition = (System.currentTimeMillis() - lastSyncTimestamp.get()) > 1000;
+        boolean bufferHasData = !workingBuffer.isEmpty();
+        if (timeCondition && bufferHasData) {
+            LOG.debug("Consumer offset sync triggered by time interval.");
+            this.triggerBufferSwapAndSync();
+        }
+    }
 
     public void cleanOffset(String group) {
         Iterator<Entry<String, ConcurrentMap<Integer, Long>>> it = this.offsetTable.entrySet().iterator();
@@ -216,9 +261,12 @@ private void commitOffset(final String clientHost, final String key, final int q
                 LOG.warn("[NOTIFYME]update consumer offset less than store. clientHost={}, key={}, queueId={}, requestOffset={}, storeOffset={}", clientHost, key, queueId, offset, storeOffset);
             }
         }
+        ConcurrentMap<Integer, Long> workingMap = this.workingBuffer.computeIfAbsent(key, k -> new ConcurrentHashMap<>(32));
+        workingMap.put(queueId, offset);
         if (versionChangeCounter.incrementAndGet() % brokerController.getBrokerConfig().getConsumerOffsetUpdateVersionStep() == 0) {
             long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
             dataVersion.nextVersion(stateMachineVersion);
+            this.triggerBufferSwapAndSync();
         }
     }
 
@@ -370,7 +418,10 @@ public Map<Integer, Long> queryOffset(final String group, final String topic) {
     public void cloneOffset(final String srcGroup, final String destGroup, final String topic) {
         ConcurrentMap<Integer, Long> offsets = this.offsetTable.get(topic + TOPIC_GROUP_SEPARATOR + srcGroup);
         if (offsets != null) {
-            this.offsetTable.put(topic + TOPIC_GROUP_SEPARATOR + destGroup, new ConcurrentHashMap<>(offsets));
+            ConcurrentMap<Integer, Long> newOffsets = new ConcurrentHashMap<>(offsets);
+            this.offsetTable.put(topic + TOPIC_GROUP_SEPARATOR + destGroup, newOffsets);
+            this.workingBuffer.put(topic + TOPIC_GROUP_SEPARATOR + destGroup, new ConcurrentHashMap<>(newOffsets));
+            this.triggerBufferSwapAndSync();
         }
     }
 
@@ -418,13 +469,15 @@ public void removeOffset(final String group) {
             }
             return removed;
         };
-
         boolean clearOffset = deleteFunction.apply(this.offsetTable.entrySet().iterator());
+        boolean clearWorking = deleteFunction.apply(this.workingBuffer.entrySet().iterator());
+        boolean clearSync = deleteFunction.apply(this.syncBuffer.entrySet().iterator());
         boolean clearReset = deleteFunction.apply(this.resetOffsetTable.entrySet().iterator());
         boolean clearPull = deleteFunction.apply(this.pullOffsetTable.entrySet().iterator());
 
         LOG.info("Consumer offset manager clean group offset, groupName={}, " +
-            "offsetTable={}, resetOffsetTable={}, pullOffsetTable={}", group, clearOffset, clearReset, clearPull);
+                        "offsetTable={}, workingBuffer={}, syncBuffer={}, resetOffsetTable={}, pullOffsetTable={}",
+                group, clearOffset, clearWorking, clearSync, clearReset, clearPull);
     }
 
     public void assignResetOffset(String topic, String group, int queueId, long offset) {
@@ -443,6 +496,8 @@ public void assignResetOffset(String topic, String group, int queueId, long offs
         // 2, Our overriding here may get overridden by the client instantly in concurrent cases; But it still makes
         // sense in cases like clients are offline.
         offsetTable.computeIfAbsent(key, k -> Maps.newConcurrentMap()).put(queueId, offset);
+        workingBuffer.computeIfAbsent(key, k -> new ConcurrentHashMap<>()).put(queueId, offset);
+        this.triggerBufferSwapAndSync();
     }
 
     public boolean hasOffsetReset(String topic, String group, int queueId) {
@@ -463,4 +518,19 @@ public Long queryThenEraseResetOffset(String topic, String group, Integer queueI
             return map.remove(queueId);
         }
     }
+
+    public ConcurrentMap<String, ConcurrentMap<Integer, Long>> deepCopyOffsetTableSnapshot() {
+        ConcurrentMap<String, ConcurrentMap<Integer, Long>> newTable = new ConcurrentHashMap<>(this.offsetTable.size());
+        for (Map.Entry<String, ConcurrentMap<Integer, Long>> entry : this.offsetTable.entrySet()) {
+            String topicAtGroup = entry.getKey();
+            ConcurrentMap<Integer, Long> originalInnerMap = entry.getValue();
+
+            if (originalInnerMap != null) {
+                ConcurrentMap<Integer, Long> newInnerMap = new ConcurrentHashMap<>(originalInnerMap.size());
+                newInnerMap.putAll(originalInnerMap);
+                newTable.put(topicAtGroup, newInnerMap);
+            }
+        }
+        return newTable;
+    }
 }