GH-4321: Add error handling to share consumer container by sobychacko · Pull Request #4338 · spring-projects/spring-kafka

sobychacko · 2026-03-10T01:56:31Z

A single record that failed deserialization or had a corrupt batch was stopping the whole consumer thread. Catch RecordDeserializationException and CorruptRecordException from poll(), REJECT or log and continue, so the thread keeps running and later records are still processed.
Every listener exception was always REJECTing the record, with no way to RELEASE for transient failures. Introduce ShareConsumerRecordRecoverer (default: DefaultShareConsumerRecordRecoverer, REJECT) so callers can plug in ACCEPT/RELEASE/REJECT per failure; wire it on AbstractShareKafkaMessageListenerContainer and ShareKafkaListenerContainerFactory.
In explicit mode, failed records were only removed from pending acks and not from acknowledgmentTimestamps, which could leak. Clear both on error.

Fixes: spring-projects#4321 - A single record that failed deserialization or had a corrupt batch was stopping the whole consumer thread. Catch `RecordDeserializationException` and `CorruptRecordException` from `poll()`, `REJECT` or log and continue, so the thread keeps running and later records are still processed. - Every listener exception was always REJECTing the record, with no way to RELEASE for transient failures. Introduce `ShareConsumerRecordRecoverer` (default: `DefaultShareConsumerRecordRecoverer`, `REJECT`) so callers can plug in `ACCEPT/RELEASE/REJECT` per failure; wire it on `AbstractShareKafkaMessageListenerContainer` and `ShareKafkaListenerContainerFactory`. - In explicit mode, failed records were only removed from pending acks and not from `acknowledgmentTimestamps`, which could leak. Clear both on error. Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>

artembilan

Looks cool!
Nothing critical, just some cosmetic suggestions.

Thanks

artembilan · 2026-03-10T17:36:36Z

+Share consumer containers now provide configurable error handling:
+
+* **Poll-level**: `RecordDeserializationException` and `CorruptRecordException` from `poll()` are caught so the consumer thread continues; undeserializable records are REJECTed and the next poll proceeds.
+* **Listener-level**: A `ShareConsumerRecordRecoverer` interface decides whether to ACCEPT, RELEASE, or REJECT when the listener throws. The default implementation REJECTs all failures; you can set a custom recoverer on the factory or container to RELEASE transient errors for redelivery.


One sentence per line.

artembilan · 2026-03-10T18:01:46Z

+ *
+ * @author Soby Chacko
+ * @since 4.1
+ * @see DefaultShareConsumerRecordRecoverer


Why all of these are stuck together?
Cannot it be made more readable?
Like each block over blank line?

artembilan · 2026-03-10T18:07:12Z

+public class DefaultShareConsumerRecordRecoverer implements ShareConsumerRecordRecoverer {
+
+	private static final LogAccessor logger = new LogAccessor(
+			LogFactory.getLog(DefaultShareConsumerRecordRecoverer.class));


We don't need LogFactory to use here.
See LogAccessor(Class<?>) ctor.

Do you foresee any evolution of this class (DefaultShareConsumerRecordRecoverer) at all?
I wonder if it could just make it into a property in the ShareConsumerRecordRecoverer interface.
Something like:

@FunctionalInterface public interface ShareConsumerRecordRecoverer { private LogAccessor logger = new LogAccessor(ShareConsumerRecordRecoverer .class); ShareConsumerRecordRecoverer REJECTING = (record, ex) -> { logger.error(exception, () -> "Share consumer record processing failed; rejecting record from " + record.topic() + "-" + record.partition() + "@" + record.offset()); return AcknowledgeType.REJECT; }

And with that you may consider to add a RELEASING one, too...

artembilan · 2026-03-10T18:09:37Z

+	 * @see ShareConsumerRecordRecoverer
+	 */
+	public void setShareConsumerRecordRecoverer(ShareConsumerRecordRecoverer recoverer) {
+		Assert.notNull(recoverer, "'recoverer' must not be null");


I think we can just stop from now checking for null on non-null arguments.
And even don't mention that in Javadocs.
It becomes obvious from signature and from Nullability tools.

artembilan · 2026-03-10T18:10:27Z

+	 * @return the recoverer
+	 * @since 4.1
+	 */
+	public ShareConsumerRecordRecoverer getShareConsumerRecordRecoverer() {


Why do we need public getter?
The more methods are visible from the class for me in IDE while typing, the more I get frustrated.

artembilan · 2026-03-10T18:11:43Z

 	private int concurrency = 1;

+	@Nullable
+	private ShareConsumerRecordRecoverer recordRecoverer;


Put annotation next to type.
Not saying this doesn't work, but for visual reasoning behind the annotation.

artembilan · 2026-03-10T18:12:34Z

 		instance.setApplicationContext(this.applicationContext);
 		instance.setApplicationEventPublisher(this.applicationEventPublisher);

+		if (this.recordRecoverer != null) {


Why does this go to the separate if, but not as a part of the JavaUtils.INSTANCE chain below?

artembilan · 2026-03-10T18:22:42Z

+		containerProps.setMessageListener((MessageListener<String, String>) record -> {
+			if (goodRecordValue.equals(record.value())) {
+				if (goodRecordsReceived.incrementAndGet() == 2) {
+					twoGoodRecordsLatch.countDown();


Looks like you don't verify anything else afterwards, so we probably don't need to counters.
You just can make CountDownLatch twoGoodRecordsLatch = new CountDownLatch(2);.
Wait exactly for two counts. Kinda the same what you do with the increment, but more straightforward and without extra code.

artembilan · 2026-03-10T18:25:09Z

+				new ShareKafkaMessageListenerContainer<>(shareConsumerFactory, containerProperties);
+
+		ShareConsumerRecordRecoverer recoverer = container.getShareConsumerRecordRecoverer();
+		assertThat(recoverer).isNotNull();


Redundant. The isInstanceOf() covers us.

…view - Wire record recoverer in the `JavaUtils` chain with other optional config for consistent style. - Drop null check from the setter; rely on signature and nullability. - Make recoverer getter protected so subclasses in other packages can use it without expanding the public API. - Put default behavior on the interface as `REJECTING` and `RELEASING` so one extension point, no extra class, and a built-in for log-and-RELEASE; use `LogAccessor(Class)` instead of `LogFactory`. - Use a plain string for the `CorruptRecordException` log; the message is fixed so a supplier adds nothing. - Test changes. Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>

artembilan

Just couple nit-picks.

Thanks

artembilan · 2026-03-11T14:19:20Z

 * **Poll-level**: `RecordDeserializationException` and `CorruptRecordException` from `poll()` are caught so the consumer thread continues; undeserializable records are REJECTed and the next poll proceeds.
-* **Listener-level**: A `ShareConsumerRecordRecoverer` interface decides whether to ACCEPT, RELEASE, or REJECT when the listener throws. The default implementation REJECTs all failures; you can set a custom recoverer on the factory or container to RELEASE transient errors for redelivery.
+* **Listener-level**: A `ShareConsumerRecordRecoverer` interface decides whether to ACCEPT, RELEASE, or REJECT when the listener throws.
+The default is `ShareConsumerRecordRecoverer.REJECTING`; `RELEASING` is also available. You can set a custom recoverer on the factory or container.


Still: one sentence per line.

artembilan · 2026-03-11T14:20:34Z

+	/**
+	 * Logger used by built-in implementations.
+	 */
+	LogAccessor LOGGER = new LogAccessor(ShareConsumerRecordRecoverer.class);


Cannot this be private?

its an interface and i don't think we can have private instance fields in interfaces. All fields declared in an interface implicitly become public, static and final. No?

Right. I see now.
So, if you are OK having this to be available for target implementations of this interface, then we can live it as is.
Otherwise we would need to push this property down to every lambda.
WDYT?

Keeping it as is. I think its ok to have a logger living on the interface; hiding it would mean adding extra components for little benefit.

Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>

artembilan requested changes Mar 10, 2026

View reviewed changes

artembilan requested changes Mar 11, 2026

View reviewed changes

sobychacko added 2 commits March 11, 2026 11:29

Single line per sentence in whats-new asciidoc

b61b43d

Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>

Adding a unit test for ShareConsumerRecordRecoverer

4c98864

Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>

artembilan merged commit 8ee8d0f into spring-projects:main Mar 11, 2026
3 checks passed

Conversation

sobychacko commented Mar 10, 2026

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants