Enforcing container to stop on transaction fencing

[Rajh]

Affects Version(s): 2.5.6.RELEASE

Regarding this stackoverflow question :
https://stackoverflow.com/questions/64665725/why-i-lost-messages-when-using-kafka-with-eos-beta/64665928

When a ProducerFencedException | FencedInstanceIdException occurs, the framework assumes that the consumer will stop, and thus is just logging :

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

Lines 1519 to 1523 in fb1977c

	catch (ProducerFencedException | FencedInstanceIdException e) {
	this.logger.error(e, "Producer or '"
	+ ConsumerConfig.GROUP_INSTANCE_ID_CONFIG
	+ "' fenced during transaction");
	}

It would be usefull to have a safeguard , enforcing the consumer container to stop to prevent losing message in an unexpected behavior concerning this exception.

[garyrussell]

The consumer cannot tell if the fenced exception is caused by a rebalance or timeout. This is a known limitation of Kafka itself, and will be addressed in a future release.

We can certainly add an option to stop the container on these exceptions, but the proper fix can only be done when the new exception is thrown for a timeout.

https://issues.apache.org/jira/browse/KAFKA-9803

[garyrussell]

In the meantime, you should make sure the transaction.timeout.ms is long enough to process a batch of records without timing out.

[Rajh]

Reading https://issues.apache.org/jira/browse/KAFKA-9803 I can see :

Currently the coordinator does not distinguish these two cases. Both will end up as a ProducerFencedException, which means the producer needs to shut itself down.

Actually the producer is not shutting down, is it ? Or if it is, what happens to the container ? Since it is not rollbacking, does it try to re produce the message and then the ProducerFactory creates a new one ?

[garyrussell]

The producer factory should recycle the producer (it does for me with your test application with the short tx timeout). Any error on begin/abort/commitTransaction() closes the producer and a new one is created.

[Rajh]

Yes but since the exception is ignored the consumer skip the messages with the failed producer and continues with a new batch with a new producer.

[garyrussell]

Right, so until they fix it, the only thing we can do is stop the container.

Here's another work-around:

Set the container's idleEventInterval. Add an event listener to catch ListenerContainerIdleEvent; then perform a dummy (no-op) transaction; e.g. once a day.

[Rajh]

Which kind of no-op can be done to keep a transaction up ? (without producing a message :D)

[garyrussell]

template.executeInTransaction(t -> {
      return null;
});

It will just do beginTransaction(), commitTransaction().

[Rajh]

Does handling ContainerStoppedEvent in order to restart the container can have side effects ?

Say we have 3 instances of the application running on a topic with 3 partitions (1 partition for each consumer).
When a consumer transaction timeout, the container is stopped, this will start a rebalance for the others.
Then the container is restarted, and it will start a new rebalance.

Does those rebalances can trigger a ProducerFencedException but this time due to a rebalance and not a transaction timeout ?
And if this is the case, does restarting everytime can cause a sort of dead lock, or disturb the rebalances ?

[garyrussell]

Yes, it will cause a rebalance, although that can be avoided by setting a unique ConsumerConfig.GROUP_INSTANCE_ID_CONFIG on each instance (using static membership).

[Rajh]

We are using dynamic auto scaling. This means our consumer partitions are not static.

Causing a rebalance does not really matters, this FencedException should be rare and causing a rebalance to recover is acceptable.
My concern was more about the dead lock having rebalance causing FencedException causing rebalance, etc ...

[garyrussell]

It won't deadlock, even if the other instances are not well-behaved (if they don't handle the rebalance in a timely manner).