[GitHub] [rocketmq] RongtongJin commented on a diff in pull request #4823: [ISSUE #4817] Add a command to clear broker data from controller for CLI

[GitBox <gi...@apache.org>]

RongtongJin commented on code in PR #4823:
URL: https://github.com/apache/rocketmq/pull/4823#discussion_r945407081


##########
controller/src/main/java/org/apache/rocketmq/controller/impl/manager/ReplicasInfoManager.java:
##########
@@ -373,6 +374,19 @@ public ControllerResult<Void> getSyncStateData(final List<String> brokerNames) {
         return result;
     }
 
+    public ControllerResult<Void> cleanBrokerData(final CleanControllerBrokerDataRequestHeader requestHeader) {
+        final ControllerResult<Void> result = new ControllerResult<>();
+        String brokerName = requestHeader.getBrokerName();
+        String brokerAddress = requestHeader.getBrokerAddress();

Review Comment:
   Multiple brokerAddress are not handled here, eg: 192.168.0.1:30911;192.168.0.2:30911



##########
controller/src/main/java/org/apache/rocketmq/controller/impl/DLedgerController.java:
##########
@@ -193,6 +194,32 @@ public RemotingServer getRemotingServer() {
         return this.dLedgerServer.getRemotingServer();
     }
 
+    @Override
+    public CompletableFuture<RemotingCommand> cleanBrokerData(
+        final CleanControllerBrokerDataRequestHeader requestHeader,
+        final BiPredicate<String, String> brokerAlivePredicate) {
+
+        String brokerAddrs = requestHeader.getBrokerAddress();
+        String clusterName = requestHeader.getClusterName();
+        if ((null == brokerAddrs || brokerAddrs.trim().isEmpty()) && brokerAlivePredicate.test(clusterName, null)) {
+            RemotingCommand responseCommand = RemotingCommand.createResponseCommand(ResponseCode.CONTROLLER_INVALID_REQUEST,
+                String.format("Broker %s is still alive, clean up failure", requestHeader.getBrokerName()));
+            return CompletableFuture.completedFuture(responseCommand);
+        } else {
+            String[] brokerAddrArray = brokerAddrs.split(";");
+            for (String brokerAddr : brokerAddrArray) {
+                if (brokerAlivePredicate.test(clusterName, brokerAddr)) {
+                    RemotingCommand responseCommand = RemotingCommand.createResponseCommand(ResponseCode.CONTROLLER_INVALID_REQUEST,
+                        String.format("Broker [%s,  %s] is still alive, clean up failure", requestHeader.getBrokerName(), brokerAddr));
+                    return CompletableFuture.completedFuture(responseCommand);
+                }
+            }
+        }

Review Comment:
   How about moving these to replicasInfoManager just like other methods do？



##########
controller/src/main/java/org/apache/rocketmq/controller/impl/DefaultBrokerHeartbeatManager.java:
##########
@@ -112,6 +115,24 @@ public void registerBroker(String clusterName, String brokerName, String brokerA
         }
     }
 
+    public boolean isBrokerActiveOfBrokerName(String clusterName, String brokerName) {
+
+        Set<BrokerAddrInfo> addrInfoSet = this.brokerLiveTable.keySet().stream()
+            .filter(item -> StringUtils.equals(clusterName, item.getClusterName())).collect(Collectors.toSet());
+        if (null == addrInfoSet || addrInfoSet.size() == 0) {
+            return false;
+        }
+        for (BrokerAddrInfo addrInfo : addrInfoSet) {
+            BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.get(addrInfo);
+            if (brokerLiveInfo != null && StringUtils.equals(brokerLiveInfo.brokerName, brokerName)) {
+                long last = brokerLiveInfo.lastUpdateTimestamp;
+                long timeoutMillis = brokerLiveInfo.heartbeatTimeoutMillis;
+                return (last + timeoutMillis) >= System.currentTimeMillis();
+            }
+        }
+        return false;
+    }

Review Comment:
   Is it necessary to add this function? Maybe we can do this in replicasInfoManager



##########
controller/src/main/java/org/apache/rocketmq/controller/Controller.java:
##########
@@ -106,4 +108,11 @@ CompletableFuture<RemotingCommand> alterSyncStateSet(
      * Get the remotingServer used by the controller, the upper layer will reuse this remotingServer.
      */
     RemotingServer getRemotingServer();
+
+    /**
+     * Clean controller broker data, the broker which  shut down or offline
+     * @return
+     */
+    CompletableFuture<RemotingCommand> cleanBrokerData(final CleanControllerBrokerDataRequestHeader requestHeader,
+        final BiPredicate<String,String> brokerAlivePredicate);

Review Comment:
   brokerAlivePredicate may not be required. The implementation class DLedgerController contains brokerAlivePredicate.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@rocketmq.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org