kafka exactly-once语义下,从svaepoint恢复报错怎么办？

您好：kafka在exactly-once语义下,从svaepoint恢复报错。初步排查认为是kafka事务使用了旧的epoch。请问这个问题怎么处理？

//todo 通过配置传进来

env.setParallelism(1);

env.enableCheckpointing(600000L, CheckpointingMode.EXACTLY_ONCE);

// checkpoint的清除策略(即使任务被显示地取消也会保留checkpoint)

env.getCheckpointConfig()

.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

env.getCheckpointConfig().setCheckpointTimeout(60000);

env.getCheckpointConfig().setTolerableCheckpointFailureNumber(10);

//TODO 生产中必须使用 HDFS

env.setStateBackend(new FsStateBackend("hdfs://10.10.98.226:8020/tmp/checkpoint66"));

StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

public static final String TABLE_NAME = "KafkaTable";

public static final String COLUMN_NAME = "source_value";

public static final String KAFKA_TABLE_FORMAT =

"CREATE TABLE "+TABLE_NAME+" (\n" +

" "+COLUMN_NAME+" STRING\n" +

") WITH (\n" +

" 'connector' = 'kafka',\n" +

" 'topic' = '%s',\n" +

" 'properties.bootstrap.servers' = '%s',\n" +

" 'sink.semantic' = 'exactly-once',\n" +

" 'properties.transaction.timeout.ms' = '30000',\n" +

" 'format' = 'dbz-json'\n" +

")\n";

org.apache.kafka.common.errors.ProducerFencedException: Producer attempted an operation with an old epoch. Either there is a newer producer with the same transactionalId,

or the producer's transaction has been expired by the broker. while recovering transaction KafkaTransactionState [transactionalId=Source:

TableSourceScan(table=[[default_catalog, default_database, debezium_source]], fields=[data]) -> Sink: Sink(table=[default_catalog.default_database.KafkaTable],

fields=[data])-7df19f87deec5680128845fd9a6ca18d-0, producerId=239009, epoch=216]. Presumably this transaction has been already committed before*来自志愿者整理的flink邮件归档