[
https://issues.apache.org/jira/browse/KAFKA-13292?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17415251#comment-17415251
]
Matthias J. Sax commented on KAFKA-13292:
-----------------------------------------
KIP-671 won't avoid the exception, but it gives you the ability to avoid that
the thread dies, by using the newly added exception handler. (Well, technically
the thread dies, but you can just restart a new one.)
Also KIP-691 won't really _avoid_ the exception, but we plan to have build-it
support to handle the exception within Kafka Streams such that the thread won't
die and thus it does not surface to the exception handler.
If you upgrade client side, you need to bump all versions for
producer/consumer/admin/KafkaStreams within the same application, because you
cannot mix-and-match those, but they must always be the same. – You don't need
to upgrade the brokers though. Clients are (forward and) backward compatible to
(newer) older brokers in general (for details, read the docs – there are some
limitation what version work together – for your particular case, there should
be no limitations IIRC).
> 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)
