"org.apache.kafka.common.errors.UnknownProducerIdException" occurs in KafkaStreams and the KafkaStreams application shutdown

Issue

• "org.apache.kafka.common.errors.UnknownProducerIdException" occurs in KafkaStreams and the KafkaStreams application shutdown.

2020-01-20 17:53:17.908 WARN  [31636][kafka-producer-network-thread | example.example01-0b40803f-2869-4ddc-8f36-f63758c4c14b-StreamThread-1-0_0-producer][RecordCollectorImpl] task [0_0] Error sending records topic=[ObeRequest] and partition=[null]; The exception handler chose to CONTINUE processing in spite of this error. Enable TRACE logging to view failed messages key and value.
org.apache.kafka.common.errors.UnknownProducerIdException: This exception is raised by the broker if it could not locate the producer metadata associated with the producerId in question. This could happen if, for instance, the producer's records were deleted because their retention time had elapsed. Once the last records of the producerId are removed, the producer's metadata is removed from the broker, and future appends by the producer will return this exception.

2020-01-20 17:53:17.918 ERROR [31636][example.example01-0b40803f-2869-4ddc-8f36-f63758c4c14b-StreamThread-1][AssignedStreamsTasks] stream-thread [example.example01-0b40803f-2869-4ddc-8f36-f63758c4c14b-StreamThread-1] Failed to commit stream task 0_0 due to the following error:
org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state
        at org.apache.kafka.clients.producer.internals.TransactionManager.maybeFailWithError(TransactionManager.java:924) ~[kafka-clients-2.3.1.jar:?]
        at org.apache.kafka.clients.producer.internals.TransactionManager.sendOffsetsToTransaction(TransactionManager.java:327) ~[kafka-clients-2.3.1.jar:?]
        at org.apache.kafka.clients.producer.KafkaProducer.sendOffsetsToTransaction(KafkaProducer.java:673) ~[kafka-clients-2.3.1.jar:?]
        at org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:486) ~[kafka-streams-2.3.1.jar:?]
        at org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:458) ~[kafka-streams-2.3.1.jar:?]
        at org.apache.kafka.streams.processor.internals.AssignedTasks.commit(AssignedTasks.java:286) [kafka-streams-2.3.1.jar:?]
        at org.apache.kafka.streams.processor.internals.TaskManager.commitAll(TaskManager.java:417) [kafka-streams-2.3.1.jar:?]
        at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1079) [kafka-streams-2.3.1.jar:?]
        at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:933) [kafka-streams-2.3.1.jar:?]
        at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:819) [kafka-streams-2.3.1.jar:?]
        at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:788) [kafka-streams-2.3.1.jar:?]
Caused by: org.apache.kafka.common.errors.UnknownProducerIdException: This exception is raised by the broker if it could not locate the producer metadata associated with the producerId in question. This could happen if, for instance, the producer's records were deleted because their retention time had elapsed. Once the last records of the producerId are removed, the producer's metadata is removed from the broker, and future appends by the producer will return this exception.