Flink kafka

[周瑞]

程序用于测试 flink kafka exactly once, 普通提交可以正常运行, 但如果从 savepoint 中恢复就会报下面的错误
kafka server 端, 配置了  transaction.timeout.ms = 900000

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' = '900000',\n" +
                "   'sink.partitioner' = 
'com.woqutench.qmatrix.cdc.extractor.PkPartitioner',\n" +
                "   'properties.max.in.flight.requests.per.connection' = 
'1',\n" +
                "   'properties.enable.idempotence' = 'true',\n" +
                "   'properties.transactional.id' = '%s',\n" +
                "   'format' = 'dbz-json'\n" +
                ")\n";
2021-05-30 19:38:57,513 WARN  org.apache.flink.runtime.taskmanager.Task         
           [] - Source: TableSourceScan(table=[[default_catalog, 
default_database, debezium_source]], fields=[data]) -> Sink: 
Sink(table=[default_catalog.default_database.KafkaTable2], fields=[data]) 
(1/1)#144518 (4739d37a5f82268901f8fb51b39735e9) switched from INITIALIZING to 
FAILED with failure cause: org.apache.kafka.common.KafkaException: Unexpected 
error in InitProducerIdResponse; 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.
    at 
org.apache.kafka.clients.producer.internals.TransactionManager$InitProducerIdHandler.handleResponse(TransactionManager.java:1352)
    at 
org.apache.kafka.clients.producer.internals.TransactionManager$TxnRequestHandler.onComplete(TransactionManager.java:1260)
    at 
org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse.java:109)
    at 
org.apache.kafka.clients.NetworkClient.completeResponses(NetworkClient.java:572)
    at 
org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:564)
    at 
org.apache.kafka.clients.producer.internals.Sender.maybeSendAndPollTransactionalRequest(Sender.java:414)
    at 
org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:312)
    at 
org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:239)
    at java.lang.Thread.run(Thread.java:748)


查阅了 google 上的相关资料, 但是仍然无法解决, 有人遇到过类似的问题? 或者能提供排查思路么?