You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Fabian Paul (Jira)" <ji...@apache.org> on 2021/08/03 12:35:00 UTC
[jira] [Assigned] (FLINK-23509) FlinkKafkaInternalProducer
overrides static final ProducerIdAndEpoch#NONE during transaction recovery
(fails)
[ https://issues.apache.org/jira/browse/FLINK-23509?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Fabian Paul reassigned FLINK-23509:
-----------------------------------
Assignee: Fabian Paul
> FlinkKafkaInternalProducer overrides static final ProducerIdAndEpoch#NONE during transaction recovery (fails)
> -------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-23509
> URL: https://issues.apache.org/jira/browse/FLINK-23509
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Affects Versions: 1.13.0
> Reporter: Matthias Schwalbe
> Assignee: Fabian Paul
> Priority: Major
> Attachments: 2021-07-26_16-47-48.png
>
>
> When recovering Kafka transactions from a snapshot, FlinkKafkaInternalProducer overrides static final ProducerIdAndEpoch#NONE here:
> [FlinkKafkaInternalProducer#resumeTransaction|https://github.com/apache/flink/blob/f06faf13930f2e8acccf1e04e2c250b85bdbf48e/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/FlinkKafkaInternalProducer.java#L229]
> and consequently TransactionManager initializes transactions as new transactions instead of recovered ones. Here:
> [TransactionManager#initializeTransactions|https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java#L332]
> TransactionManager log (edited for readability):
> {quote}{{[Sink: trxRollupKafkaSink (1/1)#3|#3] INFO org.apache.kafka.clients.producer.KafkaProducer - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Overriding the default enable.idempotence to true since transactional.id is specified.
> [Sink: trxRollupKafkaSink (1/1)#3|#3] INFO org.apache.kafka.clients.producer.KafkaProducer - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Instantiated a transactional producer.
> [Sink: trxRollupKafkaSink (1/1)#3|#3] INFO org.apache.kafka.clients.producer.KafkaProducer - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Overriding the default retries config to the recommended value of 2147483647 since the idempotent producer is enabled.
> [Sink: trxRollupKafkaSink (1/1)#3|#3] INFO org.apache.kafka.clients.producer.KafkaProducer - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Overriding the default acks to all since idempotence is enabled.
> [Sink: trxRollupKafkaSink (1/1)#3|#3] DEBUG org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Transition from state UNINITIALIZED to INITIALIZING
> [Sink: trxRollupKafkaSink (1/1)#3|#3] INFO org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Invoking InitProducerId for the first time in order to acquire a producer ID
> [Sink: trxRollupKafkaSink (1/1)#3|#3] DEBUG org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Enqueuing transactional request InitProducerIdRequestData(transactionalId='Sink: trxRollupKafkaSink-...8b6-2', transactionTimeoutMs=60000, producerId=1545118, producerEpoch=17)
> [kafka-producer-network-thread | producer-Sink: trxRollupKafkaSink-...8b6-2] TRACE org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Request InitProducerIdRequestData(transactionalId='Sink: trxRollupKafkaSink-...8b6-2', transactionTimeoutMs=60000, producerId=1545118, producerEpoch=17) dequeued for sending
> [kafka-producer-network-thread | producer-Sink: trxRollupKafkaSink-...8b6-2] DEBUG org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Enqueuing transactional request FindCoordinatorRequestData(key='Sink: trxRollupKafkaSink-...8b6-2', keyType=1)
> [kafka-producer-network-thread | producer-Sink: trxRollupKafkaSink-...8b6-2] DEBUG org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Enqueuing transactional request InitProducerIdRequestData(transactionalId='Sink: trxRollupKafkaSink-...8b6-2', transactionTimeoutMs=60000, producerId=1545118, producerEpoch=17)
> [kafka-producer-network-thread | producer-Sink: trxRollupKafkaSink-...8b6-2] TRACE org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Request FindCoordinatorRequestData(key='Sink: trxRollupKafkaSink-...8b6-2', keyType=1) dequeued for sending
> [kafka-producer-network-thread | producer-Sink: trxRollupKafkaSink-...8b6-2] TRACE org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Received transactional response FindCoordinatorResponseData(throttleTimeMs=0, errorCode=0, errorMessage='NONE', nodeId=3, host='ulxxtkafbrk03.adgr.net', port=9093) for request FindCoordinatorRequestData(key='Sink: trxRollupKafkaSink-...8b6-2', keyType=1)
> [kafka-producer-network-thread | producer-Sink: trxRollupKafkaSink-...8b6-2] INFO org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Discovered transaction coordinator ulxxtkafbrk03.adgr.net:9093 (id: 3 rack: null)
> [kafka-producer-network-thread | producer-Sink: trxRollupKafkaSink-...8b6-2] TRACE org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Request InitProducerIdRequestData(transactionalId='Sink: trxRollupKafkaSink-...8b6-2', transactionTimeoutMs=60000, producerId=1545118, producerEpoch=17) dequeued for sending
> [kafka-producer-network-thread | producer-Sink: trxRollupKafkaSink-...8b6-2] TRACE org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Received transactional response InitProducerIdResponseData(throttleTimeMs=0, errorCode=47, producerId=-1, producerEpoch=-1) for request InitProducerIdRequestData(transactionalId='Sink: trxRollupKafkaSink-...8b6-2', transactionTimeoutMs=60000, producerId=1545118, producerEpoch=17)
> [kafka-producer-network-thread | producer-Sink: trxRollupKafkaSink-...8b6-2] DEBUG org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Transition from state INITIALIZING to error state FATAL_ERROR
> [Sink: trxRollupKafkaSink (1/1)#3|#3] INFO org.apache.kafka.clients.producer.KafkaProducer - [Producer clientId=producer-Sink: trxRollupKafkaSink-...8b6-2, transactionalId=Sink: trxRollupKafkaSink-...8b6-2] Closing the Kafka producer with timeoutMillis = 0 ms.
> 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)
> }}
> {quote}
> Notice here "Invoking InitProducerId for the first time in order to acquire a producer ID" indicates a request for a new transaction (-1, -1) but below we see instead: "Enqueuing transactional request InitProducerIdRequestData(transactionalId='Sink: ...', ..., producerId=1545118, producerEpoch=17)" because of changed ProducerIdAndEpoch#NONE
>
> TransactionManager#initializeTransactions variables:
> !2021-07-26_16-47-48.png!
> Notice the values above which should be -1, -1.
> Stack trace of TransactionManager#initializeTransactions:
> {quote}initializeTransactions:314, TransactionManager (org.apache.kafka.clients.producer.internals)
> initializeTransactions:310, TransactionManager (org.apache.kafka.clients.producer.internals)
> initTransactions:591, KafkaProducer (org.apache.kafka.clients.producer)
> initTransactions:88, FlinkKafkaInternalProducer (org.apache.flink.streaming.connectors.kafka.internals)
> recoverAndAbort:1060, FlinkKafkaProducer (org.apache.flink.streaming.connectors.kafka)
> recoverAndAbort:99, FlinkKafkaProducer (org.apache.flink.streaming.connectors.kafka)
> initializeState:371, TwoPhaseCommitSinkFunction (org.apache.flink.streaming.api.functions.sink)
> initializeState:1195, FlinkKafkaProducer (org.apache.flink.streaming.connectors.kafka)
> tryRestoreFunction:189, StreamingFunctionUtils (org.apache.flink.streaming.util.functions)
> restoreFunctionState:171, StreamingFunctionUtils (org.apache.flink.streaming.util.functions)
> initializeState:96, AbstractUdfStreamOperator (org.apache.flink.streaming.api.operators)
> initializeOperatorState:118, StreamOperatorStateHandler (org.apache.flink.streaming.api.operators)
> initializeState:290, AbstractStreamOperator (org.apache.flink.streaming.api.operators)
> initializeStateAndOpenOperators:436, OperatorChain (org.apache.flink.streaming.runtime.tasks)
> restoreGates:574, StreamTask (org.apache.flink.streaming.runtime.tasks)
> call:-1, 412600778 (org.apache.flink.streaming.runtime.tasks.StreamTask$$Lambda$745)
> call:55, StreamTaskActionExecutor$1 (org.apache.flink.streaming.runtime.tasks)
> restore:554, StreamTask (org.apache.flink.streaming.runtime.tasks)
> doRun:756, Task (org.apache.flink.runtime.taskmanager)
> run:563, Task (org.apache.flink.runtime.taskmanager)
> run:748, Thread (java.lang)
> {quote}
>
> Stack trace of FlinkKafkaInternalProducer#resumeTransaction when the values are overridden:
> {quote}resumeTransaction:204, FlinkKafkaInternalProducer (org.apache.flink.streaming.connectors.kafka.internals)
> resumeTransaction:196, KafkaProducerJobSink$$anon$1$$anon$2 (ch.viseca.flink.connectors.kafka.sinks)
> recoverAndCommit:1029, FlinkKafkaProducer (org.apache.flink.streaming.connectors.kafka)
> recoverAndCommit:99, FlinkKafkaProducer (org.apache.flink.streaming.connectors.kafka)
> recoverAndCommitInternal:414, TwoPhaseCommitSinkFunction (org.apache.flink.streaming.api.functions.sink)
> initializeState:364, TwoPhaseCommitSinkFunction (org.apache.flink.streaming.api.functions.sink)
> initializeState:1195, FlinkKafkaProducer (org.apache.flink.streaming.connectors.kafka)
> tryRestoreFunction:189, StreamingFunctionUtils (org.apache.flink.streaming.util.functions)
> restoreFunctionState:171, StreamingFunctionUtils (org.apache.flink.streaming.util.functions)
> initializeState:96, AbstractUdfStreamOperator (org.apache.flink.streaming.api.operators)
> initializeOperatorState:118, StreamOperatorStateHandler (org.apache.flink.streaming.api.operators)
> initializeState:290, AbstractStreamOperator (org.apache.flink.streaming.api.operators)
> initializeStateAndOpenOperators:436, OperatorChain (org.apache.flink.streaming.runtime.tasks)
> restoreGates:574, StreamTask (org.apache.flink.streaming.runtime.tasks)
> call:-1, 412600778 (org.apache.flink.streaming.runtime.tasks.StreamTask$$Lambda$745)
> call:55, StreamTaskActionExecutor$1 (org.apache.flink.streaming.runtime.tasks)
> restore:554, StreamTask (org.apache.flink.streaming.runtime.tasks)
> doRun:756, Task (org.apache.flink.runtime.taskmanager)
> run:563, Task (org.apache.flink.runtime.taskmanager)
> run:748, Thread (java.lang)
> {quote}
>
> Background:
> * we recently upgraded from Flink 1.8.0 to 1.13.0
> * FlinkKafkaInternalProducer#resumeTransaction has not changed between these versions, however
> * in Flink 1.8.0 we never observed any resumable transaction as part of a checkpoint
> * we could not determine what actually made the change that causes the failure, however:
> * it would probably be much saver to instead of assigning new values to an arbitrary ProducerIdAndEpoch held by TransactionManager to directly assign a fresh ProducerIdAndEpoch and thus avoid overriding ProducerIdAndEpoch#NONE
> Sample workaround (scala):
> {quote}val sink = new FlinkKafkaProducer[T](val sink = new FlinkKafkaProducer[T]( defaultTopic, schema, getProperties, getSemantic, getProducerPoolSize) {
> override protected def createProducer: FlinkKafkaInternalProducer[Array[Byte], Array[Byte]] =
> Unknown macro: \{ new FlinkKafkaInternalProducer[Array[Byte], Array[Byte]](this.producerConfig) Unknown macro}
> override def resumeTransaction(producerId: Long, epoch: Short): Unit =
> Unknown macro: \{ val transactionManager = FlinkKafkaInternalProducer.getField(kafkaProducer, "transactionManager") transactionManager.synchronized Unknown macro}
> super.resumeTransaction(producerId, epoch) } } }}
> {quote}
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)