You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by zh...@apache.org on 2021/09/24 09:07:08 UTC
[rocketmq] branch 5.0.0-preview updated (48d3c7e -> dd09ba1)
This is an automated email from the ASF dual-hosted git repository.
zhoubo pushed a change to branch 5.0.0-preview
in repository https://gitbox.apache.org/repos/asf/rocketmq.git.
from 48d3c7e [maven-release-plugin] prepare for next development iteration
add a027f8d [maven-release-plugin] prepare release rocketmq-all-4.9.0
add a1babab [maven-release-plugin] prepare for next development iteration
add f42db53 Changing the comma input method
add 317fd7e Changing the comma input method
add 52348b8 Merge pull request #2977 from ferris-cx/master
add 2fe6dc3 #2990 Add message tracing switch in sendMessage command of mqadmin
add 57c166b Merge pull request #2991 from yuz10/develop4
add b9c108b Merge remote-tracking branch 'origin/develop_include_3_commits' into develop
add 93dd086 AsyncTraceDispatcher unique groupName
add 0c36d14 remove SubAfter field
add a92996a Merge pull request #3005 from lfz757077613/develop
add 5fe0041 Change default value of some parameters: sendMessageThreadPoolNums/useReentrantLockWhenPutMessage/flushCommitLogTimed/endTransactionThreadPoolNums
add e5a3f7d Merge pull request #2888 from areyouok/dev_speed_D
add e126310 [ISSUE #2997] Add a Chinese version of Configuration_System.md
add 2b236c3 [ISSUE 3019] Fixed bug with batch-sent message statistics
add 160c577 test(benchmark): support delay message test (#3010)
add a0a2192 [ISSUE #2873] add benchmark for batch message
add 6b519ef [ISSUE #3068] remove jaeger tracing dependency in release package
add a2f8810 [ISSUE #2988] fix fail to send trace of last message before shutting down producer
add fcc20de [ISSUE #3059] typo transactional message wiki
add 95a6275 typo
add df35edf Merge pull request #3088 from sunbufu/fix-a-typo
add 9a5ec90 [ISSUE #3109]Return a no-support value when not match any request code
add 56c3ad8 update the option's description of deleteAccessConfig command
add 22647de update the option's description of updateAccessConfig command
add b186bcd update the option's description of consumer.sh
add 4f33895 Merge branch 'develop' into RocketMQ-zhurunhua
add 2a63717 Merge pull request #2882 from zhurunhua/RocketMQ-zhurunhua
add 6591589 [ISSUE #3118] Fix type conversion exception
add 7c36468 [ISSUE #3089] fix Print log format is incorrect in NettyRemotingAbstract class (#3126)
add 6cee569 duplicate doAfterRpcHooks
add 7029545 Merge branch 'apache:develop' into develop
add b74c48b add trace data format test case
add cba3e05 Merge pull request #3124 from lizhiboo/add_trace_test_case
add d65778f [ISSUE #2883] [Part B] Improve produce performance in M/S mode. (#2885)
add 3183122 Eliminate array copy (#2886)
add 7f1cb79 remove useless code
add 2fc1148 Merge pull request #3111 from codersfarm/develop
add 589e87c [ISSUE #3099] Fix the incorrect judegment 'writePos - lastCommittedPosition > commitLeastPages' in the method 'commit0' of class 'MappedFile' (#3125)
add f4b3d4f fix namesrv endpoint
add 3efcd85 Merge pull request #3122 from panzhi33/master
add 94e0899 [ISSUE #3097] docs remove obsolete/error configuration
add 2bd0676 Merge pull request #3116 from trdmm/develop
add d8dec65 add different send examples
add d0d2f15 add chinese version of FAQ.md
add e1f94aa Merge pull request #3034 from PatrickkZ/develop
add 278d328 add Chinese version
add 0cc5ad2 add Chinese version of Example_Batch
add 69e8185 Merge pull request #3036 from Cu-Cu/master
add 622bbbf add some doc of the API_Reference_DefaultMQProducer of English.
add 5908ca7 Merge pull request #3056 from zplovekq/doc_add
add 23f3893 Add a Chinese version of Deployment.md
add 30058cf Merge pull request #3038 from IntroDrop/develop
add 92f2d03 log exception stack
add 83efcce Merge remote-tracking branch 'upstream/develop' into fix-issue-603
add bc73c32 use String.format() in warn log.
add 89d9ed7 Merge pull request #3100 from sunbufu/fix-issue-603
add a99cf37 Add Chinses Version Doc of Example_Simple.md
add e4daba5 Merge pull request #3042 from haoran-pku/master
add 17a8e93 fix spell error
add cf0819e Merge pull request #3144 from A-Zee029/master
add 44e2669 [ISSUE #3128] fix the delete global white addresses
add 4c1998a retrigger ci
add a315833 Merge pull request #3132 from zhangjidi2016/fix_update_globalWhiteAddresses
add 04c7354 delete the unuse code
add 36333ce Merge pull request #3157 from yglovewms/yeyi
add 61cd395 code style(client):polish lite pull consumer code style
add 893e8fd code style(client):polish lite pull consumer code style
add 6aabf77 Merge pull request #3047 from qingjiyuji/develop
add 18e2a26 [ISSUE #3164]Fix the checkstyle error
add bb9a70c Merge pull request #3165 from zhangjidi2016/fix_checkstyle_error
add d7c2d25 [ISSUE #2708] fix offset rollback when fetch offset from broker exception
add 5b945a9 Merge pull request #3158 from Zanglei06/fix_rmq_client_offset_dev
add 5808772 fix checkstyle fail
add df88bce add msgTraceEnable config in transaction producer benchmark
add 011035f Merge pull request #3163 from yuz10/develop6
add ece364a Improve the comment of maxReconsumeTimes
add 35a15b6 Merge pull request #3022 from heihaozi/maxReconsumeTimes
add 25f6711 [statistics] Fix the inconsistency of putMessageDistributeTime statistics
add 765cac3 Merge pull request #3025 from Git-Yang/statistics_bug
add bc3eeed fix: fix issue 2055
add 74aa3e0 Update SendMessageProcessor.java
add c63ec20 Merge pull request #2058 from RealBar/fix/issue2055
add 40e1435 #2990 Add document
add 1e982af Merge pull request #2995 from yuz10/develop4
add 361eb8d Upgrade netty-all dependency version to 4.1.65.Final
add 9a9039e Merge pull request #3032 from messense/upgrade-netty-all
add b954d35 feat: issue2715, make netty parameter configure by system property
add 33a533b feat: add some test case
add 03adb63 feat: add some test case
add 98237b1 feat: add some test case
add 67ceb64 fix: add license
add 59cbbc4 feat: add some test case
add d60643a Merge pull request #2720 from EZLippi/feature/issue2715
add 36422c5 [ISSUE #3066]producing a ghost consumer
add d93dc4d Merge branch 'apache:develop' into develop
add 4e5a60c [ISSUE #2442] when happening system busy, retry
add 062e6f3 revert [ISSUE #2442] when happening system busy, retry
add c75ef50 Merge branch 'apache:develop' into develop
add 8eb275f Merge branch 'apache:develop' into develop
add 36e6687 Merge pull request #3120 from lwclover/develop
add 360b616 [ISSUE #3185]remove the arm machine test in travis ci
add 0325772 [ISSUE #2883] [Part E] Improve produce performance in M/S mode. (#2889)
add b68724b [ISSUE #3030]Remove the -Xmn JVM parameter when using G1 (#3170)
add 800421e registerProducer & unregisterProducer add synchronization too, how about add for producer
add 72128f7 Merge branch 'develop' of https://github.com/lwclover/rocketmq-2.git into develop
add 6f7b0d6 Merge pull request #3182 from lwclover/develop
add 17b9efe [ISSUE #3177]updateAclConfig in all brokers. (#3184)
add 1b757b4 [ISSUE #3196] Check if "requestHeader" is null before "requestHeader.getTopic()"
add 1eaa38f [ISSUE #3200] Fix code style in 'NettySystemConfig.java'
add d067bfc [ISSUE #3192] Make free disk value more accurate
add 20b6cdd [ISSUE 2883] Remove an duplicate MessageDecoder.string2messageProperties for each message, and prevent store "WAIT=true" property (in most case) to save 9 bytes for each message.
add a23df96 [ISSUE 2883] Improve performance of string2messageProperties/messageProperties2String, and save 1 byte for each message.
add 2509b0e Merge pull request #2961 from areyouok/dev_speed_F
add b13a4c8 [ISSUE 2883] [Part G] Optimise parse performance for SendMessageRequestHeaderV2
add a137e48 [ISSUE 2883] [Part G] Optimise parse performance for SendMessageRequestHeaderV2, unit test
add 3e01424 Merge pull request #3134 from areyouok/dev_speed_G
add 94abced [ISSUE #3199] Two timed task for RequestFutureTable (#3202)
add 66eeaed #3175 check acl config before update. otherwise broker may fail to start
add 3aa4943 fix checkstyle
add d21266d Merge branch 'develop' into develop3
add 9e8fea9 Merge pull request #3176 from yuz10/develop3
add 5e58893 prepare to release rocketmq 4.9.1 version
add 73db297 [#Issue3205]Fix the issue of code style in client and broker modules.
add ec985ee Merge pull request #3206 from zongtanghu/develop_fix_4.9.1_codestyle
add a1bfe1a [maven-release-plugin] prepare release rocketmq-all-4.9.1
add a7e71cb [maven-release-plugin] prepare for next development iteration
add 2820e47 [ISSUE #3215] polish litePullConsumer seek logic #3216
add 5bd7ccc Add little rocketmq robot
add d1f9915 Place into workflows directory
add 15b0cde Update greetings.yml
add 5a2aa7e Update greetings.yml
add d83b7e5 Fix typo
add cfadd4d Update greetings.yml
add 39f549a Update greetings.yml
add a20f31b Limit event trigger
add fb8bc64 Merge branch 'develop'
add e9c4253 [maven-release-plugin] prepare release rocketmq-all-4.9.1
add bb34171 [maven-release-plugin] prepare for next development iteration
add 9d4e8de Looking at the Chinese document, I found that the subtraction operation of countdown timer is missing in asynchronous message push, so I downloaded the code. I found that someone added subtraction operation to the demo code in 18 years, but the document was not repaired, so I did this repair
add 0fdc73d Merge pull request #3033 from zyh-future/zyh_asuncproducer_doc_fix
add 3a2b172 [ISSUE #3006]Replace ScheduledExecutorService instead of Timer to avoid affecting other tasks during exception (#3001)
add 44bdeed [ISSUE 3203] Replace the class 'StringBuffer' by 'StringBuilder' (#3204)
add 9e66da4 Merge branch 'develop_include_prs_after_4.9.1_version', remote-tracking branch 'origin' into develop
add 7c7e9ac [ISSUE #3284]Optimizing benchmark code (#3285)
add 2cac866 [ISSUE #3136] TraceDataEncoder add SubAfter trace bean timestamp (#3137)
add 857d28d [ISSUE #3284]Optimizing benchmark code (#3317)
add d7a830c [ISSUE 3194] [PART A] Use LongAdder instead of AtomicLong in BrokerStatsService to improve performance. (#3195)
add fea769d [ISSUE #3284]Optimize the buildMessage method (#3324)
add 7519a53 [ISSUE 3333] Remove commons-codes dependency (#3334)
add ef7685d [ISSUE #3341](doc)fix typo in update operation.md
add b2aa85d [ISSUS #3340] (doc)add hyperlink for client repos in the introduction. (#3338)
add 4b7a90a [#3326] fix send trace fail if useTLS=true (#3325)
add 8434658 [ISSUE #2724] Resend message to DLQ directly when max reconsume times reach (#3318)
add 564ee29 [ISSUE] Add get stats and single queue stats for schedule topic
add 3604100 [ISSUE #3314] Make mqClientApi request timeout settable
add 53c55f7 [ISSUE #3245] Use df algorithm to calculate the disk used ratio
add 83b04d0 [ISSUE #3346]Avoid double parse remoteAddr in rpcHook call
add bf343a3 [ISSUE #3225]change Random to ThreadLocalRandom in broker
add 8d964f4 close #1848
add 1f2b84b fix #1883 fix wipeWritePerm doc desc
add f9a0f53 Merge pull request #1858 from coder-zzzz/github_features/#1848
add 147d23e [ISSUE #3281 ][acl] fix fail to delete topic perm list and global white address(#3128) (#3280)
add 311d76f [ISSUE 3194] [PART C] Replace AtomicLong with LongAdder in StatsItem.java to improve performance (#3351)
add 0a16ff6 DLQ topic default perm change
add b822db0 Fix DLQ can't be query sometimes
add 502ca60 fix ci
add a9ae546 Merge pull request #2437 from maixiaohai/DLQ_perm
add 0de4038 [ISSUE #3361]DefaultMQProducer's constructor can call the overloaded constructor. (#3362)
add 0363be1 add response code for retrying
add d6da221 fix import issue
add 67ccb7f Merge pull request #2729 from wz2cool/master
add 46bd4be [ISSUE #2964] Add a query consumer config command in mqadmin.
add a9837d9 Merge pull request #2965 from zhangjidi2016/add_getConsumerConfig_command
add ced5843 Update rocketmq client maven version
add ee55441 Merge pull request #3335 from zhouwentong1993/patch-1
add 2baae14 Improve performance of expandIP, and readable
add 7a6c374 Revert "Improve performance of expandIP, and readable"
add 6358b3c Revert "Improve performance of expandIP, and readable"
add 97da4ac Improve performance of expandIP, and readable
add 7314183 netadress toUpperCase beforehand
add 0748aa0 Merge pull request #3322 from 844392610/develop_xm
add 67f4357 add checkstyle to travis ci
add 401dc8e production level pull api demo
add 1e8e728 Merge pull request #3295 from lwclover/develop
new 2bf133b Merge branch 'develop' into 5.0.0-preview
new dd09ba1 merger from develop branch and fix compile error
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/greetings.yml | 29 +
README.md | 2 +-
.../org/apache/rocketmq/acl/common/AclUtils.java | 68 +-
.../rocketmq/acl/plain/PlainPermissionManager.java | 53 +-
.../acl/plain/RemoteAddressStrategyFactory.java | 2 +-
.../apache/rocketmq/acl/common/AclUtilsTest.java | 1 +
.../acl/plain/PlainAccessValidatorTest.java | 53 +-
.../broker/pagecache/ManyMessageTransfer.java | 27 +
.../broker/pagecache/OneMessageTransfer.java | 27 +
.../broker/pagecache/QueryMessageTransfer.java | 27 +
.../broker/plugin/AbstractPluginMessageStore.java | 4 +-
.../processor/AbstractSendMessageProcessor.java | 87 ++-
.../broker/processor/AdminBrokerProcessor.java | 20 +-
.../broker/processor/ReplyMessageProcessor.java | 4 +-
.../broker/processor/SendMessageProcessor.java | 120 ++-
.../AbstractTransactionalMessageCheckListener.java | 2 -
.../DefaultTransactionalMessageCheckListener.java | 4 +-
.../AbstractSendMessageProcessorTest.java | 82 ++
client/pom.xml | 4 -
.../org/apache/rocketmq/client/ClientConfig.java | 13 +-
.../client/consumer/DefaultLitePullConsumer.java | 4 +-
.../client/consumer/DefaultMQPushConsumer.java | 18 +-
.../rocketmq/client/impl/MQClientAPIImpl.java | 28 +-
.../ConsumeMessageConcurrentlyService.java | 28 +-
.../consumer/ConsumeMessageOrderlyService.java | 4 +-
.../impl/consumer/DefaultLitePullConsumerImpl.java | 64 +-
.../impl/consumer/DefaultMQPushConsumerImpl.java | 2 +-
.../client/impl/consumer/RebalanceImpl.java | 2 +-
.../impl/consumer/RebalanceLitePullImpl.java | 3 +-
.../client/impl/factory/MQClientInstance.java | 42 +-
.../impl/producer/DefaultMQProducerImpl.java | 60 +-
.../client/producer/DefaultMQProducer.java | 54 +-
.../client/producer/RequestFutureTable.java | 7 +
.../client/trace/AsyncTraceDispatcher.java | 74 +-
.../rocketmq/client/trace/TraceDataEncoder.java | 1 -
.../apache/rocketmq/client/trace/TraceView.java | 4 +-
.../rocketmq/client/impl/MQClientAPIImplTest.java | 27 +
.../ConsumeMessageConcurrentlyServiceTest.java | 2 +-
.../trace/DefaultMQConsumerWithTraceTest.java | 9 +
.../DefaultMQLitePullConsumerWithTraceTest.java | 12 +-
.../trace/DefaultMQProducerWithTraceTest.java | 10 +
.../client/trace/TraceDataEncoderTest.java | 106 ++-
.../rocketmq/client/trace/TraceViewTest.java | 4 +-
.../org/apache/rocketmq/common/BrokerConfig.java | 8 +-
.../java/org/apache/rocketmq/common/MixAll.java | 1 +
.../java/org/apache/rocketmq/common/UtilAll.java | 33 +-
.../apache/rocketmq/common/message/Message.java | 11 +-
.../rocketmq/common/message/MessageDecoder.java | 41 +-
.../rocketmq/common/protocol/NamespaceUtil.java | 2 +-
.../rocketmq/common/protocol/RequestCode.java | 2 +
...java => AddWritePermOfBrokerRequestHeader.java} | 2 +-
...ava => AddWritePermOfBrokerResponseHeader.java} | 13 +-
.../apache/rocketmq/common/stats/StatsItem.java | 23 +-
.../apache/rocketmq/common/stats/StatsItemSet.java | 8 +-
.../common/utils/NameServerAddressUtils.java | 1 +
.../org/apache/rocketmq/common/UtilAllTest.java | 14 +
.../common/message/MessageDecoderTest.java | 108 +++
.../rocketmq/common/stats/StatsItemSetTest.java | 4 +-
.../rocketmq/common/utils/IOTinyUtilsTest.java | 4 +-
.../benchmark/{consumer.sh => batchproducer.sh} | 2 +-
distribution/bin/runbroker.cmd | 2 +-
distribution/bin/runbroker.sh | 2 +-
distribution/bin/runserver.cmd | 2 +-
distribution/bin/runserver.sh | 6 +-
distribution/release.xml | 4 +
docs/cn/Configuration_System.md | 70 ++
docs/cn/Deployment.md | 159 ++++
docs/{en => cn}/Example_Batch.md | 15 +-
docs/{en => cn}/Example_Delay.md | 16 +-
.../Example_Simple.md => cn/Example_Simple_cn.md} | 24 +-
docs/cn/FAQ.md | 110 +++
docs/cn/RocketMQ_Example.md | 8 +-
docs/cn/best_practice.md | 2 +-
docs/cn/design.md | 4 +-
docs/cn/image/rocketmq_architecture_1.png | Bin 89290 -> 62810 bytes
docs/cn/image/rocketmq_architecture_3.png | Bin 106758 -> 74884 bytes
docs/cn/msg_trace/user_guide.md | 19 +-
docs/cn/operation.md | 10 +-
docs/en/CLITools.md | 8 +
docs/en/Example_Transaction.md | 2 +-
docs/en/best_practice.md | 2 +-
.../client/java/API_Reference_DefaultMQProducer.md | 71 ++
docs/en/image/rocketmq_architecture_1.png | Bin 89290 -> 62810 bytes
docs/en/image/rocketmq_architecture_3.png | Bin 106758 -> 74884 bytes
docs/en/msg_trace/user_guide.md | 16 +
docs/en/operation.md | 2 +-
.../rocketmq/example/benchmark/BatchProducer.java | 403 ++++++++++
.../rocketmq/example/benchmark/Consumer.java | 66 +-
.../rocketmq/example/benchmark/Producer.java | 157 ++--
.../example/benchmark/TransactionProducer.java | 117 +--
.../rocketmq/example/quickstart/Producer.java | 33 +
.../rocketmq/example/simple/PullConsumer.java | 154 ++--
.../filter/expression/UnaryExpression.java | 16 +-
.../rocketmq/filter/parser/ParseException.java | 7 +-
.../rocketmq/filter/parser/TokenMgrError.java | 3 +-
.../org/apache/rocketmq/filter/ParserTest.java | 2 +-
.../namesrv/processor/DefaultRequestProcessor.java | 26 +-
.../namesrv/routeinfo/RouteInfoManager.java | 43 +-
.../namesrv/routeinfo/RouteInfoManagerTest.java | 52 +-
pom.xml | 2 +-
.../rocketmq/remoting/common/RemotingHelper.java | 2 +-
.../rocketmq/remoting/netty/NettyClientConfig.java | 8 +-
.../rocketmq/remoting/netty/NettyLogger.java | 46 ++
.../remoting/netty/NettyRemotingAbstract.java | 7 +-
.../rocketmq/remoting/netty/NettySystemConfig.java | 16 +
.../remoting/netty/NettyClientConfigTest.java | 64 ++
.../rocketmq/store/AppendMessageCallback.java | 5 +-
.../apache/rocketmq/store/AppendMessageResult.java | 17 +
.../java/org/apache/rocketmq/store/CommitLog.java | 835 ++++++++-------------
.../apache/rocketmq/store/DefaultMessageStore.java | 73 +-
.../java/org/apache/rocketmq/store/MappedFile.java | 26 +-
.../rocketmq/store/MessageExtBrokerInner.java | 12 +
.../org/apache/rocketmq/store/MessageStore.java | 4 +-
.../rocketmq/store/SelectMappedBufferResult.java | 7 -
.../apache/rocketmq/store/StoreStatsService.java | 109 +--
.../rocketmq/store/config/MessageStoreConfig.java | 7 +-
.../rocketmq/store/dledger/DLedgerCommitLog.java | 250 +-----
.../org/apache/rocketmq/store/ha/HAService.java | 63 +-
.../apache/rocketmq/store/ha/WaitNotifyObject.java | 64 +-
.../store/schedule/ScheduleMessageService.java | 5 +
.../apache/rocketmq/store/stats/BrokerStats.java | 4 +-
.../rocketmq/store/stats/BrokerStatsManager.java | 50 +-
.../apache/rocketmq/store/AppendCallbackTest.java | 28 +-
.../apache/rocketmq/store/BatchPutMessageTest.java | 18 +-
.../rocketmq/store/StoreStatsServiceTest.java | 18 +-
.../store/schedule/ScheduleMessageServiceTest.java | 6 +-
.../test/java/stats/BrokerStatsManagerTest.java | 47 ++
.../rocketmq/tools/admin/DefaultMQAdminExt.java | 8 +-
.../tools/admin/DefaultMQAdminExtImpl.java | 12 +-
.../apache/rocketmq/tools/admin/MQAdminExt.java | 5 +-
.../rocketmq/tools/command/MQAdminStartup.java | 4 +
.../command/acl/DeleteAccessConfigSubCommand.java | 14 +-
.../command/acl/UpdateAccessConfigSubCommand.java | 8 +-
.../acl/UpdateGlobalWhiteAddrSubCommand.java | 6 +-
.../consumer/GetConsumerConfigSubCommand.java | 146 ++++
.../message/QueryMsgByUniqueKeySubCommand.java | 5 +-
.../tools/command/message/SendMessageCommand.java | 17 +-
...SubCommand.java => AddWritePermSubCommand.java} | 29 +-
.../command/namesrv/WipeWritePermSubCommand.java | 2 +-
.../tools/admin/DefaultMQAdminExtTest.java | 7 +
.../GetConsumerConfigSubCommandTest.java} | 11 +-
.../AddWritePermSubCommandTest.java} | 19 +-
142 files changed, 3531 insertions(+), 1659 deletions(-)
create mode 100644 .github/workflows/greetings.yml
create mode 100644 broker/src/test/java/org/apache/rocketmq/broker/processor/AbstractSendMessageProcessorTest.java
copy common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/{WipeWritePermOfBrokerRequestHeader.java => AddWritePermOfBrokerRequestHeader.java} (94%)
copy common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/{WipeWritePermOfBrokerRequestHeader.java => AddWritePermOfBrokerResponseHeader.java} (81%)
copy distribution/benchmark/{consumer.sh => batchproducer.sh} (91%)
create mode 100644 docs/cn/Configuration_System.md
create mode 100644 docs/cn/Deployment.md
copy docs/{en => cn}/Example_Batch.md (79%)
copy docs/{en => cn}/Example_Delay.md (76%)
copy docs/{en/Example_Simple.md => cn/Example_Simple_cn.md} (82%)
create mode 100644 docs/cn/FAQ.md
create mode 100644 docs/en/client/java/API_Reference_DefaultMQProducer.md
create mode 100644 example/src/main/java/org/apache/rocketmq/example/benchmark/BatchProducer.java
create mode 100644 remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java
create mode 100644 tools/src/main/java/org/apache/rocketmq/tools/command/consumer/GetConsumerConfigSubCommand.java
copy tools/src/main/java/org/apache/rocketmq/tools/command/namesrv/{WipeWritePermSubCommand.java => AddWritePermSubCommand.java} (77%)
copy tools/src/test/java/org/apache/rocketmq/tools/command/{broker/UpdateBrokerConfigSubCommandTest.java => consumer/GetConsumerConfigSubCommandTest.java} (92%)
copy tools/src/test/java/org/apache/rocketmq/tools/command/{acl/GetAccessConfigSubCommandTest.java => namesrv/AddWritePermSubCommandTest.java} (68%)
[rocketmq] 01/02: Merge branch 'develop' into 5.0.0-preview
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhoubo pushed a commit to branch 5.0.0-preview
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit 2bf133be29c444f0aea648133123a2d9a61ffb07
Merge: 48d3c7e 1e8e728
Author: odbozhou <87...@qq.com>
AuthorDate: Fri Sep 24 15:35:46 2021 +0800
Merge branch 'develop' into 5.0.0-preview
# Conflicts:
# .travis.yml
# acl/pom.xml
# broker/pom.xml
# broker/src/main/java/org/apache/rocketmq/broker/processor/AbstractSendMessageProcessor.java
# broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
# broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
# client/pom.xml
# client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalanceImpl.java
# client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
# client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
# common/pom.xml
# common/src/main/java/org/apache/rocketmq/common/MQVersion.java
# common/src/main/java/org/apache/rocketmq/common/protocol/RequestCode.java
# distribution/pom.xml
# example/pom.xml
# filter/pom.xml
# logappender/pom.xml
# logging/pom.xml
# namesrv/pom.xml
# namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java
# namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java
# openmessaging/pom.xml
# pom.xml
# remoting/pom.xml
# srvutil/pom.xml
# store/pom.xml
# store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
# test/pom.xml
# tools/pom.xml
# tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
.github/workflows/greetings.yml | 29 +
README.md | 2 +-
.../org/apache/rocketmq/acl/common/AclUtils.java | 69 +-
.../rocketmq/acl/plain/PlainPermissionManager.java | 53 +-
.../acl/plain/RemoteAddressStrategyFactory.java | 2 +-
.../apache/rocketmq/acl/common/AclUtilsTest.java | 1 +
.../acl/plain/PlainAccessValidatorTest.java | 53 +-
.../broker/pagecache/ManyMessageTransfer.java | 27 +
.../broker/pagecache/OneMessageTransfer.java | 27 +
.../broker/pagecache/QueryMessageTransfer.java | 27 +
.../broker/plugin/AbstractPluginMessageStore.java | 4 +-
.../processor/AbstractSendMessageProcessor.java | 89 ++-
.../broker/processor/AdminBrokerProcessor.java | 20 +-
.../broker/processor/ReplyMessageProcessor.java | 4 +-
.../broker/processor/SendMessageProcessor.java | 120 ++-
.../AbstractTransactionalMessageCheckListener.java | 2 -
.../DefaultTransactionalMessageCheckListener.java | 4 +-
.../AbstractSendMessageProcessorTest.java | 82 ++
client/pom.xml | 4 -
.../org/apache/rocketmq/client/ClientConfig.java | 13 +-
.../client/consumer/DefaultLitePullConsumer.java | 4 +-
.../client/consumer/DefaultMQPushConsumer.java | 18 +-
.../rocketmq/client/impl/MQClientAPIImpl.java | 28 +-
.../ConsumeMessageConcurrentlyService.java | 28 +-
.../consumer/ConsumeMessageOrderlyService.java | 4 +-
.../impl/consumer/DefaultLitePullConsumerImpl.java | 64 +-
.../impl/consumer/DefaultMQPushConsumerImpl.java | 2 +-
.../client/impl/consumer/RebalanceImpl.java | 2 +-
.../impl/consumer/RebalanceLitePullImpl.java | 3 +-
.../client/impl/factory/MQClientInstance.java | 42 +-
.../impl/producer/DefaultMQProducerImpl.java | 61 +-
.../client/producer/DefaultMQProducer.java | 54 +-
.../client/producer/RequestFutureTable.java | 7 +
.../client/trace/AsyncTraceDispatcher.java | 74 +-
.../rocketmq/client/trace/TraceDataEncoder.java | 1 -
.../apache/rocketmq/client/trace/TraceView.java | 4 +-
.../rocketmq/client/impl/MQClientAPIImplTest.java | 27 +
.../ConsumeMessageConcurrentlyServiceTest.java | 2 +-
.../trace/DefaultMQConsumerWithTraceTest.java | 9 +
.../DefaultMQLitePullConsumerWithTraceTest.java | 12 +-
.../trace/DefaultMQProducerWithTraceTest.java | 10 +
.../client/trace/TraceDataEncoderTest.java | 106 ++-
.../rocketmq/client/trace/TraceViewTest.java | 4 +-
.../org/apache/rocketmq/common/BrokerConfig.java | 8 +-
.../java/org/apache/rocketmq/common/MixAll.java | 1 +
.../java/org/apache/rocketmq/common/UtilAll.java | 33 +-
.../apache/rocketmq/common/message/Message.java | 11 +-
.../rocketmq/common/message/MessageDecoder.java | 41 +-
.../rocketmq/common/protocol/NamespaceUtil.java | 2 +-
.../rocketmq/common/protocol/RequestCode.java | 2 +
.../namesrv/AddWritePermOfBrokerRequestHeader.java | 39 +
.../AddWritePermOfBrokerResponseHeader.java | 38 +
.../apache/rocketmq/common/stats/StatsItem.java | 22 +-
.../apache/rocketmq/common/stats/StatsItemSet.java | 8 +-
.../common/utils/NameServerAddressUtils.java | 1 +
.../org/apache/rocketmq/common/UtilAllTest.java | 13 +
.../common/message/MessageDecoderTest.java | 108 +++
.../rocketmq/common/stats/StatsItemSetTest.java | 4 +-
.../rocketmq/common/utils/IOTinyUtilsTest.java | 4 +-
distribution/benchmark/batchproducer.sh | 18 +
distribution/bin/runbroker.cmd | 2 +-
distribution/bin/runbroker.sh | 2 +-
distribution/bin/runserver.cmd | 2 +-
distribution/bin/runserver.sh | 6 +-
distribution/release.xml | 4 +
docs/cn/Configuration_System.md | 70 ++
docs/cn/Deployment.md | 159 ++++
docs/cn/Example_Batch.md | 82 ++
docs/cn/Example_Delay.md | 85 +++
docs/cn/Example_Simple_cn.md | 136 ++++
docs/cn/FAQ.md | 110 +++
docs/cn/RocketMQ_Example.md | 8 +-
docs/cn/best_practice.md | 2 +-
docs/cn/design.md | 4 +-
docs/cn/image/rocketmq_architecture_1.png | Bin 89290 -> 62810 bytes
docs/cn/image/rocketmq_architecture_3.png | Bin 106758 -> 74884 bytes
docs/cn/msg_trace/user_guide.md | 19 +-
docs/cn/operation.md | 10 +-
docs/en/CLITools.md | 8 +
docs/en/Example_Transaction.md | 2 +-
docs/en/best_practice.md | 2 +-
.../client/java/API_Reference_DefaultMQProducer.md | 71 ++
docs/en/image/rocketmq_architecture_1.png | Bin 89290 -> 62810 bytes
docs/en/image/rocketmq_architecture_3.png | Bin 106758 -> 74884 bytes
docs/en/msg_trace/user_guide.md | 16 +
docs/en/operation.md | 2 +-
.../rocketmq/example/benchmark/BatchProducer.java | 403 ++++++++++
.../rocketmq/example/benchmark/Consumer.java | 66 +-
.../rocketmq/example/benchmark/Producer.java | 157 ++--
.../example/benchmark/TransactionProducer.java | 117 +--
.../rocketmq/example/quickstart/Producer.java | 33 +
.../rocketmq/example/simple/PullConsumer.java | 154 ++--
.../filter/expression/UnaryExpression.java | 16 +-
.../rocketmq/filter/parser/ParseException.java | 7 +-
.../rocketmq/filter/parser/TokenMgrError.java | 3 +-
.../org/apache/rocketmq/filter/ParserTest.java | 2 +-
.../namesrv/processor/DefaultRequestProcessor.java | 24 +
.../namesrv/routeinfo/RouteInfoManager.java | 41 +-
.../namesrv/routeinfo/RouteInfoManagerTest.java | 52 +-
pom.xml | 2 +-
.../rocketmq/remoting/common/RemotingHelper.java | 2 +-
.../rocketmq/remoting/netty/NettyClientConfig.java | 8 +-
.../rocketmq/remoting/netty/NettyLogger.java | 46 ++
.../remoting/netty/NettyRemotingAbstract.java | 7 +-
.../rocketmq/remoting/netty/NettySystemConfig.java | 16 +
.../remoting/netty/NettyClientConfigTest.java | 64 ++
.../rocketmq/store/AppendMessageCallback.java | 5 +-
.../apache/rocketmq/store/AppendMessageResult.java | 17 +
.../java/org/apache/rocketmq/store/CommitLog.java | 835 ++++++++-------------
.../apache/rocketmq/store/DefaultMessageStore.java | 73 +-
.../java/org/apache/rocketmq/store/MappedFile.java | 26 +-
.../rocketmq/store/MessageExtBrokerInner.java | 12 +
.../org/apache/rocketmq/store/MessageStore.java | 4 +-
.../rocketmq/store/SelectMappedBufferResult.java | 7 -
.../apache/rocketmq/store/StoreStatsService.java | 109 +--
.../rocketmq/store/config/MessageStoreConfig.java | 7 +-
.../rocketmq/store/dledger/DLedgerCommitLog.java | 250 +-----
.../org/apache/rocketmq/store/ha/HAService.java | 63 +-
.../apache/rocketmq/store/ha/WaitNotifyObject.java | 64 +-
.../store/schedule/ScheduleMessageService.java | 5 +
.../apache/rocketmq/store/stats/BrokerStats.java | 4 +-
.../rocketmq/store/stats/BrokerStatsManager.java | 50 +-
.../apache/rocketmq/store/AppendCallbackTest.java | 28 +-
.../apache/rocketmq/store/BatchPutMessageTest.java | 18 +-
.../rocketmq/store/StoreStatsServiceTest.java | 18 +-
.../store/schedule/ScheduleMessageServiceTest.java | 6 +-
.../test/java/stats/BrokerStatsManagerTest.java | 47 ++
.../rocketmq/tools/admin/DefaultMQAdminExt.java | 8 +-
.../tools/admin/DefaultMQAdminExtImpl.java | 12 +-
.../apache/rocketmq/tools/admin/MQAdminExt.java | 5 +-
.../rocketmq/tools/command/MQAdminStartup.java | 4 +
.../command/acl/DeleteAccessConfigSubCommand.java | 14 +-
.../command/acl/UpdateAccessConfigSubCommand.java | 8 +-
.../acl/UpdateGlobalWhiteAddrSubCommand.java | 6 +-
.../consumer/GetConsumerConfigSubCommand.java | 146 ++++
.../message/QueryMsgByUniqueKeySubCommand.java | 5 +-
.../tools/command/message/SendMessageCommand.java | 17 +-
...SubCommand.java => AddWritePermSubCommand.java} | 29 +-
.../command/namesrv/WipeWritePermSubCommand.java | 2 +-
.../tools/admin/DefaultMQAdminExtTest.java | 7 +
.../consumer/GetConsumerConfigSubCommandTest.java | 83 ++
.../namesrv/AddWritePermSubCommandTest.java | 37 +
142 files changed, 3998 insertions(+), 1606 deletions(-)
diff --cc broker/src/main/java/org/apache/rocketmq/broker/processor/AbstractSendMessageProcessor.java
index 4daa832,9d26e99..d2f751b
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/AbstractSendMessageProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/AbstractSendMessageProcessor.java
@@@ -21,19 -19,15 +21,22 @@@ import com.google.common.collect.Maps
import io.netty.channel.ChannelHandlerContext;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
+ import java.util.HashMap;
import java.util.List;
-import java.util.Map;
+import java.util.Locale;
- import java.util.Map;
++import java.util.Map;Optional
++
+ import java.util.concurrent.ThreadLocalRandom;
+
+import java.util.Optional;
- import java.util.Random;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.LongAdder;
import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.domain.LogicalQueuesInfoInBroker;
import org.apache.rocketmq.broker.mqtrace.SendMessageContext;
import org.apache.rocketmq.broker.mqtrace.SendMessageHook;
-import org.apache.rocketmq.common.topic.TopicValidator;
+import org.apache.rocketmq.broker.topic.TopicConfigManager;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.TopicFilterType;
diff --cc broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
index b3309e1,86aab63..e7b7949
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
@@@ -154,7 -116,7 +154,8 @@@ import org.apache.rocketmq.remoting.net
import org.apache.rocketmq.remoting.protocol.LanguageCode;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+ import org.apache.rocketmq.remoting.protocol.RemotingSysResponseCode;
+import org.apache.rocketmq.srvutil.ConcurrentHashMapUtil;
import org.apache.rocketmq.store.ConsumeQueue;
import org.apache.rocketmq.store.ConsumeQueueExt;
import org.apache.rocketmq.store.DefaultMessageStore;
@@@ -261,31 -235,9 +262,29 @@@ public class AdminBrokerProcessor exten
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_TOPIC_LOGICAL_QUEUE_MAPPING:
+ return updateTopicLogicalQueueMapping(ctx, request);
+ case RequestCode.DELETE_TOPIC_LOGICAL_QUEUE_MAPPING:
+ return deleteTopicLogicalQueueMapping(ctx, request);
+ case RequestCode.QUERY_TOPIC_LOGICAL_QUEUE_MAPPING:
+ return queryTopicLogicalQueueMapping(ctx, request);
+ case RequestCode.SEAL_TOPIC_LOGICAL_QUEUE:
+ return sealTopicLogicalQueue(ctx, request);
+ case RequestCode.REUSE_TOPIC_LOGICAL_QUEUE:
+ return reuseTopicLogicalQueue(ctx, request);
+ case RequestCode.CREATE_MESSAGE_QUEUE_FOR_LOGICAL_QUEUE:
+ return createMessageQueueForLogicalQueue(ctx, request);
+ case RequestCode.MIGRATE_TOPIC_LOGICAL_QUEUE_PREPARE:
+ return migrateTopicLogicalQueuePrepare(ctx, request);
+ case RequestCode.MIGRATE_TOPIC_LOGICAL_QUEUE_COMMIT:
+ return migrateTopicLogicalQueueCommit(ctx, request);
+ case RequestCode.MIGRATE_TOPIC_LOGICAL_QUEUE_NOTIFY:
+ return migrateTopicLogicalQueueNotify(ctx, request);
default:
- break;
+ return getUnknownCmdResponse(ctx, request);
}
-
- return null;
}
@Override
diff --cc broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
index 97b7e62,3a401e1..692c98c
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
@@@ -285,16 -292,18 +292,24 @@@ public class SendMessageProcessor exten
msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes());
String clusterName = this.brokerController.getBrokerConfig().getBrokerClusterName();
MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_CLUSTER, clusterName);
- msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
+ if (origProps.containsKey(MessageConst.PROPERTY_WAIT_STORE_MSG_OK)) {
+ // There is no need to store "WAIT=true", remove it from propertiesString to save 9 bytes for each message.
+ // It works for most case. In some cases msgInner.setPropertiesString invoked later and replace it.
+ String waitStoreMsgOKValue = origProps.remove(MessageConst.PROPERTY_WAIT_STORE_MSG_OK);
+ msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
+ // Reput to properties, since msgInner.isWaitStoreMsgOK() will be invoked later
+ origProps.put(MessageConst.PROPERTY_WAIT_STORE_MSG_OK, waitStoreMsgOKValue);
+ } else {
+ msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
+ }
+ LogicalQueueContext logicalQueueContext = super.buildLogicalQueueContext(msgInner.getTopic(), msgInner.getQueueId(), response);
+ CompletableFuture<RemotingCommand> future = logicalQueueContext.hookBeforePut(ctx, requestHeader, request, response);
+ if (future != null) {
+ return future;
+ }
+
CompletableFuture<PutMessageResult> putMessageResult = null;
- Map<String, String> origProps = MessageDecoder.string2messageProperties(requestHeader.getProperties());
String transFlag = origProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
if (transFlag != null && Boolean.parseBoolean(transFlag)) {
if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {
diff --cc client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalanceImpl.java
index 7a457c1,7677d8b..ab0d885
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalanceImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalanceImpl.java
@@@ -557,201 -401,9 +557,201 @@@ public abstract class RebalanceImpl
log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq);
}
}
+
+ }
+
+ if (!allMQLocked) {
+ mQClientFactory.rebalanceLater(500);
+ }
+
+ this.dispatchPullRequest(pullRequestList, 500);
+
+ return changed;
+ }
+
+ private boolean updateMessageQueueAssignment(final String topic, final Set<MessageQueueAssignment> assignments,
+ final boolean isOrder) {
+ boolean changed = false;
+
+ Map<MessageQueue, MessageQueueAssignment> mq2PushAssignment = new HashMap<MessageQueue, MessageQueueAssignment>();
+ Map<MessageQueue, MessageQueueAssignment> mq2PopAssignment = new HashMap<MessageQueue, MessageQueueAssignment>();
+ for (MessageQueueAssignment assignment : assignments) {
+ MessageQueue messageQueue = assignment.getMessageQueue();
+ if (messageQueue == null) {
+ continue;
+ }
+ if (MessageRequestMode.POP == assignment.getMode()) {
+ mq2PopAssignment.put(messageQueue, assignment);
+ } else {
+ mq2PushAssignment.put(messageQueue, assignment);
+ }
+ }
+
+ if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
+ if (mq2PopAssignment.isEmpty() && !mq2PushAssignment.isEmpty()) {
+ //pop switch to push
+ //subscribe pop retry topic
+ try {
+ final String retryTopic = KeyBuilder.buildPopRetryTopic(topic, getConsumerGroup());
+ SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(retryTopic, SubscriptionData.SUB_ALL);
+ getSubscriptionInner().put(retryTopic, subscriptionData);
+ } catch (Exception ignored) {
+ }
+
+ } else if (!mq2PopAssignment.isEmpty() && mq2PushAssignment.isEmpty()) {
+ //push switch to pop
+ //unsubscribe pop retry topic
+ try {
+ final String retryTopic = KeyBuilder.buildPopRetryTopic(topic, getConsumerGroup());
+ getSubscriptionInner().remove(retryTopic);
+ } catch (Exception ignored) {
+ }
+
+ }
+ }
+
+ {
+ // drop process queues no longer belong me
+ HashMap<MessageQueue, ProcessQueue> removeQueueMap = new HashMap<MessageQueue, ProcessQueue>(this.processQueueTable.size());
+ Iterator<Entry<MessageQueue, ProcessQueue>> it = this.processQueueTable.entrySet().iterator();
+ while (it.hasNext()) {
+ Entry<MessageQueue, ProcessQueue> next = it.next();
+ MessageQueue mq = next.getKey();
+ ProcessQueue pq = next.getValue();
+
+ if (mq.getTopic().equals(topic)) {
+ if (!mq2PushAssignment.containsKey(mq)) {
+ pq.setDropped(true);
+ removeQueueMap.put(mq, pq);
+ } else if (pq.isPullExpired() && this.consumeType() == ConsumeType.CONSUME_PASSIVELY) {
+ pq.setDropped(true);
+ removeQueueMap.put(mq, pq);
+ log.error("[BUG]doRebalance, {}, try remove unnecessary mq, {}, because pull is pause, so try to fixed it",
+ consumerGroup, mq);
+ }
+ }
+ }
+ // remove message queues no longer belong me
+ for (Entry<MessageQueue, ProcessQueue> entry : removeQueueMap.entrySet()) {
+ MessageQueue mq = entry.getKey();
+ ProcessQueue pq = entry.getValue();
+
+ if (this.removeUnnecessaryMessageQueue(mq, pq)) {
+ this.processQueueTable.remove(mq);
+ changed = true;
+ log.info("doRebalance, {}, remove unnecessary mq, {}", consumerGroup, mq);
+ }
+ }
+ }
+
+ {
+ HashMap<MessageQueue, PopProcessQueue> removeQueueMap = new HashMap<MessageQueue, PopProcessQueue>(this.popProcessQueueTable.size());
+ Iterator<Entry<MessageQueue, PopProcessQueue>> it = this.popProcessQueueTable.entrySet().iterator();
+ while (it.hasNext()) {
+ Entry<MessageQueue, PopProcessQueue> next = it.next();
+ MessageQueue mq = next.getKey();
+ PopProcessQueue pq = next.getValue();
+
+ if (mq.getTopic().equals(topic)) {
+ if (!mq2PopAssignment.containsKey(mq)) {
+ //the queue is no longer your assignment
+ pq.setDropped(true);
+ removeQueueMap.put(mq, pq);
+ } else if (pq.isPullExpired() && this.consumeType() == ConsumeType.CONSUME_PASSIVELY) {
+ pq.setDropped(true);
+ removeQueueMap.put(mq, pq);
+ log.error("[BUG]doRebalance, {}, try remove unnecessary pop mq, {}, because pop is pause, so try to fixed it",
+ consumerGroup, mq);
+ }
+ }
+ }
+ // remove message queues no longer belong me
+ for (Entry<MessageQueue, PopProcessQueue> entry : removeQueueMap.entrySet()) {
+ MessageQueue mq = entry.getKey();
+ PopProcessQueue pq = entry.getValue();
+
+ if (this.removeUnnecessaryPopMessageQueue(mq, pq)) {
+ this.popProcessQueueTable.remove(mq);
+ changed = true;
+ log.info("doRebalance, {}, remove unnecessary pop mq, {}", consumerGroup, mq);
+ }
+ }
+ }
+
+ {
+ // add new message queue
+ boolean allMQLocked = true;
+ List<PullRequest> pullRequestList = new ArrayList<PullRequest>();
+ for (MessageQueue mq : mq2PushAssignment.keySet()) {
+ if (!this.processQueueTable.containsKey(mq)) {
+ if (isOrder && !this.lock(mq)) {
+ log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);
+ allMQLocked = false;
+ continue;
+ }
+
+ this.removeDirtyOffset(mq);
+ ProcessQueue pq = createProcessQueue();
+ pq.setLocked(true);
+ long nextOffset = -1L;
+ try {
+ nextOffset = this.computePullFromWhereWithException(mq);
- } catch (MQClientException e) {
++ } catch (Exception e) {
+ log.info("doRebalance, {}, compute offset failed, {}", consumerGroup, mq);
+ continue;
+ }
+
+ if (nextOffset >= 0) {
+ ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
+ if (pre != null) {
+ log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);
+ } else {
+ log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);
+ PullRequest pullRequest = new PullRequest();
+ pullRequest.setConsumerGroup(consumerGroup);
+ pullRequest.setNextOffset(nextOffset);
+ pullRequest.setMessageQueue(mq);
+ pullRequest.setProcessQueue(pq);
+ pullRequestList.add(pullRequest);
+ changed = true;
+ }
+ } else {
+ log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq);
+ }
+ }
+ }
+
+ if (!allMQLocked) {
+ mQClientFactory.rebalanceLater(500);
+ }
+ this.dispatchPullRequest(pullRequestList, 500);
}
- this.dispatchPullRequest(pullRequestList);
+ {
+ // add new message queue
+ List<PopRequest> popRequestList = new ArrayList<PopRequest>();
+ for (MessageQueue mq : mq2PopAssignment.keySet()) {
+ if (!this.popProcessQueueTable.containsKey(mq)) {
+ PopProcessQueue pq = createPopProcessQueue();
+ PopProcessQueue pre = this.popProcessQueueTable.putIfAbsent(mq, pq);
+ if (pre != null) {
+ log.info("doRebalance, {}, mq pop already exists, {}", consumerGroup, mq);
+ } else {
+ log.info("doRebalance, {}, add a new pop mq, {}", consumerGroup, mq);
+ PopRequest popRequest = new PopRequest();
+ popRequest.setTopic(topic);
+ popRequest.setConsumerGroup(consumerGroup);
+ popRequest.setMessageQueue(mq);
+ popRequest.setPopProcessQueue(pq);
+ popRequest.setInitMode(getConsumeInitMode());
+ popRequestList.add(popRequest);
+ changed = true;
+ }
+ }
+ }
+
+ this.dispatchPopPullRequest(popRequestList, 500);
+ }
return changed;
}
diff --cc client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
index c1a50dd,e897d49..dafc4f8
--- 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
@@@ -676,7 -619,7 +676,7 @@@ public class MQClientInstance
}
}
} else {
- topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3, true, logicalQueueIdsFilter);
- topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, clientConfig.getMqClientApiTimeout());
++ topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, clientConfig.getMqClientApiTimeout(), true, logicalQueueIdsFilter);
}
if (topicRouteData != null) {
TopicRouteData old = this.topicRouteTable.get(topic);
diff --cc client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
index 8802a9c,bdc103f..ba479d2
--- a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
@@@ -22,14 -20,10 +22,12 @@@ import java.io.IOException
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
+import java.util.Locale;
import java.util.Random;
import java.util.Set;
- import java.util.Timer;
- import java.util.TimerTask;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
diff --cc client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
index c91d55a,e1b3bed..34b34f9
--- 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,42 -25,14 +37,44 @@@ import org.apache.rocketmq.client.produ
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.DataVersion;
import org.apache.rocketmq.common.PlainAccessConfig;
+import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageConst;
+ import org.apache.rocketmq.common.protocol.RequestCode;
+import org.apache.rocketmq.common.message.MessageDecoder;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueueAssignment;
+import org.apache.rocketmq.common.message.MessageRequestMode;
import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.apache.rocketmq.common.protocol.body.ClusterAclVersionInfo;
+import org.apache.rocketmq.common.protocol.body.QueryAssignmentResponseBody;
+import org.apache.rocketmq.common.protocol.header.AckMessageRequestHeader;
+import org.apache.rocketmq.common.protocol.header.ChangeInvisibleTimeRequestHeader;
+import org.apache.rocketmq.common.protocol.header.ChangeInvisibleTimeResponseHeader;
+import org.apache.rocketmq.common.protocol.header.ExtraInfoUtil;
+import org.apache.rocketmq.common.protocol.header.GetBrokerAclConfigResponseHeader;
+import org.apache.rocketmq.common.protocol.header.GetBrokerClusterAclConfigResponseBody;
+import org.apache.rocketmq.common.protocol.header.GetBrokerClusterAclConfigResponseHeader;
+import org.apache.rocketmq.common.protocol.header.GetConsumerListByGroupResponseBody;
+import org.apache.rocketmq.common.protocol.header.GetConsumerListByGroupResponseHeader;
+import org.apache.rocketmq.common.protocol.header.GetEarliestMsgStoretimeResponseHeader;
+import org.apache.rocketmq.common.protocol.header.GetMaxOffsetResponseHeader;
+import org.apache.rocketmq.common.protocol.header.GetMinOffsetResponseHeader;
+import org.apache.rocketmq.common.protocol.header.PopMessageRequestHeader;
+import org.apache.rocketmq.common.protocol.header.PopMessageResponseHeader;
+import org.apache.rocketmq.common.protocol.header.QueryConsumerOffsetRequestHeader;
+import org.apache.rocketmq.common.protocol.header.QueryConsumerOffsetResponseHeader;
+import org.apache.rocketmq.common.protocol.header.SearchOffsetResponseHeader;
import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
import org.apache.rocketmq.common.protocol.header.SendMessageResponseHeader;
+ import org.apache.rocketmq.common.protocol.header.namesrv.AddWritePermOfBrokerResponseHeader;
+import org.apache.rocketmq.common.protocol.header.UpdateConsumerOffsetRequestHeader;
+import org.apache.rocketmq.common.protocol.header.UpdateConsumerOffsetResponseHeader;
+import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
+import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.remoting.InvokeCallback;
import org.apache.rocketmq.remoting.RemotingClient;
import org.apache.rocketmq.remoting.exception.RemotingException;
diff --cc common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
index 17cc2a1,f710cdb..bf2355c
--- a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
+++ b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
@@@ -58,12 -57,10 +58,11 @@@ public class BrokerConfig
@ImportantField
private boolean traceTopicEnable = false;
/**
- * thread numbers for send message thread pool, since spin lock will be used by default since 4.0.x, the default
- * value is 1.
+ * thread numbers for send message thread pool.
*/
- private int sendMessageThreadPoolNums = 1; //16 + Runtime.getRuntime().availableProcessors() * 4;
+ private int sendMessageThreadPoolNums = Math.min(Runtime.getRuntime().availableProcessors(), 4);
private int pullMessageThreadPoolNums = 16 + Runtime.getRuntime().availableProcessors() * 2;
+ private int ackMessageThreadPoolNums = 3;
private int processReplyMessageThreadPoolNums = 16 + Runtime.getRuntime().availableProcessors() * 2;
private int queryMessageThreadPoolNums = 8 + Runtime.getRuntime().availableProcessors();
diff --cc common/src/main/java/org/apache/rocketmq/common/protocol/RequestCode.java
index 3613049,5624a7e..04f126b
--- a/common/src/main/java/org/apache/rocketmq/common/protocol/RequestCode.java
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/RequestCode.java
@@@ -193,18 -189,5 +193,20 @@@ public class RequestCode
public static final int PUSH_REPLY_MESSAGE_TO_CLIENT = 326;
+ public static final int ADD_WRITE_PERM_OF_BROKER = 327;
++
+ public static final int GET_TOPIC_CONFIG = 351;
+
+ public static final int QUERY_ASSIGNMENT = 400;
+ public static final int SET_MESSAGE_REQUEST_MODE = 401;
+
+ public static final int UPDATE_TOPIC_LOGICAL_QUEUE_MAPPING = 411;
+ public static final int DELETE_TOPIC_LOGICAL_QUEUE_MAPPING = 422;
+ public static final int QUERY_TOPIC_LOGICAL_QUEUE_MAPPING = 413;
+ public static final int SEAL_TOPIC_LOGICAL_QUEUE = 414;
+ public static final int REUSE_TOPIC_LOGICAL_QUEUE = 415;
+ public static final int CREATE_MESSAGE_QUEUE_FOR_LOGICAL_QUEUE = 416;
+ public static final int MIGRATE_TOPIC_LOGICAL_QUEUE_PREPARE = 417;
+ public static final int MIGRATE_TOPIC_LOGICAL_QUEUE_COMMIT = 418;
+ public static final int MIGRATE_TOPIC_LOGICAL_QUEUE_NOTIFY = 419;
}
diff --cc store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
index b11eb49,5bf68ac..152af7b
--- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+++ b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
@@@ -35,7 -34,7 +35,8 @@@ import java.util.Set
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+ import java.util.concurrent.ExecutionException;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@@ -675,8 -634,8 +638,8 @@@ public class DefaultMessageStore implem
continue;
}
- this.storeStatsService.getGetMessageTransferedMsgCount().incrementAndGet();
+ this.storeStatsService.getGetMessageTransferedMsgCount().add(1);
- getResult.addMessage(selectResult);
+ getResult.addMessage(selectResult, offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE));
status = GetMessageStatus.FOUND;
nextPhyFileStartOffset = Long.MIN_VALUE;
}
diff --cc store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java
index 9057ebe,1164ab8..c127515
--- a/store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java
+++ b/store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java
@@@ -26,10 -24,13 +26,11 @@@ import java.util.TimerTask
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
import org.apache.rocketmq.common.ConfigManager;
+ import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicFilterType;
import org.apache.rocketmq.common.constant.LoggerName;
-import org.apache.rocketmq.common.topic.TopicValidator;
-import org.apache.rocketmq.logging.InternalLogger;
-import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
diff --cc tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java
index c1b42f5,d701056..0da449d
--- a/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java
@@@ -98,10 -90,9 +98,10 @@@ public interface MQAdminExt extends MQA
final SubscriptionGroupConfig config) throws RemotingException,
MQBrokerException, InterruptedException, MQClientException;
- SubscriptionGroupConfig examineSubscriptionGroupConfig(final String addr, final String group);
+ SubscriptionGroupConfig examineSubscriptionGroupConfig(final String addr, final String group) throws InterruptedException, RemotingException, MQClientException, MQBrokerException;
- TopicConfig examineTopicConfig(final String addr, final String topic);
+ TopicConfig examineTopicConfig(final String addr,
+ final String topic) throws InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException;
TopicStatsTable examineTopicStats(
final String topic) throws RemotingException, MQClientException, InterruptedException,
diff --cc tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
index 3e98939,4411a6c..1d29959
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
@@@ -49,14 -49,9 +49,15 @@@ import org.apache.rocketmq.tools.comman
import org.apache.rocketmq.tools.command.consumer.ConsumerProgressSubCommand;
import org.apache.rocketmq.tools.command.consumer.ConsumerStatusSubCommand;
import org.apache.rocketmq.tools.command.consumer.DeleteSubscriptionGroupCommand;
+ import org.apache.rocketmq.tools.command.consumer.GetConsumerConfigSubCommand;
+import org.apache.rocketmq.tools.command.consumer.SetConsumeModeSubCommand;
import org.apache.rocketmq.tools.command.consumer.StartMonitoringSubCommand;
import org.apache.rocketmq.tools.command.consumer.UpdateSubGroupSubCommand;
+import org.apache.rocketmq.tools.command.logicalqueue.DeleteTopicLogicalQueueMappingCommand;
+import org.apache.rocketmq.tools.command.logicalqueue.MigrateTopicLogicalQueueCommand;
+import org.apache.rocketmq.tools.command.logicalqueue.QueryTopicLogicalQueueMappingCommand;
+import org.apache.rocketmq.tools.command.logicalqueue.UpdateTopicLogicalQueueMappingCommand;
+import org.apache.rocketmq.tools.command.logicalqueue.UpdateTopicLogicalQueueNumCommand;
import org.apache.rocketmq.tools.command.message.CheckMsgSendRTCommand;
import org.apache.rocketmq.tools.command.message.ConsumeMessageCommand;
import org.apache.rocketmq.tools.command.message.PrintMessageByQueueCommand;
[rocketmq] 02/02: merger from develop branch and fix compile error
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhoubo pushed a commit to branch 5.0.0-preview
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit dd09ba1ec869acb1538d2c73f98d44066bbbb00f
Author: odbozhou <87...@qq.com>
AuthorDate: Fri Sep 24 17:06:48 2021 +0800
merger from develop branch and fix compile error
---
acl/src/main/java/org/apache/rocketmq/acl/common/AclUtils.java | 7 +++----
.../rocketmq/broker/processor/AbstractSendMessageProcessor.java | 2 +-
.../rocketmq/client/impl/producer/DefaultMQProducerImpl.java | 1 +
.../src/main/java/org/apache/rocketmq/common/stats/StatsItem.java | 1 -
common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java | 3 ++-
.../apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java | 2 --
.../org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java | 2 --
7 files changed, 7 insertions(+), 11 deletions(-)
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/common/AclUtils.java b/acl/src/main/java/org/apache/rocketmq/acl/common/AclUtils.java
index a03d136..b801c69 100644
--- a/acl/src/main/java/org/apache/rocketmq/acl/common/AclUtils.java
+++ b/acl/src/main/java/org/apache/rocketmq/acl/common/AclUtils.java
@@ -23,7 +23,6 @@ import java.io.FileNotFoundException;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
-import java.util.ArrayList;
import java.util.Map;
import java.util.SortedMap;
import org.apache.commons.lang3.StringUtils;
@@ -210,9 +209,9 @@ public class AclUtils {
// expand netaddress
int separatorCount = StringUtils.countMatches(netaddress, ":");
int padCount = part - separatorCount;
- if(padCount > 0){
+ if (padCount > 0) {
StringBuilder padStr = new StringBuilder(":");
- for(int i = 0; i < padCount; i++){
+ for (int i = 0; i < padCount; i++) {
padStr.append(":");
}
netaddress = StringUtils.replace(netaddress, "::", padStr.toString());
@@ -221,7 +220,7 @@ public class AclUtils {
// pad netaddress
String[] strArray = StringUtils.splitPreserveAllTokens(netaddress, ":");
for (int i = 0; i < strArray.length; i++) {
- if(strArray[i].length() < 4){
+ if (strArray[i].length() < 4) {
strArray[i] = StringUtils.leftPad(strArray[i], 4, '0');
}
}
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/AbstractSendMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/AbstractSendMessageProcessor.java
index d2f751b..cd279b9 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/AbstractSendMessageProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/AbstractSendMessageProcessor.java
@@ -24,7 +24,7 @@ import java.net.SocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
-import java.util.Map;Optional
+import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
index ba479d2..046f27d 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
@@ -88,6 +88,7 @@ import org.apache.rocketmq.common.message.MessageId;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.message.MessageType;
import org.apache.rocketmq.common.protocol.NamespaceUtil;
+import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.header.CheckTransactionStateRequestHeader;
import org.apache.rocketmq.common.protocol.header.EndTransactionRequestHeader;
import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
diff --git a/common/src/main/java/org/apache/rocketmq/common/stats/StatsItem.java b/common/src/main/java/org/apache/rocketmq/common/stats/StatsItem.java
index 6007cb0..d016662 100644
--- a/common/src/main/java/org/apache/rocketmq/common/stats/StatsItem.java
+++ b/common/src/main/java/org/apache/rocketmq/common/stats/StatsItem.java
@@ -20,7 +20,6 @@ package org.apache.rocketmq.common.stats;
import java.util.LinkedList;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
import org.apache.rocketmq.common.UtilAll;
diff --git a/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java b/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java
index ffbcd33..32d2ba6 100644
--- a/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java
+++ b/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java
@@ -119,7 +119,8 @@ public class UtilAllTest {
String comma = ",";
assertEquals("groupA=DENY,groupB=PUB|SUB,groupC=SUB", UtilAll.join(list, comma));
assertEquals(null, UtilAll.join(null, comma));
- assertEquals("", UtilAll.join(Collections.emptyList(), comma));
+ List<String> objects = Collections.emptyList();
+ assertEquals("", UtilAll.join(objects, comma));
}
static class DemoConfig {
diff --git a/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java b/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java
index f73705f..85e3883 100644
--- a/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java
+++ b/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java
@@ -56,8 +56,6 @@ import org.apache.rocketmq.common.protocol.header.namesrv.WipeWritePermOfBrokerR
import org.apache.rocketmq.common.protocol.route.LogicalQueuesInfoUnordered;
import org.apache.rocketmq.common.protocol.route.TopicRouteDataNameSrv;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
-import org.apache.rocketmq.logging.InternalLogger;
-import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.namesrv.NamesrvController;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
diff --git a/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java b/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java
index 5586603..ab9e77f 100644
--- a/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java
+++ b/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java
@@ -50,8 +50,6 @@ import org.apache.rocketmq.common.protocol.route.LogicalQueuesInfoUnordered;
import org.apache.rocketmq.common.protocol.route.QueueData;
import org.apache.rocketmq.common.protocol.route.TopicRouteDataNameSrv;
import org.apache.rocketmq.common.sysflag.TopicSysFlag;
-import org.apache.rocketmq.logging.InternalLogger;
-import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.common.RemotingUtil;
import org.apache.rocketmq.srvutil.ConcurrentHashMapUtil;