[ISSUE #617] retrieve transactionid from property first (#620)

apache · Mar 16, 2021 · 05929da · 05929da
1 parent f698f32
commit 05929da
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 1 deletion.
diff --git a/primitive/message.go b/primitive/message.go
@@ -59,6 +59,7 @@ const (
 	PropertyTranscationCheckTimes          = "TRANSACTION_CHECK_TIMES"
 	PropertyCheckImmunityTimeInSeconds     = "CHECK_IMMUNITY_TIME_IN_SECONDS"
 	PropertyShardingKey                    = "SHARDING_KEY"
+	PropertyTransactionID                  = "__transactionId__"
 )
 
 type Message struct {

diff --git a/producer/producer.go b/producer/producer.go
@@ -462,13 +462,20 @@ func (tp *transactionProducer) checkTransactionState() {
 			if uniqueKey == "" {
 				uniqueKey = callback.Msg.MsgId
 			}
+			transactionId := callback.Msg.GetProperty(primitive.PropertyTransactionID)
+			if transactionId == "" {
+				transactionId = callback.Header.TransactionId
+			}
+			if transactionId == "" {
+				transactionId = callback.Msg.TransactionId
+			}
 			header := &internal.EndTransactionRequestHeader{
 				CommitLogOffset:      callback.Header.CommitLogOffset,
 				ProducerGroup:        tp.producer.group,
 				TranStateTableOffset: callback.Header.TranStateTableOffset,
 				FromTransactionCheck: true,
 				MsgID:                uniqueKey,
-				TransactionId:        callback.Header.TransactionId,
+				TransactionId:        transactionId,
 				CommitOrRollback:     tp.transactionState(localTransactionState),
 			}