bugfix: use per-branch qualified gtrid in XA mode for DM

rm-datasource/src/main/java/org/apache/seata/rm/datasource/xa/ConnectionProxyXA.java

@@ -132,8 +132,17 @@ private void xaEnd(XAXid xaXid, int flags) throws XAException {
      */
     public void xaCommit(String xid, long branchId, String applicationData) throws XAException {
         try (ResourceLock ignored = resourceLock.obtain()) {
-            XAXid xaXid = XAXidBuilder.build(xid, branchId);
-            xaResource.commit(xaXid, false);
+            XAXid xaXid = XAXidBuilder.build(xid, branchId, resource.getDbType());
+            try {
+                xaResource.commit(xaXid, false);
+            } catch (XAException e) {
+                if (isLegacyPreparedBranch(e, xaXid)) {
+                    // resolve the branch with its original physical XA identity
+                    xaResource.commit(XAXidBuilder.build(xid, branchId), false);
+                } else {
+                    throw e;
+                }
+            }
             releaseIfNecessary();
         }
     }
@@ -149,8 +158,17 @@ public void xaRollback(String xid, long branchId, String applicationData) throws
             if (this.xaBranchXid != null) {
                 xaRollback(xaBranchXid);
             } else {
-                XAXid xaXid = XAXidBuilder.build(xid, branchId);
-                xaRollback(xaXid);
+                XAXid xaXid = XAXidBuilder.build(xid, branchId, resource.getDbType());
+                try {
+                    xaRollback(xaXid);
+                } catch (XAException e) {
+                    if (isLegacyPreparedBranch(e, xaXid)) {
+                        // resolve the branch with its original physical XA identity
+                        xaRollback(XAXidBuilder.build(xid, branchId));
+                    } else {
+                        throw e;
+                    }
+                }
             }
         }
     }
@@ -166,6 +184,19 @@ public void xaRollback(XAXid xaXid) throws XAException {
         releaseIfNecessary();
     }
 
+    /**
+     * A DM branch prepared by an RM running a version before the per-branch-gtrid format keeps
+     * the legacy physical XA identity (formatId 9752, gtrid = xid). When the rebuilt
+     * per-branch-qualified Xid misses such a branch (XAER_NOTA) during phase two, the commit or
+     * rollback must be retried once with the legacy Xid, so that an upgraded RM can still resolve
+     * branches that were prepared before the upgrade.
+     */
+    private boolean isLegacyPreparedBranch(XAException e, XAXid xaXid) {
+        return e.errorCode == XAException.XAER_NOTA
+                && xaXid.getFormatId() == XABranchXid.SEATA_XA_XID_FORMAT_ID_PER_BRANCH_GTRID
+                && JdbcConstants.DM.equalsIgnoreCase(resource.getDbType());
+    }
+
     @Override
     public void setAutoCommit(boolean autoCommit) throws SQLException {
         if (currentAutoCommitStatus == autoCommit) {
@@ -204,7 +235,7 @@ public void setAutoCommit(boolean autoCommit) throws SQLException {
                         "failed to register xa branch " + xid + " since " + te.getCode() + ":" + te.getMessage(), te);
             }
             // 2. build XA-Xid with xid and branchId
-            this.xaBranchXid = XAXidBuilder.build(xid, branchId);
+            this.xaBranchXid = XAXidBuilder.build(xid, branchId, resource.getDbType());
             // Keep the Connection if necessary
             keepIfNecessary();
             try {

rm-datasource/src/main/java/org/apache/seata/rm/datasource/xa/ResourceManagerXA.java

@@ -143,8 +143,10 @@ private BranchStatus finishBranch(
             String resourceId,
             String applicationData)
             throws TransactionException {
-        XAXid xaBranchXid = XAXidBuilder.build(xid, branchId);
         Resource resource = dataSourceCache.get(resourceId);
+        String dbType =
+                resource instanceof BaseDataSourceResource ? ((BaseDataSourceResource) resource).getDbType() : null;
+        XAXid xaBranchXid = XAXidBuilder.build(xid, branchId, dbType);
         if (resource instanceof AbstractDataSourceProxyXA) {
             try (ConnectionProxyXA connectionProxyXA =
                     ((AbstractDataSourceProxyXA) resource).getConnectionForXAFinish(xaBranchXid)) {

rm-datasource/src/main/java/org/apache/seata/rm/datasource/xa/XABranchXid.java

@@ -17,6 +17,8 @@
 package org.apache.seata.rm.datasource.xa;
 
 import java.io.UnsupportedEncodingException;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
 import java.util.Objects;
 
 /**
@@ -28,7 +30,24 @@ public class XABranchXid implements XAXid {
     private static final String DEFAULT_ENCODE_CHARSET = "UTF-8";
     private static final String BRANCH_ID_PREFIX = "-";
 
-    private static final int SEATA_XA_XID_FORMAT_ID = 9752;
+    static final int SEATA_XA_XID_FORMAT_ID = 9752;
+
+    /**
+     * Format id for XA Xids whose gtrid is qualified with the branch id
+     * (gtrid = xid + "-" + branchId), so that every branch of one Seata global transaction
+     * owns a distinct gtrid. This is required by databases (e.g. DM) that reject a new
+     * xa start with XAER_RMERR while an in-doubt branch of the same gtrid exists.
+     * If the readable form would exceed the X/Open maximum gtrid length, a bounded
+     * (md5-hex based) encoding of the xid is used instead.
+     */
+    public static final int SEATA_XA_XID_FORMAT_ID_PER_BRANCH_GTRID = 9753;
+
+    /**
+     * The maximum length of the global transaction id defined by the X/Open CAE Specification.
+     */
+    static final int MAX_GTRID_LENGTH = 64;
+
+    private final int formatId;
 
     private String xid;
 
@@ -39,14 +58,29 @@ public class XABranchXid implements XAXid {
     private byte[] branchQualifier;
 
     XABranchXid(String xid, long branchId) {
+        this(xid, branchId, SEATA_XA_XID_FORMAT_ID);
+    }
+
+    /**
+     * Build a XAXid with an explicit format id, which selects the gtrid encoding:
+     * {@link #SEATA_XA_XID_FORMAT_ID_PER_BRANCH_GTRID} qualifies the gtrid with the branch id,
+     * {@link #SEATA_XA_XID_FORMAT_ID} keeps the legacy gtrid (= xid).
+     */
+    XABranchXid(String xid, long branchId, int formatId) {
         this.xid = xid;
         this.branchId = branchId;
+        this.formatId = formatId;
         encode();
     }
 
+    /**
+     * Rebuild a XAXid from raw XA bytes. Only the legacy encoding can be reconstructed this way;
+     * per-branch-qualified gtrids must be rebuilt via {@link XAXidBuilder#build(String, long, String)}.
+     */
     XABranchXid(byte[] globalTransactionId, byte[] branchQualifier) {
         this.globalTransactionId = globalTransactionId;
         this.branchQualifier = branchQualifier;
+        this.formatId = SEATA_XA_XID_FORMAT_ID;
         decode();
     }
 
@@ -62,7 +96,7 @@ public long getBranchId() {
 
     @Override
     public int getFormatId() {
-        return SEATA_XA_XID_FORMAT_ID;
+        return formatId;
     }
 
     @Override
@@ -94,6 +128,8 @@ private String byteArray2String(byte[] bytes) {
     private void encode() {
         if (xid == null) {
             globalTransactionId = new byte[0];
+        } else if (formatId == SEATA_XA_XID_FORMAT_ID_PER_BRANCH_GTRID) {
+            globalTransactionId = encodeQualifiedGtrid();
         } else {
             globalTransactionId = string2byteArray(xid);
         }
@@ -105,6 +141,36 @@ private void encode() {
         }
     }
 
+    /**
+     * Encode the per-branch-qualified gtrid as {@code xid + "-" + branchId}. When the readable
+     * form would exceed the X/Open maximum gtrid length, fall back to the bounded form
+     * {@code md5hex(xid) + "-" + branchId}, which always fits. Both forms are deterministic pure
+     * functions of the xid and the branch id, so phase two rebuilds exactly the same bytes, and
+     * distinct branch ids still own distinct gtrids.
+     */
+    private byte[] encodeQualifiedGtrid() {
+        byte[] encodedGtrid = string2byteArray(xid + BRANCH_ID_PREFIX + branchId);
+        if (encodedGtrid.length > MAX_GTRID_LENGTH) {
+            return string2byteArray(md5Hex(string2byteArray(xid)) + BRANCH_ID_PREFIX + branchId);
+        }
+        return encodedGtrid;
+    }
+
+    private static String md5Hex(byte[] bytes) {
+        try {
+            byte[] md5 = MessageDigest.getInstance("MD5").digest(bytes);
+            StringBuilder hex = new StringBuilder(md5.length * 2);
+            for (byte b : md5) {
+                hex.append(Character.forDigit((b >> 4) & 0xF, 16));
+                hex.append(Character.forDigit(b & 0xF, 16));
+            }
+            return hex.toString();
+        } catch (NoSuchAlgorithmException e) {
+            // the MD5 algorithm is required to be available in every JRE
+            throw new RuntimeException(e);
+        }
+    }
+
     private void decode() {
         if (globalTransactionId == null || globalTransactionId.length == 0) {
             xid = null;

rm-datasource/src/main/java/org/apache/seata/rm/datasource/xa/XAXidBuilder.java

@@ -16,6 +16,8 @@
  */
 package org.apache.seata.rm.datasource.xa;
 
+import org.apache.seata.sqlparser.util.JdbcConstants;
+
 /**
  * XA-Xid builder.
  *
@@ -25,9 +27,48 @@ public class XAXidBuilder {
     private XAXidBuilder() {}
 
     public static final XAXid build(String xid, long branchId) {
+        return build(xid, branchId, null);
+    }
+
+    /**
+     * Build a XAXid from the given Seata global transaction xid and branch id.
+     * <p>
+     * For databases that reject a new xa start with XAER_RMERR while an in-doubt branch of the
+     * same gtrid exists (e.g. DM), the returned XAXid uses a per-branch-qualified gtrid
+     * ({@code xid + "-" + branchId}) with a dedicated format id, so that every branch of one
+     * Seata global transaction owns a distinct gtrid and can be started, prepared and decided
+     * independently by the database. If the readable form would exceed the X/Open maximum gtrid
+     * length, a bounded (md5-hex based) encoding of the xid is used instead, so the build never
+     * fails for a valid Seata xid. The rebuild on the phase-2 path is deterministic, since the
+     * gtrid is derived from the xid and the branch id only, both of which are carried by the
+     * branch commit/rollback requests of the TC.
+     * <p>
+     * For all other databases (or if dbType is null) the legacy encoding (gtrid = xid) is used,
+     * which stays byte-to-byte identical to previous versions.
+     *
+     * @param xid the Seata global transaction xid
+     * @param branchId the branch id
+     * @param dbType the database type of the resource this XAXid is used against, may be null
+     * @return the XAXid
+     */
+    public static final XAXid build(String xid, long branchId, String dbType) {
+        if (JdbcConstants.DM.equalsIgnoreCase(dbType)) {
+            return new XABranchXid(xid, branchId, XABranchXid.SEATA_XA_XID_FORMAT_ID_PER_BRANCH_GTRID);
+        }
         return new XABranchXid(xid, branchId);
     }
 
+    /**
+     * Rebuild a XAXid from raw XA bytes.
+     * <p>
+     * Note: only the legacy encoding (format id {@code 9752}, gtrid = xid) can be reconstructed
+     * from raw bytes. Per-branch-qualified gtrids must be rebuilt via
+     * {@link #build(String, long, String)}.
+     *
+     * @param globalTransactionId the raw global transaction id bytes
+     * @param branchQualifier the raw branch qualifier bytes
+     * @return the XAXid
+     */
     public static final XAXid build(byte[] globalTransactionId, byte[] branchQualifier) {
         return new XABranchXid(globalTransactionId, branchQualifier);
     }

rm-datasource/src/test/java/org/apache/seata/rm/datasource/xa/ConnectionProxyXABehaviorTest.java

@@ -25,11 +25,14 @@
 import org.apache.seata.rm.datasource.util.SeataXAResource;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 import org.mockito.MockedStatic;
 
 import javax.sql.XAConnection;
 import javax.transaction.xa.XAException;
 import javax.transaction.xa.XAResource;
+import javax.transaction.xa.Xid;
+import java.nio.charset.StandardCharsets;
 import java.sql.Connection;
 import java.sql.SQLException;
 
@@ -161,4 +164,89 @@ void oracleUsesLooseCouplingAndIgnoresFailedStatusReport() throws Exception {
             verify(original, times(2)).close();
         }
     }
+
+    @Test
+    void dmPhaseTwoCommitRetriesLegacyIdentityWhenPerBranchXidIsMissing() throws Exception {
+        when(resource.getDbType()).thenReturn("dm");
+        try (MockedStatic<DefaultResourceManager> managers = mockStatic(DefaultResourceManager.class)) {
+            managers.when(DefaultResourceManager::get).thenReturn(manager);
+            proxy.setAutoCommit(false);
+            proxy.close();
+            // a branch prepared by an RM before the per-branch-gtrid format keeps the legacy
+            // physical identity: the rebuilt per-branch Xid misses it (XAER_NOTA) and the legacy
+            // identity must be retried once, so an upgraded RM can still resolve the branch
+            doThrow(new XAException(XAException.XAER_NOTA)).doNothing().when(xa).commit(any(), eq(false));
+            assertDoesNotThrow(() -> proxy.xaCommit(xid, 42, null));
+
+            ArgumentCaptor<Xid> committed = ArgumentCaptor.forClass(Xid.class);
+            verify(xa, times(2)).commit(committed.capture(), eq(false));
+            Xid perBranch = committed.getAllValues().get(0);
+            Xid legacy = committed.getAllValues().get(1);
+            assertEquals(XABranchXid.SEATA_XA_XID_FORMAT_ID_PER_BRANCH_GTRID, perBranch.getFormatId());
+            assertEquals(XABranchXid.SEATA_XA_XID_FORMAT_ID, legacy.getFormatId());
+            assertArrayEquals((xid + "-" + 42).getBytes(StandardCharsets.UTF_8), perBranch.getGlobalTransactionId());
+            assertArrayEquals(xid.getBytes(StandardCharsets.UTF_8), legacy.getGlobalTransactionId());
+        }
+    }
+
+    @Test
+    void dmPhaseTwoRollbackRetriesLegacyIdentityWhenPerBranchXidIsMissing() throws Exception {
+        when(resource.getDbType()).thenReturn("dm");
+        try (MockedStatic<DefaultResourceManager> managers = mockStatic(DefaultResourceManager.class)) {
+            managers.when(DefaultResourceManager::get).thenReturn(manager);
+            proxy.setAutoCommit(false);
+            proxy.close();
+            doThrow(new XAException(XAException.XAER_NOTA)).doNothing().when(xa).rollback(any());
+            assertDoesNotThrow(() -> proxy.xaRollback(xid, 42, null));
+
+            ArgumentCaptor<Xid> rolledBack = ArgumentCaptor.forClass(Xid.class);
+            verify(xa, times(2)).rollback(rolledBack.capture());
+            Xid legacy = rolledBack.getAllValues().get(1);
+            assertEquals(XABranchXid.SEATA_XA_XID_FORMAT_ID, legacy.getFormatId());
+            assertArrayEquals(xid.getBytes(StandardCharsets.UTF_8), legacy.getGlobalTransactionId());
+        }
+    }
+
+    @Test
+    void xaerNotaStaysPropagatingForDatabasesWithoutTheLegacyIdentity() throws Exception {
+        // non-DM databases keep the legacy encoding in phase one: XAER_NOTA must not trigger any
+        // retry and propagates to the TC retryable handling exactly as before
+        try (MockedStatic<DefaultResourceManager> managers = mockStatic(DefaultResourceManager.class)) {
+            managers.when(DefaultResourceManager::get).thenReturn(manager);
+            proxy.setAutoCommit(false);
+            proxy.close();
+            doThrow(new XAException(XAException.XAER_NOTA)).when(xa).commit(any(), eq(false));
+            assertThrows(XAException.class, () -> proxy.xaCommit(xid, 42, null));
+            verify(xa, times(1)).commit(any(), eq(false));
+        }
+    }
+
+    @Test
+    void longXidDmBranchSurvivesRegistrationPhaseOneAndPhaseTwoRollback() throws Exception {
+        // the overflow case from the review: the readable per-branch gtrid would exceed the X/Open
+        // limit, yet registration, xa start, prepare and the TC driven rollback must all work
+        String longXid = "2001:db8:1234:5678:90ab:cdef:1234:5678:8091:812345678901234567";
+        when(resource.getDbType()).thenReturn("dm");
+        when(manager.branchRegister(BranchType.XA, "resource", null, longXid, null, null))
+                .thenReturn(43L);
+        try (MockedStatic<DefaultResourceManager> managers = mockStatic(DefaultResourceManager.class)) {
+            managers.when(DefaultResourceManager::get).thenReturn(manager);
+            XAConnection xaConnection = mock(XAConnection.class);
+            when(xaConnection.getXAResource()).thenReturn(xa);
+            ConnectionProxyXA longXidProxy = new ConnectionProxyXA(original, xaConnection, resource, longXid);
+            longXidProxy.init();
+
+            // phase one: registration succeeds and the physical xa branch starts with a bounded gtrid
+            assertDoesNotThrow(() -> longXidProxy.setAutoCommit(false));
+            ArgumentCaptor<Xid> started = ArgumentCaptor.forClass(Xid.class);
+            verify(xa).start(started.capture(), eq(XAResource.TMNOFLAGS));
+            assertTrue(started.getValue().getGlobalTransactionId().length <= XABranchXid.MAX_GTRID_LENGTH);
+
+            longXidProxy.close();
+            // phase two: the TC driven rollback resolves without any encoder failure
+            doThrow(new XAException(XAException.XAER_NOTA)).doNothing().when(xa).rollback(any());
+            assertDoesNotThrow(() -> longXidProxy.xaRollback(longXid, 43L, null));
+            verify(xa, times(2)).rollback(any());
+        }
+    }
 }