Recently I’ve updated Spring Boot to 2.7.10 and Spring Kafka to 2.8.11 (it was 2.4.5 and 2.6.7 accordingly) and ran into a different behavior in exception handling. I use the RecordInterceptor that checks some cases and if that check is false then throws an exception. In earlier version it led to message re-read until case check becomes true or amount of retries is exhausted. The new version behavior is different: a message re-read is never occurred.
I’ve found in the code difference:
The earlyRecordInterceptor is now not null (The reason is that isInterceptBeforeTx is now true by default). Running checkEarlyIntercept() leads to following case: when Exception is thrown within RecordInterceptor it’s caught by that block in run() method:
catch (Exception e) {
handleConsumerException(e);
}
It leads to logging the exception and stop of further process.
In earlier version earlyRecordInterceptor was null so the checkEarlyIntercept() didn’t invoke RecordInterceptor’s intercept. it was invoked later in the doInvokeRecordListener method. In that case a thrown exception within RecordInterceptor was handled by the invokeErrorHandler, where the exception was decorated in ListenerExecutionFailedException, which led to further message re-read process as I suppose.
I’ve tried to set interceftBeforeTx = false to emulate previous behavior:
factory.setContainerCustomizer(container -> container.setInterceptBeforeTx(false));
But there is second check in earlyRecordInterceptor assignment kafkaTxManager == null, so the interceftBeforeTx property didn’t change anything.
Is the behavior correct? What should I do to get previous version behavior?
Fixed in 2.9.x (current is 2.9.9).
https://github.com/spring-projects/spring-kafka/issues/2329
2.8.x is out of OSS support; you should use 2.9.x with Boot 2.7.x.