Mirror of Apache RocketMQ

clinet exception:

org.apache.rocketmq.client.exception.MQClientException: No route info of this topic, ffff
See http://rocketmq.apache.org/docs/faq/ for further details.
	at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:634)
	at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1253)
	at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1203)
	at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:214)
	at com.xx.rmq.Producer.main(Producer.java:55)

rocketmq server config：

autoCreateTopicEnable=true

问题： 我们手动在rokcetmq中创建topic时，客户端访问时正常的，如果不提前手动创建将会出现上面的异常，如果需要让服务端自动创建topic，应该如何设置

JIRA: https://issues.apache.org/jira/browse/ROCKETMQ-107

No Syncrinations in start or shutdown which may result in :

1. Start twice if two thread starts the same consumer in the same time, which leads to initizize the resources twice.
2. Shutdown may not be effective if thread A start and Thread b tries to shutdown in a very short time after A's starting.

Implemenations:

1. Add lock in getter and setter, and replace all direct read/write operations to getter and setter
2. Add lock before start and shutdown

This commit is in reference to Story number 44 from the Kanban Board, which refers to removing duplicated logic from DefaultMessageStore.getEarliestMessageTime(). Please let me know if you need any improvements.

Support message filtering based on SQL92

So far, RocketMQ only support message filtering feature by TAG, but one message only can own one tag, this is too limited to meet complex business requirements.

So, we want to define and implement a reasonable filter language based on a subset of the SQL 92 expression syntax to support customized message filtering.

Why subset of SQL92

Let RocketMQ has the ability of message filtering is the purpose of this issue, no matter SQL92 or any other languages.

As I know, ActiveMQ already impllement this functionality based on JavaCC, it's simple and exntensible.So I just extract it and integrate into RocketMQ, only some grammars:

1. numeric comparison, like >, >=, <,<=, BETWEEN
2. character comparison, like =, <>, IN
3. check NULL
4. logical AND, logical OR, logical NOT

Design

• New Module, rocketmq-filter

The implementation of SQL92 language is placed in this module which have dependency on common module.

Broker compile or evaluate expression through the interface of FilterSpi contained in FilterFactory that manage all FilterSpi and also support new one to register.

• How to manage consumer's expression data

Different from tag filtering, expression of SQL92 should be compiled first to check whether is leagal and then use the complied expression to compute. This procedure is designed to take place at broker.

ConsumerManager manage the suscriptions of push consumer, and ConsumerFilterManager manage the expression info of push consumer who wish to filter message by special language, the info includes data version, expression, compiled expression, alive time and etc.

• How to filter message by expression

I redesign the interface getMessage of MessageStore by replace the last parameter SubscriptionData to MessageFilter that is also refactored. The purpose is to make module rocketmq-store has no relation with protocol.

When get message, the implementation ExpressionMessageFilter would check whether the message is matched by BitsArray which will be refered later or evaluation, just as the mechanism of tag filtering.

• Optimization, pre-calculate the filtering result when build consume queue

It's poor performance to do filter when pull message:

1. off-heap to heap, once every consumer subscribed same topic pull message.
2. decode message properties, once every consumer subscribed same topic pull message.

BloomFilter and pre-calculation are adopted to optimize the situation:

1. Every consumer has been asigned some bit position of BloomFilter when register to broker.
2. When broker build queue after message into CommitLog, the consumer's filtering result would be calculated, and all resuls are assembled as a BitsArray saved in ConsumeQueueExt.
3. ConsumeQueueExt is a store file linked to ConsumeQueue, ConsumeQueue could find the data by the tagsCode whitch is already replaced by the address(for compitable, the range is Long.MIN_VALUE to Integer.MIN_VALUE) generated by ConsumeQueueExt.
4. ExpressionMessageFilter could use the BitsArray to check whether the message is matched. Because of BloomFilter's collision, it also need to decode properties to do calculation for matched message(may could be reduced by check the collision, not include in this edition).

This optimization is suitable for:

1. High subscription ratio.
2. Large properties.

Interface

Only push consumer could filter message by SQL92 expression in this edition.

Tests show that Kafka's million-level TPS is mainly owed to batch. When set batch size to 1, the TPS is reduced an order of magnitude. So I try to add this feature to RocketMQ.

For a minimal effort, it works as follows:

• Only add synchronous send functions to MQProducer interface, just like send(final Collection msgs)
• Use MessageBatch which extends Message and implements Iterable<Message>
• Use byte buffer instead of list of objects to avoid too much GC in Broker.
• Split the decode and encode logic from lockForPutMessage to avoid too many race conditions.

Tests: On linux with 24 Core 48G Ram and SSD, single broker, using 50 threads to send 50Byte(body) message in batch size 50, we get about 150w TPS until the disk is full.

Potential problems: Although the messages can be accumulated in the Broker very quickly, it need time to dispatch to the consume queue, which is much slower than accepting messages. So the messages may not be able to be consumed immediately.

We may need to refactor the ReputMessageService to solve this problem.

And if guys have some ideas, please let me know or just share it in this issue.

raft协议在网络抖动，机器临时故障等极小概率情况下会产生以下选举问题： 假定有编号为【1，2，3，4，5，6，7】共7个结点构成了一个选举整体，当结点1作为候选人发起term=6的投票，该term对应投票为【1，2，3，4】所接受，结点1作为leader执行leader职责前，刚好结点7发起term=7的投票(因为网络抖动原因结点7发起的term=6刚好未被结点【1，2，3，4】所接收)，抛开上帝视角具体分析，此时存在以下两种情况： 情况1：结点【4，5，6，7】多数结点接受结点7该term=7的选举，结点7作为leader执行职责，该情况下目前所用raft协议能保证结点1会自动失去leader资格。 情况2:结点【5，6，7】等少数结点接受结点7该term=7的选举，结点7未作为leader执行职责，但此时因为【5，6，7】结点已经认可了term=7的任期，会导致结点1以term=6所发送提议无法被接受，但结点1因为已经此实确实获取多数结点认可，在提议被多数接受后会进行提交。可惜目前所用的raft协议未考虑该情况下的处理方案。 解决策略:1、引入结点在接受到leader名义发起的提议后，如果term<自身已知选举认可term且term>自身已提交term时，对提议仅进行记录但拒绝接受该提议。当接受到提交请求时，如果提交所用term>自身已提交term，则找到对应提议并无条件进行提交(因为raft协议的提议提交命令发送前提为大多数已经接受了)。2、如果接受了提议，则在自身未主动探测到leader丢失下，拒绝对后续的选举term进行投票(以保证在接受结点1提议后结点7无法获取到多数结点的认可当选为leader出现双主问题). 问题扩展:实际情况会存在网络抖动引起的N个(N>=2)结点同时误认自己是leader或当前leader因为所用term并非全局视角下的最大term在履行leader职责时未受少数结点认可存在的问题

What is the purpose of the change

(1)Adding data new version mechanism for every broker's acl config file in a cluster. (2)Adding ClusterAclConfigVersionListSubCommand for querying acl config version; (3)Adding DeleteAccessConfigSubCommand for deleting account attribute in acl config file. (4)Adding UpdateAccessConfigSubCommand for updating account attribute in acl config file. (5)Adding UpdateGlobalWhiteAddrSubCommand for updating global white addrs in acl config file.

Brief changelog

(1)Adding data new version mechanism for every broker's acl config file in a cluster. (2)Adding ClusterAclConfigVersionListSubCommand for querying acl config version; (3)Adding DeleteAccessConfigSubCommand for deleting account attribute in acl config file. (4)Adding UpdateAccessConfigSubCommand for updating account attribute in acl config file. (5)Adding UpdateGlobalWhiteAddrSubCommand for updating global white addrs in acl config file.

Adding data new version mechanism in the acl yaml configuration file for acl mqadmin for acl configuration.

Verifying this change

I have already verified this changes in my local enviroment and passed acl uniting test cases.

Follow this checklist to help us incorporate your contribution quickly and easily. Notice, it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR.

• [x] Make sure there is a Github issue filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue.
• [x] Format the pull request title like [ISSUE #123] Fix UnknownException when host config not exist. Each commit in the pull request should have a meaningful subject line and body.
• [x] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
• [x] Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when cross module dependency exist. If the new feature or significant change is committed, please remember to add integration-test in test module.
• [x] Run mvn -B clean apache-rat:check findbugs:findbugs checkstyle:checkstyle to make sure basic checks pass. Run mvn clean install -DskipITs to make sure unit-test pass. Run mvn clean test-compile failsafe:integration-test to make sure integration-test pass.
• [ ] If this contribution is large, please file an Apache Individual Contributor License Agreement.

JIRA:https://issues.apache.org/jira/browse/ROCKETMQ-184?jql=project%20%3D%20ROCKETMQ

Problem, no listener is triggered when Chanel is close.

When async command sent to the server, and the server is crash before sending response to client, the callback can not be invoked in time. Instead, the callback can only be triggered by the timeout scan service.

This is obvious for pulling message since the timeout is by default 30 seconds. So if master crashes before process response to the client, the client can not repull until scan service tell it, which takes at most 30 seconds. And repull will have 3 seconds delay, so the HA to read from slave has to take 3-33 seconds when this problem occurs.

Solution: add interface for push consumer to accept await termination time to await consuming.

JIRA: https://issues.apache.org/jira/browse/ROCKETMQ-102

UnitTest in DefaultMQPushConsumerTest.java

@Test
public void testShutdownAwait() throws Exception {
    final LinkedList<Long> consumedOffset = new LinkedList<>();
    pushConsumer.setPullInterval(0);
    pushConsumer.getDefaultMQPushConsumerImpl().setConsumeMessageService(new ConsumeMessageConcurrentlyService(pushConsumer.getDefaultMQPushConsumerImpl(), new MessageListenerConcurrently() {
        @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
            for (MessageExt msg : msgs) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {e.printStackTrace();}
                synchronized (consumedOffset) {
                    consumedOffset.add(msg.getQueueOffset());
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    }));
    pushConsumer.getDefaultMQPushConsumerImpl().doRebalance();
    PullMessageService pullMessageService = mQClientFactory.getPullMessageService();
    pullMessageService.executePullRequestImmediately(createPullRequest());
    Thread.sleep(1000);
    pushConsumer.shutdown(10 * 1000);//await consume for at most 10 seconds
    //pushConsumer.shutdown();//here if we do not await, the test case will not pass
    long persitOffset =pushConsumer.getDefaultMQPushConsumerImpl().getOffsetStore().readOffset(new MessageQueue(topic, brokerName, 0), ReadOffsetType.READ_FROM_MEMORY);
    Thread.sleep(1000);//wait for thread pool to continue consume for sometime if not terminated well
    Collections.sort(consumedOffset);
    Assert.assertEquals(consumedOffset.getLast() + 1, persitOffset);//when shutdown with await, the persisted offset should be the latest message offset
}

…ctor

SelectMessageQueueByConsistentHash user the consistent hash algorithm to select msg queue, producer client can reuse the instance for the same topic

The issue tracker is ONLY used for bug report(feature request need to follow RIP process). Keep in mind, please check whether there is an existing same report before your raise a new one.

Alternately (especially if your communication is not a bug report), you can send mail to our mailing lists. We welcome any friendly suggestions, bug fixes, collaboration and other improvements.

Please ensure that your bug report is clear and that it is complete. Otherwise, we may be unable to understand it or to reproduce it, either of which would prevent us from fixing the bug. We strongly recommend the report(bug report or feature request) could include some hints as the following:

BUG REPORT

1. Please describe the issue you observed: 我配置了双主双从同步模式下的master1，默认两个namesrv都已经启动，准备开始启动第一个master broker broker-a.properties配置文件如下： #所属集群名字 brokerClusterName=rocketmq-cluster #broker名字，注意此处不同的配置文件填写的不一样 brokerName=broker-a #0 表示 Master，>0 表示 Slave brokerId=0 #nameServer地址，分号分割 namesrvAddr=192.168.5.100:9876;192.168.5.101:9876 #在发送消息时，自动创建服务器不存在的topic，默认创建的队列数 defaultTopicQueueNums=4 #是否允许 Broker 自动创建Topic，建议线下开启，线上关闭 autoCreateTopicEnable=true #是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭 autoCreateSubscriptionGroup=true #Broker 对外服务的监听端口 listenPort=10911 #删除文件时间点，默认凌晨 4点 deleteWhen=04 #文件保留时间，默认 48 小时 fileReservedTime=120 #commitLog每个文件的大小默认1G mappedFileSizeCommitLog=1073741824 #ConsumeQueue每个文件默认存30W条，根据业务情况调整 #mapedFileSizeConsumeQueue=300000 mappedFileSizeConsumeQueue=6000000 destroyMapedFileIntervalForcibly=120000 #redeleteHangedFileInterval=120000 #检测物理文件磁盘空间 diskMaxUsedSpaceRatio=88 #存储路径 storePathRootDir=/usr/local/bin/rocketmq-all-4.5.2-bin-release/store #commitLog 存储路径 storePathCommitLog=/usr/local/bin/rocketmq-all-4.5.2-bin-release/store/commitlog #消费队列存储路径存储路径 storePathConsumeQueue=/usr/local/bin/rocketmq-all-4.5.2-bin-release/store/consumequeue #消息索引存储路径 storePathIndex=/usr/local/bin/rocketmq-all-4.5.2-bin-release/store/index #checkpoint 文件存储路径 storeCheckpoint=/usr/local/bin/rocketmq-all-4.5.2-bin-release/store/checkpoint #abort 文件存储路径 abortFile=/usr/local/bin/rocketmq-all-4.5.2-bin-release/store/abort #限制的消息大小 maxMessageSize=65536 #flushCommitLogLeastPages=4 #flushConsumeQueueLeastPages=2 #flushCommitLogThoroughInterval=10000 #flushConsumeQueueThoroughInterval=60000 #Broker 的角色 #- ASYNC_MASTER 异步复制Master #- SYNC_MASTER 同步双写Master #- SLAVE brokerRole=SYNC_MASTER #刷盘方式 #- ASYNC_FLUSH 异步刷盘 #- SYNC_FLUSH 同步刷盘 flushDiskType=SYNC_FLUSH #checkTransactionMessageEnable=false #发消息线程池数量 #sendMessageThreadPoolNums=128 #拉消息线程池数量 #pullMessageThreadPoolNums=128

开始启动： nohup sh mqbroker -c /usr/local/bin/rocketmq-all-4.5.2-bin-release/conf/2m-2s-sync/broker-a.properties &

执行上述命令后,无法启动broker-a:

找到日志： 2019-11-24 01:19:42 INFO main - warmMapedFileEnable=false 2019-11-24 01:19:42 INFO main - offsetCheckInSlave=false 2019-11-24 01:19:42 INFO main - debugLockEnable=false 2019-11-24 01:19:42 INFO main - duplicationEnable=false 2019-11-24 01:19:42 INFO main - diskFallRecorded=true 2019-11-24 01:19:42 INFO main - osPageCacheBusyTimeOutMills=1000 2019-11-24 01:19:42 INFO main - defaultQueryMaxNum=32 2019-11-24 01:19:42 INFO main - transientStorePoolEnable=false 2019-11-24 01:19:42 INFO main - transientStorePoolSize=5 2019-11-24 01:19:42 INFO main - fastFailIfNoBufferInStorePool=false 2019-11-24 01:19:42 INFO main - enableDLegerCommitLog=false 2019-11-24 01:19:42 INFO main - dLegerGroup= 2019-11-24 01:19:42 INFO main - dLegerPeers= 2019-11-24 01:19:42 INFO main - dLegerSelfId= 2019-11-24 01:19:42 INFO main - load /usr/local/bin/rocketmq-all-4.5.2-bin-release/store/config/consumerOffset.json OK 2019-11-24 01:19:42 INFO main - load /usr/local/bin/rocketmq-all-4.5.2-bin-release/store/config/consumerFilter.json OK 2019-11-24 01:19:42 INFO main - Try to start service thread:AllocateMappedFileService started:false lastThread:null 2019-11-24 01:19:42 INFO main - Try to shutdown service thread:AllocateMappedFileService started:true lastThread:Thread[AllocateMappedFileService,5,main] 2019-11-24 01:19:42 INFO main - shutdown thread AllocateMappedFileService interrupt true 2019-11-24 01:19:42 INFO main - join thread AllocateMappedFileService elapsed time(ms) 1 90000 2019-11-24 01:19:42 INFO main - Try to shutdown service thread:PullRequestHoldService started:false lastThread:null

最后的日志感觉由点问题，找了好多资料，没把这个问题解决掉，请问我是哪里配置不对吗？ 2. Please tell us about your environment:

RocketMQ 4.5.2 Release版 + JDK8

3. Other information (e.g. detailed explanation, logs, related issues, suggestions how to fix, etc):

FEATURE REQUEST

1. Please describe the feature you are requesting.

2. Provide any additional detail on your proposed use case for this feature.

3. Indicate the importance of this issue to you (blocker, must-have, should-have, nice-to-have). Are you currently using any workarounds to address this issue?

4. If there are some sub-tasks using -[] for each subtask and create a corresponding issue to map to the sub task:

Make sure set the target branch to develop

What is the purpose of the change

close #5800

Brief changelog

• Fix start producer.sh can't find java execute file

Verifying this change

XXXX

Follow this checklist to help us incorporate your contribution quickly and easily. Notice, it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR.

• [x] Make sure there is a Github issue filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue.
• [x] Format the pull request title like [ISSUE #123] Fix UnknownException when host config not exist. Each commit in the pull request should have a meaningful subject line and body.
• [x] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
• [x] Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when cross module dependency exist. If the new feature or significant change is committed, please remember to add integration-test in test module.
• [x] Run mvn -B clean apache-rat:check findbugs:findbugs checkstyle:checkstyle to make sure basic checks pass. Run mvn clean install -DskipITs to make sure unit-test pass. Run mvn clean test-compile failsafe:integration-test to make sure integration-test pass.
• [ ] If this contribution is large, please file an Apache Individual Contributor License Agreement.

The issue tracker is used for bug reporting purposes ONLY whereas feature request needs to follow the RIP process. To avoid unnecessary duplication, please check whether there is a previous issue before filing a new one.

It is recommended to start a discussion thread in the mailing lists in cases of discussing your deployment plan, API clarification, and other non-bug-reporting issues. We welcome any friendly suggestions, bug fixes, collaboration, and other improvements.

Please ensure that your bug report is clear and self-contained. Otherwise, it would take additional rounds of communication, thus more time, to understand the problem itself.

Generally, fixing an issue goes through the following steps:

1. Understand the issue reported;
2. Reproduce the unexpected behavior locally;
3. Perform root cause analysis to identify the underlying problem;
4. Create test cases to cover the identified problem;
5. Work out a solution to rectify the behavior and make the newly created test cases pass;
6. Make a pull request and go through peer review;

As a result, it would be very helpful yet challenging if you could provide an isolated project reproducing your reported issue. Anyway, please ensure your issue report is informative enough for the community to pick up. At a minimum, include the following hints:

BUG REPORT

1. Please describe the issue you observed:

java installed in /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.352.b08-2.al8.x86_64/jre/bin/java

2. Please tell us about your environment:

3. Other information (e.g. detailed explanation, logs, related issues, suggestions on how to fix, etc):

FEATURE REQUEST

1. Please describe the feature you are requesting.

2. Provide any additional detail on your proposed use case for this feature.

3. Indicate the importance of this issue to you (blocker, must-have, should-have, nice-to-have). Are you currently using any workarounds to address this issue?

4. If there are some sub-tasks involved, use -[] for each sub-task and create a corresponding issue to map to the sub-task:

• sub-task1-issue-number: sub-task1 description here,
• sub-task2-issue-number: sub-task2 description here,
• ...

Apache RocketMQ, like all messaging systems, imposes a size limit on each message sent to a broker. This prevents the load per message from exceeding the maxMessageSize set in the broker, which defaults to 4 MB.

However, in application scenarios such as image processing and audio processing, many users need RocketMQ clients to send large messages to brokers. You can achieve this by adjusting maxMessageSize. However, this approach can cause many problems. For example, if a client publishes a 100 MB message, and the broker allows storing the message to disk, the broker will spend a lot of IO resources processing the message. This will affect broker flushing performance and cause a backlog of messages.

Therefore, when RocketMQ sends large messages, in addition to increasing the value of maxMessageSize, we can also support the 'message chunking' function. Through message chunking, producers can split large messages into multiple chunks according to maxMessageSize, and send each chunk to the broker as a normal message. The consumer will then assemble these chunked messages back into the original message.

For example:

Therefore, it's time to discuss whether 'message chunking' should be supported or not.

… missing messages

Make sure set the target branch to develop

What is the purpose of the change

Close #5663

Brief changelog

Add a new state 'isSynchronizingSyncStateSet' in 'AutoSwitchHAService' to Solve the problem of missing messages

Verifying this change

XXXX

Follow this checklist to help us incorporate your contribution quickly and easily. Notice, it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR.

• [x] Make sure there is a Github issue filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue.
• [x] Format the pull request title like [ISSUE #123] Fix UnknownException when host config not exist. Each commit in the pull request should have a meaningful subject line and body.
• [x] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
• [x] Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when cross module dependency exist. If the new feature or significant change is committed, please remember to add integration-test in test module.
• [x] Run mvn -B clean apache-rat:check findbugs:findbugs checkstyle:checkstyle to make sure basic checks pass. Run mvn clean install -DskipITs to make sure unit-test pass. Run mvn clean test-compile failsafe:integration-test to make sure integration-test pass.
• [ ] If this contribution is large, please file an Apache Individual Contributor License Agreement.

Make sure set the target branch to develop

What is the purpose of the change

Support DLedger Controller snapshot RIP-59: SHIMO: https://shimo.im/docs/WlArzngm9JhzY0A2 GOOGLE DOCS: https://docs.google.com/document/d/1jiKtv5WRDqlgZLGYK0oHuZww1r7KKrnTdRGf_eDIQy8/edit?usp=sharing

Brief changelog

Support DLedger Controller snapshot

Verifying this change

Follow this checklist to help us incorporate your contribution quickly and easily. Notice, it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR.

• [x] Make sure there is a Github issue filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue.
• [x] Format the pull request title like [ISSUE #123] Fix UnknownException when host config not exist. Each commit in the pull request should have a meaningful subject line and body.
• [x] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
• [x] Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when cross module dependency exist. If the new feature or significant change is committed, please remember to add integration-test in test module.
• [x] Run mvn -B clean apache-rat:check findbugs:findbugs checkstyle:checkstyle to make sure basic checks pass. Run mvn clean install -DskipITs to make sure unit-test pass. Run mvn clean test-compile failsafe:integration-test to make sure integration-test pass.
• [ ] If this contribution is large, please file an Apache Individual Contributor License Agreement.

Make sure set the target branch to develop

What is the purpose of the change

close #5795

Brief changelog

• Optimize batch unregistration broker service

Verifying this change

XXXX

Follow this checklist to help us incorporate your contribution quickly and easily. Notice, it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR.

• [x] Make sure there is a Github issue filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue.
• [x] Format the pull request title like [ISSUE #123] Fix UnknownException when host config not exist. Each commit in the pull request should have a meaningful subject line and body.
• [x] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
• [x] Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when cross module dependency exist. If the new feature or significant change is committed, please remember to add integration-test in test module.
• [x] Run mvn -B clean apache-rat:check findbugs:findbugs checkstyle:checkstyle to make sure basic checks pass. Run mvn clean install -DskipITs to make sure unit-test pass. Run mvn clean test-compile failsafe:integration-test to make sure integration-test pass.
• [ ] If this contribution is large, please file an Apache Individual Contributor License Agreement.