You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "NEERAJ VAIDYA (Jira)" <ji...@apache.org> on 2021/09/14 05:50:00 UTC
[jira] [Comment Edited] (KAFKA-13292) InvalidPidMappingException:
The producer attempted to use a producer id which is not currently assigned
to its transactional id
[ https://issues.apache.org/jira/browse/KAFKA-13292?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17414732#comment-17414732 ]
NEERAJ VAIDYA edited comment on KAFKA-13292 at 9/14/21, 5:49 AM:
-----------------------------------------------------------------
Thanks [~mjsax] for the reference to the KIP. It might not be possible for my application to be upgraded soon.
However, I still feel this is a defect because I believed that stream processing should continue even when transactional.id.expiration.ms has elapsed and should not bring down the entire application.
I noticed from the logs that the application does try to get a new epoch and producer Id, but then still suspends and stops all threads. As you will observe below towards the end of the log messages (at timestamp : 2021-09-10T12:21:59.766) , for the task[ 0_2], the call to "*Invoking InitProducerId*" is made, but still the flow ends up in the UnCaughtExceptionHandler.
I was of the understanding that once the producerId epoch is bumped, stream processing should resume.
However, it starts shutting down the entire application by an uncaught exception.
*FYI - I am using exactly_once processing and enable.idempotence=true*
{code:java}
2021-09-10T12:21:59.636 [kafka-producer-network-thread | mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-0_2-producer] INFO o.a.k.c.p.i.TransactionManager - MSG=[Producer clientId=mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-Stre
amThread-1-0_2-producer, transactionalId=mtx-caf-0_2] 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 tra
nsactional id.
2021-09-10T12:21:59.642 [kafka-producer-network-thread | mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-0_2-producer] ERROR o.a.k.s.p.i.RecordCollectorImpl - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamT
hread-1] task [0_2] Error encountered sending record to topic mtx-caf-DuplicateCheckStore-changelog for task 0_2 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.
Exception handler choose to FAIL the processing, no more records would be sent.
2021-09-10T12:21:59.740 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] ERROR o.a.k.s.p.internals.StreamThread - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] Encountered the following exception
during processing and the thread is going to shut down:
org.apache.kafka.streams.errors.StreamsException: Error encountered sending record to topic mtx-caf-DuplicateCheckStore-changelog for task 0_2 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.
Exception handler choose to FAIL the processing, no more records would be sent.
at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.recordSendError(RecordCollectorImpl.java:214)
at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.lambda$send$0(RecordCollectorImpl.java:186)
at org.apache.kafka.clients.producer.KafkaProducer$InterceptorCallback.onCompletion(KafkaProducer.java:1363)
at org.apache.kafka.clients.producer.internals.ProducerBatch.completeFutureAndFireCallbacks(ProducerBatch.java:231)
at org.apache.kafka.clients.producer.internals.ProducerBatch.abort(ProducerBatch.java:159)
at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortUndrainedBatches(RecordAccumulator.java:781)
at org.apache.kafka.clients.producer.internals.Sender.maybeSendAndPollTransactionalRequest(Sender.java:425)
at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:313)
at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:240)
at java.base/java.lang.Thread.run(Thread.java:829)
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.
2021-09-10T12:21:59.740 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.s.p.internals.StreamThread - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] State transition from RUNNING to PE
NDING_SHUTDOWN
2021-09-10T12:21:59.741 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.s.p.internals.StreamThread - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] Shutting down
2021-09-10T12:21:59.743 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO c.m.a.k.t.SessionBasedDataUsageAccumulator - MSG=Shutting Down
2021-09-10T12:21:59.744 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.s.p.internals.StreamTask - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] task [0_8] Suspended running
2021-09-10T12:21:59.747 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.clients.consumer.KafkaConsumer - MSG=[Consumer clientId=mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-restore-consumer, groupId=nu
ll] Subscribed to partition(s): mtx-caf-DuplicateCheckStore-changelog-10, mtx-caf-DuplicateCheckStore-changelog-6, mtx-caf-DuplicateCheckStore-changelog-4, mtx-caf-DuplicateCheckStore-changelog-2, mtx-caf-AggregatedRecordStore-changelog-10, m
tx-caf-DuplicateCheckStore-changelog-0, mtx-caf-AggregatedRecordStore-changelog-4, mtx-caf-AggregatedRecordStore-changelog-6, mtx-caf-AggregatedRecordStore-changelog-0, mtx-caf-AggregatedRecordStore-changelog-2, mtx-caf-DuplicateCheckStore-ch
angelog-11, mtx-caf-DuplicateCheckStore-changelog-9, mtx-caf-DuplicateCheckStore-changelog-7, mtx-caf-AggregatedRecordStore-changelog-11, mtx-caf-DuplicateCheckStore-changelog-5, mtx-caf-DuplicateCheckStore-changelog-3, mtx-caf-AggregatedReco
rdStore-changelog-7, mtx-caf-DuplicateCheckStore-changelog-1, mtx-caf-AggregatedRecordStore-changelog-9, mtx-caf-AggregatedRecordStore-changelog-3, mtx-caf-AggregatedRecordStore-changelog-5, mtx-caf-AggregatedRecordStore-changelog-1
2021-09-10T12:21:59.753 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.s.p.i.RecordCollectorImpl - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] task [0_8] Closing record collector
dirty
...
...
...
...
2021-09-10T12:21:59.763 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.s.p.internals.StreamTask - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] task [0_2] Suspended running
2021-09-10T12:21:59.764 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.clients.consumer.KafkaConsumer - MSG=[Consumer clientId=mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-restore-consumer, groupId=null] Subscribed to partition(s): mtx-caf-DuplicateCheckStore-changelog-11, mtx-caf-DuplicateCheckStore-changelog-10, mtx-caf-DuplicateCheckStore-changelog-9, mtx-caf-DuplicateCheckStore-changelog-7, mtx-caf-DuplicateCheckStore-changelog-6, mtx-caf-AggregatedRecordStore-changelog-11, mtx-caf-DuplicateCheckStore-changelog-4, mtx-caf-AggregatedRecordStore-changelog-7, mtx-caf-AggregatedRecordStore-changelog-10, mtx-caf-DuplicateCheckStore-changelog-0, mtx-caf-AggregatedRecordStore-changelog-9, mtx-caf-AggregatedRecordStore-changelog-4, mtx-caf-AggregatedRecordStore-changelog-6, mtx-caf-AggregatedRecordStore-changelog-0
2021-09-10T12:21:59.765 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.s.p.i.RecordCollectorImpl - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] task [0_2] Closing record collector dirty
2021-09-10T12:21:59.765 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.clients.producer.KafkaProducer - MSG=[Producer clientId=mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-0_2-producer, transactionalId=mtx-caf-0_2] Aborting incomplete transaction
2021-09-10T12:21:59.766 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.c.p.i.TransactionManager - MSG=[Producer clientId=mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-0_2-producer, transactionalId=mtx-caf-0_2] Invoking InitProducerId with current producer ID and epoch (producerId=1005, epoch=16) in order to bump the epoch
2021-09-10T12:21:59.769 [kafka-producer-network-thread | mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-0_2-producer] INFO o.a.k.c.p.i.TransactionManager - MSG=[Producer clientId=mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-0_2-producer, transactionalId=mtx-caf-0_2] ProducerId set to 1006 with epoch 0
...
...
...
2021-09-10T12:21:59.798 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.clients.producer.KafkaProducer - MSG=[Producer clientId=mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-0_2-producer, transactionalId=mtx-caf-0_2] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms.
{code}
was (Author: neeraj.vaidya):
Thanks [~mjsax] for the reference to the KIP. It might not be possible for my application to be upgraded soon.
However, I still feel this is a defect because I believed that stream processing should continue even when transactional.id.expiration.ms has elapsed and should not bring down the entire application.
I noticed from the logs that the application does try to get a new epoch and producer Id, but then still suspends and stops all threads. As you will observe below towards the end of the log messages, for the task[ 0_2], the call to "*Invoking InitProducerId*" is made, but still the flow ends up in the UnCaughtExceptionHandler.
I was of the understanding that once the producerId epoch is bumped, stream processing should resume.
However, it starts shutting down the entire application by an uncaught exception.
*FYI - I am using exactly_once processing and enable.idempotence=true*
{code:java}
2021-09-10T12:21:59.636 [kafka-producer-network-thread | mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-0_2-producer] INFO o.a.k.c.p.i.TransactionManager - MSG=[Producer clientId=mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-Stre
amThread-1-0_2-producer, transactionalId=mtx-caf-0_2] 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 tra
nsactional id.
2021-09-10T12:21:59.642 [kafka-producer-network-thread | mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-0_2-producer] ERROR o.a.k.s.p.i.RecordCollectorImpl - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamT
hread-1] task [0_2] Error encountered sending record to topic mtx-caf-DuplicateCheckStore-changelog for task 0_2 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.
Exception handler choose to FAIL the processing, no more records would be sent.
2021-09-10T12:21:59.740 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] ERROR o.a.k.s.p.internals.StreamThread - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] Encountered the following exception
during processing and the thread is going to shut down:
org.apache.kafka.streams.errors.StreamsException: Error encountered sending record to topic mtx-caf-DuplicateCheckStore-changelog for task 0_2 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.
Exception handler choose to FAIL the processing, no more records would be sent.
at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.recordSendError(RecordCollectorImpl.java:214)
at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.lambda$send$0(RecordCollectorImpl.java:186)
at org.apache.kafka.clients.producer.KafkaProducer$InterceptorCallback.onCompletion(KafkaProducer.java:1363)
at org.apache.kafka.clients.producer.internals.ProducerBatch.completeFutureAndFireCallbacks(ProducerBatch.java:231)
at org.apache.kafka.clients.producer.internals.ProducerBatch.abort(ProducerBatch.java:159)
at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortUndrainedBatches(RecordAccumulator.java:781)
at org.apache.kafka.clients.producer.internals.Sender.maybeSendAndPollTransactionalRequest(Sender.java:425)
at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:313)
at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:240)
at java.base/java.lang.Thread.run(Thread.java:829)
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.
2021-09-10T12:21:59.740 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.s.p.internals.StreamThread - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] State transition from RUNNING to PE
NDING_SHUTDOWN
2021-09-10T12:21:59.741 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.s.p.internals.StreamThread - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] Shutting down
2021-09-10T12:21:59.743 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO c.m.a.k.t.SessionBasedDataUsageAccumulator - MSG=Shutting Down
2021-09-10T12:21:59.744 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.s.p.internals.StreamTask - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] task [0_8] Suspended running
2021-09-10T12:21:59.747 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.clients.consumer.KafkaConsumer - MSG=[Consumer clientId=mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-restore-consumer, groupId=nu
ll] Subscribed to partition(s): mtx-caf-DuplicateCheckStore-changelog-10, mtx-caf-DuplicateCheckStore-changelog-6, mtx-caf-DuplicateCheckStore-changelog-4, mtx-caf-DuplicateCheckStore-changelog-2, mtx-caf-AggregatedRecordStore-changelog-10, m
tx-caf-DuplicateCheckStore-changelog-0, mtx-caf-AggregatedRecordStore-changelog-4, mtx-caf-AggregatedRecordStore-changelog-6, mtx-caf-AggregatedRecordStore-changelog-0, mtx-caf-AggregatedRecordStore-changelog-2, mtx-caf-DuplicateCheckStore-ch
angelog-11, mtx-caf-DuplicateCheckStore-changelog-9, mtx-caf-DuplicateCheckStore-changelog-7, mtx-caf-AggregatedRecordStore-changelog-11, mtx-caf-DuplicateCheckStore-changelog-5, mtx-caf-DuplicateCheckStore-changelog-3, mtx-caf-AggregatedReco
rdStore-changelog-7, mtx-caf-DuplicateCheckStore-changelog-1, mtx-caf-AggregatedRecordStore-changelog-9, mtx-caf-AggregatedRecordStore-changelog-3, mtx-caf-AggregatedRecordStore-changelog-5, mtx-caf-AggregatedRecordStore-changelog-1
2021-09-10T12:21:59.753 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.s.p.i.RecordCollectorImpl - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] task [0_8] Closing record collector
dirty
...
...
...
...
2021-09-10T12:21:59.763 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.s.p.internals.StreamTask - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] task [0_2] Suspended running
2021-09-10T12:21:59.764 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.clients.consumer.KafkaConsumer - MSG=[Consumer clientId=mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-restore-consumer, groupId=null] Subscribed to partition(s): mtx-caf-DuplicateCheckStore-changelog-11, mtx-caf-DuplicateCheckStore-changelog-10, mtx-caf-DuplicateCheckStore-changelog-9, mtx-caf-DuplicateCheckStore-changelog-7, mtx-caf-DuplicateCheckStore-changelog-6, mtx-caf-AggregatedRecordStore-changelog-11, mtx-caf-DuplicateCheckStore-changelog-4, mtx-caf-AggregatedRecordStore-changelog-7, mtx-caf-AggregatedRecordStore-changelog-10, mtx-caf-DuplicateCheckStore-changelog-0, mtx-caf-AggregatedRecordStore-changelog-9, mtx-caf-AggregatedRecordStore-changelog-4, mtx-caf-AggregatedRecordStore-changelog-6, mtx-caf-AggregatedRecordStore-changelog-0
2021-09-10T12:21:59.765 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.s.p.i.RecordCollectorImpl - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] task [0_2] Closing record collector dirty
2021-09-10T12:21:59.765 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.clients.producer.KafkaProducer - MSG=[Producer clientId=mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-0_2-producer, transactionalId=mtx-caf-0_2] Aborting incomplete transaction
2021-09-10T12:21:59.766 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.c.p.i.TransactionManager - MSG=[Producer clientId=mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-0_2-producer, transactionalId=mtx-caf-0_2] Invoking InitProducerId with current producer ID and epoch (producerId=1005, epoch=16) in order to bump the epoch
2021-09-10T12:21:59.769 [kafka-producer-network-thread | mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-0_2-producer] INFO o.a.k.c.p.i.TransactionManager - MSG=[Producer clientId=mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-0_2-producer, transactionalId=mtx-caf-0_2] ProducerId set to 1006 with epoch 0
{code}
> InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id
> -------------------------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-13292
> URL: https://issues.apache.org/jira/browse/KAFKA-13292
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 2.7.0
> Reporter: NEERAJ VAIDYA
> Priority: Major
>
> I have a KafkaStreams application which consumes from a topic which has 12 partitions. The incoming message rate into this topic is very low, perhaps 3-4 per minute. Also, some partitions will not receive messages for more than 7 days.
>
> Exactly after 7 days of starting this application, I seem to be getting the following exception and the application shuts down, without processing anymore messages :
>
> {code:java}
> 2021-09-10T12:21:59.636 [kafka-producer-network-thread | mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-0_2-producer] INFO o.a.k.c.p.i.TransactionManager - MSG=[Producer clientId=mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-0_2-producer, transactionalId=mtx-caf-0_2] 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.
> 2021-09-10T12:21:59.642 [kafka-producer-network-thread | mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1-0_2-producer] ERROR o.a.k.s.p.i.RecordCollectorImpl - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] task [0_2] Error encountered sending record to topic mtx-caf-DuplicateCheckStore-changelog for task 0_2 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.
> Exception handler choose to FAIL the processing, no more records would be sent.
> 2021-09-10T12:21:59.740 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] ERROR o.a.k.s.p.internals.StreamThread - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] Encountered the following exception during processing and the thread is going to shut down:
> org.apache.kafka.streams.errors.StreamsException: Error encountered sending record to topic mtx-caf-DuplicateCheckStore-changelog for task 0_2 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.
> Exception handler choose to FAIL the processing, no more records would be sent.
> at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.recordSendError(RecordCollectorImpl.java:214)
> at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.lambda$send$0(RecordCollectorImpl.java:186)
> at org.apache.kafka.clients.producer.KafkaProducer$InterceptorCallback.onCompletion(KafkaProducer.java:1363)
> at org.apache.kafka.clients.producer.internals.ProducerBatch.completeFutureAndFireCallbacks(ProducerBatch.java:231)
> at org.apache.kafka.clients.producer.internals.ProducerBatch.abort(ProducerBatch.java:159)
> at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortUndrainedBatches(RecordAccumulator.java:781)
> at org.apache.kafka.clients.producer.internals.Sender.maybeSendAndPollTransactionalRequest(Sender.java:425)
> at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:313)
> at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:240)
> at java.base/java.lang.Thread.run(Thread.java:829)
> 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.
> 2021-09-10T12:21:59.740 [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] INFO o.a.k.s.p.internals.StreamThread - MSG=stream-thread [mtx-caf-53dc7e96-90f1-4ae9-8af6-236d22c88e08-StreamThread-1] State transition from RUNNING to PENDING_SHUTDOWN
> {code}
>
> After this, I can see that all 12 tasks (because there are 12 partitions for all topics) get shutdown and this brings down the whole application.
>
> I understand that the transactional.id.expiration.ms = 7 days (default) will likely cause the application thread from getting expired, but why does this specific thread/task not get fenced or respawned.
> Why shutdown the entire Streams processing application just because one task has been idle ??
>
> Is there a way to keep my application up and running without causing it to shutdown ?
--
This message was sent by Atlassian Jira
(v8.3.4#803005)