[RIP-83] Lite Topic: A New Message Model

BUILDING

@@ -4,7 +4,7 @@ Build Instructions for Apache RocketMQ
 
 (1) Prerequisites
 
-    JDK 1.7+ is required in order to compile and run RocketMQ.
+    JDK 1.8+ is required in order to compile and run RocketMQ.
 
     RocketMQ utilizes Maven as a distribution management and packaging tool. Version 3.0.3 or later is required.
     Maven installation and configuration instructions can be found here:

README.md

@@ -5,8 +5,8 @@
 [![Maven Central][maven-central-image]][maven-central-url]
 [![Release][release-image]][release-url]
 [![License][license-image]][license-url]
-[![Average Time to Resolve An Issue][percentage-of-issues-still-open-image]][percentage-of-issues-still-open-url]
-[![Percentage of Issues Still Open][average-time-to-resolve-an-issue-image]][average-time-to-resolve-an-issue-url]
+[![Average Time to Resolve An Issue][average-time-to-resolve-an-issue-image]][average-time-to-resolve-an-issue-url]
+[![Percentage of Issues Still Open][percentage-of-issues-still-open-image]][percentage-of-issues-still-open-url]
 [![Twitter Follow][twitter-follow-image]][twitter-follow-url]
 
 **[Apache RocketMQ](https://rocketmq.apache.org) is a distributed messaging and streaming platform with low latency, high performance and reliability, trillion-level capacity and flexible scalability.**
@@ -49,21 +49,21 @@ $ java -version
 java version "1.8.0_121"
 ```
 
-For Windows users, click [here](https://dist.apache.org/repos/dist/release/rocketmq/5.3.3/rocketmq-all-5.3.3-bin-release.zip) to download the 5.3.3 RocketMQ binary release,
+For Windows users, click [here](https://dist.apache.org/repos/dist/release/rocketmq/5.3.4/rocketmq-all-5.3.4-bin-release.zip) to download the 5.3.4 RocketMQ binary release,
 unpack it to your local disk, such as `D:\rocketmq`.
 For macOS and Linux users, execute following commands:
 
 ```shell
 # Download release from the Apache mirror
-$ wget https://dist.apache.org/repos/dist/release/rocketmq/5.3.3/rocketmq-all-5.3.3-bin-release.zip
+$ wget https://dist.apache.org/repos/dist/release/rocketmq/5.3.4/rocketmq-all-5.3.4-bin-release.zip
 
 # Unpack the release
-$ unzip rocketmq-all-5.3.3-bin-release.zip
+$ unzip rocketmq-all-5.3.4-bin-release.zip
 ```
 
 Prepare a terminal and change to the extracted `bin` directory:
 ```shell
-$ cd rocketmq-all-5.3.3-bin-release/bin
+$ cd rocketmq-all-5.3.4-bin-release/bin
 ```
 
 **1) Start NameServer**
@@ -98,17 +98,17 @@ The Name Server boot success...
 For macOS and Linux users:
 ```shell
 ### start Broker
-$ nohup sh bin/mqbroker -n localhost:9876 &
+$ nohup sh mqbroker -n localhost:9876 &
 
 ### check whether Broker is successfully started, eg: Broker's IP is 192.168.1.2, Broker's name is broker-a
 $ tail -f ~/logs/rocketmqlogs/broker.log
-The broker[broker-a, 192.169.1.2:10911] boot success...
+The broker[broker-a, 192.168.1.2:10911] boot success...
 ```
 
 For Windows users:
 ```shell
 $ mqbroker.cmd -n localhost:9876
-The broker[broker-a, 192.169.1.2:10911] boot success...
+The broker[broker-a, 192.168.1.2:10911] boot success...
 ```
 
 ### Run RocketMQ in Docker
@@ -239,7 +239,7 @@ services.
 [maven-central-image]: https://maven-badges.herokuapp.com/maven-central/org.apache.rocketmq/rocketmq-all/badge.svg
 [maven-central-url]: http://search.maven.org/#search%7Cga%7C1%7Corg.apache.rocketmq
 [release-image]: https://img.shields.io/badge/release-download-orange.svg
-[release-url]: https://www.apache.org/licenses/LICENSE-2.0.html
+[release-url]: https://rocketmq.apache.org/download/
 [license-image]: https://img.shields.io/badge/license-Apache%202-4EB1BA.svg
 [license-url]: https://www.apache.org/licenses/LICENSE-2.0.html
 [average-time-to-resolve-an-issue-image]: http://isitmaintained.com/badge/resolution/apache/rocketmq.svg

WORKSPACE

@@ -71,7 +71,7 @@ maven_install(
         "org.bouncycastle:bcpkix-jdk15on:1.69",
         "com.google.code.gson:gson:2.8.9",
         "com.googlecode.concurrentlinkedhashmap:concurrentlinkedhashmap-lru:1.4.2",
-        "org.apache.rocketmq:rocketmq-proto:2.0.4",
+        "org.apache.rocketmq:rocketmq-proto:2.1.0",
         "com.google.protobuf:protobuf-java:3.20.1",
         "com.google.protobuf:protobuf-java-util:3.20.1",
         "com.conversantmedia:disruptor:1.2.10",

auth/src/main/java/org/apache/rocketmq/auth/authentication/provider/LocalAuthenticationMetadataProvider.java

@@ -46,14 +46,16 @@ public class LocalAuthenticationMetadataProvider implements AuthenticationMetada
 
     private LoadingCache<String, User> userCache;
 
+    protected ThreadPoolExecutor cacheRefreshExecutor;
+
     @Override
     public void initialize(AuthConfig authConfig, Supplier<?> metadataService) {
         this.storage = ConfigRocksDBStorage.getStore(authConfig.getAuthConfigPath() + File.separator + "users", false);
         if (!this.storage.start()) {
             throw new RuntimeException("Failed to load rocksdb for auth_user, please check whether it is occupied");
         }
 
-        ThreadPoolExecutor cacheRefreshExecutor = ThreadPoolMonitor.createAndMonitor(
+        this.cacheRefreshExecutor = ThreadPoolMonitor.createAndMonitor(
             1,
             1,
             1000 * 60,
@@ -144,6 +146,9 @@ public void shutdown() {
         if (this.storage != null) {
             this.storage.shutdown();
         }
+        if (this.cacheRefreshExecutor != null) {
+            this.cacheRefreshExecutor.shutdown();
+        }
     }
 
     private static class UserCacheLoader implements CacheLoader<String, User> {

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

@@ -31,6 +31,7 @@
 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;
@@ -124,6 +125,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());
@@ -192,11 +200,7 @@ public List<DefaultAuthorizationContext> build(ChannelHandlerContext context, Re
                     break;
                 case RequestCode.SEND_MESSAGE:
                     if (NamespaceUtil.isRetryTopic(fields.get(TOPIC))) {
-                        if (StringUtils.isNotBlank(fields.get(GROUP))) {
-                            group = Resource.ofGroup(fields.get(GROUP));
-                        } else {
-                            group = Resource.ofGroup(fields.get(TOPIC));
-                        }
+                        group = Resource.ofGroup(fields.get(TOPIC));
                         result.add(DefaultAuthorizationContext.of(subject, group, Action.SUB, sourceIp));
                     } else {
                         topic = Resource.ofTopic(fields.get(TOPIC));
@@ -206,11 +210,7 @@ public List<DefaultAuthorizationContext> build(ChannelHandlerContext context, Re
                 case RequestCode.SEND_MESSAGE_V2:
                 case RequestCode.SEND_BATCH_MESSAGE:
                     if (NamespaceUtil.isRetryTopic(fields.get(B))) {
-                        if (StringUtils.isNotBlank(fields.get(A))) {
-                            group = Resource.ofGroup(fields.get(A));
-                        } else {
-                            group = Resource.ofGroup(fields.get(B));
-                        }
+                        group = Resource.ofGroup(fields.get(B));
                         result.add(DefaultAuthorizationContext.of(subject, group, Action.SUB, sourceIp));
                     } else {
                         topic = Resource.ofTopic(fields.get(B));

auth/src/main/java/org/apache/rocketmq/auth/authorization/manager/AuthorizationMetadataManagerImpl.java

@@ -176,19 +176,26 @@ public CompletableFuture<Void> deleteAcl(Subject subject, PolicyType policyType,
 
     @Override
     public CompletableFuture<Acl> getAcl(Subject subject) {
-        CompletableFuture<? extends Subject> subjectFuture;
-        if (subject.isSubject(SubjectType.USER)) {
-            User user = (User) subject;
-            subjectFuture = this.getAuthenticationMetadataProvider().getUser(user.getUsername());
-        } else {
-            subjectFuture = CompletableFuture.completedFuture(subject);
-        }
-        return subjectFuture.thenCompose(sub -> {
-            if (sub == null) {
-                throw new AuthorizationException("The subject is not exist.");
+        try {
+            if (subject == null) {
+                throw new AuthorizationException("The subject is null.");
             }
-            return this.getAuthorizationMetadataProvider().getAcl(subject);
-        });
+            CompletableFuture<? extends Subject> subjectFuture;
+            if (subject.isSubject(SubjectType.USER)) {
+                User user = (User) subject;
+                subjectFuture = this.getAuthenticationMetadataProvider().getUser(user.getUsername());
+            } else {
+                subjectFuture = CompletableFuture.completedFuture(subject);
+            }
+            return subjectFuture.thenCompose(sub -> {
+                if (sub == null) {
+                    throw new AuthorizationException("The subject is not exist.");
+                }
+                return this.getAuthorizationMetadataProvider().getAcl(sub);
+            });
+        } catch (Exception e) {
+            return this.handleException(e);
+        }
     }
 
     @Override

auth/src/main/java/org/apache/rocketmq/auth/authorization/provider/LocalAuthorizationMetadataProvider.java

@@ -51,13 +51,15 @@ public class LocalAuthorizationMetadataProvider implements AuthorizationMetadata
 
     private LoadingCache<String, Acl> aclCache;
 
+    protected ThreadPoolExecutor cacheRefreshExecutor;
+
     @Override
     public void initialize(AuthConfig authConfig, Supplier<?> metadataService) {
         this.storage = ConfigRocksDBStorage.getStore(authConfig.getAuthConfigPath() + File.separator + "acls", false);
         if (!this.storage.start()) {
             throw new RuntimeException("Failed to load rocksdb for auth_acl, please check whether it is occupied.");
         }
-        ThreadPoolExecutor cacheRefreshExecutor = ThreadPoolMonitor.createAndMonitor(
+        this.cacheRefreshExecutor = ThreadPoolMonitor.createAndMonitor(
             1,
             1,
             1000 * 60,
@@ -172,6 +174,9 @@ public void shutdown() {
         if (this.storage != null) {
             this.storage.shutdown();
         }
+        if (this.cacheRefreshExecutor != null) {
+            this.cacheRefreshExecutor.shutdown();
+        }
     }
 
     private static class AclCacheLoader implements CacheLoader<String, Acl> {

auth/src/main/java/org/apache/rocketmq/auth/authorization/strategy/AbstractAuthorizationStrategy.java

@@ -16,8 +16,8 @@
  */
 package org.apache.rocketmq.auth.authorization.strategy;
 
-import java.util.ArrayList;
-import java.util.List;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.function.Supplier;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.auth.authorization.context.AuthorizationContext;
@@ -30,7 +30,7 @@
 public abstract class AbstractAuthorizationStrategy implements AuthorizationStrategy {
 
     protected final AuthConfig authConfig;
-    protected final List<String> authorizationWhitelist = new ArrayList<>();
+    protected final Set<String> authorizationWhiteSet = new HashSet<>();
     protected final AuthorizationProvider<AuthorizationContext> authorizationProvider;
 
     public AbstractAuthorizationStrategy(AuthConfig authConfig, Supplier<?> metadataService) {
@@ -42,7 +42,7 @@ public AbstractAuthorizationStrategy(AuthConfig authConfig, Supplier<?> metadata
         if (StringUtils.isNotBlank(authConfig.getAuthorizationWhitelist())) {
             String[] whitelist = StringUtils.split(authConfig.getAuthorizationWhitelist(), ",");
             for (String rpcCode : whitelist) {
-                this.authorizationWhitelist.add(StringUtils.trim(rpcCode));
+                this.authorizationWhiteSet.add(StringUtils.trim(rpcCode));
             }
         }
     }
@@ -57,7 +57,7 @@ public void doEvaluate(AuthorizationContext context) {
         if (this.authorizationProvider == null) {
             return;
         }
-        if (this.authorizationWhitelist.contains(context.getRpcCode())) {
+        if (this.authorizationWhiteSet.contains(context.getRpcCode())) {
             return;
         }
         try {

auth/src/test/java/org/apache/rocketmq/auth/authentication/provider/LocalAuthenticationMetadataProviderTest.java

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.auth.authentication.provider;
+
+import org.apache.rocketmq.auth.config.AuthConfig;
+import org.apache.rocketmq.auth.helper.AuthTestHelper;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class LocalAuthenticationMetadataProviderTest {
+
+    @Rule
+    public TemporaryFolder tempFolder = new TemporaryFolder();
+
+    @Test
+    public void testShutdownReleasesCacheExecutor() throws Exception {
+        AuthConfig authConfig = AuthTestHelper.createDefaultConfig();
+        authConfig.setAuthConfigPath(tempFolder.newFolder("auth-test").getAbsolutePath());
+
+        LocalAuthenticationMetadataProvider provider = new LocalAuthenticationMetadataProvider();
+        // Initialize provider to create the internal cache refresh executor
+        provider.initialize(authConfig, () -> null);
+
+        // After initialization, the executor should exist and not be shutdown
+        Assert.assertNotNull(provider.cacheRefreshExecutor);
+        Assert.assertFalse(provider.cacheRefreshExecutor.isShutdown());
+
+        // Shutdown provider should also shutdown its executor to release resources
+        provider.shutdown();
+
+        // Verify that the cache refresh executor has been shutdown
+        Assert.assertTrue(provider.cacheRefreshExecutor.isShutdown());
+    }
+}

auth/src/test/java/org/apache/rocketmq/auth/authorization/manager/AuthorizationMetadataManagerTest.java

@@ -203,6 +203,21 @@ public void getAcl() {
         });
     }
 
+    @Test
+    public void testGetAclWithNullSubject() {
+        if (MixAll.isMac()) {
+            return;
+        }
+        AuthorizationException authorizationException = Assert.assertThrows(AuthorizationException.class, () -> {
+            try {
+                this.authorizationMetadataManager.getAcl(null).join();
+            } catch (Exception e) {
+                AuthTestHelper.handleException(e);
+            }
+        });
+        Assert.assertEquals("The subject is null.", authorizationException.getMessage());
+    }
+
     @Test
     public void listAcl() {
         if (MixAll.isMac()) {

auth/src/test/java/org/apache/rocketmq/auth/authorization/provider/LocalAuthorizationMetadataProviderTest.java

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.auth.authorization.provider;
+
+import org.apache.rocketmq.auth.config.AuthConfig;
+import org.apache.rocketmq.auth.helper.AuthTestHelper;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class LocalAuthorizationMetadataProviderTest {
+
+    @Rule
+    public TemporaryFolder tempFolder = new TemporaryFolder();
+
+    @Test
+    public void testShutdownReleasesCacheExecutor() throws Exception {
+        AuthConfig authConfig = AuthTestHelper.createDefaultConfig();
+        authConfig.setAuthConfigPath(tempFolder.newFolder("auth-test").getAbsolutePath());
+
+        LocalAuthorizationMetadataProvider provider = new LocalAuthorizationMetadataProvider();
+        // Initialize provider to create the internal cache refresh executor
+        provider.initialize(authConfig, () -> null);
+
+        // After initialization, the executor should exist and not be shutdown
+        Assert.assertNotNull(provider.cacheRefreshExecutor);
+        Assert.assertFalse(provider.cacheRefreshExecutor.isShutdown());
+
+        // Shutdown provider should also shutdown its executor to release resources
+        provider.shutdown();
+
+        // Verify that the cache refresh executor has been shutdown
+        Assert.assertTrue(provider.cacheRefreshExecutor.isShutdown());
+    }
+}