KafkaMessageListenerContainer.doInvokeBatchListener()

[Gunju-Ko]

Please understand that I'm not good at english.

Even though the transaction is rolled back in the doInvokeBatchListener, the offset can be committed.

Below code is a doInvokeBatchListener's catch

catch (RuntimeException e) {
	if (this.containerProperties.isAckOnError() && !this.autoCommit) {
		for (ConsumerRecord<K, V> record : getHighestOffsetRecords(recordList)) {
			this.acks.add(record);
		}
	}
	if (this.batchErrorHandler == null) {
		throw e;
	}
	.... skip
}		

If ackOnError is true and autoCommit is false, Add ConsumerRecord to acks for committing offset.

If batchErrorHandler is null, exception is throw. So Kafka Transaction will be rollback. (If KafkaMessageListenerContainer is provided KafkaTransactionManager)

problem is that before call KafkaConsumer.poll method() processsCommits() can be called.

below is code

while (isRunning()) {
	try {
		if (!this.autoCommit && !this.isRecordAck) {
			processCommits();
		}

As a result even through Kafka Transaction is rollbacked but offset is committed.

I think before add ConsumerRecord to acks, checks producer and batchErrorHandler like below code

if (producer != null && this.batchErrorHandler == null) {
	throw e;
}
if (this.containerProperties.isAckOnError() && !this.autoCommit) {
	for (ConsumerRecord<K, V> record : getHighestOffsetRecords(recordList)) {
		this.acks.add(record);
	}
}
if (this.batchErrorHandler == null) {
	throw e;
}

Thanks

[garyrussell]

Yes, I noticed this yesterday; we should never commit when we are using transactions.

I will address it when fixing this: #656