Solace场景下Debezium嵌入式引擎能否获取MySQL事务元数据？

核心逻辑说明 #

Debezium不会单独生成事务开始/结束的独立事件，但每个SourceRecord的source元数据字段中包含了事务标识信息，我们可以通过跟踪这些元数据来识别事务的起始与结束：

• 同一个事务内的所有变更事件共享唯一的transaction_id
• 开启特定配置后，事务的最后一个变更事件会携带last_event标记

具体实现步骤 #

1. 开启事务元数据采集 #

在Debezium MySQL连接器配置中添加以下属性，确保能获取到事务相关元数据：

include.transaction.metadata=true

2. 修改事件处理器，跟踪事务状态 #

通过维护一个本地缓存来跟踪活跃事务，结合元数据判断事务边界：

// 并发安全的缓存，用于跟踪当前活跃事务，key为事务ID，value为事务起始时间
private final Map<String, Long> activeTransactions = new ConcurrentHashMap<>();

private void handleChangeEvent(RecordChangeEvent<SourceRecord> event) {
    SourceRecord sourceRecord = event.record();
    Struct sourceStruct = (Struct) sourceRecord.source();
    
    // 提取事务ID、事件时间戳、事务结束标记
    String transactionId = sourceStruct.getString("transaction_id");
    Long eventTs = sourceStruct.getInt64("ts_ms");
    Boolean isLastTransactionEvent = sourceStruct.getBoolean("last_event");

    if (transactionId != null) {
        // 判断事务开始：首次出现该事务ID
        if (!activeTransactions.containsKey(transactionId)) {
            activeTransactions.put(transactionId, eventTs);
            handleTransactionStart(transactionId, eventTs, sourceRecord);
        }

        // 判断事务结束：当前事件是事务最后一条变更
        if (Boolean.TRUE.equals(isLastTransactionEvent)) {
            activeTransactions.remove(transactionId);
            handleTransactionEnd(transactionId, eventTs, sourceRecord);
        }
    }

    // 原有行变更处理逻辑
    processRowChange(sourceRecord);
}

// 事务开始时的业务处理
private void handleTransactionStart(String transactionId, Long startTime, SourceRecord sourceRecord) {
    String gtid = sourceStruct.getString("gtid");
    // 这里可以添加自定义逻辑，比如记录事务启动日志、初始化事务上下文
    System.out.printf("事务启动 | ID: %s | GTID: %s | 时间: %d%n", transactionId, gtid, startTime);
}

// 事务结束时的业务处理
private void handleTransactionEnd(String transactionId, Long endTime, SourceRecord sourceRecord) {
    // 这里可以添加自定义逻辑，比如发送事务结束通知到Solace、统计事务耗时
    System.out.printf("事务结束 | ID: %s | 时间: %d%n", transactionId, endTime);
}

// 原有行变更处理方法
private void processRowChange(SourceRecord sourceRecord) {
    // 解析INSERT/UPDATE/DELETE数据并发送到Solace的逻辑
}

关键注意点 #

• 同一个事务的所有变更事件会按顺序连续发送，可通过transaction_id全程跟踪
• 大事务的变更事件会分批发送，但transaction_id保持一致，不影响事务边界判断
• 必须开启include.transaction.metadata=true配置，否则无法获取transaction_id和last_event标记

内容的提问来源于stack exchange，提问作者geeinhub