Merge branch 'refs/heads/develop' into dev-tofastjson2

Author: yx9o

CONTRIBUTING.md

@@ -3,7 +3,7 @@
 We are always very happy to have contributions, whether for trivial cleanups or big new features.
 We want to have high quality, well documented codes for each programming language, as well as the surrounding [ecosystem](https://github.com/apache/rocketmq-externals) of integration tools that people use with RocketMQ.
 
-Nor is code the only way to contribute to the project. We strongly value documentation, integration with other project, and gladly accept improvements for these aspects.
+Nor is code the only way to contribute to the project. We strongly value documentation, integration with other projects, and gladly accept improvements for these aspects.
 
 Recommend reading:
  * [Contributors Tech Guide](http://www.apache.org/dev/contributors)
@@ -34,15 +34,15 @@ More details of squash can be found at [stackoverflow](https://stackoverflow.com
 
 We are always interested in adding new contributors. What we look for are series of contributions, good taste and ongoing interest in the project. If you are interested in becoming a committer, please let one of the existing committers know and they can help you walk through the process.
 
-Nowadays,we have several important contribution points:
+Nowadays, we have several important contribution points:
 #### Wiki & JavaDoc
 #### RocketMQ SDK(C++\.Net\Php\Python\Go\Node.js)
 #### RocketMQ Connectors
 
-##### Prerequisite
-If you want to contribute the above listing points, you must abide our some prerequisites:
+##### Prerequisites
+If you want to contribute to the above listed points, you must abide by the following prerequisites:
 
-###### Readability - API must have Javadoc, some very important methods also must have javadoc
-###### Testability - 80% above unit test coverage about main process
-###### Maintainability - Comply with our [checkstyle spec](style/rmq_checkstyle.xml), and at least 3 month update frequency
+###### Readability - API must have Javadoc, and some very important methods must also have Javadoc
+###### Testability - Above 80% unit test coverage for the main process
+###### Maintainability - Comply with our [checkstyle spec](style/rmq_checkstyle.xml), and at least a 3-month update frequency
 ###### Deployability - We encourage you to deploy into [maven repository](http://search.maven.org/)

README.md

@@ -16,8 +16,8 @@ It offers a variety of features:
 
 * Messaging patterns including publish/subscribe, request/reply and streaming
 * Financial grade transactional message
-* Built-in fault tolerance and high availability configuration options base on [DLedger Controller](docs/en/controller/quick_start.md)
-* Built-in message tracing capability, also support opentracing
+* Built-in fault tolerance and high availability configuration options based on [DLedger Controller](docs/en/controller/quick_start.md)
+* Built-in message tracing capability, also supports opentracing
 * Versatile big-data and streaming ecosystem integration
 * Message retroactivity by time or offset
 * Reliable FIFO and strict ordered messaging in the same queue
@@ -80,7 +80,7 @@ $ tail -f ~/logs/rocketmqlogs/namesrv.log
 The Name Server boot success...
 ```
 
-For Windows users, you need set environment variables first:
+For Windows users, you need to set environment variables first:
 - From the desktop, right click the Computer icon.
 - Choose Properties from the context menu.
 - Click the Advanced system settings link.
@@ -139,7 +139,7 @@ Before your operations, make sure that `kubectl` and related kubeconfig file ins
 $ git clone https://github.com/apache/rocketmq-operator
 $ cd rocketmq-operator && make deploy
 
-### check whether CRDs is successfully installed
+### check whether CRDs are successfully installed
 $ kubectl get crd | grep rocketmq.apache.org
 brokers.rocketmq.apache.org                 2022-05-12T09:23:18Z
 consoles.rocketmq.apache.org                2022-05-12T09:23:19Z
@@ -156,7 +156,7 @@ rocketmq-operator-6f65c77c49-8hwmj   1/1     Running   0          93s
 ### create RocketMQ cluster resource
 $ cd example && kubectl create -f rocketmq_v1alpha1_rocketmq_cluster.yaml
 
-### check whether cluster resources is running
+### check whether cluster resources are running
 $ kubectl get sts
 NAME                 READY   AGE
 broker-0-master      1/1     107m
@@ -182,7 +182,7 @@ name-service         1/1     107m
 * [RocketMQ Dashboard](https://github.com/apache/rocketmq-dashboard): Operation and maintenance console of Apache RocketMQ.
 * [RocketMQ Connect](https://github.com/apache/rocketmq-connect): A tool for scalably and reliably streaming data between Apache RocketMQ and other systems.
 * [RocketMQ MQTT](https://github.com/apache/rocketmq-mqtt): A new MQTT protocol architecture model, based on which Apache RocketMQ can better support messages from terminals such as IoT devices and Mobile APP.
-* [RocketMQ EventBridge](https://github.com/apache/rocketmq-eventbridge): EventBridge make it easier to build a event-driven application.
+* [RocketMQ EventBridge](https://github.com/apache/rocketmq-eventbridge): EventBridge makes it easier to build an event-driven application.
 * [RocketMQ Incubating Community Projects](https://github.com/apache/rocketmq-externals): Incubator community projects of Apache RocketMQ, including [logappender](https://github.com/apache/rocketmq-externals/tree/master/logappender), [rocketmq-ansible](https://github.com/apache/rocketmq-externals/tree/master/rocketmq-ansible), [rocketmq-beats-integration](https://github.com/apache/rocketmq-externals/tree/master/rocketmq-beats-integration), [rocketmq-cloudevents-binding](https://github.com/apache/rocketmq-externals/tree/master/rocketmq-cloudevents-binding), etc.
 * [RocketMQ Site](https://github.com/apache/rocketmq-site): The repository for Apache RocketMQ website.
 * [RocketMQ E2E](https://github.com/apache/rocketmq-e2e): A project for testing Apache RocketMQ, including end-to-end, performance, compatibility tests.

auth/src/main/java/org/apache/rocketmq/auth/authentication/manager/AuthenticationMetadataManagerImpl.java

@@ -208,14 +208,14 @@ private void handleException(Exception e, CompletableFuture<?> result) {
 
     private AuthenticationMetadataProvider getAuthenticationMetadataProvider() {
         if (authenticationMetadataProvider == null) {
-            throw new IllegalStateException("The authenticationMetadataProvider is not configured");
+            throw new IllegalStateException("The authenticationMetadataProvider is not configured.");
         }
         return authenticationMetadataProvider;
     }
 
     private AuthorizationMetadataProvider getAuthorizationMetadataProvider() {
         if (authorizationMetadataProvider == null) {
-            throw new IllegalStateException("The authorizationMetadataProvider is not configured");
+            throw new IllegalStateException("The authorizationMetadataProvider is not configured.");
         }
         return authorizationMetadataProvider;
     }

auth/src/main/java/org/apache/rocketmq/auth/authentication/strategy/AbstractAuthenticationStrategy.java

@@ -16,8 +16,8 @@
  */
 package org.apache.rocketmq.auth.authentication.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.authentication.context.AuthenticationContext;
@@ -30,7 +30,7 @@
 public abstract class AbstractAuthenticationStrategy implements AuthenticationStrategy {
 
     protected final AuthConfig authConfig;
-    protected final List<String> authenticationWhitelist = new ArrayList<>();
+    protected final Set<String> authenticationWhiteSet = new HashSet<>();
     protected final AuthenticationProvider<AuthenticationContext> authenticationProvider;
 
     public AbstractAuthenticationStrategy(AuthConfig authConfig, Supplier<?> metadataService) {
@@ -42,7 +42,7 @@ public AbstractAuthenticationStrategy(AuthConfig authConfig, Supplier<?> metadat
         if (StringUtils.isNotBlank(authConfig.getAuthenticationWhitelist())) {
             String[] whitelist = StringUtils.split(authConfig.getAuthenticationWhitelist(), ",");
             for (String rpcCode : whitelist) {
-                this.authenticationWhitelist.add(StringUtils.trim(rpcCode));
+                this.authenticationWhiteSet.add(StringUtils.trim(rpcCode));
             }
         }
     }
@@ -57,7 +57,7 @@ protected void doEvaluate(AuthenticationContext context) {
         if (this.authenticationProvider == null) {
             return;
         }
-        if (this.authenticationWhitelist.contains(context.getRpcCode())) {
+        if (this.authenticationWhiteSet.contains(context.getRpcCode())) {
             return;
         }
         try {

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

@@ -269,15 +269,15 @@ private <T> CompletableFuture<T> handleException(Exception e) {
     }
 
     private AuthenticationMetadataProvider getAuthenticationMetadataProvider() {
-        if (authorizationMetadataProvider == null) {
+        if (authenticationMetadataProvider == null) {
             throw new IllegalStateException("The authenticationMetadataProvider is not configured.");
         }
         return authenticationMetadataProvider;
     }
 
     private AuthorizationMetadataProvider getAuthorizationMetadataProvider() {
-        if (authenticationMetadataProvider == null) {
-            throw new IllegalStateException("The authenticationMetadataProvider is not configured.");
+        if (authorizationMetadataProvider == null) {
+            throw new IllegalStateException("The authorizationMetadataProvider is not configured.");
         }
         return authorizationMetadataProvider;
     }

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

@@ -148,7 +148,7 @@ public CompletableFuture<List<Acl>> listAcl(String subjectFilter, String resourc
                     if (CollectionUtils.isEmpty(entries)) {
                         continue;
                     }
-                    if (StringUtils.isNotBlank(resourceFilter) && !subjectKey.contains(resourceFilter)) {
+                    if (StringUtils.isNotBlank(resourceFilter)) {
                         entries.removeIf(entry -> !entry.toResourceStr().contains(resourceFilter));
                     }
                     if (CollectionUtils.isEmpty(entries)) {

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

@@ -28,6 +28,7 @@
 import org.apache.rocketmq.auth.authorization.factory.AuthorizationFactory;
 import org.apache.rocketmq.auth.authorization.model.Acl;
 import org.apache.rocketmq.auth.authorization.model.Policy;
+import org.apache.rocketmq.auth.authorization.model.PolicyEntry;
 import org.apache.rocketmq.auth.authorization.model.Resource;
 import org.apache.rocketmq.auth.config.AuthConfig;
 import org.apache.rocketmq.auth.helper.AuthTestHelper;
@@ -220,6 +221,10 @@ public void listAcl() {
             "192.168.0.0/24,10.10.0.0/24", Decision.ALLOW);
         this.authorizationMetadataManager.createAcl(acl2).join();
 
+        Acl acl3 = AuthTestHelper.buildAcl("User:test-2", "Topic:acl-2,Group:acl-2", "PUB,SUB",
+            "192.168.0.0/24,10.10.0.0/24", Decision.ALLOW);
+        this.authorizationMetadataManager.createAcl(acl3).join();
+
         List<Acl> acls1 = this.authorizationMetadataManager.listAcl(null, null).join();
         Assert.assertEquals(acls1.size(), 2);
 
@@ -235,13 +240,21 @@ public void listAcl() {
 
         List<Acl> acls5 = this.authorizationMetadataManager.listAcl(null, "test-1").join();
         Assert.assertEquals(acls5.size(), 1);
-        Assert.assertEquals(acls4.get(0).getPolicy(PolicyType.CUSTOM).getEntries().size(), 1);
+        Assert.assertEquals(acls5.get(0).getPolicy(PolicyType.CUSTOM).getEntries().size(), 2);
 
         List<Acl> acls6 = this.authorizationMetadataManager.listAcl("User:abc", null).join();
         Assert.assertTrue(CollectionUtils.isEmpty(acls6));
 
         List<Acl> acls7 = this.authorizationMetadataManager.listAcl(null, "Topic:abc").join();
         Assert.assertTrue(CollectionUtils.isEmpty(acls7));
+
+        List<Acl> acls8 = this.authorizationMetadataManager.listAcl("test-2", "test-2").join();
+        Assert.assertEquals(acls8.size(), 1);
+        List<PolicyEntry> policyEntries = acls8.get(0).getPolicy(PolicyType.CUSTOM).getEntries();
+        Assert.assertEquals(policyEntries.size(), 2);
+        for (PolicyEntry policyEntry : policyEntries) {
+            Assert.assertTrue(policyEntry.toResourceStr().contains("test-2"));
+        }
     }
 
     private void clearAllUsers() {

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

@@ -124,6 +124,10 @@ public static ConfigContext configFileToConfigContext(String filePath) throws Ex
         NettyClientConfig nettyClientConfig = new NettyClientConfig();
         MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
         AuthConfig authConfig = new AuthConfig();
+
+        nettyServerConfig.setListenPort(10911);
+        messageStoreConfig.setHaListenPort(0);
+
         Properties properties = new Properties();
         if (StringUtils.isNotBlank(filePath)) {
             systemConfigFileHelper.setFile(filePath);
@@ -161,9 +165,6 @@ public static BrokerController buildBrokerController(ConfigContext configContext
         AuthConfig authConfig = configContext.getAuthConfig();
         Properties properties = configContext.getProperties();
 
-        nettyServerConfig.setListenPort(10911);
-        configContext.getMessageStoreConfig().setHaListenPort(0);
-
         if (null == brokerConfig.getRocketmqHome()) {
             System.out.printf("Please set the %s variable in your environment " +
                 "to match the location of the RocketMQ installation", MixAll.ROCKETMQ_HOME_ENV);

broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java

@@ -356,7 +356,7 @@ public RemotingCommand sendMessage(final ChannelHandlerContext ctx,
             }
             handlePutMessageResult(putMessageResult, response, request, msgInner, responseHeader, sendMessageContext, ctx, queueIdInt, beginTimeMillis, mappingContext, BrokerMetricsManager.getMessageType(requestHeader));
             // record the transaction metrics
-            if (putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK && putMessageResult.getAppendMessageResult().isOk()) {
+            if (sendTransactionPrepareMessage && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK && putMessageResult.getAppendMessageResult().isOk()) {
                 this.brokerController.getTransactionalMessageService().getTransactionMetrics().addAndGet(msgInner.getProperty(MessageConst.PROPERTY_REAL_TOPIC), 1);
             }
             sendMessageCallback.onComplete(sendMessageContext, response);

broker/src/test/java/org/apache/rocketmq/broker/config/v2/ConsumerOffsetManagerV2Test.java

@@ -26,7 +26,6 @@
 import org.apache.rocketmq.store.config.MessageStoreConfig;
 import org.junit.After;
 import org.junit.Assert;
-import org.junit.Assume;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -60,7 +59,6 @@ public void cleanUp() {
 
     @Before
     public void setUp() throws IOException {
-        Assume.assumeFalse(MixAll.isMac());
         BrokerConfig brokerConfig = new BrokerConfig();
         Mockito.doReturn(brokerConfig).when(controller).getBrokerConfig();
 
@@ -77,7 +75,6 @@ public void setUp() throws IOException {
      */
     @Test
     public void testCommitOffset_Standard() {
-        Assume.assumeFalse(MixAll.isMac());
         Assert.assertTrue(consumerOffsetManagerV2.load());
 
         String clientHost = "localhost";
@@ -104,7 +101,6 @@ public void testCommitOffset_Standard() {
      */
     @Test
     public void testCommitOffset_LMQ() {
-        Assume.assumeFalse(MixAll.isMac());
         Assert.assertTrue(consumerOffsetManagerV2.load());
 
         String clientHost = "localhost";
@@ -130,7 +126,6 @@ public void testCommitOffset_LMQ() {
      */
     @Test
     public void testCommitPullOffset_LMQ() {
-        Assume.assumeFalse(MixAll.isMac());
         Assert.assertTrue(consumerOffsetManagerV2.load());
 
         String clientHost = "localhost";
@@ -155,7 +150,6 @@ public void testCommitPullOffset_LMQ() {
      */
     @Test
     public void testRemoveByTopicAtGroup() {
-        Assume.assumeFalse(MixAll.isMac());
         Assert.assertTrue(consumerOffsetManagerV2.load());
 
         String clientHost = "localhost";
@@ -188,7 +182,6 @@ public void testRemoveByTopicAtGroup() {
      */
     @Test
     public void testRemoveByGroup() {
-        Assume.assumeFalse(MixAll.isMac());
         Assert.assertTrue(consumerOffsetManagerV2.load());
 
         String clientHost = "localhost";