[ISSUE #10876] Fix BrokerIdentityInfo.equals for partial identities

broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java

@@ -97,6 +97,9 @@ public void addPartialSubscription(String clientId, String group, String topic,
             if (!liteLifecycleManager.isSubscriptionActive(topic, lmqName)) {
                 continue;
             }
+            if (!isLiteTopicSubscribed(clientGroup, lmqName) && getActiveSubscriptionNum() >= maxCount) {
+                throw new LiteQuotaException("lite subscription quota exceeded " + maxCount);
+            }
             thisSub.addLiteTopic(lmqName);
             // First remove the old subscription
             if (LiteMetadataUtil.isSubLiteExclusive(group, brokerController)) {
@@ -147,6 +150,10 @@ public void addCompleteSubscription(String clientId, String group, String topic,
             removeTopicGroup(clientGroup, lmqName, false);
         });
         lmqNameNew.forEach(lmqName -> {
+            long maxCount = brokerController.getBrokerConfig().getMaxLiteSubscriptionCount();
+            if (!isLiteTopicSubscribed(clientGroup, lmqName) && getActiveSubscriptionNum() >= maxCount) {
+                throw new LiteQuotaException("lite subscription quota exceeded " + maxCount);
+            }
             thisSub.addLiteTopic(lmqName);
             addTopicGroup(clientGroup, lmqName);
         });
@@ -269,6 +276,11 @@ public void cleanSubscription(String lmqName, boolean notifyClient) {
         }
     }
 
+    protected boolean isLiteTopicSubscribed(ClientGroup clientGroup, String lmqName) {
+        Set<ClientGroup> topicGroupSet = liteTopic2Group.get(lmqName);
+        return topicGroupSet != null && topicGroupSet.contains(clientGroup);
+    }
+
     protected void addTopicGroup(ClientGroup clientGroup, String lmqName) {
         Set<ClientGroup> topicGroupSet = liteTopic2Group
             .computeIfAbsent(lmqName, k -> ConcurrentHashMap.newKeySet());

broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerRocksdbStore.java

@@ -147,9 +147,11 @@ public List<PopConsumerRecord> scanExpiredRecords(long lower, long upper, int ma
         // configure prefix indexing to improve the performance of scans.
         // However, in the current implementation, this is not the bottleneck.
         List<PopConsumerRecord> consumerRecordList = new ArrayList<>();
-        try (ReadOptions scanOptions = new ReadOptions()
-            .setIterateLowerBound(new Slice(ByteBuffer.allocate(Long.BYTES).putLong(lower).array()))
-            .setIterateUpperBound(new Slice(ByteBuffer.allocate(Long.BYTES).putLong(upper).array()));
+        try (Slice lowerBound = new Slice(ByteBuffer.allocate(Long.BYTES).putLong(lower).array());
+             Slice upperBound = new Slice(ByteBuffer.allocate(Long.BYTES).putLong(upper).array());
+             ReadOptions scanOptions = new ReadOptions()
+                 .setIterateLowerBound(lowerBound)
+                 .setIterateUpperBound(upperBound);
              RocksIterator iterator = db.newIterator(this.columnFamilyHandle, scanOptions)) {
             iterator.seek(ByteBuffer.allocate(Long.BYTES).putLong(lower).array());
             while (iterator.isValid() && consumerRecordList.size() < maxCount) {

controller/src/main/java/org/apache/rocketmq/controller/impl/heartbeat/BrokerIdentityInfo.java

@@ -63,7 +63,9 @@ public boolean equals(Object obj) {
 
         if (obj instanceof BrokerIdentityInfo) {
             BrokerIdentityInfo addr = (BrokerIdentityInfo) obj;
-            return clusterName.equals(addr.clusterName) && brokerName.equals(addr.brokerName) && brokerId.equals(addr.brokerId);
+            return Objects.equals(clusterName, addr.clusterName)
+                && Objects.equals(brokerName, addr.brokerName)
+                && Objects.equals(brokerId, addr.brokerId);
         }
         return false;
     }

proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsManager.java

@@ -148,9 +148,9 @@ public void start() throws Exception {
         if (StringUtils.isNotBlank(labels)) {
             List<String> kvPairs = Splitter.on(',').omitEmptyStrings().splitToList(labels);
             for (String item : kvPairs) {
-                String[] split = item.split(":");
+                String[] split = item.split(":", 2);
                 if (split.length != 2) {
-                    log.warn("metricsLabel is not valid: {}", labels);
+                    log.warn("metricsLabel is not valid: {}", item);
                     continue;
                 }
                 LABEL_MAP.put(split[0], split[1]);
@@ -188,9 +188,9 @@ public void start() throws Exception {
                 Map<String, String> headerMap = new HashMap<>();
                 List<String> kvPairs = Splitter.on(',').omitEmptyStrings().splitToList(headers);
                 for (String item : kvPairs) {
-                    String[] split = item.split(":");
+                    String[] split = item.split(":", 2);
                     if (split.length != 2) {
-                        log.warn("metricsGrpcExporterHeader is not valid: {}", headers);
+                        log.warn("metricsGrpcExporterHeader is not valid: {}", item);
                         continue;
                     }
                     headerMap.put(split[0], split[1]);

proxy/src/main/java/org/apache/rocketmq/proxy/processor/ClientProcessor.java

@@ -197,8 +197,10 @@ protected void validateLiteSubTopic(ProxyContext ctx, String group, Set<Subscrip
         if (CollectionUtils.isEmpty(subList)) {
             return;
         }
-        // check bindTopic for sub list
-        validateLiteBindTopic(ctx, group, subList.iterator().next().getTopic());
+        // check bindTopic for every subscription in the set
+        for (SubscriptionData sub : subList) {
+            validateLiteBindTopic(ctx, group, sub.getTopic());
+        }
     }
 
     protected void validateLiteBindTopic(ProxyContext ctx, String group, String bindTopic) {

proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java

@@ -211,13 +211,23 @@ protected SendMessageRequestHeader buildSendMessageRequestHeader(List<Message> m
         if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
             String reconsumeTimes = MessageAccessor.getReconsumeTime(message);
             if (reconsumeTimes != null) {
-                requestHeader.setReconsumeTimes(Integer.valueOf(reconsumeTimes));
+                try {
+                    requestHeader.setReconsumeTimes(Integer.parseInt(reconsumeTimes.trim()));
+                } catch (NumberFormatException e) {
+                    log.warn("parse reconsumeTimes error, with value:{}", reconsumeTimes);
+                    requestHeader.setReconsumeTimes(0);
+                }
                 MessageAccessor.clearProperty(message, MessageConst.PROPERTY_RECONSUME_TIME);
             }
 
             String maxReconsumeTimes = MessageAccessor.getMaxReconsumeTimes(message);
             if (maxReconsumeTimes != null) {
-                requestHeader.setMaxReconsumeTimes(Integer.valueOf(maxReconsumeTimes));
+                try {
+                    requestHeader.setMaxReconsumeTimes(Integer.parseInt(maxReconsumeTimes.trim()));
+                } catch (NumberFormatException e) {
+                    log.warn("parse maxReconsumeTimes error, with value:{}", maxReconsumeTimes);
+                    requestHeader.setMaxReconsumeTimes(0);
+                }
                 MessageAccessor.clearProperty(message, MessageConst.PROPERTY_MAX_RECONSUME_TIMES);
             }
         }

remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java

@@ -784,7 +784,7 @@ public void run() {
             while (!this.isStopped()) {
                 try {
                     NettyEvent event = this.eventQueue.poll(3000, TimeUnit.MILLISECONDS);
-                    if (event != null && listener != null) {
+                    if (event != null && event.getType() != null && listener != null) {
                         switch (event.getType()) {
                             case IDLE:
                                 listener.onChannelIdle(event.getRemoteAddr(), event.getChannel());
@@ -814,6 +814,14 @@ public void run() {
             log.info(this.getServiceName() + " service end");
         }
 
+        @Override
+        public void shutdown(final boolean interrupt) {
+            // wake up the thread blocked on eventQueue.poll() so it observes the stopped flag
+            // immediately instead of waiting for the next poll timeout
+            this.eventQueue.offer(new NettyEvent(null, null, null));
+            super.shutdown(interrupt);
+        }
+
         @Override
         public String getServiceName() {
             return NettyEventExecutor.class.getSimpleName();

remoting/src/main/java/org/apache/rocketmq/remoting/protocol/route/BrokerData.java

@@ -89,6 +89,9 @@ public BrokerData(String cluster, String brokerName, HashMap<Long, String> broke
      * @return Broker address.
      */
     public String selectBrokerAddr() {
+        if (this.brokerAddrs == null || this.brokerAddrs.isEmpty()) {
+            return null;
+        }
         String masterAddress = this.brokerAddrs.get(MixAll.MASTER_ID);
 
         if (masterAddress == null) {