Merge remote-tracking branch 'apache/develop' into fix-slave-sync

Author: RongtongJin

.github/PULL_REQUEST_TEMPLATE.md

@@ -4,7 +4,7 @@
 
 <!-- Please ensure that the related issue has already been created, and [link this pull request to that issue using keywords](<https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword>) to ensure automatic closure. -->
 
-Fixes #issue_id
+- Fixes #issue_id
 
 ### Brief Description
 

.gitignore

@@ -18,4 +18,5 @@ bazel-bin
 bazel-rocketmq
 bazel-testlogs
 .vscode
-MODULE.bazel.lock
+MODULE.bazel.lock
+*.flattened-pom.xml

WORKSPACE

@@ -41,7 +41,7 @@ maven_install(
     artifacts = [
         "junit:junit:4.13.2",
         "com.alibaba:fastjson:1.2.76",
-        "com.alibaba.fastjson2:fastjson2:2.0.43",
+        "com.alibaba.fastjson2:fastjson2:2.0.59",
         "org.hamcrest:hamcrest-library:1.3",
         "io.netty:netty-all:4.1.65.Final",
         "org.assertj:assertj-core:3.22.0",
@@ -107,12 +107,12 @@ maven_install(
         "com.fasterxml.jackson.core:jackson-databind:2.13.4.2",
         "com.adobe.testing:s3mock-junit4:2.11.0",
         "io.github.aliyunmq:rocketmq-grpc-netty-codec-haproxy:1.0.0",
-        "org.apache.rocketmq:rocketmq-rocksdb:1.0.2",
+        "org.apache.rocketmq:rocketmq-rocksdb:1.0.6",
         "com.alipay.sofa:jraft-core:1.3.14",
         "com.alipay.sofa:hessian:3.3.6",
         "io.netty:netty-tcnative-boringssl-static:2.0.48.Final",
         "org.mockito:mockito-junit-jupiter:4.11.0",
-        "com.alibaba.fastjson2:fastjson2:2.0.43",
+        "com.alibaba.fastjson2:fastjson2:2.0.59",
         "org.junit.jupiter:junit-jupiter-api:5.9.1",
     ],
     fetch_sources = True,

auth/src/main/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilder.java

@@ -31,13 +31,15 @@
 import apache.rocketmq.v2.Subscription;
 import apache.rocketmq.v2.SubscriptionEntry;
 import apache.rocketmq.v2.TelemetryCommand;
+import apache.rocketmq.v2.SyncLiteSubscriptionRequest;
 import com.google.protobuf.GeneratedMessageV3;
 import io.grpc.Metadata;
 import io.netty.channel.ChannelHandlerContext;
 import java.lang.reflect.Field;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.List;
 import org.apache.commons.collections.CollectionUtils;
@@ -84,6 +86,8 @@ public class DefaultAuthorizationContextBuilder implements AuthorizationContextB
     private static final String B = "b";
     private static final String CONSUMER_GROUP = "consumerGroup";
     private final AuthConfig authConfig;
+    private static final EnumSet<ClientType> CONSUMER_CLIENT_TYPES =
+            EnumSet.of(ClientType.PUSH_CONSUMER, ClientType.SIMPLE_CONSUMER, ClientType.PULL_CONSUMER);
 
     private final RequestHeaderRegistry requestHeaderRegistry;
 
@@ -124,6 +128,13 @@ public List<DefaultAuthorizationContext> build(Metadata metadata, GeneratedMessa
             }
             result = newSubContexts(metadata, request.getGroup(), request.getMessageQueue().getTopic());
         }
+        if (message instanceof SyncLiteSubscriptionRequest) {
+            SyncLiteSubscriptionRequest request = (SyncLiteSubscriptionRequest) message;
+            if (request.getLiteTopicSetCount() <= 0) {
+                return null;
+            }
+            result = newSubContexts(metadata, request.getGroup(), request.getTopic());
+        }
         if (message instanceof AckMessageRequest) {
             AckMessageRequest request = (AckMessageRequest) message;
             result = newSubContexts(metadata, request.getGroup(), request.getTopic());
@@ -430,8 +441,7 @@ private static List<DefaultAuthorizationContext> newContext(Metadata metadata, T
     }
 
     private boolean isConsumerClientType(ClientType clientType) {
-        return Arrays.asList(ClientType.PUSH_CONSUMER, ClientType.SIMPLE_CONSUMER, ClientType.PULL_CONSUMER)
-            .contains(clientType);
+        return CONSUMER_CLIENT_TYPES.contains(clientType);
     }
 
     private static List<DefaultAuthorizationContext> newPubContext(Metadata metadata, apache.rocketmq.v2.Resource topic) {

broker/pom.xml

@@ -72,7 +72,7 @@
         </dependency>
         <dependency>
             <groupId>org.bouncycastle</groupId>
-            <artifactId>bcpkix-jdk15on</artifactId>
+            <artifactId>bcpkix-jdk18on</artifactId>
         </dependency>
         <dependency>
             <groupId>com.googlecode.concurrentlinkedhashmap</groupId>

broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java

@@ -73,6 +73,14 @@
 import org.apache.rocketmq.broker.filter.CommitLogDispatcherCalcBitMap;
 import org.apache.rocketmq.broker.filter.ConsumerFilterManager;
 import org.apache.rocketmq.broker.latency.BrokerFastFailure;
+import org.apache.rocketmq.broker.lite.AbstractLiteLifecycleManager;
+import org.apache.rocketmq.broker.lite.LiteEventDispatcher;
+import org.apache.rocketmq.broker.lite.LiteSubscriptionRegistry;
+import org.apache.rocketmq.broker.lite.LiteSubscriptionRegistryImpl;
+import org.apache.rocketmq.broker.lite.LiteLifecycleManager;
+import org.apache.rocketmq.broker.lite.LiteSharding;
+import org.apache.rocketmq.broker.lite.LiteShardingImpl;
+import org.apache.rocketmq.broker.lite.RocksDBLiteLifecycleManager;
 import org.apache.rocketmq.broker.longpolling.LmqPullRequestHoldService;
 import org.apache.rocketmq.broker.longpolling.NotifyMessageArrivingListener;
 import org.apache.rocketmq.broker.longpolling.PullRequestHoldService;
@@ -93,10 +101,13 @@
 import org.apache.rocketmq.broker.processor.ClientManageProcessor;
 import org.apache.rocketmq.broker.processor.ConsumerManageProcessor;
 import org.apache.rocketmq.broker.processor.EndTransactionProcessor;
+import org.apache.rocketmq.broker.processor.LiteManagerProcessor;
+import org.apache.rocketmq.broker.processor.LiteSubscriptionCtlProcessor;
 import org.apache.rocketmq.broker.processor.NotificationProcessor;
 import org.apache.rocketmq.broker.processor.PeekMessageProcessor;
 import org.apache.rocketmq.broker.processor.PollingInfoProcessor;
 import org.apache.rocketmq.broker.processor.PopInflightMessageCounter;
+import org.apache.rocketmq.broker.processor.PopLiteMessageProcessor;
 import org.apache.rocketmq.broker.processor.PopMessageProcessor;
 import org.apache.rocketmq.broker.processor.PullMessageProcessor;
 import org.apache.rocketmq.broker.processor.QueryAssignmentProcessor;
@@ -206,12 +217,19 @@ public class BrokerController {
     protected final PullMessageProcessor pullMessageProcessor;
     protected final PeekMessageProcessor peekMessageProcessor;
     protected final PopMessageProcessor popMessageProcessor;
+    protected final PopLiteMessageProcessor popLiteMessageProcessor;
     protected final AckMessageProcessor ackMessageProcessor;
     protected final ChangeInvisibleTimeProcessor changeInvisibleTimeProcessor;
     protected final NotificationProcessor notificationProcessor;
     protected final PollingInfoProcessor pollingInfoProcessor;
     protected final QueryAssignmentProcessor queryAssignmentProcessor;
     protected final ClientManageProcessor clientManageProcessor;
+    protected final LiteSubscriptionCtlProcessor liteSubscriptionCtlProcessor;
+    protected final LiteSharding liteSharding;
+    protected final AbstractLiteLifecycleManager liteLifecycleManager;
+    protected final LiteSubscriptionRegistry liteSubscriptionRegistry;
+    protected final LiteEventDispatcher liteEventDispatcher;
+    protected final LiteManagerProcessor liteManagerProcessor;
     protected final SendMessageProcessor sendMessageProcessor;
     protected final RecallMessageProcessor recallMessageProcessor;
     protected final ReplyMessageProcessor replyMessageProcessor;
@@ -376,18 +394,27 @@ public BrokerController(
         this.topicQueueMappingManager = new TopicQueueMappingManager(this);
         this.authenticationMetadataManager = AuthenticationFactory.getMetadataManager(this.authConfig);
         this.authorizationMetadataManager = AuthorizationFactory.getMetadataManager(this.authConfig);
+        this.topicRouteInfoManager = new TopicRouteInfoManager(this);
+        this.liteSharding = new LiteShardingImpl(this, this.topicRouteInfoManager);
+        this.liteLifecycleManager = this.messageStoreConfig.isEnableRocksDBStore() ?
+            new RocksDBLiteLifecycleManager(this, this.liteSharding) : new LiteLifecycleManager(this, this.liteSharding);
+        this.liteSubscriptionRegistry = new LiteSubscriptionRegistryImpl(this, liteLifecycleManager);
+        this.liteSubscriptionCtlProcessor = new LiteSubscriptionCtlProcessor(this, liteSubscriptionRegistry);
+        this.liteEventDispatcher = new LiteEventDispatcher(this, this.liteSubscriptionRegistry, this.liteLifecycleManager);
+        this.liteManagerProcessor = new LiteManagerProcessor(this, liteLifecycleManager, liteSharding);
         this.pullMessageProcessor = new PullMessageProcessor(this);
         this.peekMessageProcessor = new PeekMessageProcessor(this);
         this.pullRequestHoldService = messageStoreConfig.isEnableLmq() ? new LmqPullRequestHoldService(this) : new PullRequestHoldService(this);
         this.popMessageProcessor = new PopMessageProcessor(this);
+        this.popLiteMessageProcessor = new PopLiteMessageProcessor(this, this.liteEventDispatcher);
         this.notificationProcessor = new NotificationProcessor(this);
         this.pollingInfoProcessor = new PollingInfoProcessor(this);
         this.ackMessageProcessor = new AckMessageProcessor(this);
         this.changeInvisibleTimeProcessor = new ChangeInvisibleTimeProcessor(this);
         this.sendMessageProcessor = new SendMessageProcessor(this);
         this.recallMessageProcessor = new RecallMessageProcessor(this);
         this.replyMessageProcessor = new ReplyMessageProcessor(this);
-        this.messageArrivingListener = new NotifyMessageArrivingListener(this.pullRequestHoldService, this.popMessageProcessor, this.notificationProcessor);
+        this.messageArrivingListener = new NotifyMessageArrivingListener(this.pullRequestHoldService, this.popMessageProcessor, this.notificationProcessor, this.liteEventDispatcher);
         this.consumerIdsChangeListener = new DefaultConsumerIdsChangeListener(this);
         this.consumerManager = new ConsumerManager(this.consumerIdsChangeListener, this.brokerStatsManager, this.brokerConfig);
         this.producerManager = new ProducerManager(this.brokerStatsManager);
@@ -466,8 +493,6 @@ public boolean online(String instanceId, String group, String topic) {
 
         this.escapeBridge = new EscapeBridge(this);
 
-        this.topicRouteInfoManager = new TopicRouteInfoManager(this);
-
         if (this.brokerConfig.isEnableSlaveActingMaster() && !this.brokerConfig.isSkipPreOnline()) {
             this.brokerPreOnlineService = new BrokerPreOnlineService(this);
         }
@@ -950,6 +975,8 @@ public boolean recoverAndInitService() throws CloneNotSupportedException {
 
             initialRequestPipeline();
 
+            initLiteService();
+
             if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {
                 // Register a listener to reload SslContext
                 try {
@@ -1045,6 +1072,21 @@ public PutMessageResult executeBeforePutMessage(MessageExt msg) {
             }
         });
 
+        putMessageHookList.add(new PutMessageHook() {
+            @Override
+            public String hookName() {
+                return "handleLmqQuota";
+            }
+
+            @Override
+            public PutMessageResult executeBeforePutMessage(MessageExt msg) {
+                if (msg instanceof MessageExtBrokerInner) {
+                    return HookUtils.handleLmqQuota(BrokerController.this, (MessageExtBrokerInner) msg);
+                }
+                return null;
+            }
+        });
+
         SendMessageBackHook sendMessageBackHook = new SendMessageBackHook() {
             @Override
             public boolean executeSendMessageBack(List<MessageExt> msgList, String brokerName, String brokerAddr) {
@@ -1111,6 +1153,11 @@ private void initialRequestPipeline() {
         }
     }
 
+    private void initLiteService() {
+        this.liteEventDispatcher.init();
+        this.liteLifecycleManager.init();
+    }
+
     public void registerProcessor() {
         RemotingServer remotingServer = remotingServerMap.get(TCP_REMOTING_SERVER);
         RemotingServer fastRemotingServer = remotingServerMap.get(FAST_REMOTING_SERVER);
@@ -1145,6 +1192,7 @@ public void registerProcessor() {
          * PopMessageProcessor
          */
         remotingServer.registerProcessor(RequestCode.POP_MESSAGE, this.popMessageProcessor, this.pullMessageExecutor);
+        remotingServer.registerProcessor(RequestCode.POP_LITE_MESSAGE, this.popLiteMessageProcessor, this.pullMessageExecutor);
 
         /**
          * AckMessageProcessor
@@ -1196,10 +1244,12 @@ public void registerProcessor() {
         remotingServer.registerProcessor(RequestCode.HEART_BEAT, clientManageProcessor, this.heartbeatExecutor);
         remotingServer.registerProcessor(RequestCode.UNREGISTER_CLIENT, clientManageProcessor, this.clientManageExecutor);
         remotingServer.registerProcessor(RequestCode.CHECK_CLIENT_CONFIG, clientManageProcessor, this.clientManageExecutor);
+        remotingServer.registerProcessor(RequestCode.LITE_SUBSCRIPTION_CTL, liteSubscriptionCtlProcessor, this.clientManageExecutor);
 
         fastRemotingServer.registerProcessor(RequestCode.HEART_BEAT, clientManageProcessor, this.heartbeatExecutor);
         fastRemotingServer.registerProcessor(RequestCode.UNREGISTER_CLIENT, clientManageProcessor, this.clientManageExecutor);
         fastRemotingServer.registerProcessor(RequestCode.CHECK_CLIENT_CONFIG, clientManageProcessor, this.clientManageExecutor);
+        fastRemotingServer.registerProcessor(RequestCode.LITE_SUBSCRIPTION_CTL, liteSubscriptionCtlProcessor, this.clientManageExecutor);
 
         /**
          * ConsumerManageProcessor
@@ -1227,6 +1277,23 @@ public void registerProcessor() {
         remotingServer.registerProcessor(RequestCode.END_TRANSACTION, endTransactionProcessor, this.endTransactionExecutor);
         fastRemotingServer.registerProcessor(RequestCode.END_TRANSACTION, endTransactionProcessor, this.endTransactionExecutor);
 
+        /*
+         * lite admin
+         */
+        remotingServer.registerProcessor(RequestCode.GET_BROKER_LITE_INFO, liteManagerProcessor, adminBrokerExecutor);
+        remotingServer.registerProcessor(RequestCode.GET_PARENT_TOPIC_INFO, liteManagerProcessor, adminBrokerExecutor);
+        remotingServer.registerProcessor(RequestCode.GET_LITE_TOPIC_INFO, liteManagerProcessor, adminBrokerExecutor);
+        remotingServer.registerProcessor(RequestCode.GET_LITE_CLIENT_INFO, liteManagerProcessor, adminBrokerExecutor);
+        remotingServer.registerProcessor(RequestCode.GET_LITE_GROUP_INFO, liteManagerProcessor, adminBrokerExecutor);
+        remotingServer.registerProcessor(RequestCode.TRIGGER_LITE_DISPATCH, liteManagerProcessor, adminBrokerExecutor);
+
+        fastRemotingServer.registerProcessor(RequestCode.GET_BROKER_LITE_INFO, liteManagerProcessor, adminBrokerExecutor);
+        fastRemotingServer.registerProcessor(RequestCode.GET_PARENT_TOPIC_INFO, liteManagerProcessor, adminBrokerExecutor);
+        fastRemotingServer.registerProcessor(RequestCode.GET_LITE_TOPIC_INFO, liteManagerProcessor, adminBrokerExecutor);
+        fastRemotingServer.registerProcessor(RequestCode.GET_LITE_CLIENT_INFO, liteManagerProcessor, adminBrokerExecutor);
+        fastRemotingServer.registerProcessor(RequestCode.GET_LITE_GROUP_INFO, liteManagerProcessor, adminBrokerExecutor);
+        fastRemotingServer.registerProcessor(RequestCode.TRIGGER_LITE_DISPATCH, liteManagerProcessor, adminBrokerExecutor);
+
         /*
          * Default
          */
@@ -1409,6 +1476,10 @@ public PopMessageProcessor getPopMessageProcessor() {
         return popMessageProcessor;
     }
 
+    public PopLiteMessageProcessor getPopLiteMessageProcessor() {
+        return popLiteMessageProcessor;
+    }
+
     public NotificationProcessor getNotificationProcessor() {
         return notificationProcessor;
     }
@@ -1437,6 +1508,14 @@ public ChangeInvisibleTimeProcessor getChangeInvisibleTimeProcessor() {
         return changeInvisibleTimeProcessor;
     }
 
+    public LiteSubscriptionRegistry getLiteSubscriptionRegistry() {
+        return liteSubscriptionRegistry;
+    }
+
+    public AbstractLiteLifecycleManager getLiteLifecycleManager() {
+        return liteLifecycleManager;
+    }
+
     protected void shutdownBasicService() {
 
         shutdown = true;
@@ -1474,6 +1553,13 @@ protected void shutdownBasicService() {
             this.popMessageProcessor.getPopLongPollingService().shutdown();
         }
 
+        if (this.popLiteMessageProcessor != null) {
+            this.popLiteMessageProcessor.stopPopLiteLockManager();
+            if (this.popLiteMessageProcessor.getPopLiteLongPollingService() != null) {
+                this.popLiteMessageProcessor.getPopLiteLongPollingService().shutdown();
+            }
+        }
+
         if (this.popMessageProcessor.getQueueLockManager() != null) {
             this.popMessageProcessor.getQueueLockManager().shutdown();
         }
@@ -1637,6 +1723,18 @@ protected void shutdownBasicService() {
             this.coldDataCgCtrService.shutdown();
         }
 
+        if (this.liteEventDispatcher != null) {
+            this.liteEventDispatcher.shutdown();
+        }
+
+        if (this.liteLifecycleManager != null) {
+            this.liteLifecycleManager.shutdown();
+        }
+
+        if (this.liteSubscriptionRegistry != null) {
+            this.liteSubscriptionRegistry.shutdown();
+        }
+
         shutdownScheduledExecutorService(this.syncBrokerMemberGroupExecutorService);
         shutdownScheduledExecutorService(this.brokerHeartbeatExecutorService);
 
@@ -1777,6 +1875,13 @@ protected void startBasicService() throws Exception {
             this.popMessageProcessor.getQueueLockManager().start();
         }
 
+        if (this.popLiteMessageProcessor != null) {
+            this.popLiteMessageProcessor.startPopLiteLockManager();
+            if (this.popLiteMessageProcessor.getPopLiteLongPollingService() != null) {
+                this.popLiteMessageProcessor.getPopLiteLongPollingService().start();
+            }
+        }
+
         if (this.ackMessageProcessor != null) {
             if (brokerConfig.isPopConsumerFSServiceInit()) {
                 this.ackMessageProcessor.startPopReviveService();
@@ -1838,6 +1943,18 @@ protected void startBasicService() throws Exception {
         if (this.coldDataCgCtrService != null) {
             this.coldDataCgCtrService.start();
         }
+
+        if (this.liteEventDispatcher != null) {
+            this.liteEventDispatcher.start();
+        }
+
+        if (this.liteLifecycleManager != null) {
+            this.liteLifecycleManager.start();
+        }
+
+        if (this.liteSubscriptionRegistry != null) {
+            this.liteSubscriptionRegistry.start();
+        }
     }
 
     public void start() throws Exception {
@@ -2699,4 +2816,8 @@ public ConfigContext getConfigContext() {
     public void setConfigContext(ConfigContext configContext) {
         this.configContext = configContext;
     }
+
+    public LiteEventDispatcher getLiteEventDispatcher() {
+        return liteEventDispatcher;
+    }
 }