Merge branch 'develop' into dev_gyy_crc

apache · Oct 25, 2023 · 147d3d7 · 147d3d7
2 parents 2b9bff7 + f90c553
commit 147d3d7
Show file tree

Hide file tree

Showing 33 changed files with 145 additions and 401 deletions.
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/plain/PlainPermissionManager.java b/acl/src/main/java/org/apache/rocketmq/acl/plain/PlainPermissionManager.java
@@ -484,7 +484,7 @@ public boolean updateGlobalWhiteAddrsConfig(List<String> globalWhiteAddrsList, S
             return false;
         }
 
-        if (!fileName.startsWith(fileHome)) {
+        if (!file.getAbsolutePath().startsWith(fileHome)) {
             log.error("Parameter value " + fileName + " is not in the directory rocketmq.home.dir " + fileHome);
             return false;
         }

diff --git a/broker/src/main/java/org/apache/rocketmq/broker/client/ClientHousekeepingService.java b/broker/src/main/java/org/apache/rocketmq/broker/client/ClientHousekeepingService.java
@@ -87,4 +87,9 @@ public void onChannelIdle(String remoteAddr, Channel channel) {
         this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, channel);
         this.brokerController.getBrokerStatsManager().incChannelIdleNum();
     }
+
+    @Override
+    public void onChannelActive(String remoteAddr, Channel channel) {
+
+    }
 }
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
@@ -49,7 +49,6 @@
 import org.apache.rocketmq.broker.plugin.BrokerAttachedPlugin;
 import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
 import org.apache.rocketmq.broker.transaction.queue.TransactionalMessageUtil;
-import org.apache.rocketmq.common.AclConfig;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.KeyBuilder;
 import org.apache.rocketmq.common.LockCallback;
@@ -130,8 +129,6 @@
 import org.apache.rocketmq.remoting.protocol.header.GetAllProducerInfoRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.GetAllTopicConfigResponseHeader;
 import org.apache.rocketmq.remoting.protocol.header.GetBrokerAclConfigResponseHeader;
-import org.apache.rocketmq.remoting.protocol.header.GetBrokerClusterAclConfigResponseBody;
-import org.apache.rocketmq.remoting.protocol.header.GetBrokerClusterAclConfigResponseHeader;
 import org.apache.rocketmq.remoting.protocol.header.GetBrokerConfigResponseHeader;
 import org.apache.rocketmq.remoting.protocol.header.GetConsumeStatsInBrokerHeader;
 import org.apache.rocketmq.remoting.protocol.header.GetConsumeStatsRequestHeader;
@@ -311,8 +308,6 @@ public RemotingCommand processRequest(ChannelHandlerContext ctx,
                 return updateGlobalWhiteAddrsConfig(ctx, request);
             case RequestCode.RESUME_CHECK_HALF_MESSAGE:
                 return resumeCheckHalfMessage(ctx, request);
-            case RequestCode.GET_BROKER_CLUSTER_ACL_CONFIG:
-                return getBrokerClusterAclConfig(ctx, request);
             case RequestCode.GET_TOPIC_CONFIG:
                 return getTopicConfig(ctx, request);
             case RequestCode.UPDATE_AND_CREATE_STATIC_TOPIC:
@@ -699,27 +694,6 @@ private RemotingCommand getBrokerAclConfigVersion(ChannelHandlerContext ctx, Rem
         return null;
     }
 
-    private RemotingCommand getBrokerClusterAclConfig(ChannelHandlerContext ctx, RemotingCommand request) {
-
-        final RemotingCommand response = RemotingCommand.createResponseCommand(GetBrokerClusterAclConfigResponseHeader.class);
-
-        try {
-            AccessValidator accessValidator = this.brokerController.getAccessValidatorMap().get(PlainAccessValidator.class);
-            GetBrokerClusterAclConfigResponseBody body = new GetBrokerClusterAclConfigResponseBody();
-            AclConfig aclConfig = accessValidator.getAllAclConfig();
-            body.setGlobalWhiteAddrs(aclConfig.getGlobalWhiteAddrs());
-            body.setPlainAccessConfigs(aclConfig.getPlainAccessConfigs());
-            response.setCode(ResponseCode.SUCCESS);
-            response.setBody(body.encode());
-            response.setRemark(null);
-            return response;
-        } catch (Exception e) {
-            LOGGER.error("Failed to generate a proper getBrokerClusterAclConfig response", e);
-        }
-
-        return null;
-    }
-
     private RemotingCommand getUnknownCmdResponse(ChannelHandlerContext ctx, RemotingCommand request) {
         String error = " request type " + request.getCode() + " not supported";
         final RemotingCommand response =

diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java
@@ -204,7 +204,7 @@ public RemotingCommand processRequest(final ChannelHandlerContext ctx, RemotingC
         RemotingCommand response = RemotingCommand.createResponseCommand(PopMessageResponseHeader.class);
         final PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) response.readCustomHeader();
         final PopMessageRequestHeader requestHeader =
-            (PopMessageRequestHeader) request.decodeCommandCustomHeader(PopMessageRequestHeader.class);
+            (PopMessageRequestHeader) request.decodeCommandCustomHeader(PopMessageRequestHeader.class, true);
         StringBuilder startOffsetInfo = new StringBuilder(64);
         StringBuilder msgOffsetInfo = new StringBuilder(64);
         StringBuilder orderCountInfo = null;

diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
@@ -53,7 +53,6 @@
 import org.apache.rocketmq.client.producer.SendCallback;
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.client.producer.SendStatus;
-import org.apache.rocketmq.common.AclConfig;
 import org.apache.rocketmq.common.BoundaryType;
 import org.apache.rocketmq.common.MQVersion;
 import org.apache.rocketmq.common.MixAll;
@@ -154,7 +153,6 @@
 import org.apache.rocketmq.remoting.protocol.header.ExtraInfoUtil;
 import org.apache.rocketmq.remoting.protocol.header.GetAllProducerInfoRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.GetBrokerAclConfigResponseHeader;
-import org.apache.rocketmq.remoting.protocol.header.GetBrokerClusterAclConfigResponseBody;
 import org.apache.rocketmq.remoting.protocol.header.GetConsumeStatsInBrokerHeader;
 import org.apache.rocketmq.remoting.protocol.header.GetConsumeStatsRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.GetConsumerConnectionListRequestHeader;
@@ -520,31 +518,6 @@ public ClusterAclVersionInfo getBrokerClusterAclInfo(final String addr,
 
     }
 
-    public AclConfig getBrokerClusterConfig(final String addr,
-        final long timeoutMillis) throws InterruptedException, RemotingTimeoutException,
-        RemotingSendRequestException, RemotingConnectException, MQBrokerException {
-        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_CLUSTER_ACL_CONFIG, null);
-
-        RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr), request, timeoutMillis);
-        assert response != null;
-        switch (response.getCode()) {
-            case ResponseCode.SUCCESS: {
-                if (response.getBody() != null) {
-                    GetBrokerClusterAclConfigResponseBody body =
-                        GetBrokerClusterAclConfigResponseBody.decode(response.getBody(), GetBrokerClusterAclConfigResponseBody.class);
-                    AclConfig aclConfig = new AclConfig();
-                    aclConfig.setGlobalWhiteAddrs(body.getGlobalWhiteAddrs());
-                    aclConfig.setPlainAccessConfigs(body.getPlainAccessConfigs());
-                    return aclConfig;
-                }
-            }
-            default:
-                break;
-        }
-        throw new MQBrokerException(response.getCode(), response.getRemark(), addr);
-
-    }
-
     public SendResult sendMessage(
         final String addr,
         final String brokerName,

diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java b/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
@@ -159,14 +159,6 @@ public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String cli
                 private final ConcurrentMap<String, HashMap<Long, String>> brokerAddrTable = MQClientInstance.this.brokerAddrTable;
                 @Override
                 public void onChannelConnect(String remoteAddr, Channel channel) {
-                    for (Map.Entry<String, HashMap<Long, String>> addressEntry : brokerAddrTable.entrySet()) {
-                        for (String address : addressEntry.getValue().values()) {
-                            if (address.equals(remoteAddr)) {
-                                sendHeartbeatToAllBrokerWithLockV2(false);
-                                break;
-                            }
-                        }
-                    }
                 }
 
                 @Override
@@ -180,6 +172,18 @@ public void onChannelException(String remoteAddr, Channel channel) {
                 @Override
                 public void onChannelIdle(String remoteAddr, Channel channel) {
                 }
+
+                @Override
+                public void onChannelActive(String remoteAddr, Channel channel) {
+                    for (Map.Entry<String, HashMap<Long, String>> addressEntry : brokerAddrTable.entrySet()) {
+                        for (String address : addressEntry.getValue().values()) {
+                            if (address.equals(remoteAddr)) {
+                                sendHeartbeatToAllBrokerWithLockV2(false);
+                                break;
+                            }
+                        }
+                    }
+                }
             };
         } else {
             channelEventListener = null;

diff --git a/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java b/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
@@ -37,7 +37,6 @@
 import org.apache.rocketmq.client.producer.SendCallback;
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.client.producer.SendStatus;
-import org.apache.rocketmq.common.AclConfig;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.TopicConfig;
@@ -62,8 +61,6 @@
 import org.apache.rocketmq.remoting.protocol.header.ChangeInvisibleTimeRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.ChangeInvisibleTimeResponseHeader;
 import org.apache.rocketmq.remoting.protocol.header.ExtraInfoUtil;
-import org.apache.rocketmq.remoting.protocol.header.GetBrokerClusterAclConfigResponseBody;
-import org.apache.rocketmq.remoting.protocol.header.GetBrokerClusterAclConfigResponseHeader;
 import org.apache.rocketmq.remoting.protocol.header.GetConsumerListByGroupResponseBody;
 import org.apache.rocketmq.remoting.protocol.header.GetConsumerListByGroupResponseHeader;
 import org.apache.rocketmq.remoting.protocol.header.GetEarliestMsgStoretimeResponseHeader;
@@ -700,30 +697,6 @@ public RemotingCommand answer(InvocationOnMock mock) {
         mqClientAPI.createTopic(brokerAddr, topic, new TopicConfig(), 10000);
     }
 
-    @Test
-    public void testGetBrokerClusterConfig() throws Exception {
-        doAnswer(new Answer<RemotingCommand>() {
-            @Override
-            public RemotingCommand answer(InvocationOnMock mock) {
-                RemotingCommand request = mock.getArgument(1);
-
-                RemotingCommand response = RemotingCommand.createResponseCommand(GetBrokerClusterAclConfigResponseHeader.class);
-                GetBrokerClusterAclConfigResponseBody body = new GetBrokerClusterAclConfigResponseBody();
-                body.setGlobalWhiteAddrs(Collections.singletonList("1.1.1.1"));
-                body.setPlainAccessConfigs(Collections.singletonList(new PlainAccessConfig()));
-                response.setBody(body.encode());
-                response.makeCustomHeaderToNet();
-                response.setCode(ResponseCode.SUCCESS);
-                response.setOpaque(request.getOpaque());
-                return response;
-            }
-        }).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong());
-
-        AclConfig aclConfig = mqClientAPI.getBrokerClusterConfig(brokerAddr, 10000);
-        assertThat(aclConfig.getPlainAccessConfigs()).size().isGreaterThan(0);
-        assertThat(aclConfig.getGlobalWhiteAddrs()).size().isGreaterThan(0);
-    }
-
     @Test
     public void testViewMessage() throws Exception {
         doAnswer(new Answer<RemotingCommand>() {

diff --git a/container/src/main/java/org/apache/rocketmq/container/BrokerContainer.java b/container/src/main/java/org/apache/rocketmq/container/BrokerContainer.java
@@ -164,7 +164,7 @@ public void run0() {
                         LOG.error("ScheduledTask fetchNameServerAddr exception", e);
                     }
                 }
-            }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
+            }, 1000 * 10, this.brokerContainerConfig.getUpdateNamesrvAddrInterval(), TimeUnit.MILLISECONDS);
         } else if (this.brokerContainerConfig.isFetchNamesrvAddrByAddressServer()) {
             this.scheduledExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(BrokerIdentity.BROKER_CONTAINER_IDENTITY) {
 

diff --git a/container/src/main/java/org/apache/rocketmq/container/BrokerContainerConfig.java b/container/src/main/java/org/apache/rocketmq/container/BrokerContainerConfig.java
@@ -44,6 +44,11 @@ public class BrokerContainerConfig {
      */
     private long fetchNamesrvAddrInterval = 10 * 1000;
 
+    /**
+     * The interval to update namesrv addr, default value is 120 second
+     */
+    private long updateNamesrvAddrInterval = 60 * 2 * 1000;
+
     public String getRocketmqHome() {
         return rocketmqHome;
     }
@@ -95,4 +100,12 @@ public long getFetchNamesrvAddrInterval() {
     public void setFetchNamesrvAddrInterval(final long fetchNamesrvAddrInterval) {
         this.fetchNamesrvAddrInterval = fetchNamesrvAddrInterval;
     }
+
+    public long getUpdateNamesrvAddrInterval() {
+        return updateNamesrvAddrInterval;
+    }
+
+    public void setUpdateNamesrvAddrInterval(long updateNamesrvAddrInterval) {
+        this.updateNamesrvAddrInterval = updateNamesrvAddrInterval;
+    }
 }
diff --git a/container/src/main/java/org/apache/rocketmq/container/BrokerContainerProcessor.java b/container/src/main/java/org/apache/rocketmq/container/BrokerContainerProcessor.java
@@ -91,11 +91,10 @@ private synchronized RemotingCommand addBroker(ChannelHandlerContext ctx,
                 LOGGER.error("addBroker load config from {} failed, {}", configPath, e);
             }
         } else {
-            byte[] body = request.getBody();
-            if (body != null) {
-                String bodyStr = new String(body, MixAll.DEFAULT_CHARSET);
-                brokerProperties = MixAll.string2Properties(bodyStr);
-            }
+            LOGGER.error("addBroker config path is empty");
+            response.setCode(ResponseCode.SYSTEM_ERROR);
+            response.setRemark("addBroker config path is empty");
+            return response;
         }
 
         if (brokerProperties == null) {

diff --git a/...ainer/src/main/java/org/apache/rocketmq/container/ContainerClientHouseKeepingService.java b/...ainer/src/main/java/org/apache/rocketmq/container/ContainerClientHouseKeepingService.java
@@ -49,6 +49,11 @@ public void onChannelIdle(String remoteAddr, Channel channel) {
         onChannelOperation(CallbackCode.IDLE, remoteAddr, channel);
     }
 
+    @Override
+    public void onChannelActive(String remoteAddr, Channel channel) {
+        onChannelOperation(CallbackCode.ACTIVE, remoteAddr, channel);
+    }
+
     private void onChannelOperation(CallbackCode callbackCode, String remoteAddr, Channel channel) {
         Collection<InnerBrokerController> masterBrokers = this.brokerContainer.getMasterBrokers();
         Collection<InnerSalveBrokerController> slaveBrokers = this.brokerContainer.getSlaveBrokers();
@@ -103,6 +108,10 @@ public enum CallbackCode {
         /**
          * onChannelIdle
          */
-        IDLE
+        IDLE,
+        /**
+         * onChannelActive
+         */
+        ACTIVE
     }
 }
diff --git a/controller/src/main/java/org/apache/rocketmq/controller/BrokerHousekeepingService.java b/controller/src/main/java/org/apache/rocketmq/controller/BrokerHousekeepingService.java
@@ -48,4 +48,9 @@ public void onChannelException(String remoteAddr, Channel channel) {
     public void onChannelIdle(String remoteAddr, Channel channel) {
         this.controllerManager.getHeartbeatManager().onBrokerChannelClose(channel);
     }
+
+    @Override
+    public void onChannelActive(String remoteAddr, Channel channel) {
+
+    }
 }
diff --git a/example/src/main/java/org/apache/rocketmq/example/simple/PullConsumer.java b/example/src/main/java/org/apache/rocketmq/example/simple/PullConsumer.java
@@ -75,7 +75,7 @@ public void run() {
 
                                             if (msgs != null && !msgs.isEmpty()) {
                                                 this.doSomething(msgs);
-                                                //update offset to broker
+                                                //update offset to local memory, eventually to broker
                                                 consumer.updateConsumeOffset(messageQueue, pullResult.getNextBeginOffset());
                                                 //print pull tps
                                                 this.incPullTPS(topic, pullResult.getMsgFoundList().size());

diff --git a/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/BrokerHousekeepingService.java b/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/BrokerHousekeepingService.java
@@ -46,4 +46,9 @@ public void onChannelException(String remoteAddr, Channel channel) {
     public void onChannelIdle(String remoteAddr, Channel channel) {
         this.namesrvController.getRouteInfoManager().onChannelDestroy(channel);
     }
+
+    @Override
+    public void onChannelActive(String remoteAddr, Channel channel) {
+
+    }
 }
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/ClientHousekeepingService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/ClientHousekeepingService.java
@@ -49,5 +49,9 @@ public void onChannelIdle(String remoteAddr, Channel channel) {
         this.clientManagerActivity.doChannelCloseEvent(remoteAddr, channel);
     }
 
+    @Override
+    public void onChannelActive(String remoteAddr, Channel channel) {
+
+    }
 }