[ISSUE #598] Enhance transaction by putting messages that exceed max check times to system topic #633

xiangwangcheng · 2018-12-25T15:19:22Z

What is the purpose of the change

Brief changelog

For half messages that check times exceed the max check times(15 by default), not doing anything in here

    @Override
    public void resolveDiscardMsg(MessageExt msgExt) {
        log.error("MsgExt:{} has been checked too many times, so discard it", msgExt);
    }

and instead of this, put it into a system topic will be more meaningful. And you can also reopen message check logic manually.
(The logic of reopen message check will be submitted later.)

Brief Design：

Complement the logic of DefaultTransactionalMessageCheckListener#resolveDiscardMsg to put half messages to a system topic TRANS_CHECK_MAX_TIME_TOPIC.
add a new RequestCode RESUME_CHECK_HALF_MESSAGE(323) to enable resuming check logic by using DefaultMQAdminExt

Modification:

Implement DefaultTransactionalMessageCheckListener#resolveDiscardMsg
add resumeCheckHalfMessage method to AdminBrokerProcessor.java to recieve resume request from client.
add resumeCheckHalfMessage in MQClientAPIImpl.java to send resume request to broker.

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.

[√] 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.
[√] 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.
[√] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
[√] 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.
[√] 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.

coveralls · 2018-12-25T16:31:00Z

Coverage increased (+0.1%) to 50.479% when pulling 8f973de on xiangwangcheng:enhance_transaction into bf2a7e5 on apache:develop.

duhenglucky · 2019-01-04T05:41:52Z

@xiangwangcheng please rebase first and make sure this PR only contains your commit history.

xiangwangcheng · 2019-01-04T06:28:29Z

@duhenglucky I already did rebase and now the commit history is much better.

zongtanghu · 2019-03-07T12:21:49Z

...a/org/apache/rocketmq/broker/transaction/queue/DefaultTransactionalMessageCheckListener.java

+        TopicConfig topicConfig = this.getBrokerController().getTopicConfigManager()
+                .createTopicOfTranCheckMaxTime(TCMT_QUEUE_NUMS,
+                        PermName.PERM_READ | PermName.PERM_WRITE);
+        int queueId = Math.abs(random.nextInt() % 99999999) % TCMT_QUEUE_NUMS;


The TRANS_CHECK_MAX_TIME_TOPIC's queue num is 1,so the variable queueId should be set to 1,directly.

Yes, as @zongtanghu comment, it has the difference with sendBackMessage method, so if only one queue in this inner topic, please use a constant instead.

duhenglucky

Thanks for @xiangwangcheng's excellent PR, and we have put forward some comments for reference.

duhenglucky · 2019-03-11T01:31:40Z

common/src/main/java/org/apache/rocketmq/common/protocol/body/ResumeCheckHalfMessageResult.java

+
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+
+public class ResumeCheckHalfMessageResult extends RemotingSerializable {


Put this part of the content into the header and it seems that we don't need an API return some metrics data in response.

duhenglucky · 2019-03-11T01:32:27Z

common/src/main/java/org/apache/rocketmq/common/protocol/body/ResumeResult.java

+
+package org.apache.rocketmq.common.protocol.body;
+
+public enum ResumeResult {


IMO, it would be nice if we can use the ResponseCode directly.

duhenglucky · 2019-03-11T01:33:09Z

pom.xml

@@ -454,7 +454,19 @@
        <dependency>
            <groupId>org.mockito</groupId>
            <artifactId>mockito-core</artifactId>
-            <version>2.6.3</version>
+            <version>2.23.0</version>


Why need to import other mockito packages here?

duhenglucky · 2019-03-11T01:36:00Z

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

-
-        if (requestHeader.getTopic().equals(this.brokerController.getBrokerConfig().getBrokerClusterName())) {
-            String errorMsg = "the topic[" + requestHeader.getTopic() + "] is conflict with system reserved words.";
+                (CreateTopicRequestHeader) request


If no other changes, please don't reformat the code :)

duhenglucky · 2019-03-11T01:37:13Z

...a/org/apache/rocketmq/broker/transaction/queue/DefaultTransactionalMessageCheckListener.java


    public DefaultTransactionalMessageCheckListener() {
        super();
    }

    @Override
    public void resolveDiscardMsg(MessageExt msgExt) {
-        log.error("MsgExt:{} has been checked too many times, so discard it", msgExt);
+        log.error(
+                "MsgExt:{} has been checked too many times, so discard it by moving it to system topic TRANS_CHECK_MAXTIME_TOPIC",


I wonder if we can change the log format to "[NOTIFYME]MsgExt: ...."?

duhenglucky · 2019-03-11T01:38:25Z

...a/org/apache/rocketmq/broker/transaction/queue/DefaultTransactionalMessageCheckListener.java

+        TopicConfig topicConfig = this.getBrokerController().getTopicConfigManager()
+                .createTopicOfTranCheckMaxTime(TCMT_QUEUE_NUMS,
+                        PermName.PERM_READ | PermName.PERM_WRITE);
+        int queueId = Math.abs(random.nextInt() % 99999999) % TCMT_QUEUE_NUMS;


Yes, as @zongtanghu comment, it has the difference with sendBackMessage method, so if only one queue in this inner topic, please use a constant instead.

broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java

zongtanghu · 2019-06-17T12:19:09Z

Firstly,please resolve the confict codes in this pr. @xiangwangcheng And then,please help to review this pr. @ShannonDing @jonnxu @wlliqipeng Thanks very much.

…opic TRANS_CHECK_MAXTIME_TOPIC

ShannonDing

LGTM

zongtanghu · 2019-06-18T01:52:06Z

LGTM

jonnxu · 2019-06-18T02:13:09Z

good idea, LGTM

@after

…d max check times to system topic (apache#633) * add logic of putting message that exceeds max-check-times to system topic TRANS_CHECK_MAXTIME_TOPIC * add test case:testResolveDiscardMsg * add @after logic to test case * comment brokerController.shutdown and use mock * add logic of resuming half message check * add test case:resumeCheckHalfMessage * delete commented codes

@after

…d max check times to system topic (apache#633) * add logic of putting message that exceeds max-check-times to system topic TRANS_CHECK_MAXTIME_TOPIC * add test case:testResolveDiscardMsg * add @after logic to test case * comment brokerController.shutdown and use mock * add logic of resuming half message check * add test case:resumeCheckHalfMessage * delete commented codes

xiangwangcheng force-pushed the enhance_transaction branch from 8cf0d6a to 5034136 Compare January 4, 2019 06:02

xiangwangcheng force-pushed the enhance_transaction branch from 36aff5b to c8f1fe3 Compare January 14, 2019 15:00

xiangwangcheng force-pushed the enhance_transaction branch from c8f1fe3 to 8b56256 Compare February 17, 2019 02:46

zongtanghu reviewed Mar 7, 2019

View reviewed changes

zongtanghu added the type/enhancement label Mar 7, 2019

zongtanghu added this to the 4.5.0 milestone Mar 7, 2019

duhenglucky reviewed Mar 11, 2019

View reviewed changes

xiangwangcheng force-pushed the enhance_transaction branch 3 times, most recently from 780b2bb to 76d700e Compare March 11, 2019 13:20

zongtanghu reviewed Mar 12, 2019

View reviewed changes

broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java Outdated Show resolved Hide resolved

broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java Outdated Show resolved Hide resolved

ShannonDing modified the milestones: 4.5.0, 4.6.0 Apr 10, 2019

zongtanghu modified the milestones: 4.6.0, 4.5.2 Jun 17, 2019

chengxiangwang added 6 commits June 17, 2019 20:31

add logic of putting message that exceeds max-check-times to system t…

f124ac3

…opic TRANS_CHECK_MAXTIME_TOPIC

add test case:testResolveDiscardMsg

3a1fd0f

add @after logic to test case

c2d538c

comment brokerController.shutdown and use mock

8202b7b

add logic of resuming half message check

80be5e9

add test case:resumeCheckHalfMessage

584a31c

xiangwangcheng force-pushed the enhance_transaction branch from 71c1df1 to 729f33f Compare June 17, 2019 13:07

ShannonDing approved these changes Jun 18, 2019

View reviewed changes

delete commented codes

8f973de

xiangwangcheng force-pushed the enhance_transaction branch from 729f33f to 8f973de Compare June 18, 2019 09:37

duhenglucky merged commit d66243c into apache:develop Jun 19, 2019

zongtanghu mentioned this pull request Jul 12, 2019

About transactional message, some more meaningful logic should be implemented when check times exceeds transactionCheckMax(15 by default). #598

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[ISSUE #598] Enhance transaction by putting messages that exceed max check times to system topic #633

[ISSUE #598] Enhance transaction by putting messages that exceed max check times to system topic #633

xiangwangcheng commented Dec 25, 2018 •

edited

Loading

coveralls commented Dec 25, 2018 •

edited

Loading

duhenglucky commented Jan 4, 2019

xiangwangcheng commented Jan 4, 2019

zongtanghu Mar 7, 2019

duhenglucky Mar 11, 2019

duhenglucky left a comment

duhenglucky Mar 11, 2019

duhenglucky Mar 11, 2019

duhenglucky Mar 11, 2019

duhenglucky Mar 11, 2019

duhenglucky Mar 11, 2019

ahuazhu Jun 18, 2019

duhenglucky Mar 11, 2019

zongtanghu commented Jun 17, 2019

ShannonDing left a comment

zongtanghu commented Jun 18, 2019

jonnxu commented Jun 18, 2019


		import org.apache.rocketmq.remoting.protocol.RemotingSerializable;

		public class ResumeCheckHalfMessageResult extends RemotingSerializable {


		package org.apache.rocketmq.common.protocol.body;

		public enum ResumeResult {

[ISSUE #598] Enhance transaction by putting messages that exceed max check times to system topic #633

[ISSUE #598] Enhance transaction by putting messages that exceed max check times to system topic #633

Conversation

xiangwangcheng commented Dec 25, 2018 • edited Loading

What is the purpose of the change

Brief changelog

Verifying this change

coveralls commented Dec 25, 2018 • edited Loading

duhenglucky commented Jan 4, 2019

xiangwangcheng commented Jan 4, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

duhenglucky left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zongtanghu commented Jun 17, 2019

ShannonDing left a comment

Choose a reason for hiding this comment

zongtanghu commented Jun 18, 2019

jonnxu commented Jun 18, 2019

xiangwangcheng commented Dec 25, 2018 •

edited

Loading

coveralls commented Dec 25, 2018 •

edited

Loading