Seatunnel同步Kafka灾备集群时重复消息问题求助

将源Kafka集群消息同步至灾备目标Kafka集群，要求重启后从最后偏移量读取、不产生重复消息，但测试发现重启后偶尔出现重复记录。以下是相关配置及解决方案：

相关配置信息 #

创建Kafka主题命令 #

bin/kafka-topics.sh --create --topic source --bootstrap-server instance-3:9092 --partitions 1 --replication-factor 1
bin/kafka-topics.sh --create --topic sink --bootstrap-server instance-4:9092 --partitions 1 --replication-factor 1

Seatunnel执行命令 #

bin/seatunnel.sh --config ./config/kafka-cdc.conf -e local

原kafka-cdc.conf配置 #

# Defining the runtime environment
env {
  # You can set flink configuration here
  job.mode = "STREAMING"
}
source {
  Kafka {
    result_table_name="kafka-cdc"
    topic = "source"
    bootstrap.servers = "instance-3:9092"
    kafka.config = {
      auto.offset.reset = "earliest"
      enable.auto.commit = "true"
    }
}
}
sink {
  Console {
  source_table_name="kafka-cdc"
  }
  kafka {
      source_table_name="kafka-cdc"
      topic = "sink"
      bootstrap.servers = "instance-4:9092"
      kafka.config = {
        acks = "all"
        request.timeout.ms = 60000
        buffer.memory = 33554432
    }
      format = text
  }
}

问题根源 #

1. 自动提交偏移量不可靠：Kafka Source开启enable.auto.commit = "true"后，偏移量由Kafka客户端异步提交，若任务在消息处理完成但偏移量未提交时重启，会重复消费已处理的消息。
2. local模式状态丢失：local模式下任务状态默认存于内存，重启后状态重置，偏移量可能回到上次提交位置甚至最早位置。
3. Sink未实现Exactly-Once：当前Kafka Sink仅配置acks = "all"，未开启事务，无法配合Checkpoint实现端到端的Exactly-Once语义。

解决方案 #

1. 关闭Kafka Source自动提交，改用Checkpoint管理偏移量 #

修改Source配置，禁用自动提交，让Seatunnel通过Checkpoint统一管理偏移量：

source {
  Kafka {
    result_table_name="kafka-cdc"
    topic = "source"
    bootstrap.servers = "instance-3:9092"
    kafka.config = {
      auto.offset.reset = "earliest"
      enable.auto.commit = "false"  # 关闭自动提交
    }
  }
}

2. 配置Checkpoint持久化 #

在env块中添加Checkpoint配置，将状态持久化到本地或分布式存储，确保重启后可恢复：

env {
  job.mode = "STREAMING"
  # 每30秒触发一次Checkpoint
  execution.checkpointing.interval = 30000
  # 任务取消或失败后保留Checkpoint
  execution.checkpointing.externalized-checkpoint-retention = "RETAIN_ON_CANCELLATION"
  # 使用文件系统作为状态后端
  execution.checkpointing.state.backend = "filesystem"
  # 替换为实际持久化目录（本地路径或HDFS路径均可）
  execution.checkpointing.state.backend.fs.checkpointdir = "/data/seatunnel/checkpoints"
}

3. 配置Kafka Sink实现Exactly-Once #

开启Kafka Sink的事务和幂等性，配合Checkpoint确保消息仅写入一次：

sink {
  Console {
    source_table_name="kafka-cdc"
  }
  kafka {
      source_table_name="kafka-cdc"
      topic = "sink"
      bootstrap.servers = "instance-4:9092"
      kafka.config = {
        acks = "all"
        request.timeout.ms = 60000
        buffer.memory = 33554432
        # 唯一事务ID，多并行度时每个实例需配置不同值
        transactional.id = "sink-transaction-001"
        # 开启幂等性
        enable.idempotence = "true"
      }
      format = text
      # 事务模式下并行度建议设为1，多并行度需保证每个实例transactional.id唯一
      sink.parallelism = 1
  }
}

4. 生产环境避免使用local模式 #

local模式仅适用于开发调试，生产环境建议将任务提交到Flink集群运行，确保状态管理的稳定性。

验证方法 #

1. 修改配置后启动任务，向source主题发送10条测试消息
2. 等待至少一次Checkpoint触发（按配置的30秒间隔）
3. 手动停止任务后重新启动
4. 检查sink主题消息数量，确认无重复且重启后从停止位置继续消费

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