[ https://issues.apache.org/jira/browse/FLINK-22328?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Flink Jira Bot updated FLINK-22328: ----------------------------------- Labels: Transactional auto-deprioritized-major kafka stale-minor (was: Transactional auto-deprioritized-major kafka) I am the [Flink Jira Bot|https://github.com/apache/flink-jira-bot/] and I help the community manage its development. I see this issues has been marked as Minor but is unassigned and neither itself nor its Sub-Tasks have been updated for 180 days. I have gone ahead and marked it "stale-minor". If this ticket is still Minor, please either assign yourself or give an update. Afterwards, please remove the label or in 7 days the issue will be deprioritized. > Failed to send data to Kafka: Producer attempted an operation with an old > epoch > ------------------------------------------------------------------------------- > > Key: FLINK-22328 > URL: https://issues.apache.org/jira/browse/FLINK-22328 > Project: Flink > Issue Type: Bug > Components: Connectors / Kafka > Affects Versions: 1.11.1 > Environment: kafka 3.0.0 > Flink 1.11.1 > Reporter: 孙峰 > Priority: Minor > Labels: Transactional, auto-deprioritized-major, kafka, > stale-minor > > Flink job fails occasionally.Here is the stacktrace: > {code:java} > org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Failed to > send data to Kafka: Producer attempted an operation with an old epoch.Either > there is a newer producer with the same transactionalId, or the producer's > transactioin has been expired by the broker. > at > org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.checkErroneous(FlinkKafkaProducer.java:640) > at > org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.invoke(FlinkKafkaProducer.java:157) > at > org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.invoke(FlinkKafkaProducer.java:81) > at > org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.invoke(TwoPhaseCommitSinkFunction.java:235) > at > org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:56) > .... > ... > Caused by: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 transactioin has been > expired by the broker.{code} > The job use FlinkKafkaProducer with EXACTLY_ONCE and deploy on Yarn. > In the debugging information I found the transactionalId is "Source: Custom > Source -> (Process -> Sink: errorMessageToKafka, Sink: etlMultiTopicSink) > -03f86923ea4164263684d81917202071-0". > In kafka server.log,the exception: > {code:java} > ERROR [ReplicaManager borker=1004] Error processing append on partition > ods_source-2 (kafka.server.ReplicaManager) > org.apache.kafka.common.errors.ProducerFencedExceptioin: Producer's epoch is > no longer valid.There is probably another producer with a newer epoch. 158 > (request epoch), 159 (server:epoch){code} > Here is the log that kafka increase epoch with this transactionalId "Source: > Custom Source -> (Process -> Sink: errorMessageToKafka, Sink: > etlMultiTopicSink) -03f86923ea4164263684d81917202071-0": > {code:java} > INFO [TransactionCoordinator id=1003] Initialized transactionalId Source: > Custom Source -> (Process -> Sink: errorMessageToKafka, Sink: > etlMultiTopicSink) -03f86923ea4164263684d81917202071-0 with producerId 21036 > and producer epoch 158 on partition _transaction_state-3 > (kafka.coordinator.transaction.TransactionCoordinator) > INFO [TransactionCoordinator id=1003] Initialized transactionalId Source: > Custom Source -> (Process -> Sink: errorMessageToKafka, Sink: > etlMultiTopicSink) -03f86923ea4164263684d81917202071-0 with producerId 21036 > and producer epoch 160 on partition _transaction_state-3 > (kafka.coordinator.transaction.TransactionCoordinator) > {code} > There is no info that record kafka set producer epoch to 159. -- This message was sent by Atlassian Jira (v8.20.1#820001)