dtm-labs / dtmcli-java

Official java client for distributed transaction framework dtm

dtmcli-java

English | 简体中文

dtmcli-java is the Java client SDK for distributed transaction manager dtm

What is DTM

DTM is a distributed transaction framework which provides cross-service eventual data consistency. It provides saga, tcc, xa, 2-phase message, outbox patterns for a variety of application scenarios. It also supports multiple languages and multiple store engine to form up a transaction as following:

Features

• Support for multiple languages: Go, Java, PHP, C#, Python, Nodejs SDKs
• Support for multiple transaction patterns: SAGA, TCC, XA
• Support for OutBox pattern: 2-phase messages, a more elegant solution than OutBox
• Support for multiple database transactions: Mysql, Redis, MongoDB, Postgres, TDSQL, etc.
• Support for multiple storage engines: Mysql (common), Redis (high performance), MongoDB (in planning)
• Support for multiple microservices architectures: go-zero, go-kratos/kratos, polarismesh/polaris
• Support for high availability and easy horizontal scaling

Usage

Step 1: Determine the version

1. Springcloud Projects

• If the version of springboot >= 2.4.0, choose a corresponding version of dtmcli-springcloud.
• If the version of springboot <> 2.4.0, use dtmcli-java, it provide the interfaces for micro-services, please set the configuration for nacos.

2. Not Springcloud Projects.

• use dtmcli-java，and set your configurations.

artifact	version	dtmcli-java version	remark
dtmcli-springcloud	2.1.4.1	2.4.0 <= springboot version < 2.5.13	for springboot >= 2.5.0，please set spring.cloud.compatibility-verifier.enabled=false
dtmcli-springcloud	2.1.4.2	2.6.0 <= springboot version < 2.6.latest
dtmcli-java	2.1.4	others

Step 2: Add Dependencies

Maven：

<dependency>
	<groupId>io.github.dtm-labs</groupId>
	<artifactId>dtmcli-springcloud</artifactId>
	<version>${dtmcli.version}</version>
</dependency>

Gradle:

dependencies {
	        implementation 'io.github.dtm-labs:dtmcli-springcloud:${dtmcli.version}'
	}

Step 3：Configure dtmcli-java

If you are using dtmcli-java，new a file named dtm-conf.properties

• Case 1: Using naceos

serverAddr=127.0.0.1:8848
username=nacos
password=nacos
namespace=c3dc917d-906a-429d-90a9-85012b41014e
dtm.service.name=dtmService
dtm.service.registryType=nacos

• Case 2: Connect to dtmsvr directly

dtm.ipport=127.0.0.1:36789

Sample

@RequestMapping("testTcc")
    public String testTcc() {
        // new dtm clinet
        DtmClient dtmClient = new DtmClient(ipPort);
        //create TCC transaction
        try {
            dtmClient.tccGlobalTransaction(dtmClient.genGid(), TccTestController::tccTrans);
        } catch (Exception e) {
            log.error("tccGlobalTransaction error", e);
            return "fail";
        }
        return "success";
    }

/**
     * define TCC sub-transactions by calling callBranch
     *
     * @param tcc
     * @return
     * @see TransController
     */
    public static void tccTrans(Tcc tcc) throws Exception {
        Response outResponse = tcc
                .callBranch("", svc + "/TransOutTry", svc + "/TransOutConfirm", svc + "/TransOutCancel");
        log.info("outResponse:{}", outResponse);
        Response inResponse = tcc.callBranch("", svc + "/TransInTry", svc + "/TransInConfirm", svc + "/TransInCancel");
        log.info("inResponse:{}", inResponse);
    }

Complete Sample

Sample for dtmcli-java

dtmcli-java-sample dtmcli-java-sample-use-configuration

Sample for dtmcli-springcloud

dtmcli-java-spring-sample

dtmcli-java's Issues

No support for 2-phase message?

No support for 2-phase message yet?

Saga 循环依赖问题

创建saga事务的时候设置http头可能出现栈溢出。

编译jdk版本过高

阅读代码发现的4个问题

看了一下Java SDK的源码，发现如下几个问题（由大到小排序）

---大问题---

1. 为完成本地事务屏障，私自打开和关闭JDBC链接，导致和原有Spring 的连接/事务管理功能脱钩。实际项目中，用户代码不太可能直接基于指定的JDBC连接执行，一般是基于Spring技术栈提供的更高级的数据访问技术操作，这会导致SDK保证事务屏障的JDBC连接和实际业务的连接并非同一连接。

---小问题---

2. 本地事务屏障只支持MySQL

3. 不会自动创建barrier表

tcc：get response.body().string() closed

tcc模式下想要解析 response的body内容
但response.body().string()只能请求一次，在dtm内已经调用过此方法
请问是否还有其他解决方案可以获取返回值吗？

public static void checkResult(Response response) throws Exception { if (response.code() >= 400) { throw new FailureException(response.message()); } else { ResponseBody body = response.body(); String result; if (body != null && !StringUtils.isBlank(result = body.string())) { if (result.contains("FAILURE")) { throw new FailureException("Service returned failed"); } } else { throw new FailureException("response is null"); } } }

BUG: tcc query-string

BUG:
FILE: tcc.java

HttpResponse response2 = HttpRequest.post(tryUrl)
                    .body(JSONUtil.toJsonStr(body))
                    .form(paramMap2) // HERE
                    .execute();

maybe FIX :

            HttpResponse response2 = HttpRequest.post(tryUrl + "?" + mapToQueryString(paramMap2))
                    .body(JSONUtil.toJsonStr(body))
                    //.form(paramMap2)
                    .execute();

  private static String mapToQueryString(Map<String, Object> queryParams) {
        String queryString = queryParams.entrySet().stream()
                .map(e -> e.getKey() + "=" + e.getValue())
                .collect(Collectors.joining("&"));
        return URLUtil.encode(queryString);
    }

设置全局事务默认的超时时间，会出现递归调用，导致栈溢出

// pub.dtm.client.saga.Saga#setTimeoutToFail

public Saga setTimeoutToFail(long timeoutToFail) {
    this.setTimeoutToFail(timeoutToFail);
    return this;
}

这个代码，是一个死递归，会导致栈溢出

dtmcli-core没有实现Saga模式下的空补偿，只实现了TCC模式下的空补偿

// pub.dtm.client.barrier.BranchBarrier#insertBarrier

private boolean insertBarrier(Connection connection) throws SQLException {
log.info("insert barrier {}", this);
if (Objects.isNull(connection)) {
return false;
}
PreparedStatement preparedStatement = null;
try {
String sql = "insert ignore into barrier(trans_type, gid, branch_id, op, barrier_id, reason) values(?,?,?,?,?,?)";
preparedStatement = connection.prepareStatement(sql);
preparedStatement.setString(1, this.getTransTypeEnum().getValue());
preparedStatement.setString(2, this.getGid());
preparedStatement.setString(3, branchId);
preparedStatement.setString(4, op);
preparedStatement.setString(5, String.format("%02d", barrierId));
preparedStatement.setString(6, op);

        if (preparedStatement.executeUpdate() == 0) {
            return false;
        }
        if (ParamFieldConstants.CANCEL.equals(op)) {
            int opIndex = 4;
            preparedStatement.setString(opIndex, ParamFieldConstants.TRY);
            if (preparedStatement.executeUpdate() > 0) {
                return false;
            }
        }
    } finally {
        if (Objects.nonNull(preparedStatement)) {
            preparedStatement.close();
        }
    }
    return true;
}

这个地方，只判断了ParamFieldConstants.CANCEL.equals(op)，而Saga模式的补偿op是compensate，不是cancel

missing java-sample-demo sql script

CREATE TABLE barrier (
id bigint NOT NULL AUTO_INCREMENT,
user_id int DEFAULT NULL,
amount int DEFAULT NULL,
trans_type varchar(255) DEFAULT NULL,
gid varchar(100) DEFAULT NULL,
branch_id varchar(100) DEFAULT NULL,
op varchar(255) DEFAULT NULL,
barrier_id varchar(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL,
reason varchar(255) DEFAULT NULL,
PRIMARY KEY (id)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

support gRPC