fix(proxy): add PROPERTY_PRODUCER_GROUP property for transaction mess…

…age (#803) Signed-off-by: SSpirits <[email protected]> Signed-off-by: wangxye <[email protected]> Co-authored-by: wangxye <[email protected]>
AutoMQ · Dec 8, 2023 · e284f99 · e284f99
1 parent 0a7e33c
commit e284f99
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 1 deletion.
diff --git a/common/src/main/fbs/flat_message.fbs b/common/src/main/fbs/flat_message.fbs
@@ -58,4 +58,6 @@ table SystemProperties {
     orphaned_transaction_recovery_seconds: long;
     // For a pending transaction message, this field records the check times.
     orphaned_transaction_check_times:int = -1;
+    // For a pending transaction message, this field records the producer group name.
+    orphaned_transaction_producer:string;
 }
diff --git a/distribution/helm/deploy/helm_sample_values.yaml b/distribution/helm/deploy/helm_sample_values.yaml
@@ -30,6 +30,9 @@ broker:
       accessKey: "foot"
       secretKey: "bar"
 
+    transactionTimeoutMillis: 100
+    transactionCheckInterval: 1000
+
 mysql:
   enabled: false
 

diff --git a/proxy/src/main/java/com/automq/rocketmq/proxy/service/MessageServiceImpl.java b/proxy/src/main/java/com/automq/rocketmq/proxy/service/MessageServiceImpl.java
@@ -352,7 +352,11 @@ private void checkTransactionStatus(TimerTag timerTag) {
             store.scheduleCheckTransaction(message);
 
             Topic topic = metadataService.topicOf(message.topicId()).join();
-            Channel channel = producerManager.getAvailableChannel(topic.getName());
+            String producerGroup = message.systemProperties().orphanedTransactionProducer();
+            if (StringUtils.isBlank(producerGroup)) {
+                producerGroup = topic.getName();
+            }
+            Channel channel = producerManager.getAvailableChannel(producerGroup);
             if (channel != null) {
                 CheckTransactionStateRequestHeader requestHeader = new CheckTransactionStateRequestHeader();
                 requestHeader.setCommitLogOffset(0L);

diff --git a/proxy/src/main/java/com/automq/rocketmq/proxy/util/FlatMessageUtil.java b/proxy/src/main/java/com/automq/rocketmq/proxy/util/FlatMessageUtil.java
@@ -265,6 +265,11 @@ private static SystemPropertiesT splitSystemProperties(FlatMessageT flatMessageT
             }
         }
 
+        String orphanedTransactionProducer = properties.remove(MessageConst.PROPERTY_PRODUCER_GROUP);
+        if (!Strings.isNullOrEmpty(orphanedTransactionProducer)) {
+            systemPropertiesT.setOrphanedTransactionProducer(orphanedTransactionProducer);
+        }
+
         // Remove all system properties
         for (String systemPropertyKey : MessageConst.STRING_HASH_SET) {
             properties.remove(systemPropertyKey);
@@ -302,5 +307,8 @@ private static void fillSystemProperties(MessageExt messageExt, FlatMessageExt f
         if (systemProperties.dlqOriginalMessageId() != null) {
             MessageAccessor.putProperty(messageExt, MessageConst.PROPERTY_DLQ_ORIGIN_MESSAGE_ID, String.valueOf(systemProperties.dlqOriginalMessageId()));
         }
+        if (systemProperties.orphanedTransactionProducer() != null) {
+            MessageAccessor.putProperty(messageExt, MessageConst.PROPERTY_PRODUCER_GROUP, systemProperties.orphanedTransactionProducer());
+        }
     }
 }