https://flink.apache.org/ logo
Join Slack
Powered by
Hi again Flinkers, Currently I have a Flink job t...
# troubleshooting
r
Hi again Flinkers, Currently I have a Flink job that handles writing to Kafka using exactly-once semantics and thus uses transactions at the KafkaSink to handle the writes. I had a long running issue that caused the job to be down for long enough so that all of the in-flight transactions are now expired:
Copy code
2023-09-06 08:15:33
java.lang.IllegalStateException: Failed to commit KafkaCommittable{producerId=17310759, epoch=0, transactionalId=my-job-events-0-41817}
	at org.apache.flink.streaming.runtime.operators.sink.committables.CommitRequestImpl.signalFailedWithUnknownReason(CommitRequestImpl.java:77)
	at org.apache.flink.connector.kafka.sink.KafkaCommitter.commit(KafkaCommitter.java:119)
	at org.apache.flink.streaming.runtime.operators.sink.committables.CheckpointCommittableManagerImpl.commit(CheckpointCommittableManagerImpl.java:127)
	at org.apache.flink.streaming.runtime.operators.sink.CommitterOperator.commitAndEmit(CommitterOperator.java:176)
	at org.apache.flink.streaming.runtime.operators.sink.CommitterOperator.commitAndEmitCheckpoints(CommitterOperator.java:160)
	at org.apache.flink.streaming.runtime.operators.sink.CommitterOperator.initializeState(CommitterOperator.java:121)
	at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.initializeOperatorState(StreamOperatorStateHandler.java:122)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:286)
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:106)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:700)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:676)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:643)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:948)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:917)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:741)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:563)
	at java.lang.Thread.run(Thread.java:750)
Caused by: org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id.
Is there some type of configuration that would allow to tolerate/ignore these types of errors? I reran the job using a different transaction-prefix, however it seems that the in-flight messages are all still pointing to older/expired transactions and throwing an exception at the sink-level. Any advice/insight would be appreciated.
m
There isn't a nice fix for this yet
r
Thanks @Martijn Visser, so it’s likely that I’d need to restore the job sans state to handle this? This particular job has a related bootstrap job that runs to pre-fill all of the state for enrichment/lookups so I was trying to avoid that.
m
I believe so, unless this is only throwing a WARN and not an ERROR...
r
The above error is actually a WARN but it looks like it throws an exception which crashes the job:
Copy code
2023-09-06 13:30:48,910 ERROR org.apache.flink.connector.kafka.sink.KafkaCommitter         [] - Transaction (org.apache.flink.streaming.runtime.operators.sink.committables.CommitRequestImpl@17e2a69e) encountered error and data has been potentially lost.
org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id.
2023-09-06 13:30:48,911 INFO  org.apache.kafka.clients.producer.KafkaProducer              [] - [Producer clientId=producer-common-event-enrichment-pinot-investigate-events-0-41817, transactionalId=common-event-enrichment-pinot-investigate-events-0-41817] Closing the Kafka producer with timeoutMillis = 3600000 ms.
2023-09-06 13:30:48,911 INFO  org.apache.kafka.clients.producer.internals.Sender           [] - [Producer clientId=producer-common-event-enrichment-pinot-investigate-events-0-41817, transactionalId=common-event-enrichment-pinot-investigate-events-0-41817] Aborting incomplete transaction due to shutdown
2023-09-06 13:30:48,911 INFO  org.apache.kafka.clients.producer.internals.TransactionManager [] - [Producer clientId=producer-common-event-enrichment-pinot-investigate-events-0-41817, transactionalId=common-event-enrichment-pinot-investigate-events-0-41817] Invoking InitProducerId with current producer ID and epoch (producerId=17311257, epoch=0) in order to bump the epoch
2023-09-06 13:30:48,912 INFO  org.apache.kafka.clients.producer.internals.TransactionManager [] - [Producer clientId=producer-common-event-enrichment-pinot-investigate-events-0-41817, transactionalId=common-event-enrichment-pinot-investigate-events-0-41817] Transiting to fatal error state due to org.apache.kafka.common.errors.ProducerFencedException: There is a newer producer with the same transactionalId which fences the current one.
2023-09-06 13:30:48,914 INFO  org.apache.kafka.common.metrics.Metrics                      [] - Metrics scheduler closed
2023-09-06 13:30:48,914 INFO  org.apache.kafka.common.metrics.Metrics                      [] - Closing reporter org.apache.kafka.common.metrics.JmxReporter
2023-09-06 13:30:48,915 INFO  org.apache.kafka.common.metrics.Metrics                      [] - Metrics reporters closed
2023-09-06 13:30:48,916 INFO  org.apache.kafka.common.utils.AppInfoParser                  [] - App info kafka.producer for producer-common-event-enrichment-pinot-investigate-events-0-41817 unregistered
2023-09-06 13:30:48,916 WARN  org.apache.flink.runtime.taskmanager.Task                    [] - filter-immutable-events -> map-immutable-enriched-events-to-common-events -> immutable-enriched-events-sink: Writer -> immutable-enriched-events-sink: Committer (1/1)#22708 (15f3b1b211815581ed5fbde18b9c6677) switched from INITIALIZING to FAILED with failure cause: java.lang.IllegalStateException: Failed to commit KafkaCommittable{producerId=17311257, epoch=0, transactionalId=common-event-enrichment-pinot-investigate-events-0-41817}
	at org.apache.flink.streaming.runtime.operators.sink.committables.CommitRequestImpl.signalFailedWithUnknownReason(CommitRequestImpl.java:77)
	at org.apache.flink.connector.kafka.sink.KafkaCommitter.commit(KafkaCommitter.java:119)
	at org.apache.flink.streaming.runtime.operators.sink.committables.CheckpointCommittableManagerImpl.commit(CheckpointCommittableManagerImpl.java:127)
	at org.apache.flink.streaming.runtime.operators.sink.CommitterOperator.commitAndEmit(CommitterOperator.java:176)
	at org.apache.flink.streaming.runtime.operators.sink.CommitterOperator.commitAndEmitCheckpoints(CommitterOperator.java:160)
	at org.apache.flink.streaming.runtime.operators.sink.CommitterOperator.initializeState(CommitterOperator.java:121)
	at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.initializeOperatorState(StreamOperatorStateHandler.java:122)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:286)
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:106)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:700)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:676)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:643)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:948)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:917)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:741)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:563)
	at java.lang.Thread.run(Thread.java:750)
m
Ah yes that makes sense with exactly-once, since else you can't guarantee that anymore duh
r
Basically these:
Copy code
Transiting to abortable error state due to org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id.
...
Transaction (org.apache.flink.streaming.runtime.operators.sink.committables.CommitRequestImpl@17e2a69e) encountered error and data has been potentially lost.
...
The producer attempted to use a producer id which is not currently assigned to its transactional id.
...
Transiting to fatal error state due to org.apache.kafka.common.errors.ProducerFencedException: There is a newer producer with the same transactionalId which fences the current one.
Historically, we have a parameter for the transaction-prefix that we used to update to handle “flushing” these bad transactions, but for whatever reason it’s continually restoring these in a crashloop.
m
Yeah because IIRC the transaction prefix is only used when first starting the job
r
The KIP/FLIPs sound promising. I suppose technically I could pass a parameter in for the UID for the transactional sinks to not restore the older records, but then it’d get messy transitioning back to the original UID.
Hey @Martijn Visser, just circling back around to this since I ran into an issue that you might have some advice on troubleshooting (it’s related). Earlier today I did a deploy of the job above in a GKE environment. The job was running without issue, however after being suspended (taking a savepoint), and restoring from the savepoint after the deployment, I began seeing the above issue (InvalidPidMappingException) on the job. I’ve triply confirmed that nothing really changed in the job outside of the addition of a rebalance() operator that occurs upstream from the sink. Additionally, it seems that this exception is being thrown on all of the sinks that use transactions/exactly_once. The only thing I can think of that makes sense would be: • records were written to Kafka • job was suspended (possibly prior to committing those offsets) • job gets restored from a savepoint • uncommitted (but written to Kafka) records are restored from state • records fail since transaction no longer exists
I also realized that one of the sinks had a topic change recently without updating the uid/name for the sink. Since it uses exactly_once processing, I’m guessing that this could potentially cause this problem as well?
2 Views
Open in Slack
PreviousNext
Powered by