Mastering Seata Global Transactions in Spring Boot 2.2: A Step-by-Step Guide

1. Prepare the Environment

Environment: springboot 2.2.11 + seata 1.3.0

<dependency>
  <groupId>com.alibaba.cloud</groupId>
  <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
  <exclusions>
    <exclusion>
      <groupId>io.seata</groupId>
      <artifactId>seata-all</artifactId>
    </exclusion>
  </exclusions>
</dependency>
<dependency>
  <groupId>io.seata</groupId>
  <artifactId>seata-all</artifactId>
  <version>1.3.0</version>
</dependency>

Enable global transaction:

seata:
  service:
    disable-global-transaction: true

2. Proxy DataSource and Register Proxy Bean

@Bean
@DependsOn({BEAN_NAME_SPRING_APPLICATION_CONTEXT_PROVIDER, BEAN_NAME_FAILURE_HANDLER})
@ConditionalOnMissingBean(GlobalTransactionScanner.class)
public GlobalTransactionScanner globalTransactionScanner(SeataProperties seataProperties, FailureHandler failureHandler) {
    if (LOGGER.isInfoEnabled()) {
        LOGGER.info("Automatically configure Seata");
    }
    return new GlobalTransactionScanner(seataProperties.getApplicationId(), seataProperties.getTxServiceGroup(), failureHandler);
}

@Bean(BEAN_NAME_SEATA_AUTO_DATA_SOURCE_PROXY_CREATOR)
@ConditionalOnProperty(prefix = StarterConstants.SEATA_PREFIX, name = {"enableAutoDataSourceProxy", "enable-auto-data-source-proxy"}, havingValue = "true", matchIfMissing = true)
@ConditionalOnMissingBean(SeataAutoDataSourceProxyCreator.class)
public SeataAutoDataSourceProxyCreator seataAutoDataSourceProxyCreator(SeataProperties seataProperties) {
    return new SeataAutoDataSourceProxyCreator(seataProperties.isUseJdkProxy(), seataProperties.getExcludesForAutoProxying());
}

2.1 Create Proxy Bean

Seata registers all methods annotated with @GlobalTransactional through GlobalTransactionScanner.

public class GlobalTransactionScanner extends AbstractAutoProxyCreator implements InitializingBean, ApplicationContextAware, DisposableBean

Inheritance hierarchy of AbstractAutoProxyCreator (image omitted). GlobalTransactionScanner is a BeanPostProcessor. Important methods in InstantiationAwareBeanPostProcessor are:

postProcessBeforeInstantiation

postProcessAfterInstantiation

postProcessProperties

These methods are not overridden in GlobalTransactionScanner. The parent class implements two BeanPostProcessor methods:

@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) {
    return bean;
}

@Override
public Object postProcessAfterInitialization(@Nullable Object bean, String beanName) {
    if (bean != null) {
        Object cacheKey = getCacheKey(bean.getClass(), beanName);
        if (this.earlyProxyReferences.remove(cacheKey) != bean) {
            return wrapIfNecessary(bean, beanName, cacheKey);
        }
    }
    return bean;
}

The wrapIfNecessary method eventually adds the Seata interceptor to the AdvisedSupport of the bean.

if (!AopUtils.isAopProxy(bean)) {
    bean = super.wrapIfNecessary(bean, beanName, cacheKey);
} else {
    AdvisedSupport advised = SpringProxyUtils.getAdvisedSupport(bean);
    Advisor[] advisor = buildAdvisors(beanName, getAdvicesAndAdvisorsForBean(null, null, null));
    for (Advisor avr : advisor) {
        advised.addAdvisor(0, avr);
    }
}

Helper to obtain AdvisedSupport from a proxy:

public static AdvisedSupport getAdvisedSupport(Object proxy) throws Exception {
    Field h;
    if (AopUtils.isJdkDynamicProxy(proxy)) {
        h = proxy.getClass().getSuperclass().getDeclaredField("h");
    } else {
        h = proxy.getClass().getDeclaredField("CGLIB$CALLBACK_0");
    }
    h.setAccessible(true);
    Object dynamicAdvisedInterceptor = h.get(proxy);
    Field advised = dynamicAdvisedInterceptor.getClass().getDeclaredField("advised");
    advised.setAccessible(true);
    return (AdvisedSupport) advised.get(dynamicAdvisedInterceptor);
}

JDK dynamic proxy uses InvocationHandler:

final class JdkDynamicAopProxy implements AopProxy, InvocationHandler, Serializable {
    private static final Log logger = LogFactory.getLog(JdkDynamicAopProxy.class);
    private final AdvisedSupport advised;
}

CGLIB proxy contains CGLIB$CALLBACK_0 which is a MethodInterceptor:

public class PersonDAOImpl$$EnhancerBySpringCGLIB$$d4658dad extends PersonDAOImpl implements SpringProxy, Advised, Factory {
    private boolean CGLIB$BOUND;
    public static Object CGLIB$FACTORY_DATA;
    private static final ThreadLocal CGLIB$THREAD_CALLBACKS;
    private static final Callback[] CGLIB$STATIC_CALLBACKS;
    private MethodInterceptor CGLIB$CALLBACK_0;
}

Seata’s interceptor is added to AdvisedSupport:

Advisor[] advisor = buildAdvisors(beanName, getAdvicesAndAdvisorsForBean(null, null, null));
for (Advisor avr : advisor) {
    advised.addAdvisor(0, avr);
}
protected Object[] getAdvicesAndAdvisorsForBean(Class beanClass, String beanName, TargetSource customTargetSource) throws BeansException {
    return new Object[]{interceptor};
}

2.2 Create Proxy DataSource

public class SeataAutoDataSourceProxyCreator extends AbstractAutoProxyCreator {
    private static final Logger LOGGER = LoggerFactory.getLogger(SeataAutoDataSourceProxyCreator.class);
    private final String[] excludes;
    private final Advisor advisor = new DefaultIntroductionAdvisor(new SeataAutoDataSourceProxyAdvice());

    public SeataAutoDataSourceProxyCreator(boolean useJdkProxy, String[] excludes) {
        this.excludes = excludes;
        setProxyTargetClass(!useJdkProxy);
    }

    @Override
    protected Object[] getAdvicesAndAdvisorsForBean(Class<?> beanClass, String beanName, TargetSource customTargetSource) {
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Auto proxy of [{}]", beanName);
        }
        return new Object[]{advisor};
    }

    @Override
    protected boolean shouldSkip(Class<?> beanClass, String beanName) {
        return SeataProxy.class.isAssignableFrom(beanClass) || !DataSource.class.isAssignableFrom(beanClass) || Arrays.asList(excludes).contains(beanClass.getName());
    }
}

public class SeataAutoDataSourceProxyAdvice implements MethodInterceptor, IntroductionInfo {
    @Override
    public Object invoke(MethodInvocation invocation) throws Throwable {
        DataSourceProxy dataSourceProxy = DataSourceProxyHolder.get().putDataSource((DataSource) invocation.getThis());
        Method method = invocation.getMethod();
        Object[] args = invocation.getArguments();
        Method m = BeanUtils.findDeclaredMethod(DataSourceProxy.class, method.getName(), method.getParameterTypes());
        if (m != null) {
            return m.invoke(dataSourceProxy, args);
        } else {
            return invocation.proceed();
        }
    }

    @Override
    public Class<?>[] getInterfaces() {
        return new Class[]{SeataProxy.class};
    }
}

The shouldSkip method ensures that only the default DataSource is proxied.

3 Global Transaction Interceptor

Methods annotated with @GlobalTransactional are intercepted by GlobalTransactionalInterceptor.

Core method handleGlobalTransaction uses TransactionalExecutor to collect annotation data into TransactionInfo and then executes the business logic via TransactionalTemplate.execute:

public Object execute(TransactionalExecutor business) throws Throwable {
    TransactionInfo txInfo = business.getTransactionInfo();
    if (txInfo == null) {
        throw new ShouldNeverHappenException("transactionInfo does not exist");
    }
    GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate();
    Propagation propagation = txInfo.getPropagation();
    SuspendedResourcesHolder suspendedResourcesHolder = null;
    try {
        switch (propagation) {
            case NOT_SUPPORTED:
                suspendedResourcesHolder = tx.suspend(true);
                return business.execute();
            case REQUIRES_NEW:
                suspendedResourcesHolder = tx.suspend(true);
                break;
            case SUPPORTS:
                if (!existingTransaction()) {
                    return business.execute();
                }
                break;
            case REQUIRED:
                break;
            case NEVER:
                if (existingTransaction()) {
                    throw new TransactionException(String.format("Existing transaction found for transaction marked with propagation 'never',xid = %s", RootContext.getXID()));
                } else {
                    return business.execute();
                }
            case MANDATORY:
                if (!existingTransaction()) {
                    throw new TransactionException("No existing transaction found for transaction marked with propagation 'mandatory'");
                }
                break;
            default:
                throw new TransactionException("Not Supported Propagation:" + propagation);
        }
        // begin transaction
        beginTransaction(txInfo, tx);
        Object rs = null;
        try {
            rs = business.execute();
        } catch (Throwable ex) {
            completeTransactionAfterThrowing(txInfo, tx, ex);
            throw ex;
        }
        commitTransaction(tx);
        return rs;
    } finally {
        triggerAfterCompletion();
        cleanUp();
        tx.resume(suspendedResourcesHolder);
    }
}

3.1 Obtain Global Transaction

GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate();
public static GlobalTransaction getCurrentOrCreate() {
    GlobalTransaction tx = getCurrent();
    if (tx == null) {
        return createNew();
    }
    return tx;
}
private static GlobalTransaction getCurrent() {
    String xid = RootContext.getXID();
    if (xid == null) {
        return null;
    }
    return new DefaultGlobalTransaction(xid, GlobalStatus.Begin, GlobalTransactionRole.Participant);
}
private static GlobalTransaction createNew() {
    return new DefaultGlobalTransaction();
}

3.2 Begin Global Transaction

private void beginTransaction(TransactionInfo txInfo, GlobalTransaction tx) throws TransactionalExecutor.ExecutionException {
    try {
        triggerBeforeBegin();
        tx.begin(txInfo.getTimeOut(), txInfo.getName());
        triggerAfterBegin();
    } catch (TransactionException txe) {
        throw new TransactionalExecutor.ExecutionException(tx, txe, TransactionalExecutor.Code.BeginFailure);
    }
}
public void begin(int timeout, String name) throws TransactionException {
    if (role != GlobalTransactionRole.Launcher) {
        assertXIDNotNull();
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Ignore Begin(): just involved in global transaction [{}]", xid);
        }
        return;
    }
    assertXIDNull();
    if (RootContext.getXID() != null) {
        throw new IllegalStateException();
    }
    xid = transactionManager.begin(null, null, name, timeout);
    status = GlobalStatus.Begin;
    RootContext.bind(xid);
    if (LOGGER.isInfoEnabled()) {
        LOGGER.info("Begin new global transaction [{}]", xid);
    }
}

Binding XID to the current thread uses SPI to load ContextCore implementation (ThreadLocal or FastThreadLocal).

private static ContextCore CONTEXT_HOLDER = ContextCoreLoader.load();
public static void bind(String xid) {
    if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("bind {}", xid);
    }
    CONTEXT_HOLDER.put(KEY_XID, xid);
}
// META-INF/services/io.seata.core.context.ContextCore
io.seata.core.context.ThreadLocalContextCore
io.seata.core.context.FastThreadLocalContextCore

3.3 Execute Local Business

During execution, a branch is registered and an undo log is recorded.

Object rs = business.execute();

3.3.1 Commit Local Transaction

@Override
public void commit() throws SQLException {
    try {
        LOCK_RETRY_POLICY.execute(() -> {
            doCommit();
            return null;
        });
    } catch (SQLException e) {
        throw e;
    } catch (Exception e) {
        throw new SQLException(e);
    }
}
private void doCommit() throws SQLException {
    if (context.inGlobalTransaction()) {
        processGlobalTransactionCommit();
    } else if (context.isGlobalLockRequire()) {
        processLocalCommitWithGlobalLocks();
    } else {
        targetConnection.commit();
    }
}

3.3.2 Register Branch

private void processGlobalTransactionCommit() throws SQLException {
    try {
        register();
    } catch (TransactionException e) {
        recognizeLockKeyConflictException(e, context.buildLockKeys());
    }
    try {
        UndoLogManagerFactory.getUndoLogManager(this.getDbType()).flushUndoLogs(this);
        targetConnection.commit();
    } catch (Throwable ex) {
        report(false);
        throw new SQLException(ex);
    }
    if (IS_REPORT_SUCCESS_ENABLE) {
        report(true);
    }
    context.reset();
}
private void register() throws TransactionException {
    if (!context.hasUndoLog() || context.getLockKeysBuffer().isEmpty()) {
        return;
    }
    Long branchId = DefaultResourceManager.get().branchRegister(BranchType.AT, getDataSourceProxy().getResourceId(), null, context.getXid(), null, context.buildLockKeys());
    context.setBranchId(branchId);
}

Branch registration uses RPC to TM:

@Override
public Long branchRegister(BranchType branchType, String resourceId, String clientId, String xid, String applicationData, String lockKeys) throws TransactionException {
    BranchRegisterRequest request = new BranchRegisterRequest();
    request.setXid(xid);
    request.setLockKey(lockKeys);
    request.setResourceId(resourceId);
    request.setBranchType(branchType);
    request.setApplicationData(applicationData);
    BranchRegisterResponse response = (BranchRegisterResponse) RmNettyRemotingClient.getInstance().sendSyncRequest(request);
    if (response.getResultCode() == ResultCode.Failed) {
        throw new RmTransactionException(response.getTransactionExceptionCode(), String.format("Response[ %s ]", response.getMsg()));
    }
    return response.getBranchId();
}

3.3.3 Record Undo Log

UndoLogManagerFactory.getUndoLogManager(this.getDbType()).flushUndoLogs(this);
public void flushUndoLogs(ConnectionProxy cp) throws SQLException {
    ConnectionContext connectionContext = cp.getContext();
    if (!connectionContext.hasUndoLog()) {
        return;
    }
    String xid = connectionContext.getXid();
    long branchId = connectionContext.getBranchId();
    BranchUndoLog branchUndoLog = new BranchUndoLog();
    branchUndoLog.setXid(xid);
    branchUndoLog.setBranchId(branchId);
    branchUndoLog.setSqlUndoLogs(connectionContext.getUndoItems());
    UndoLogParser parser = UndoLogParserFactory.getInstance();
    byte[] undoLogContent = parser.encode(branchUndoLog);
    insertUndoLogWithNormal(xid, branchId, buildContext(parser.getName()), undoLogContent, cp.getTargetConnection());
}
@Override
protected void insertUndoLogWithNormal(String xid, long branchId, String rollbackCtx, byte[] undoLogContent, Connection conn) throws SQLException {
    insertUndoLog(xid, branchId, rollbackCtx, undoLogContent, State.Normal, conn);
}
private void insertUndoLog(String xid, long branchId, String rollbackCtx, byte[] undoLogContent, State state, Connection conn) throws SQLException {
    try (PreparedStatement pst = conn.prepareStatement(INSERT_UNDO_LOG_SQL)) {
        pst.setLong(1, branchId);
        pst.setString(2, xid);
        pst.setString(3, rollbackCtx);
        pst.setBlob(4, BlobUtils.bytes2Blob(undoLogContent));
        pst.setInt(5, state.getValue());
        pst.executeUpdate();
    } catch (Exception e) {
        if (!(e instanceof SQLException)) {
            e = new SQLException(e);
        }
        throw (SQLException) e;
    }
}

3.3.4 Commit Local Transaction

targetConnection.commit();

3.3.5 Reset Context

context.reset();
void reset(String xid) {
    this.xid = xid;
    branchId = null;
    this.isGlobalLockRequire = false;
    lockKeysBuffer.clear();
    sqlUndoItemsBuffer.clear();
}

First phase of the global transaction is now complete.

3.4 Global Transaction Commit (Second Phase)

private void commitTransaction(GlobalTransaction tx) throws TransactionalExecutor.ExecutionException {
    try {
        triggerBeforeCommit();
        tx.commit();
        triggerAfterCommit();
    } catch (TransactionException txe) {
        throw new TransactionalExecutor.ExecutionException(tx, txe, TransactionalExecutor.Code.CommitFailure);
    }
}
public void commit() throws TransactionException {
    int retry = COMMIT_RETRY_COUNT <= 0 ? DEFAULT_TM_COMMIT_RETRY_COUNT : COMMIT_RETRY_COUNT;
    try {
        while (retry > 0) {
            try {
                status = transactionManager.commit(xid);
                break;
            } catch (Throwable ex) {
                retry--;
                if (retry == 0) {
                    throw new TransactionException("Failed to report global commit", ex);
                }
            }
        }
    } finally {
        if (RootContext.getXID() != null && xid.equals(RootContext.getXID())) {
            suspend(true);
        }
    }
}
@Override
public GlobalStatus commit(String xid) throws TransactionException {
    GlobalCommitRequest globalCommit = new GlobalCommitRequest();
    globalCommit.setXid(xid);
    GlobalCommitResponse response = (GlobalCommitResponse) syncCall(globalCommit);
    return response.getGlobalStatus();
}
private AbstractTransactionResponse syncCall(AbstractTransactionRequest request) throws TransactionException {
    try {
        return (AbstractTransactionResponse) TmNettyRemotingClient.getInstance().sendSyncRequest(request);
    } catch (TimeoutException toe) {
        throw new TmTransactionException(TransactionExceptionCode.IO, "RPC timeout", toe);
    }
}

Second phase commit finishes.

3.5 Global Transaction Rollback

If any branch throws an exception, the whole transaction rolls back.

3.5.1 Initiator Exception

try {
    rs = business.execute();
} catch (Throwable ex) {
    completeTransactionAfterThrowing(txInfo, tx, ex);
    throw ex;
}

private void completeTransactionAfterThrowing(TransactionInfo txInfo, GlobalTransaction tx, Throwable originalException) throws TransactionalExecutor.ExecutionException {
    if (txInfo != null && txInfo.rollbackOn(originalException)) {
        try {
            rollbackTransaction(tx, originalException);
        } catch (TransactionException txe) {
            throw new TransactionalExecutor.ExecutionException(tx, txe, TransactionalExecutor.Code.RollbackFailure, originalException);
        }
    } else {
        commitTransaction(tx);
    }
}
private void rollbackTransaction(GlobalTransaction tx, Throwable originalException) throws TransactionException, TransactionalExecutor.ExecutionException {
    triggerBeforeRollback();
    tx.rollback();
    triggerAfterRollback();
    throw new TransactionalExecutor.ExecutionException(tx, GlobalStatus.RollbackRetrying.equals(tx.getLocalStatus()) ? TransactionalExecutor.Code.RollbackRetrying : TransactionalExecutor.Code.RollbackDone, originalException);
}
public void rollback() throws TransactionException {
    if (role == GlobalTransactionRole.Participant) {
        return;
    }
    assertXIDNotNull();
    int retry = ROLLBACK_RETRY_COUNT <= 0 ? DEFAULT_TM_ROLLBACK_RETRY_COUNT : ROLLBACK_RETRY_COUNT;
    try {
        while (retry > 0) {
            try {
                status = transactionManager.rollback(xid);
                break;
            } catch (Throwable ex) {
                retry--;
                if (retry == 0) {
                    throw new TransactionException("Failed to report global rollback", ex);
                }
            }
        }
    } finally {
        if (RootContext.getXID() != null && xid.equals(RootContext.getXID())) {
            suspend(true);
        }
    }
}
@Override
public GlobalStatus rollback(String xid) throws TransactionException {
    GlobalRollbackRequest globalRollback = new GlobalRollbackRequest();
    globalRollback.setXid(xid);
    GlobalRollbackResponse response = (GlobalRollbackResponse) syncCall(globalRollback);
    return response.getGlobalStatus();
}

4 XID Propagation

4.1 RestTemplate

@Configuration(proxyBeanMethods = false)
public class SeataRestTemplateAutoConfiguration {
    @Bean
    public SeataRestTemplateInterceptor seataRestTemplateInterceptor() {
        return new SeataRestTemplateInterceptor();
    }
    @Autowired(required = false)
    private Collection<RestTemplate> restTemplates;
    @Autowired
    private SeataRestTemplateInterceptor seataRestTemplateInterceptor;
    @PostConstruct
    public void init() {
        if (this.restTemplates != null) {
            for (RestTemplate restTemplate : restTemplates) {
                List<ClientHttpRequestInterceptor> interceptors = new ArrayList<>(restTemplate.getInterceptors());
                interceptors.add(this.seataRestTemplateInterceptor);
                restTemplate.setInterceptors(interceptors);
            }
        }
    }
}

public class SeataRestTemplateInterceptor implements ClientHttpRequestInterceptor {
    @Override
    public ClientHttpResponse intercept(HttpRequest httpRequest, byte[] bytes, ClientHttpRequestExecution clientHttpRequestExecution) throws IOException {
        HttpRequestWrapper requestWrapper = new HttpRequestWrapper(httpRequest);
        String xid = RootContext.getXID();
        if (!StringUtils.isEmpty(xid)) {
            requestWrapper.getHeaders().add(RootContext.KEY_XID, xid);
        }
        return clientHttpRequestExecution.execute(requestWrapper, bytes);
    }
}

The interceptor adds the TX_XID header to every outgoing request.

4.2 Feign

@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(Client.class)
@AutoConfigureBefore(FeignAutoConfiguration.class)
public class SeataFeignClientAutoConfiguration {
    @Bean
    @Scope("prototype")
    @ConditionalOnClass(name = "com.netflix.hystrix.HystrixCommand")
    @ConditionalOnProperty(name = "feign.hystrix.enabled", havingValue = "true")
    Feign.Builder feignHystrixBuilder(BeanFactory beanFactory) {
        return SeataHystrixFeignBuilder.builder(beanFactory);
    }
}

static Feign.Builder builder(BeanFactory beanFactory) {
    return HystrixFeign.builder().retryer(Retryer.NEVER_RETRY).client(new SeataFeignClient(beanFactory));
}

public class SeataFeignClient implements Client {
    private final Client delegate;
    private final BeanFactory beanFactory;
    private static final int MAP_SIZE = 16;
    SeataFeignClient(BeanFactory beanFactory) {
        this.beanFactory = beanFactory;
        this.delegate = new Client.Default(null, null);
    }
    @Override
    public Response execute(Request request, Request.Options options) throws IOException {
        Request modifiedRequest = getModifyRequest(request);
        return this.delegate.execute(modifiedRequest, options);
    }
    private Request getModifyRequest(Request request) {
        String xid = RootContext.getXID();
        if (StringUtils.isEmpty(xid)) {
            return request;
        }
        Map<String, Collection<String>> headers = new HashMap<>(MAP_SIZE);
        headers.putAll(request.headers());
        List<String> seataXid = new ArrayList<>();
        seataXid.add(xid);
        headers.put(RootContext.KEY_XID, seataXid);
        return Request.create(request.method(), request.url(), headers, request.body(), request.charset());
    }
}

5 Participant Integration

Even without explicit annotations, participants join the global transaction because the auto‑configured SeataAutoDataSourceProxyCreator proxies the default DataSource into a DataSourceProxy. When a method executes, it obtains a ConnectionProxy from the proxy data source.

@Override
public ConnectionProxy getConnection() throws SQLException {
    Connection targetConnection = targetDataSource.getConnection();
    return new ConnectionProxy(this, targetConnection);
}

The ConnectionProxy holds a ConnectionContext that tracks XID, branch ID, lock requirements, and undo logs.

public class ConnectionProxy extends AbstractConnectionProxy {
    private ConnectionContext context = new ConnectionContext();
}

During statement execution, BaseTransactionalExecutor binds the current XID to the ConnectionProxy:

if (RootContext.inGlobalTransaction()) {
    String xid = RootContext.getXID();
    statementProxy.getConnectionProxy().bind(xid);
}
statementProxy.getConnectionProxy().setGlobalLockRequire(RootContext.requireGlobalLock());

public void bind(String xid) {
    context.bind(xid);
}

Thus the XID is available for the later commit/rollback logic.

Finally, the article ends with a call for readers to follow and share.