Fix conflict,merge develop (#2906)

* [ISSUE #1233] Fix CVE-2011-1473

* fix Multiple instances in the same application share MQClientInstance

* [ISSUE #2748] Fix deleteSubscriptionGroup not remove consumer offset

* [ISSUE #2745] Changed the support time of the request/reply feature to 4.6.0.

Co-authored-by: von gosling <[email protected]>

* [ISSUE #2729] Replace with Math.min method call

* [ISSUE #2801]Fix NamesrvAddr connot set in Producer

* [ISSUE 2800] optimize: the spelling of topicSynFlag

Co-authored-by: ph3636 <[email protected]>

* [ISSUE #2803] Fix the endpoint cannot get instanceId without http (#2804)

* fix the endpoint cannot get instanceId without http

* fix the endpoint cannot get instanceId without http

* add unit test

* add unit test

* add unit test

Co-authored-by: panzhi33 <[email protected]>

* fix messageArrivingListener NPE

* [ISSUE #2538]Optimize log output when message trace saving fails

* [ISSUE #2811] Fix the wrong topic was consumed in the DefaultMessageStoreTest test program

* [ISSUE #2821] Overriding the ServiceThread#shutdown in HAClient class

* [ISSUE #2805] remove redundant package imports

* [ISSUE #2833] Support trace for TranscationProducer (#2834)

* [ISSUE #2732] Fix message loss problem when rebalance with LitePullConsumer (#2832)

* [ISSUE #2732] Fix message loss problem when rebalance with LitePullConsumer

* Fix message loss problem when rebalance with LitePullConsumer, update 2

* [ISSUE #2846]fix -E might not port to other systems

* fix some nonconformity after checkstyle

* Support OpenTracing(#2861)

* [ISSUE #2872] remove log files created by integration test when mvn clean

* [ISSUE #2872] move log files created by integration test to target dir

* Change log level to debug: "Half offset {} has been committed/rolled back"

* Fix unit test stability

Bump mockito-core to 3.10.0, remove powermock dependency, suppress useless logging

* [ISSUE #2898] Resolve rocketmq-example project failed during checkstyle execution (#2899)

Co-authored-by: SSpirits <[email protected]>
Co-authored-by: panzhi33 <[email protected]>
Co-authored-by: panzhi <[email protected]>
Co-authored-by: ArronHuang <[email protected]>
Co-authored-by: von gosling <[email protected]>
Co-authored-by: drgnchan <[email protected]>
Co-authored-by: zhangjidi2016 <[email protected]>
Co-authored-by: ph3636 <[email protected]>
Co-authored-by: ph3636 <[email protected]>
Co-authored-by: BurningCN <[email protected]>
Co-authored-by: francis lee <[email protected]>
Co-authored-by: 灼华 <[email protected]>
Co-authored-by: yuz10 <[email protected]>
Co-authored-by: huangli <[email protected]>
Co-authored-by: chenrl <[email protected]>
Co-authored-by: ayanamist <[email protected]>
Co-authored-by: zhangjidi2016 <[email protected]>

Author: 16 people

broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java

@@ -232,4 +232,20 @@ public void cloneOffset(final String srcGroup, final String destGroup, final Str
         }
     }
 
+    public void removeOffset(final String group) {
+        Iterator<Entry<String, ConcurrentMap<Integer, Long>>> it = this.offsetTable.entrySet().iterator();
+        while (it.hasNext()) {
+            Entry<String, ConcurrentMap<Integer, Long>> next = it.next();
+            String topicAtGroup = next.getKey();
+            if (topicAtGroup.contains(group)) {
+                String[] arrays = topicAtGroup.split(TOPIC_GROUP_SEPARATOR);
+                if (arrays.length == 2 && group.equals(arrays[1])) {
+                    it.remove();
+                    log.warn("clean group offset {}", topicAtGroup);
+                }
+            }
+        }
+
+    }
+
 }

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

@@ -714,6 +714,10 @@ private RemotingCommand deleteSubscriptionGroup(ChannelHandlerContext ctx,
 
         this.brokerController.getSubscriptionGroupManager().deleteSubscriptionGroupConfig(requestHeader.getGroupName());
 
+        if (requestHeader.isRemoveOffset()) {
+            this.brokerController.getConsumerOffsetManager().removeOffset(requestHeader.getGroupName());
+        }
+
         if (this.brokerController.getBrokerConfig().isAutoDeleteUnusedStats()) {
             this.brokerController.getBrokerStatsManager().onGroupDeleted(requestHeader.getGroupName());
         }

broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java

@@ -176,9 +176,7 @@ public TopicConfig createTopicInSendMessageMethod(final String topic, final Stri
                         if (PermName.isInherited(defaultTopicConfig.getPerm())) {
                             topicConfig = new TopicConfig(topic);
 
-                            int queueNums =
-                                clientDefaultTopicQueueNums > defaultTopicConfig.getWriteQueueNums() ? defaultTopicConfig
-                                    .getWriteQueueNums() : clientDefaultTopicQueueNums;
+                            int queueNums = Math.min(clientDefaultTopicQueueNums, defaultTopicConfig.getWriteQueueNums());
 
                             if (queueNums < 0) {
                                 queueNums = 0;

broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java

@@ -164,7 +164,7 @@ public void check(long transactionTimeout, int transactionCheckMax,
                         break;
                     }
                     if (removeMap.containsKey(i)) {
-                        log.info("Half offset {} has been committed/rolled back", i);
+                        log.debug("Half offset {} has been committed/rolled back", i);
                         Long removedOpOffset = removeMap.remove(i);
                         doneOpOffset.add(removedOpOffset);
                     } else {

client/pom.xml

@@ -47,6 +47,18 @@
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-lang3</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.opentracing</groupId>
+            <artifactId>opentracing-api</artifactId>
+            <version>0.33.0</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.opentracing</groupId>
+            <artifactId>opentracing-mock</artifactId>
+            <version>0.33.0</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.logging.log4j</groupId>
             <artifactId>log4j-core</artifactId>

client/src/main/java/org/apache/rocketmq/client/ClientConfig.java

@@ -94,7 +94,7 @@ public void setInstanceName(String instanceName) {
 
     public void changeInstanceNameToPID() {
         if (this.instanceName.equals("DEFAULT")) {
-            this.instanceName = String.valueOf(UtilAll.getPid());
+            this.instanceName = UtilAll.getPid() + "#" + System.nanoTime();
         }
     }
 
@@ -178,8 +178,8 @@ public ClientConfig cloneClientConfig() {
     }
 
     public String getNamesrvAddr() {
-        if (StringUtils.isNotEmpty(namesrvAddr) && NameServerAddressUtils.NAMESRV_ENDPOINT_PATTERN.matcher(namesrvAddr.trim()).matches()) {
-            return namesrvAddr.substring(NameServerAddressUtils.ENDPOINT_PREFIX.length());
+        if (StringUtils.isNotEmpty(namesrvAddr) && NameServerAddressUtils.validateInstanceEndpoint(namesrvAddr.trim())) {
+            return NameServerAddressUtils.getNameSrvAddrFromNamesrvEndpoint(namesrvAddr);
         }
         return namesrvAddr;
     }

client/src/main/java/org/apache/rocketmq/client/hook/EndTransactionContext.java

@@ -0,0 +1,86 @@
+/*
+ * 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.client.hook;
+
+import org.apache.rocketmq.client.producer.LocalTransactionState;
+import org.apache.rocketmq.common.message.Message;
+
+public class EndTransactionContext {
+    private String producerGroup;
+    private Message message;
+    private String brokerAddr;
+    private String msgId;
+    private String transactionId;
+    private LocalTransactionState transactionState;
+    private boolean fromTransactionCheck;
+
+    public String getProducerGroup() {
+        return producerGroup;
+    }
+
+    public void setProducerGroup(String producerGroup) {
+        this.producerGroup = producerGroup;
+    }
+
+    public Message getMessage() {
+        return message;
+    }
+
+    public void setMessage(Message message) {
+        this.message = message;
+    }
+
+    public String getBrokerAddr() {
+        return brokerAddr;
+    }
+
+    public void setBrokerAddr(String brokerAddr) {
+        this.brokerAddr = brokerAddr;
+    }
+
+    public String getMsgId() {
+        return msgId;
+    }
+
+    public void setMsgId(String msgId) {
+        this.msgId = msgId;
+    }
+
+    public String getTransactionId() {
+        return transactionId;
+    }
+
+    public void setTransactionId(String transactionId) {
+        this.transactionId = transactionId;
+    }
+
+    public LocalTransactionState getTransactionState() {
+        return transactionState;
+    }
+
+    public void setTransactionState(LocalTransactionState transactionState) {
+        this.transactionState = transactionState;
+    }
+
+    public boolean isFromTransactionCheck() {
+        return fromTransactionCheck;
+    }
+
+    public void setFromTransactionCheck(boolean fromTransactionCheck) {
+        this.fromTransactionCheck = fromTransactionCheck;
+    }
+}

client/src/main/java/org/apache/rocketmq/client/hook/EndTransactionHook.java

@@ -0,0 +1,23 @@
+/*
+ * 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.client.hook;
+
+public interface EndTransactionHook {
+    String hookName();
+
+    void endTransaction(final EndTransactionContext context);
+}

client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java

@@ -1721,10 +1721,11 @@ public void deleteTopicInNameServer(final String addr, final String topic, final
         throw new MQClientException(response.getCode(), response.getRemark());
     }
 
-    public void deleteSubscriptionGroup(final String addr, final String groupName, final long timeoutMillis)
+    public void deleteSubscriptionGroup(final String addr, final String groupName, final boolean removeOffset, final long timeoutMillis)
         throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
         DeleteSubscriptionGroupRequestHeader requestHeader = new DeleteSubscriptionGroupRequestHeader();
         requestHeader.setGroupName(groupName);
+        requestHeader.setRemoveOffset(removeOffset);
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.DELETE_SUBSCRIPTIONGROUP, requestHeader);
 
         RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr),

client/src/main/java/org/apache/rocketmq/client/impl/consumer/AssignedMessageQueue.java

@@ -83,9 +83,12 @@ public long getPullOffset(MessageQueue messageQueue) {
         return -1;
     }
 
-    public void updatePullOffset(MessageQueue messageQueue, long offset) {
+    public void updatePullOffset(MessageQueue messageQueue, long offset, ProcessQueue processQueue) {
         MessageQueueState messageQueueState = assignedMessageQueueState.get(messageQueue);
         if (messageQueueState != null) {
+            if (messageQueueState.getProcessQueue() != processQueue) {
+                return;
+            }
             messageQueueState.setPullOffset(offset);
         }
     }