Spring kafka should handle some fatal exceptions like InvalidPidMappingException, UnknownProducerIdException.

[bartosz-stasikowski-projectdrgn]

Affects Version(s): 2.3.1.RELEASE

🎁 Enhancement
Spring kafka should handle some fatal exceptions like InvalidPidMappingException, UnknownProducerIdException.

Context: processing messages with transactions in non batch mode
Problem: Some exceptions like InvalidPidMappingException, UnknownProducerIdException are fatal and will cause KafkaMessageListenerContainer to fail every time polled record is processed

Usecase:

1. Records are polled
2. Transaction fails because of InvalidPidMappingException or UnknownProducerIdException when processing record
3. Rollback processing fails for InvalidPidMappingException or will send message to DLT for UnknownProducerIdException (if configured to do so)
4. Next poll will get another record which will have the same problem because exception is fatal

• in case of InvalidPidMappingException records will be processed after service is restarted or the is a rebalance
• in case of UnknownProducerIdException records wont be processed again since offsets were commited in rollback processing

Currently InvalidPidMappingException exception can be handled in ErrorHandler but the only thing I can think of is to restart listener container.
For UnknownProducerIdException it is possible to not perform rollback processing and handle in the same way as InvalidPidMappingException in error handler.
This is strange and it should be handled by spring-kafka since it is a common problem when using transactions.

[garyrussell]

Please show some logs and stack traces - any failures related to transactions are considered fatal (txFailed contains the exception) from the producer perspective and the producer is closed...

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

Lines 637 to 660 in 909af16

	public void commitTransaction() throws ProducerFencedException {
	LOGGER.debug(() -> toString() + " commitTransaction()");
	try {
	this.delegate.commitTransaction();
	}
	catch (RuntimeException e) {
	LOGGER.error(e, () -> "commitTransaction failed: " + this);
	this.txFailed = e;
	throw e;
	}
	}
	@Override
	public void abortTransaction() throws ProducerFencedException {
	LOGGER.debug(() -> toString() + " abortTransaction()");
	try {
	this.delegate.abortTransaction();
	}
	catch (RuntimeException e) {
	LOGGER.error(e, () -> "Abort failed: " + this);
	this.txFailed = e;
	throw e;
	}
	}

and

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

Lines 681 to 689 in 909af16

	if (this.txFailed != null) {
	LOGGER.warn(() -> "Error during transactional operation; producer removed from cache; "
	+ "possible cause: "
	+ "broker restarted during transaction: " + this);
	this.delegate.close(closeTimeout);
	if (this.removeConsumerProducer != null) {
	this.removeConsumerProducer.accept(this);
	}
	}

[bartosz-stasikowski-projectdrgn]

Yes it works exactly as you want, but producer is not closed because its cache is null.

@Override
		public void close(@Nullable Duration timeout) {
			LOGGER.trace(() -> toString() + " close(" + (timeout == null ? "null" : timeout) + ")");
			if (this.cache != null) {

CloseSafeProducer is created in doCreateTxProducer() where cache is taken from this.cache.get(prefix) which is null as this.cache is empty:

return new CloseSafeProducer<>(newProducer, this.cache.get(prefix), remover,
				(String) newProducerConfigs.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG));

Should it be getCache(prefix) rather that this.cache.get(prefix)?

This is my InvalidPidMappingException stack:

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)
at org.apache.kafka.clients.producer.internals.TransactionManager.beginTransaction(TransactionManager.java:290)
at org.apache.kafka.clients.producer.KafkaProducer.beginTransaction(KafkaProducer.java:642)
at org.springframework.kafka.core.DefaultKafkaProducerFactory$CloseSafeProducer.beginTransaction(DefaultKafkaProducerFactory.java:597)
at org.springframework.kafka.core.ProducerFactoryUtils.getTransactionalResourceHolder(ProducerFactoryUtils.java:101)
at org.springframework.kafka.transaction.KafkaTransactionManager.doBegin(KafkaTransactionManager.java:160)
at com.projectdrgn.common.messaging.config.KafkaProducerConfig$kafkaTransactionManager$kafkaTransactionManager$1.doBegin(KafkaProducerConfig.kt:65)
at org.springframework.transaction.support.AbstractPlatformTransactionManager.getTransaction(AbstractPlatformTransactionManager.java:376)
at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:137)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.recordAfterRollback(KafkaMessageListenerContainer.java:1463)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListenerInTx(KafkaMessageListenerContainer.java:1438)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:1398)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:1165)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:949)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:884)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.kafka.common.KafkaException: org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id.
at org.apache.kafka.clients.producer.internals.TransactionManager$AddPartitionsToTxnHandler.handleResponse(TransactionManager.java:1204)
at org.apache.kafka.clients.producer.internals.TransactionManager$TxnRequestHandler.onComplete(TransactionManager.java:1069)
at org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse.java:109)
at org.apache.kafka.clients.NetworkClient.completeResponses(NetworkClient.java:561)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:553)
at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:307)
at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:238)
... 1 common frames omitted
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.

[garyrussell]

Yes, that looks like a bug - investigating.

[garyrussell]

Hmmm - no; the cache should have been previously populated by the previous call to getCache() here, before doCreateTxProducer() is called.

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

Lines 463 to 473 in 909af16

	protected Producer<K, V> createTransactionalProducer(String txIdPrefix) {
	BlockingQueue<CloseSafeProducer<K, V>> queue = getCache(txIdPrefix);
	Assert.notNull(queue, () -> "No cache found for " + txIdPrefix);
	Producer<K, V> cachedProducer = queue.poll();
	if (cachedProducer == null) {
	return doCreateTxProducer(txIdPrefix, "" + this.transactionIdSuffix.getAndIncrement(), null);
	}
	else {
	return cachedProducer;
	}
	}

Can you show all your kafka-related configuration (and possibly a TRACE level log that shows this behavior)?

[bartosz-stasikowski-projectdrgn]

But doCreateTxProducer is called from createTransactionalProducerForPartition() in my case (suffix != null):

protected Producer<K, V> createTransactionalProducerForPartition(String txIdPrefix) {
		String suffix = TransactionSupport.getTransactionIdSuffix();
		if (suffix == null) {
			return createTransactionalProducer(txIdPrefix);
		}
		else {
			synchronized (this.consumerProducers) {
				if (!this.consumerProducers.containsKey(suffix)) {
					CloseSafeProducer<K, V> newProducer = doCreateTxProducer(txIdPrefix, suffix,
							this::removeConsumerProducer);

[garyrussell]

I'd still like to see your configuration; I don't see how we can get there will null.

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

Lines 373 to 380 in 909af16

	if (txIdPrefix != null) {
	if (this.producerPerConsumerPartition) {
	return createTransactionalProducerForPartition(txIdPrefix);
	}
	else {
	return createTransactionalProducer(txIdPrefix);
	}
	}

[bartosz-stasikowski-projectdrgn]

I am not sure if I understand: in my code txIdPrefix != null and this.producerPerConsumerPartition is true.

If by configuration you mean producer/consumer config then:

2020-01-29 18:28:47.346 INFO  17368 --- [rocessor-0-C-1] o.a.k.clients.producer.ProducerConfig    : ProducerConfig values: 
	acks = all
	batch.size = 16384
	bootstrap.servers = [localhost:9092]
	buffer.memory = 33554432
	client.dns.lookup = default
	client.id = 
	compression.type = none
	connections.max.idle.ms = 540000
	delivery.timeout.ms = 120000
	enable.idempotence = true
	interceptor.classes = [my.example.logging.MyProducerInterceptor]
	key.serializer = class org.apache.kafka.common.serialization.StringSerializer
	linger.ms = 1
	max.block.ms = 30000
	max.in.flight.requests.per.connection = 1
	max.request.size = 1048576
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
	receive.buffer.bytes = 32768
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retries = 2147483647
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = null
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism = GSSAPI
	security.protocol = PLAINTEXT
	send.buffer.bytes = 131072
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
	ssl.endpoint.identification.algorithm = https
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLS
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	transaction.timeout.ms = 60000
	transactional.id = my-service-tx-my-messaging-processor.my.topic.0
	value.serializer = class my.example.converter.MySerializer
	
	
2020-01-29 18:28:45.109 INFO  17368 --- [           main] o.a.k.clients.consumer.ConsumerConfig    : ConsumerConfig values: 
	allow.auto.create.topics = true
	auto.commit.interval.ms = 5000
	auto.offset.reset = earliest
	bootstrap.servers = [localhost:9092]
	check.crcs = true
	client.dns.lookup = default
	client.id = my-messaging-processor-0
	client.rack = 
	connections.max.idle.ms = 540000
	default.api.timeout.ms = 60000
	enable.auto.commit = false
	exclude.internal.topics = true
	fetch.max.bytes = 52428800
	fetch.max.wait.ms = 500
	fetch.min.bytes = 1
	group.id = my-messaging-processor
	group.instance.id = null
	heartbeat.interval.ms = 3000
	interceptor.classes = [my.example.logging.MyConsumerInterceptor]
	internal.leave.group.on.close = true
	isolation.level = read_committed
	key.deserializer = class org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2
	max.partition.fetch.bytes = 1048576
	max.poll.interval.ms = 300000
	max.poll.records = 100
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
	receive.buffer.bytes = 65536
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = null
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism = GSSAPI
	security.protocol = PLAINTEXT
	send.buffer.bytes = 131072
	session.timeout.ms = 10000
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
	ssl.endpoint.identification.algorithm = https
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLS
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	value.deserializer = class org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2

[garyrussell]

Got it; sorry; now I see it.