# InvalidProducerEpochException causing duplicate despite of EOS

**URL:** <https://forum.confluent.io/t/invalidproducerepochexception-causing-duplicate-despite-of-eos/10896>\
**Category:** Kafka Streams\
**Created:** [14 June 2024 08:38 UTC](https://forum.confluent.io/t/invalidproducerepochexception-causing-duplicate-despite-of-eos/10896 "2024-06-14T08:38:38Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![aleksandrakoziel](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/aleksandrakoziel/32/3826_2.png) [@aleksandrakoziel](https://forum.confluent.io/u/aleksandrakoziel)\
**Post date:** [14 June 2024 08:38 UTC](https://forum.confluent.io/t/invalidproducerepochexception-causing-duplicate-despite-of-eos/10896/1 "2024-06-14T08:38:38Z")

</div>

Please, help me understand what is wrong with the code or the config as the behaviour seems to be pretty bizarre for me and I am not sure if I understand it correctly.

We are having Kafka Streams application. The main idea is to forward original data if the given message reaches the end of the flow with success. That’s why we are waiting for the message to be present on both topics, the first one and the last one of the flow. Solution is based on Processor API and State Stores.  
And the problem is, we are observing `InvalidProducerEpochException`. It is a rare case, e.g. during performance testing 50 messages per second, 50k in total. One in five runs this error occurs. But the implication is disturbing. After the stream restart, one message (MESSAGE\_1) is getting duplicated, the rest is sent successful after the stream restart. Logs:

> 2024-06-10 11:53:49,661 INFO \*.processor.MatchingProcessor [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Message with key MESSAGE\_1 was sent to output.  
> 2024-06-10 11:53:49,661 INFO \*.processor.InputSentProcessor [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Message MESSAGE\_2: Adding sent data  
> 2024-06-10 11:54:18,574 INFO \*.services.id$Companion [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Message MESSAGE\_2: Consumed id package from topic \*  
> 2024-06-10 11:54:18,594 INFO \*.services.id$Companion [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Send package consumed info to topic \*  
> 2024-06-10 11:54:18,594 INFO \*.services.EventFactory [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Message MESSAGE\_2: Getting data for message  
> 2024-06-10 11:54:18,594 INFO \*.services.EventFactory [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Message MESSAGE\_2: Corresponding data: \*  
> 2024-06-10 11:54:18,595 INFO \*.processor.MatchingProcessor [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Message with key MESSAGE\_2 was sent to output.  
> 2024-06-10 11:54:18,595 INFO \*.processor.InputSentProcessor [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Message MESSAGE\_3: Adding sent data  
> 2024-06-10 11:54:18,599 INFO \*.services.id$Companion [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Message MESSAGE\_3: Consumed id package from topic \*  
> 2024-06-10 11:54:18,630 INFO \*.services.id$Companion [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Send package consumed info to topic \*  
> 2024-06-10 11:54:18,630 INFO \*.services.EventFactory [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Message MESSAGE\_3: Getting data for message  
> 2024-06-10 11:54:18,630 INFO \*.services.EventFactory [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Message MESSAGE\_3: Corresponding data: \*  
> 2024-06-10 11:54:18,630 INFO \*.processor.MatchingProcessor [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Message with key MESSAGE\_3 was sent to output.  
> 2024-06-10 11:54:18,630 INFO \*.processor.InputSentProcessor [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Message MESSAGE\_4: Adding sent data  
> 2024-06-10 11:54:18,634 INFO \*.services.id$Companion [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Message MESSAGE\_4: Consumed id package from topic \*  
> 2024-06-10 11:54:18,642 INFO \*.services.id$Companion [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Send package consumed info to topic \*  
> 2024-06-10 11:54:18,643 INFO \*.services.EventFactory [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Message MESSAGE\_4: Getting data for message  
> 2024-06-10 11:54:18,643 INFO \*.services.EventFactory [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Message MESSAGE\_4: Corresponding data: \*  
> 2024-06-10 11:54:18,643 INFO \*.processor.MatchingProcessor [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Message with key MESSAGE\_4 was sent to output.  
> 2024-06-10 11:54:18,653 ERROR org.apache.kafka.streams.processor.internals.RecordCollectorImpl [kafka-producer-network-thread | service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1-producer] stream-thread [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] stream-task [0\_2] Error encountered sending record to topic service-application-SENT-changelog for task 0\_2 due to:  
> org.apache.kafka.common.errors.InvalidProducerEpochException: Producer attempted to produce with an old epoch.  
> Written offsets would not be recorded and no more records would be sent since the producer is fenced, indicating the task may be migrated out  
> org.apache.kafka.common.errors.InvalidProducerEpochException: Producer attempted to produce with an old epoch.  
> 2024-06-10 11:54:18,656 INFO org.apache.kafka.clients.producer.internals.TransactionManager [kafka-producer-network-thread | service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1-producer] [Producer clientId=service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1-producer, transactionalId=service-application-d29124b7-ee33-456f-82fc-05d58665120e-1] Transiting to abortable error state due to org.apache.kafka.common.errors.InvalidProducerEpochException: Producer attempted to produce with an old epoch.  
> 2024-06-10 11:54:18,659 INFO org.apache.kafka.clients.producer.internals.TransactionManager [kafka-producer-network-thread | service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1-producer] [Producer clientId=service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1-producer, transactionalId=service-application-d29124b7-ee33-456f-82fc-05d58665120e-1] Transiting to abortable error state due to org.apache.kafka.common.errors.InvalidProducerEpochException: Producer attempted to produce with an old epoch.  
> 2024-06-10 11:54:18,659 INFO org.apache.kafka.clients.producer.internals.TransactionManager [kafka-producer-network-thread | service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1-producer] [Producer clientId=service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1-producer, transactionalId=service-application-d29124b7-ee33-456f-82fc-05d58665120e-1] Transiting to abortable error state due to org.apache.kafka.common.errors.InvalidProducerEpochException: Producer attempted to produce with an old epoch.  
> 2024-06-10 11:54:18,659 WARN org.apache.kafka.streams.processor.internals.StreamThread [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] stream-thread [service-application-d29124b7-ee33-456f-82fc-05d58665120e-StreamThread-1] Detected that the thread is being fenced. This implies that this thread missed a rebalance and dropped out of the consumer group. Will close out all assigned tasks and rejoin the consumer group.  
> org.apache.kafka.streams.errors.TaskMigratedException: Error encountered sending record to topic service-application-SENT-changelog for task 0\_2 due to:  
> org.apache.kafka.common.errors.InvalidProducerEpochException: Producer attempted to produce with an old epoch.  
> Written offsets would not be recorded and no more records would be sent since the producer is fenced, indicating the task may be migrated out; it means all tasks belonging to this thread should be migrated.  
> at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.recordSendError(RecordCollectorImpl.java:303)  
> at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.lambda$send$1(RecordCollectorImpl.java:284)  
> at io.opentelemetry.javaagent.instrumentation.kafkaclients.ProducerCallback.onCompletion(ProducerCallback.java:36)  
> at org.apache.kafka.clients.producer.KafkaProducer$AppendCallbacks.onCompletion(KafkaProducer.java:1538)  
> at org.apache.kafka.clients.producer.internals.ProducerBatch.completeFutureAndFireCallbacks(ProducerBatch.java:312)  
> at org.apache.kafka.clients.producer.internals.ProducerBatch.done(ProducerBatch.java:273)  
> at org.apache.kafka.clients.producer.internals.ProducerBatch.completeExceptionally(ProducerBatch.java:237)  
> at org.apache.kafka.clients.producer.internals.Sender.failBatch(Sender.java:830)  
> at org.apache.kafka.clients.producer.internals.Sender.failBatch(Sender.java:819)  
> at org.apache.kafka.clients.producer.internals.Sender.failBatch(Sender.java:771)  
> at org.apache.kafka.clients.producer.internals.Sender.completeBatch(Sender.java:702)  
> at org.apache.kafka.clients.producer.internals.Sender.lambda$null$1(Sender.java:627)  
> at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)  
> at org.apache.kafka.clients.producer.internals.Sender.lambda$handleProduceResponse$2(Sender.java:612)  
> at java.base/java.lang.Iterable.forEach(Iterable.java:75)  
> at org.apache.kafka.clients.producer.internals.Sender.handleProduceResponse(Sender.java:612)  
> at org.apache.kafka.clients.producer.internals.Sender.lambda$sendProduceRequest$8(Sender.java:917)  
> at org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse.java:154)  
> at org.apache.kafka.clients.NetworkClient.completeResponses(NetworkClient.java:608)  
> at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:600)  
> at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:349)  
> at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:252)  
> at java.base/java.lang.Thread.run(Thread.java:840)  
> Caused by: org.apache.kafka.common.errors.InvalidProducerEpochException: Producer attempted to produce with an old epoch.

This changelog topic is connected to the store, so this issue is similar to [InvalidProducerEpochException when running EOS and utillizing state stores](https://forum.confluent.io/t/invalidproducerepochexception-when-running-eos-and-utillizing-state-stores/9169), but for us is a pretty rare case and is causing this duplicate despite EOS. Also, we are not performing any significant operations on these stores, just get, put and delete. Here is the code:

Config:

```auto
Properties().apply {
            put(StreamsConfig.APPLICATION_ID_CONFIG, kafkaProperties.applicationId)
            put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.bootstrapServers)
            put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().javaClass.name)
            put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde::class.java.name)
            put(
                KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG,
                kafkaProperties.specificAvroReader.toBoolean()
            )
            put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, kafkaProperties.schemaRegistry)
            put(
                StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
                LogAndContinueExceptionHandler()::class.java
            )
            put(
                StreamsConfig.STATE_DIR_CONFIG,
                kafkaProperties.stateDirectory
            )
            put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2)
        }

```

Saving to the state store:

```auto
open class InputSentProcessor(
    private val kafkaSentDlqStringProducer: KafkaProducer<String, String>
) : ProcessorSupplier<String, String, String, Unit> {

    override fun get(): Processor<String, String, String, Unit> {

        return object : Processor<String, String, String, Unit> {

            lateinit var context: ProcessorContext<String, Unit>
            lateinit var sentStore: KeyValueStore<String, String>

            override fun init(context: ProcessorContext<String, Unit>) {
                this.context = context
                this.sentStore = context.getStateStore(StoreTypes.SENT.name)
            }

            override fun process(record: Record<String, String>) {
                try {
                    processSentRecord(record)
                } catch (e: Exception) {
                    logger.error("Message ${record.key()}: Exception while processing the message: ${e.message}")
                    sendMessageSentToDLQ(record)
                }
            }

            private fun sendMessageSentToDLQ(record: Record<String, String>) {
                kafkaSentDlqStringProducer.sentToDLQ(
                    record.value(),
                    record.key(),
                    "topicSentDLQ",
                    logger
                )
            }

            private fun processSentRecord(record: Record<String, String>) {
                sentStore.put(record.key(), record.value())
                logger.info("Message ${record.key()}: Adding sent data")
                context.forward(record.withValue(null))
            }
        }
    }

    companion object {
        private val logger = LoggerFactory.getLogger(InputSentProcessor::class.java)
    }
}

```

Matching messages and cleaning the stores:

```auto
open class MatchingProcessor(private val dlqStringProducer: KafkaProducer<String, String>) : ProcessorSupplier<String, Any, String, Event> {

    override fun get(): Processor<String, Any, String, Event> {
        return object : Processor<String, Any, String, Event> {

            lateinit var context: ProcessorContext<String, Event>
            lateinit var idStore: KeyValueStore<String, Event>
            lateinit var sentStore: KeyValueStore<String, String>
            lateinit var failedStore: KeyValueStore<String, String>

            override fun init(context: ProcessorContext<String, Event>) {
                this.context = context
                this.idStore = context.getStateStore(StoreTypes.ID.name)
                this.sentStore = context.getStateStore(StoreTypes.SENT.name)
                this.failedStore = context.getStateStore(StoreTypes.FAILED.name)
            }

            override fun process(record: Record<String, Any>) {
                try {
                    record.key().let { key ->
                        when {
                            idStore[key] != null && sentStore[key] != null -> {
                                // composeEvent method is containing separate consumer (outside of the stream scope) to get the new, unique id from the dedicated topic
                                context.forward(record.withValue(composeEvent(key, sentStore[key], idStore[key])))

                                idStore.delete(key)
                                sentStore.delete(key)

                                logger.info("Message with key $key was sent to output")
                            }

                            idStore[key] != null && failedStore[key] != null -> {
                                idStore.delete(key)
                                failedStore.delete(key)
                                logger.info("Message with key $key was failed to send")
                            }

                            else -> logger.info("Message with key $key - waiting for matching records")
                        }
                    }
                } catch (exception: Exception) {
                    sentStore[record.key()]?.let { message ->
                        dlqStringProducer.sentToDLQ(message, record.key(), "topicSentDLQ", logger)
                    }

                    failedStore[record.key()]?.let { message ->
                        dlqStringProducer.sentToDLQ(message, record.key(), "topicFailedDLQ", logger)
                    }
                }
            }
        }
    }

    companion object {
        private val logger = LoggerFactory.getLogger(MatchingProcessor::class.java)
    }
}

```

So, I think the delete the one which is not getting send to the changelog topic, thus the record is still there. But I am not sure why it was not deleted from the second store and why only the first message is getting duplicated (maybe the reason is this is the one, which struggle to delete? Forwarding to the sink is before the cleaning process, so it was sent but it failed to delete). Also, could the delete operation be this long? Or maybe, the fault is inside this additional consumer under the hood (I know that this kind of approach does not guarantee exactly once, [source](https://www.confluent.io/blog/exactly-once-semantics-are-possible-heres-how-apache-kafka-does-it/)).

I will be grateful for some hints to fully understand the issue. We are planning to simply switch the order or operations to move deleting before sending, but probably our long-term goal will be to switch to Stream and getting rid of these state stores as our business requirements demand no duplicates.

Thank you in advance for all answers.

---

<div class="post-metadata">

**Author:** ![mjsax](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mjsax/32/3113_2.png) [@mjsax](https://forum.confluent.io/u/mjsax)\
**Post date:** [14 June 2024 16:51 UTC](https://forum.confluent.io/t/invalidproducerepochexception-causing-duplicate-despite-of-eos/10896/2 "2024-06-14T16:51:41Z")

</div>

Not 100% sure… Maybe you are hitting: [[KAFKA-16017] Checkpointed offset is incorrect when task is revived and restoring - ASF JIRA](https://issues.apache.org/jira/browse/KAFKA-16017) ?