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

参考代码:

https://gist.github.com/giraffe-tree/15c5f707d9dfe3221959ae37b4e9d786
2020-11-17 15:24:51
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:1370)
    at org.apache.kafka.clients.producer.internals.
TransactionManager$TxnRequestHandler.onComplete(TransactionManager.java:1278
)
    at org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse
.java:109)
    at org.apache.kafka.clients.NetworkClient.completeResponses(
NetworkClient.java:566)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:558)
    at org.apache.kafka.clients.producer.internals.Sender
.maybeSendAndPollTransactionalRequest(Sender.java:415)
    at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender
.java:313)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:
240)
    at java.lang.Thread.run(Thread.java:748)

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

回复