Hi, Currently, I was trying to update our kafka cluster with larger ` transaction.max.timeout.ms`. The original setting is kafka's default value (i.e. 15 minutes) and I tried to set as 3 hours.
When I was doing rolling-restart for my brokers, this exception came to me on the next checkpoint after I restarted the broker with active controller. java.lang.RuntimeException: Error while confirming checkpoint at > org.apache.flink.runtime.taskmanager.Task$2.run(Task.java:1218) at > java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at > java.util.concurrent.FutureTask.run(FutureTask.java:266) at > java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) > at > java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) > at java.lang.Thread.run(Thread.java:748) Caused by: > org.apache.flink.util.FlinkRuntimeException: Committing one of transactions > failed, logging first encountered failure at > org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.notifyCheckpointComplete(TwoPhaseCommitSinkFunction.java:296) > at > org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.notifyCheckpointComplete(AbstractUdfStreamOperator.java:130) > at > org.apache.flink.streaming.runtime.tasks.StreamTask.notifyCheckpointComplete(StreamTask.java:684) > at org.apache.flink.runtime.taskmanager.Task$2.run(Task.java:1213) ... 5 > more Caused by: org.apache.kafka.common.errors.InvalidTxnStateException: > The producer attempted a transactional operation in an invalid state I have no idea why it happened, and I didn't find any error log from brokers. Does anyone have this exception before? How can I prevent from this exception when I tried to restart kafka cluster? Does this exception mean that I will lost data in some of these transactions? flink cluster version: 1.8.1 kafka cluster version: 1.0.1 flink kafka producer version: universal producer transaction timeout: 15 minutes checkpoint interval: 5 minutes number of concurrent checkpoint: 1 max checkpoint duration before and after the exception occurred: < 2 minutes Best, Tony Wei
