[
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 (was: Transactional
kafka stale-major)
Priority: Minor (was: Major)
This issue was labeled "stale-major" 7 ago and has not received any updates so
it is being deprioritized. If this ticket is actually Major, please raise the
priority and ask a committer to assign you the issue or revive the public
discussion.
> 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
>
> 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.3.4#803005)
