ProducerFencedException after kafka consumer group rebalancing

1
votes

I cannot comment on similar topic: TransactionId prefix for producer-only and read-process-write - ProducerFencedException so I will ask a new question.

Use case:

  • One topic with 2 partitions
  • Spring @KafkaListener with concurrency=1 (default) in consumer-group "sample-consumer-group"
  • Two instances of the same application - both with the same "transaction-id-prefix"

1) I start first app instance (lets call it "instance1" - evenrything is ok - single consumer subscribes to both partitions. Log:

o.s.k.l.KafkaMessageListenerContainer : sample-consumer-group: partitions assigned: [sampleTopic-1, sampleTopic-0]

2) I start second app instance (instance2) - everything semms ok - log from this instance:

o.s.k.l.KafkaMessageListenerContainer : sample-consumer-group: partitions assigned: [sampleTopic-1]

log from "instance1":

o.s.k.l.KafkaMessageListenerContainer    : sample-consumer-group: partitions revoked: [sampleTopic-1, sampleTopic-0]
o.s.k.l.KafkaMessageListenerContainer    : sample-consumer-group: partitions assigned: [sampleTopic-0]

Still seems ok... But, when I then try to send message to any other topic (not from any kafkaListener, but from some @Transactional method - so this is producer only transaction) the following errors occur:

ERROR 4395 --- [roducer-tx-prefix-0] o.a.k.clients.producer.internals.Sender  : [Producer clientId=producer-tx-prefix-0, transactionalId=tx-prefix-0] Aborting producer batches due to fatal error

org.apache.kafka.common.errors.ProducerFencedException: Producer attempted an operation with an old epoch. Either there is a newer producer with the same transactionalId, or the producer's transaction has been expired by the broker.

ERROR 4395 --- [roducer-tx-prefix-0] o.s.k.support.LoggingProducerListener    : Exception thrown when sending a message with key='sync-register' and payload='2020-04-21T13:52:12.148412Z' to topic anotherTopic

So Is it related to the problem that I should have separate transaction-id-prefix for each instance for producer only transactions ? If yes, what is the current status of this and how to achieve this without using separate kafkaTemplate for for consumer started and producer started transactions ?

1
springapache-kafkatransactionsspring-kafkakafka-producer-api

1 Answers

2
votes

See the documentation.

As mentioned in the overview, the producer factory is configured with this property to build the producer transactional.id property. There is rather a dichotomy when specifying this property in that, when running multiple instances of the application, it must be the same on all instances to satisfy fencing zombies (also mentioned in the overview) when producing records on a listener container thread. However, when producing records using transactions that are not started by a listener container, the prefix has to be different on each instance. Version 2.3, makes this simpler to configure, especially in a Spring Boot application. In previous versions, you had to create two producer factories and KafkaTemplate s - one for producing records on a listener container thread and one for stand-alone transactions started by kafkaTemplate.executeInTransaction() or by a transaction interceptor on a @Transactional method.

Now, you can override the factory’s transactionalIdPrefix on the KafkaTemplate and the KafkaTransactionManager.

When using a transaction manager and template for a listener container, you would normally leave this to default to the producer factory’s property. This value should be the same for all application instances. For transactions started by the template (or the transaction manager for @Transaction) you should set the property on the template and transaction manager respectively. This property must have a different value on each application instance.