[ISSUE #292] Add support of transactional message feature #358

duhengforever · 2018-07-09T06:00:46Z

Please do not create a Pull Request without creating an issue first.

What is the purpose of the change

In order to implement eventual consistency in distributed system, we define a new type of message named transactional message, it can be thought of as a two-phase commit message implementation to ensure eventual consistency in distributed system. Transactional message ensures that the execution of local transaction and the sending of message can be performed atomically.

Brief changelog

In our implementation, we made full use of the mechanism of rocketMQ itself to avoid external dependencies, and we defined a pair of message: named half message and commit/rollback message. Half message means prepared message, which are not visible to consumers，and it will be sent before local transaction is executed. After the local transaction ends，producer will send a commit or rollback message to broker based on local transaction execution result

Verifying this change

This is a trivial change.

Follow this checklist to help us incorporate your contribution quickly and easily:

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 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.

duhengforever · 2018-07-09T06:13:01Z

Related Issue:
@lizhanhui @vongosling @dongeforever Have any comments and review about this PR？

coveralls · 2018-07-09T06:23:42Z

Coverage increased (+0.8%) to 42.258% when pulling 835e013 on duhengforever:develop into 94f3b5a on apache:develop.

lizhanhui · 2018-07-09T06:24:21Z

This is a big pull request, containing bulks of changes...Please wait patiently and I would join you as soon as possible.

vongosling

first review

vongosling · 2018-07-09T09:01:33Z

example/src/main/java/org/apache/rocketmq/example/benchmark/TransactionProducer.java

-}
-
-class TransactionCheckListenerBImpl implements TransactionCheckListener {
+//class TransactionExecuterBImpl implements LocalTransactionExecuter {


The extra commented code has been removed in the latest commit，thanks for your reminding.

vongosling · 2018-07-09T09:02:00Z

example/src/main/java/org/apache/rocketmq/example/benchmark/TransactionProducer.java


-    public TransactionCheckListenerBImpl(boolean ischeckffalse,
-        StatsBenchmarkTProducer statsBenchmarkTProducer) {
+    public TransactionListenerImpl(boolean ischeckffalse, boolean isCheckLocal,


ischeckffalse ？

Sorry, the last commit did not modify the original code, including variable naming, but the latest commit has been changed.

vongosling · 2018-07-09T09:03:03Z

...on/src/main/java/org/apache/rocketmq/common/protocol/header/EndTransactionRequestHeader.java

-            + tranStateTableOffset + ", commitLogOffset=" + commitLogOffset + ", commitOrRollback="
-            + commitOrRollback + ", fromTransactionCheck=" + fromTransactionCheck + ", msgId=" + msgId
-            + "]";
+        return "EndTransactionRequestHeader{" +


ToStringBuilder is easier way to reflect all field in one class

I used IDEA's automatic generation tool to override the toString method.

vongosling · 2018-07-09T09:03:37Z

client/src/main/java/org/apache/rocketmq/client/producer/LocalTransactionExecuter.java

@@ -17,7 +17,7 @@
 package org.apache.rocketmq.client.producer;

 import org.apache.rocketmq.common.message.Message;
-
+@Deprecated


vongosling · 2018-07-09T09:22:17Z

broker/src/main/java/org/apache/rocketmq/broker/transaction/TransactionStore.java

@@ -18,7 +18,7 @@
 package org.apache.rocketmq.broker.transaction;

 import java.util.List;
-
+@Deprecated


This interface has been deleted in the latest commit.

vongosling · 2018-07-09T09:24:49Z

@shroman @Jaskey Could you help us to review the pr together?

Jaskey · 2018-07-10T11:42:15Z

broker/src/main/java/org/apache/rocketmq/broker/client/net/Broker2Client.java

        final Channel channel,
        final CheckTransactionStateRequestHeader requestHeader,
-        final SelectMappedBufferResult selectMappedBufferResult) {
+        final MessageExt messageExt) throws Exception {


Can this Exception be specific?

Thanks for your review，in fact, I don't want to throw any exceptions here，but considering that the encode method may cause exception，so in this method just throws exception following with the encode method.

vongosling · 2018-07-11T09:32:16Z

broker/src/main/java/org/apache/rocketmq/broker/client/ProducerManager.java

+            Channel channel = channelList.get(index);
+            int count = 0;
+            boolean isOk = channel.isActive() && channel.isWritable();
+            while (isOk && count++ < 3) {


3 is a magic number

I will use a constant to represent this value, which represents the number of retries for fetch an available channel.

vongosling · 2018-07-11T09:33:17Z

broker/src/main/java/org/apache/rocketmq/broker/client/net/Broker2Client.java

-        } catch (Throwable e) {
-            log.error("invokeProducer exception", e);
-            selectMappedBufferResult.release();
+            this.brokerController.getRemotingServer().invokeOneway(channel, request, 10);


This is an empirical value when using the invokeOneWay method. In the check scenario , we don't need to wait for the producer‘s response, and in order to prevent too many threads waiting，we adopted this configuration，but we also considered that whether it is necessary to make this parameter configurable，what’s your opinion？

vongosling · 2018-07-11T09:35:36Z

broker/src/main/java/org/apache/rocketmq/broker/transaction/TransactionRecord.java

 package org.apache.rocketmq.broker.transaction;
-
+@Deprecated


deprecated only but missing the deadline and replacement is not a good practice

vongosling · 2018-07-11T09:37:27Z

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

+
+    private static final int TRY_PULL_MSG_NUMBER = 1;
+
+    private final int queueTime = 60000;


In order to avoid a queue check taking up too much time and leading to other queues to starve, we added this configuration，but I will change the variable name to uppercase.

vongosling · 2018-07-11T09:44:02Z

LGTM if a rational response to my point out. This is an awesome feature, @lizhanhui @Jaskey @shroman Could we speed up the feedback for my kindly proposer, I would like to integrate into our develop branch :-) BTW, @duhengforever , do we have a wiki for the architecture and user guide?

vongosling · 2018-07-12T01:48:49Z

@duhengforever Forget to comment, What is the coverage of the new code, over 80%?

duhengforever · 2018-07-13T06:18:42Z

For transactional messages, I added some documentation, including examples and design documentation., I hope these documents can help you to review this PR, thanks!

xingfudeshi · 2018-07-13T08:03:43Z

great job!!it seems the good things are coming soon!

lizhanhui · 2018-07-13T23:28:12Z

broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java

@@ -142,6 +150,9 @@
    private BrokerFastFailure brokerFastFailure;
    private Configuration configuration;
    private FileWatchService fileWatchService;
+    private TransactionalMessageCheckService transactionMsgCheckService;


Please use consistent naming. If abbreviation were used, so were these relevant.

lizhanhui · 2018-07-14T00:06:35Z

...r/src/main/java/org/apache/rocketmq/broker/transaction/TransactionalMessageCheckService.java

+    public void start() {
+        if (started.compareAndSet(false, true)) {
+            super.start();
+            this.brokerController.getTransactionalMessageService().open();


Considering that in other storage media implementations, maybe need to open some related settings, so provided this method.

lizhanhui · 2018-07-14T00:10:32Z

...r/src/main/java/org/apache/rocketmq/broker/transaction/TransactionalMessageCheckService.java

+    }
+
+    @Override
+    public void run() {


Should take advantage of what ServiceThread have already provided. No

while(true) { ... sleep(xxx); ... }

Use onWaitEnd() instead.

Do you mean waitForRunning() and onWaitEnd() methods? yeah, I used these two methods to transform this implementation in the latest commit.

lizhanhui · 2018-07-14T00:13:13Z

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

+public class TransactionalMessageServiceImpl implements TransactionalMessageService {
+    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.TRANSACTION_LOGGER_NAME);
+
+    private TransactionalMessageBridge transactionBridge;


Having TransactionalMessageService --> TransactionalMessageServiceImpl abstraction, is it necesary to add another proxy?

In fact, in the development process, I have considered this problem, I personally think it is not needed, and we provide the Serviceloader to load other implementations. and if add another proxy，it will make the call hierarchy deeper and deeper, and need to add a strategy to ensure loading to the correct implementation，not convenient for later code maintenance.

lizhanhui · 2018-07-14T00:15:38Z

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

+        return transactionBridge.putHalfMessage(messageInner);
+    }
+
+    private boolean isNeedDiscard(MessageExt msgExt, int transactionCheckMax) {


isNeedXXX naming is weird and does not follow grammar. Using needXXX suffices.

good suggestion

lizhanhui · 2018-07-14T00:20:51Z

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

+            MessageExtBrokerInner msgInner = transactionBridge.renewHalfMessageInner(messageExt);
+            putMessageResult = transactionBridge.putMessageReturnResult(msgInner);
+        } catch (Exception e) {
+            e.printStackTrace();


Just printStackTrace?
We need to recover from the error and mark tx service unavailable and propagate the unavailable state to producer clients if recovering is not possible.

Thanks for helping to point out the problem，output to the log will be a good choice, as in this place, The outer method uses this method to decide whether to send back the request ，based on the status of putMessageResult, so there's no need to propagate the unavailable state to producer.

lizhanhui · 2018-07-14T00:39:01Z

Newly added classes, member variables & methods should be well documented, explaining their role in your solution.
Corner cases are not yet properly handled. For example, if pushing back half message experiences failure, what is the recovering procedure?
How is the HA guaranteed in case the master is down?
We prefer to use full-word in the identifier, it would be better if this pull request may follow this convention.
I am wondering whether the service locator and tx service bridge is an excessive abstraction. IMO, simplicity takes precedence.
A lot of the tx related configurations are hard-coded, making them configurable is preferable.

duhengforever · 2018-07-14T03:24:00Z

@lizhanhui thanks for your excellent review firstly.

There are comments on the key methods, and the class name directly describes the role of the current class.
Some corner cases have been considered and verified，and for the problem of put half message back failed, in fact, we only send a check request after the put back is successful. and if put back failed, we will continue to process this message, and in order to avoid causing unlimited check，we also added the limit of check times for the message, and in this situation, we will also print an error log for easy monitoring and manual processing.
In fact, the HA mechanism of the transaction message depends on the HA of the rocketMQ itself. In order to avoid the message not being lost, we recommend using the synchronous double-write mechanism and manually upgrading the slave when the master down.
I have been changed the identifier to the full-word.
I don't think it's an excessive abstraction, I think transaction message is an application of ordinary message, so I need a bridge to avoid directly manipulating the relevant storage of MQ.
Some hard code have been removed, please review the latest commit.

vongosling · 2018-07-14T03:28:20Z

Glad to see the great work in the community, I would like to merge the pr if all comments are resolved

…fier

xingfudeshi · 2018-07-15T03:02:46Z

congrats!

klaus-pd · 2018-07-16T09:24:25Z

@duhengforever Why do we not implement transactions checking based on database, and how to confirm previous transaction information if the service is down?

dominicqi · 2018-07-16T12:54:40Z

broker/src/main/java/org/apache/rocketmq/broker/client/ProducerManager.java

+                index = (++index) % size;
+                channel = channelList.get(index);
+                return channel;
+            }


here if isOk= true will return next channel.
i think it should be like
while ( count++ < GET_AVALIABLE_CHANNEL_RETRY_COUNT) { if(isOk){ return channel; } index = (++index) % size; channel = channelList.get(index); isOk = channel.isActive() && channel.isWritable(); }

Thanks for your careful review，and I will fix this problem ASAP.

duhengforever · 2018-07-16T13:10:32Z

@klaus-pd In order to avoid external dependencies, such as database, we used the mechanism of rocketMQ itself to implement the check logic. and if the producer down, broker will send check request to other producers with same produceId.

klaus-pd · 2018-07-17T02:32:57Z

@duhengforever I know, but if only one producer of the same produceId, how to solve the problem of checking back?

duhengforever · 2018-07-17T03:58:39Z

@klaus-pd broker will resend the check request when the producer restart util reached a limit.

casionone · 2018-07-27T09:54:02Z

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

+        for (MessageExt opMessageExt : opMsg) {
+            Long queueOffset = getLong(new String(opMessageExt.getBody(), TransactionalMessageUtil.charset));
+            log.info("Topic: {} tags: {}, OpOffset: {}, HalfOffset: {}", opMessageExt.getTopic(),
+                opMessageExt.getTags(), opMessageExt.getQueueOffset(), queueOffset);


queueOffset = getLong(new String(opMessageExt.getBody(), TransactionalMessageUtil.charset));
why this can get the value of HalfOffset？

Details can be found in TransactionalMessageService#deletePrepareMessage method.

…he#358)

Add support of transactional message

18cea40

vongosling reviewed Jul 9, 2018

View reviewed changes

Rename class name to make it seems more standardized

6ec0c7f

Jaskey reviewed Jul 10, 2018

View reviewed changes

vongosling reviewed Jul 11, 2018

View reviewed changes

Format code with RMQ style

aeca8a1

duhengforever added 2 commits July 12, 2018 22:32

Remove some magic numbers and redefine some constants.

0ce72f1

Rename some constants

6825247

lizhanhui reviewed Jul 13, 2018

View reviewed changes

lizhanhui reviewed Jul 14, 2018

View reviewed changes

Improved transaction service implementation and formatted some identi…

835e013

…fier

vongosling merged commit b054a9d into apache:develop Jul 15, 2018

dominicqi reviewed Jul 16, 2018

View reviewed changes

vongosling added this to the 4.3.0 milestone Jul 17, 2018

vongosling added the type/new feature label Jul 17, 2018

casionone reviewed Jul 27, 2018

View reviewed changes

renshuaibing-aaron pushed a commit to renshuaibing-aaron/rocketmq that referenced this pull request Apr 13, 2020

[ISSUE apache#292] Add support of transactional message feature (apac…

d925184

…he#358)

JiaMingLiu93 pushed a commit to JiaMingLiu93/rocketmq that referenced this pull request May 28, 2020

[ISSUE apache#292] Add support of transactional message feature (apac…

321a597

…he#358)


		private static final int TRY_PULL_MSG_NUMBER = 1;

		private final int queueTime = 60000;

[ISSUE #292] Add support of transactional message feature #358

[ISSUE #292] Add support of transactional message feature #358

Conversation

duhengforever commented Jul 9, 2018

What is the purpose of the change

Brief changelog

Verifying this change

duhengforever commented Jul 9, 2018

coveralls commented Jul 9, 2018 • edited Loading

lizhanhui commented Jul 9, 2018

vongosling 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vongosling commented Jul 9, 2018

Choose a reason for hiding this comment

duhengforever Jul 10, 2018 • edited Loading

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

vongosling commented Jul 11, 2018 • edited Loading

vongosling commented Jul 12, 2018

duhengforever commented Jul 13, 2018

xingfudeshi commented Jul 13, 2018 • edited Loading

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lizhanhui commented Jul 14, 2018

duhengforever commented Jul 14, 2018

vongosling commented Jul 14, 2018

xingfudeshi commented Jul 15, 2018

klaus-pd commented Jul 16, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

duhengforever commented Jul 16, 2018

klaus-pd commented Jul 17, 2018

duhengforever commented Jul 17, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

coveralls commented Jul 9, 2018 •

edited

Loading

duhengforever Jul 10, 2018 •

edited

Loading

vongosling commented Jul 11, 2018 •

edited

Loading

xingfudeshi commented Jul 13, 2018 •

edited

Loading