https://flink.apache.org/ logo
Join Slack
Powered by
Hi all. Does anyone know what might be causing thi...
# troubleshooting
a
Hi all. Does anyone know what might be causing this exception after sometime? I'm running a script continually and loading my Kafka topics with events for processing. It seems to work for quite some time and then gives a 
InvalidProducerEpochException
with some subsequent exceptions. I am ensuring that the 
setTransactionalIdPrefix
is unique. Flink Version: 1.17.1 (same version as flink-connector-kafka) Kafka version: 3.2.3 Java version: 11
Copy code
public KafkaSink<String> getProducer(String topic) {
    // Clone properties for potential customisation
    Properties clonedProperties = (Properties) getProperties().clone();

    // Build the KafkaRecordSerializationSchema
    KafkaRecordSerializationSchema<String> recordSerializer =
        KafkaRecordSerializationSchema.builder()
            .setTopic(topic)
            .setValueSerializationSchema(new SimpleStringSchema())
            .build();

    // Get a unique producer ID
    String uuid = CommonUtils.generateUUID();

    // Build the unique producer ID using the thread ID
    String uniqueProducerId = topic + "_" + uuid;

    // Build and return the KafkaSink
    return KafkaSink.<String>builder()
        .setBootstrapServers(getKafkaServerUrl())
        .setKafkaProducerConfig(clonedProperties)
        .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
        .setTransactionalIdPrefix(uniqueProducerId)
        .setRecordSerializer(recordSerializer)
        .build();
  }
Copy code
2023-10-13 02:52:47,214 INFO  org.apache.kafka.clients.producer.internals.TransactionManager [] - [Producer clientId=producer-domain-nvidia-peoplenet-person-PersonExtractionEvent_53b445d5-0e26-4c01-8892-1a8d622b43c8-0-1, transactionalId=domain-nvidia-peoplenet-person-PersonExtractionEvent_53b445d5-0e26-4c01-8892-1a8d622b43c8-0-1] Transiting to abortable error state due to org.apache.kafka.common.errors.InvalidProducerEpochException: Producer attempted to produce with an old epoch.
2023-10-13 02:52:48,237 INFO  org.apache.kafka.clients.producer.internals.TransactionManager [] - [Producer clientId=producer-domain-nvidia-peoplenet-person-PersonExtractionEvent_53b445d5-0e26-4c01-8892-1a8d622b43c8-0-1, transactionalId=domain-nvidia-peoplenet-person-PersonExtractionEvent_53b445d5-0e26-4c01-8892-1a8d622b43c8-0-1] Transiting to abortable error state due to org.apache.kafka.common.errors.InvalidProducerEpochException: Producer with transactionalId 'domain-nvidia-peoplenet-person-PersonExtractionEvent_53b445d5-0e26-4c01-8892-1a8d622b43c8-0-1' and ProducerIdAndEpoch(producerId=1, epoch=0) attempted to produce with an old epoch
2023-10-13 02:52:48,242 INFO  org.apache.kafka.clients.producer.KafkaProducer              [] - [Producer clientId=producer-domain-nvidia-peoplenet-person-PersonExtractionEvent_53b445d5-0e26-4c01-8892-1a8d622b43c8-0-1, transactionalId=domain-nvidia-peoplenet-person-PersonExtractionEvent_53b445d5-0e26-4c01-8892-1a8d622b43c8-0-1] Aborting incomplete transaction
2023-10-13 02:52:48,243 INFO  org.apache.kafka.clients.producer.internals.TransactionManager [] - [Producer clientId=producer-domain-nvidia-peoplenet-person-PersonExtractionEvent_53b445d5-0e26-4c01-8892-1a8d622b43c8-0-1, transactionalId=domain-nvidia-peoplenet-person-PersonExtractionEvent_53b445d5-0e26-4c01-8892-1a8d622b43c8-0-1] Invoking InitProducerId with current producer ID and epoch ProducerIdAndEpoch(producerId=1, epoch=0) in order to bump the epoch
2023-10-13 02:52:48,247 INFO  org.apache.kafka.clients.producer.internals.TransactionManager [] - [Producer clientId=producer-domain-nvidia-peoplenet-person-PersonExtractionEvent_53b445d5-0e26-4c01-8892-1a8d622b43c8-0-1, transactionalId=domain-nvidia-peoplenet-person-PersonExtractionEvent_53b445d5-0e26-4c01-8892-1a8d622b43c8-0-1] 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-10-13 02:52:48,247 INFO  org.apache.kafka.clients.producer.KafkaProducer              [] - [Producer clientId=producer-domain-nvidia-peoplenet-person-PersonExtractionEvent_53b445d5-0e26-4c01-8892-1a8d622b43c8-0-1, transactionalId=domain-nvidia-peoplenet-person-PersonExtractionEvent_53b445d5-0e26-4c01-8892-1a8d622b43c8-0-1] Closing the Kafka producer with timeoutMillis = 3600000 ms.
2023-10-13 02:52:48,253 INFO  org.apache.kafka.common.metrics.Metrics                      [] - Metrics scheduler closed
2023-10-13 02:52:48,257 INFO  org.apache.kafka.common.metrics.Metrics                      [] - Closing reporter org.apache.kafka.common.metrics.JmxReporter
2023-10-13 02:52:48,257 INFO  org.apache.kafka.common.metrics.Metrics                      [] - Metrics reporters closed
2023-10-13 02:52:48,258 INFO  org.apache.kafka.common.utils.AppInfoParser                  [] - App info kafka.producer for producer-domain-nvidia-peoplenet-person-PersonExtractionEvent_53b445d5-0e26-4c01-8892-1a8d622b43c8-0-1 unregistered
2023-10-13 02:52:48,258 INFO  org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Closing Source Reader.
2023-10-13 02:52:48,259 INFO  org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - Shutting down split fetcher 0
2023-10-13 02:52:48,259 INFO  org.apache.kafka.clients.consumer.internals.ConsumerCoordinator [] - [Consumer clientId=tru_platform_events_0e717a15-901e-4580-9e57-2fda5912973e-0, groupId=tru_platform_events_0e717a15-901e-4580-9e57-2fda5912973e] Resetting generation and member id due to: consumer pro-actively leaving the group
2023-10-13 02:52:48,259 INFO  org.apache.kafka.clients.consumer.internals.ConsumerCoordinator [] - [Consumer clientId=tru_platform_events_0e717a15-901e-4580-9e57-2fda5912973e-0, groupId=tru_platform_events_0e717a15-901e-4580-9e57-2fda5912973e] Request joining group due to: consumer pro-actively leaving the group
2023-10-13 02:52:48,260 INFO  org.apache.kafka.common.metrics.Metrics                      [] - Metrics scheduler closed
2023-10-13 02:52:48,260 INFO  org.apache.kafka.common.metrics.Metrics                      [] - Closing reporter org.apache.kafka.common.metrics.JmxReporter
2023-10-13 02:52:48,260 INFO  org.apache.kafka.common.metrics.Metrics                      [] - Metrics reporters closed
2023-10-13 02:52:48,261 INFO  org.apache.kafka.common.utils.AppInfoParser                  [] - App info kafka.consumer for tru_platform_events_0e717a15-901e-4580-9e57-2fda5912973e-0 unregistered
2023-10-13 02:52:48,261 INFO  org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - Split fetcher 0 exited.
2023-10-13 02:52:48,263 WARN  org.apache.flink.runtime.taskmanager.Task                    [] - Source: Kafka Source -> Filter Null Values -> Transform Source Stream to valid PeopleEvent Data Objects -> Add CloudEvent headers to Data objects -> Map to JSON -> Kafka Sink - domain-nvidia-peoplenet-person-PersonExtractionEvent: Writer -> Kafka Sink - domain-nvidia-peoplenet-person-PersonExtractionEvent: Committer (1/1)#0 (fc61d0f237fc4271cf56abc081b3c76d_cbc357ccb763df2852fee8c4fc7d55f2_0_0) switched from RUNNING to FAILED with failure cause:
java.io.IOException: Failed to deserialize consumer record due to
	at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:56) ~[blob_p-981130016256f8fa8c6154fe4227052d8e46f8de-7557cd0279871880b5e7fa708c94eee2:?]
	at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:33) ~[blob_p-981130016256f8fa8c6154fe4227052d8e46f8de-7557cd0279871880b5e7fa708c94eee2:?]
	at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:144) ~[flink-connector-files-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:417) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:550) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:839) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:788) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:931) [flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745) [flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562) [flink-dist-1.17.1.jar:1.17.1]
	at java.lang.Thread.run(Unknown Source) [?:?]
	Suppressed: java.io.IOException: One or more Kafka Producer send requests have encountered exception
		at org.apache.flink.connector.kafka.sink.KafkaWriter.checkAsyncException(KafkaWriter.java:417) ~[blob_p-981130016256f8fa8c6154fe4227052d8e46f8de-7557cd0279871880b5e7fa708c94eee2:?]
		at org.apache.flink.connector.kafka.sink.KafkaWriter.close(KafkaWriter.java:249) ~[blob_p-981130016256f8fa8c6154fe4227052d8e46f8de-7557cd0279871880b5e7fa708c94eee2:?]
		at org.apache.flink.util.IOUtils.closeAll(IOUtils.java:255) ~[flink-dist-1.17.1.jar:1.17.1]
		at org.apache.flink.util.IOUtils.closeAll(IOUtils.java:234) ~[flink-dist-1.17.1.jar:1.17.1]
		at org.apache.flink.util.IOUtils.closeAll(IOUtils.java:223) ~[flink-dist-1.17.1.jar:1.17.1]
		at org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.close(SinkWriterOperator.java:216) ~[flink-dist-1.17.1.jar:1.17.1]
		at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:163) ~[flink-dist-1.17.1.jar:1.17.1]
		at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.closeAllOperators(RegularOperatorChain.java:125) ~[flink-dist-1.17.1.jar:1.17.1]
		at org.apache.flink.streaming.runtime.tasks.StreamTask.closeAllOperators(StreamTask.java:1043) ~[flink-dist-1.17.1.jar:1.17.1]
		at org.apache.flink.util.IOUtils.closeAll(IOUtils.java:255) ~[flink-dist-1.17.1.jar:1.17.1]
		at org.apache.flink.core.fs.AutoCloseableRegistry.doClose(AutoCloseableRegistry.java:72) ~[flink-dist-1.17.1.jar:1.17.1]
		at org.apache.flink.util.AbstractAutoCloseableRegistry.close(AbstractAutoCloseableRegistry.java:127) ~[flink-dist-1.17.1.jar:1.17.1]
		at org.apache.flink.streaming.runtime.tasks.StreamTask.cleanUp(StreamTask.java:951) ~[flink-dist-1.17.1.jar:1.17.1]
		at org.apache.flink.runtime.taskmanager.Task.lambda$restoreAndInvoke$0(Task.java:934) ~[flink-dist-1.17.1.jar:1.17.1]
		at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952) ~[flink-dist-1.17.1.jar:1.17.1]
		at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:934) [flink-dist-1.17.1.jar:1.17.1]
		at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745) [flink-dist-1.17.1.jar:1.17.1]
		at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562) [flink-dist-1.17.1.jar:1.17.1]
		at java.lang.Thread.run(Unknown Source) [?:?]
	Caused by: org.apache.flink.util.FlinkRuntimeException: Failed to send data to Kafka domain-nvidia-peoplenet-person-PersonExtractionEvent-0@-1 with FlinkKafkaInternalProducer{transactionalId='domain-nvidia-peoplenet-person-PersonExtractionEvent_53b445d5-0e26-4c01-8892-1a8d622b43c8-0-1', inTransaction=false, closed=false} 
		at org.apache.flink.connector.kafka.sink.KafkaWriter$WriterCallback.decorateException(KafkaWriter.java:470) ~[blob_p-981130016256f8fa8c6154fe4227052d8e46f8de-
a
Where is the 
InvalidProducerEpochException
?
a
@Alexander Fedulov I must have copied the part underneath. Here it is, just after the part where the healthily looking logs end:
Copy code
2023-10-13 05:30:10,016 INFO  io.trurecognition.platform.jobs.derived.nvidia.peoplenet.criminalgang.PeopleCountEventMapper [] - Processed people count: 20
2023-10-13 05:30:10,017 INFO  io.trurecognition.platform.jobs.derived.nvidia.peoplenet.criminalgang.PeopleCountEventMapper [] - Processed people average age: 31.0
2023-10-13 05:30:10,017 INFO  io.trurecognition.platform.jobs.derived.nvidia.peoplenet.criminalgang.PeopleCountEventMapper [] - Processed people male distribution: 0.6
2023-10-13 05:30:10,092 INFO  io.trurecognition.platform.jobs.domain.nvidia.peoplenet.people.PeopleCountEventAggregator [] - Creating new io.trurecognition.platform.model.json.domain.nvidia.peoplenet.people.Data accumulator
2023-10-13 05:30:20,017 INFO  org.apache.kafka.clients.producer.internals.TransactionManager [] - [Producer clientId=producer-domain-nvidia-peoplenet-people-PeopleCountEvent_a3b82c61-edfe-46cc-9896-32dbf605969f_138-0-1, transactionalId=domain-nvidia-peoplenet-people-PeopleCountEvent_a3b82c61-edfe-46cc-9896-32dbf605969f_138-0-1] Transiting to abortable error state due to org.apache.kafka.common.errors.InvalidProducerEpochException: Producer attempted to produce with an old epoch.
2023-10-13 05:30:20,157 INFO  io.trurecognition.platform.jobs.domain.nvidia.peoplenet.people.PeopleCountEventAggregator [] - Creating new io.trurecognition.platform.model.json.domain.nvidia.peoplenet.people.Data accumulator
2023-10-13 05:30:30,001 INFO  org.apache.kafka.clients.producer.internals.TransactionManager [] - [Producer clientId=producer-domain-nvidia-peoplenet-people-PeopleCountEvent_a3b82c61-edfe-46cc-9896-32dbf605969f_138-0-1, transactionalId=domain-nvidia-peoplenet-people-PeopleCountEvent_a3b82c61-edfe-46cc-9896-32dbf605969f_138-0-1] Transiting to abortable error state due to org.apache.kafka.common.errors.InvalidProducerEpochException: Producer with transactionalId 'domain-nvidia-peoplenet-people-PeopleCountEvent_a3b82c61-edfe-46cc-9896-32dbf605969f_138-0-1' and ProducerIdAndEpoch(producerId=2, epoch=0) attempted to produce with an old epoch
2023-10-13 05:30:30,278 INFO  io.trurecognition.platform.jobs.domain.nvidia.peoplenet.people.PeopleCountEventAggregator [] - Creating new io.trurecognition.platform.model.json.domain.nvidia.peoplenet.people.Data accumulator
2023-10-13 05:30:40,001 INFO  org.apache.kafka.clients.producer.internals.TransactionManager [] - [Producer clientId=producer-domain-nvidia-peoplenet-people-PeopleCountEvent_a3b82c61-edfe-46cc-9896-32dbf605969f_138-0-1, transactionalId=domain-nvidia-peoplenet-people-PeopleCountEvent_a3b82c61-edfe-46cc-9896-32dbf605969f_138-0-1] Transiting to abortable error state due to org.apache.kafka.common.errors.InvalidProducerEpochException: Producer with transactionalId 'domain-nvidia-peoplenet-people-PeopleCountEvent_a3b82c61-edfe-46cc-9896-32dbf605969f_138-0-1' and ProducerIdAndEpoch(producerId=2, epoch=0) attempted to produce with an old epoch
2023-10-13 05:30:40,429 INFO  io.trurecognition.platform.jobs.domain.nvidia.peoplenet.people.PeopleCountEventAggregator [] - Creating new io.trurecognition.platform.model.json.domain.nvidia.peoplenet.people.Data accumulator
a
Flink’s KafkaSource uses Kafka’s java library under the hood and from its perspective it is not even a warning message (INFO level). It does not seem to me the root cause of the exception. Your original stack trace is cut, so it is not clear what actually is.
a
@Alexander Fedulov Yes upon inspection of the logs shared so far, you are correct. Here is an example of it occurring in the 
Exceptions
logs.
Copy code
What might be causing these issues?

Caused by: org.apache.flink.util.FlinkRuntimeException: Failed to send data to Kafka domain-nvidia-peoplenet-person-PersonExtractionEvent-0@-1 with FlinkKafkaInternalProducer{transactionalId='domain-nvidia-peoplenet-person-PersonExtractionEvent_df622547-c5aa-4bfb-845e-73e6e945e66b-0-1', inTransaction=true, closed=false} 
	at org.apache.flink.connector.kafka.sink.KafkaWriter$WriterCallback.decorateException(KafkaWriter.java:470) ~[blob_p-b30d905e65927da6bac82c32d8510cdeda4982b9-f7a82451d6c63b6b6f71809d3f91170a:?]
	at org.apache.flink.connector.kafka.sink.KafkaWriter$WriterCallback.onCompletion(KafkaWriter.java:445) ~[blob_p-b30d905e65927da6bac82c32d8510cdeda4982b9-f7a82451d6c63b6b6f71809d3f91170a:?]
	at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:1019) ~[kafka-clients-3.2.3.jar:?]
	at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:912) ~[kafka-clients-3.2.3.jar:?]
	at org.apache.flink.connector.kafka.sink.KafkaWriter.write(KafkaWriter.java:199) ~[blob_p-b30d905e65927da6bac82c32d8510cdeda4982b9-f7a82451d6c63b6b6f71809d3f91170a:?]
	at org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.processElement(SinkWriterOperator.java:158) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:38) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:38) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:51) ~[flink-dist-1.17.1.jar:1.17.1]
	at io.trurecognition.platform.jobs.domain.nvidia.peoplenet.person.StreamFactory$DataExtractionFunction.flatMap(StreamFactory.java:118) ~[blob_p-b30d905e65927da6bac82c32d8510cdeda4982b9-f7a82451d6c63b6b6f71809d3f91170a:?]
	at io.trurecognition.platform.jobs.domain.nvidia.peoplenet.person.StreamFactory$DataExtractionFunction.flatMap(StreamFactory.java:55) ~[blob_p-b30d905e65927da6bac82c32d8510cdeda4982b9-f7a82451d6c63b6b6f71809d3f91170a:?]
	at org.apache.flink.streaming.api.operators.StreamFlatMap.processElement(StreamFlatMap.java:47) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.api.operators.StreamFilter.processElement(StreamFilter.java:39) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask$AsyncDataOutputToOutput.emitRecord(SourceOperatorStreamTask.java:309) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.streaming.api.operators.source.SourceOutputWithWatermarks.collect(SourceOutputWithWatermarks.java:110) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter$SourceOutputWrapper.collect(KafkaRecordEmitter.java:67) ~[blob_p-b30d905e65927da6bac82c32d8510cdeda4982b9-f7a82451d6c63b6b6f71809d3f91170a:?]
	at org.apache.flink.api.common.serialization.DeserializationSchema.deserialize(DeserializationSchema.java:84) ~[flink-dist-1.17.1.jar:1.17.1]
	at org.apache.flink.connector.kafka.source.reader.deserializer.KafkaValueOnlyDeserializationSchemaWrapper.deserialize(KafkaValueOnlyDeserializationSchemaWrapper.java:51) ~[blob_p-b30d905e65927da6bac82c32d8510cdeda4982b9-f7a82451d6c63b6b6f71809d3f91170a:?]
	at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:53) ~[blob_p-b30d905e65927da6bac82c32d8510cdeda4982b9-f7a82451d6c63b6b6f71809d3f91170a:?]
	... 14 more
Caused by: org.apache.kafka.common.errors.InvalidProducerEpochException: Producer with transactionalId 'domain-nvidia-peoplenet-person-PersonExtractionEvent_df622547-c5aa-4bfb-845e-73e6e945e66b-0-1' and ProducerIdAndEpoch(producerId=10, epoch=0) attempted to produce with an old epoch
2023-10-12 04:38:36,585 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - Freeing task resources for Source: Kafka Source -> Filter Null Values -> Transform Source Stream to valid PeopleEvent Data Objects -> Add CloudEvent headers to Data objects -> Map to JSON -> Kafka Sink - domain-nvidia-peoplenet-person-PersonExtractionEvent: Writer -> Kafka Sink - domain-nvidia-peoplenet-person-PersonExtractionEvent: Committer (1/1)#0 (eed711a703447fe5216fd97d8777c7da_cbc357ccb763df2852fee8c4fc7d55f2_0_0).
2023-10-12 04:38:36,597 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor           [] - Un-registering task and sending final execution state FAILED to JobManager for task Source: Kafka Source -> Filter Null Values -> Transform Source Stream to valid PeopleEvent Data Objects -> Add CloudEvent headers to Data objects -> Map to JSON -> Kafka Sink - domain-nvidia-peoplenet-person-PersonExtractionEvent: Writer -> Kafka Sink - domain-nvidia-peoplenet-person-PersonExtractionEvent: Committer (1/1)#0 eed711a703447fe5216fd97d8777c7da_cbc357ccb763df2852fee8c4fc7d55f2_0_0.
The behavior we are seeing is that some of the Flink jobs fail to produce to their configured Kafka topics after this happens.
a
Did you try bumping up 
<http://transaction.timeout.ms|transaction.timeout.ms>
? https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/connectors/datastream/kafka/#fault-tolerance
👍 1
a
Managed to get it fixed. Thanks @Alexander Fedulov
a
Did the timeout setting help?
a
@Alexander Fedulov. Yes I believe it has helped with the stability issues seen before. The jobs are not crashing now. In kafka-ui I can see that the dashboard view shows double the messages that are shown in the Flink logs and the list view. This could be a bug in the kafka-ui, however I wonder if this could be that I need to enable log compaction.
Copy code
// Build the KafkaRecordSerializationSchema
    KafkaRecordSerializationSchema<String> recordSerializer =
        KafkaRecordSerializationSchema.builder()
            .setTopic(topic)
            .setValueSerializationSchema(new SimpleStringSchema())
            .build();
Perhaps this setup is not ensuring that Kafka retains at least the last known value for each message key within the log of data for a single topic partition. What are your thoughts?
a
If the UI does not have a 
read_committed
setting this can be expected accidentally. There is a two-phase-commit protocol under the hood, so the messages are being written out into the sinks, just become committed when the overall transaction completes upon a checkpoint. This bit might help to understand the process

https://youtu.be/9pRsewtSPkQ?si=Pk3GzQdAuPVCh-Jn&amp;t=351▾

And this

https://youtu.be/Wo9jlaz8h0k?si=jemgAjLPjSCEdXoZ&t=364▾

a
Thanks for sharing. Very useful information. I've made the updates but still seeing the issue reported before. I'm unsure if this is related to Flink or Kafka settings, but I am seeing 1 "Null key", which I suspect might be the root of the issue where the overall count of messages is double what is shown in the list of messages.
5 Views
Open in Slack
PreviousNext
Powered by