GH-4321: Add error handling to share consumer container

Fixes: #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.

- 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>

Author: sobychacko

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/kafka-queues.adoc

@@ -573,7 +573,7 @@ In explicit acknowledgment mode, the container enforces important constraints:
 
  Poll Blocking: Subsequent polls are blocked until all records from the previous poll are acknowledged.
  One-time Acknowledgment: Each record can only be acknowledged once.
- Error Handling: If processing throws an exception, the record is automatically acknowledged as `REJECT`.
+ Error Handling: If processing throws an exception, the outcome is determined by the <<share-error-handling,ShareConsumerRecordRecoverer>> (default: `REJECT`).
 
 [WARNING]
 In explicit mode, failing to acknowledge records will block further message processing.
@@ -609,6 +609,75 @@ or ack.reject() for every record.
 
 This feature helps developers quickly identify when acknowledgment calls are missing from their code, preventing the common issue of "Spring Kafka does not consume new records any more" due to forgotten acknowledgments.
 
+[[share-error-handling]]
+=== Error Handling
+
+The share container handles errors at two levels: during `poll()` (poll-level) and when the listener throws (listener-level).
+
+==== Poll-Level Error Handling
+
+If `consumer.poll()` throws an exception, the container handles it so the consumer thread does not stop:
+
+* **RecordDeserializationException**: Thrown when a record cannot be deserialized (for example, invalid UTF-8 for `StringDeserializer`).
+The container catches it, REJECTs the affected record using the topic, partition, and offset from the exception, logs a warning, and continues polling.
+Subsequent records are processed normally.
+* **CorruptRecordException**: Thrown when CRC validation fails (for example, with `check.crcs` enabled).
+The Kafka client automatically rejects the corrupt batch.
+The container catches the exception, logs it, and continues polling.
+
+Without this handling, a single bad record would terminate the consumer thread.
+
+==== Listener-Level Error Handling: ShareConsumerRecordRecoverer
+
+When the listener throws an exception, the container delegates to a `ShareConsumerRecordRecoverer` to decide whether to ACCEPT, RELEASE, or REJECT the failed record.
+The default is `ShareConsumerRecordRecoverer.REJECTING` (log and REJECT); `ShareConsumerRecordRecoverer.RELEASING` is also available (log and RELEASE for redelivery).
+
+You can provide a custom recoverer to RELEASE records for transient failures (for example, downstream timeouts) so they can be redelivered to another consumer, while REJECTing permanent failures.
+
+===== Configuring a Custom Recoverer
+
+Set the recoverer on the container or on the factory so it applies to all containers created by that factory:
+
+[source,java]
+----
+@Bean
+public ShareKafkaListenerContainerFactory<String, String> shareKafkaListenerContainerFactory(
+        ShareConsumerFactory<String, String> shareConsumerFactory) {
+
+    ShareKafkaListenerContainerFactory<String, String> factory =
+        new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);
+
+    factory.setShareConsumerRecordRecoverer((record, ex) -> {
+        if (ex instanceof TransientException || ex.getCause() instanceof TimeoutException) {
+            return AcknowledgeType.RELEASE;
+        }
+        return AcknowledgeType.REJECT;
+    });
+
+    return factory;
+}
+----
+
+You can also set the recoverer on an individual container via `container.setShareConsumerRecordRecoverer(recoverer)`.
+
+===== Recoverer Behavior
+
+* The recoverer must not return `RENEW`; the container treats it as REJECT.
+* If the recoverer itself throws, the container falls back to REJECT for that record.
+* The default recoverer logs at `ERROR` and returns `REJECT` for every exception.
+
+===== Relation to Poison Message Protection
+
+Broker-side delivery count (see <<share-poison-message-protection>>) limits how many times a record can be acquired.
+Even if your recoverer always returns `RELEASE`, the broker will eventually archive the record after the configured limit (default 5), so poison messages cannot loop forever.
+
+==== Difference from Traditional Container Error Handling
+
+Share consumers do not use `CommonErrorHandler`.
+That interface is designed for partition-based consumers (seek, remaining records, offset commit).
+Share consumers use acknowledgment-based recovery: the recoverer only decides ACCEPT, RELEASE, or REJECT; the container performs the acknowledgment.
+There is no seek or dead-letter topic integration in the share container.
+
 [[share-acknowledgment-examples]]
 === Acknowledgment Examples
 

spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc

@@ -13,3 +13,14 @@ For changes in earlier versions, see xref:appendix/change-history.adoc[Change Hi
 The `@KafkaListener` annotation now supports an `ackMode` attribute, allowing individual listeners to override the container factory's default acknowledgment mode without creating separate container factory beans.
 The attribute also supports SpEL expressions and property placeholders.
 See xref:kafka/receiving-messages/listener-annotation.adoc[`@KafkaListener` Annotation] for more information.
+
+[[x41-share-consumer-error-handling]]
+=== Share Consumer Error Handling
+
+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 is `ShareConsumerRecordRecoverer.REJECTING`; `RELEASING` is also available.
+You can set a custom recoverer on the factory or container.
+* See xref:kafka/kafka-queues.adoc#share-error-handling[Share consumer error handling] in the Kafka Queues documentation.

spring-kafka/src/main/java/org/springframework/kafka/config/ShareKafkaListenerContainerFactory.java

@@ -22,6 +22,7 @@
 
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.internals.ShareAcknowledgementMode;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.beans.BeanUtils;
 import org.springframework.context.ApplicationContext;
@@ -30,6 +31,7 @@
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.kafka.core.ShareConsumerFactory;
 import org.springframework.kafka.listener.ContainerProperties;
+import org.springframework.kafka.listener.ShareConsumerRecordRecoverer;
 import org.springframework.kafka.listener.ShareKafkaMessageListenerContainer;
 import org.springframework.kafka.support.JavaUtils;
 import org.springframework.kafka.support.TopicPartitionOffset;
@@ -68,6 +70,8 @@ public class ShareKafkaListenerContainerFactory<K, V>
 
 	private int concurrency = 1;
 
+	private @Nullable ShareConsumerRecordRecoverer recordRecoverer;
+
 	@SuppressWarnings("NullAway.Init")
 	private ApplicationEventPublisher applicationEventPublisher;
 
@@ -119,6 +123,17 @@ public void setConcurrency(int concurrency) {
 		this.concurrency = concurrency;
 	}
 
+	/**
+	 * Set a {@link ShareConsumerRecordRecoverer} to use for all containers created
+	 * by this factory. If not set, the container's default
+	 * ({@link org.springframework.kafka.listener.ShareConsumerRecordRecoverer#REJECTING}) is used.
+	 * @param recordRecoverer the recoverer
+	 * @since 4.1
+	 */
+	public void setShareConsumerRecordRecoverer(ShareConsumerRecordRecoverer recordRecoverer) {
+		this.recordRecoverer = recordRecoverer;
+	}
+
 	/**
 	 * Obtain the factory-level container properties - set properties as needed
 	 * and they will be copied to each listener container instance created by this factory.
@@ -188,6 +203,7 @@ protected void initializeContainer(ShareKafkaMessageListenerContainer<K, V> inst
 		instance.setApplicationEventPublisher(this.applicationEventPublisher);
 
 		JavaUtils.INSTANCE
+				.acceptIfNotNull(this.recordRecoverer, instance::setShareConsumerRecordRecoverer)
 				.acceptIfNotNull(endpoint.getGroupId(), properties::setGroupId)
 				.acceptIfNotNull(endpoint.getClientIdPrefix(), properties::setClientId);
 	}

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

@@ -83,6 +83,8 @@ public abstract class AbstractShareKafkaMessageListenerContainer<K, V>
 
 	private volatile boolean running = false;
 
+	private ShareConsumerRecordRecoverer recordRecoverer = ShareConsumerRecordRecoverer.REJECTING;
+
 	/**
 	 * Construct an instance with the provided factory and properties.
 	 * @param shareConsumerFactory the factory.
@@ -239,6 +241,27 @@ public void setupMessageListener(Object messageListener) {
 		this.containerProperties.setMessageListener(messageListener);
 	}
 
+	/**
+	 * Set the {@link ShareConsumerRecordRecoverer} to use when a listener throws
+	 * an exception. The recoverer determines whether to ACCEPT, RELEASE, or
+	 * REJECT the failed record.
+	 * @param recoverer the recoverer
+	 * @since 4.1
+	 * @see ShareConsumerRecordRecoverer
+	 */
+	public void setShareConsumerRecordRecoverer(ShareConsumerRecordRecoverer recoverer) {
+		this.recordRecoverer = recoverer;
+	}
+
+	/**
+	 * Return the configured {@link ShareConsumerRecordRecoverer}.
+	 * @return the recoverer
+	 * @since 4.1
+	 */
+	protected ShareConsumerRecordRecoverer getShareConsumerRecordRecoverer() {
+		return this.recordRecoverer;
+	}
+
 	protected abstract void doStart();
 
 	protected abstract void doStop();

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

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2025-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.listener;
+
+import org.apache.kafka.clients.consumer.AcknowledgeType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
+import org.springframework.core.log.LogAccessor;
+
+/**
+ * A strategy interface for determining the acknowledgment action when a share
+ * consumer record fails processing. Implementations decide whether to
+ * {@link AcknowledgeType#ACCEPT ACCEPT}, {@link AcknowledgeType#RELEASE RELEASE},
+ * or {@link AcknowledgeType#REJECT REJECT} the failed record.
+ *
+ * <p>Built-in implementations: {@link #REJECTING} (log and REJECT, default),
+ * {@link #RELEASING} (log and RELEASE for redelivery). Users can provide custom
+ * implementations or use a lambda.
+ *
+ * @author Soby Chacko
+ *
+ * @since 4.1
+ *
+ * @see ShareKafkaMessageListenerContainer
+ */
+@FunctionalInterface
+public interface ShareConsumerRecordRecoverer {
+
+	/**
+	 * Logger used by built-in implementations.
+	 */
+	LogAccessor LOGGER = new LogAccessor(ShareConsumerRecordRecoverer.class);
+
+	/**
+	 * Recoverer that logs the failure and REJECTs the record (archived, no redelivery).
+	 * This is the default when none is configured.
+	 */
+	ShareConsumerRecordRecoverer REJECTING = (record, ex) -> {
+		LOGGER.error(ex, () -> "Share consumer record processing failed; rejecting record from "
+				+ record.topic() + "-" + record.partition() + "@" + record.offset());
+		return AcknowledgeType.REJECT;
+	};
+
+	/**
+	 * Recoverer that logs the failure and RELEASEs the record (available for redelivery).
+	 */
+	ShareConsumerRecordRecoverer RELEASING = (record, ex) -> {
+		LOGGER.error(ex, () -> "Share consumer record processing failed; releasing record from "
+				+ record.topic() + "-" + record.partition() + "@" + record.offset());
+		return AcknowledgeType.RELEASE;
+	};
+
+	/**
+	 * Determine the acknowledgment action for a failed record.
+	 * @param record the record that failed processing
+	 * @param exception the exception thrown during processing
+	 * @return the {@link AcknowledgeType} to use — typically REJECT or RELEASE
+	 */
+	AcknowledgeType recover(ConsumerRecord<?, ?> record, Exception exception);
+
+}

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

@@ -34,6 +34,9 @@
 import org.apache.kafka.clients.consumer.ShareConsumer;
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.CorruptRecordException;
+import org.apache.kafka.common.errors.RecordDeserializationException;
 import org.jspecify.annotations.Nullable;
 
 import org.springframework.context.ApplicationEventPublisher;
@@ -352,8 +355,31 @@ public void run() {
 					try {
 						records = this.consumer.poll(java.time.Duration.ofMillis(POLL_TIMEOUT));
 					}
+					catch (RecordDeserializationException e) {
+						// poll() throws when a record can't be deserialized. Override client's auto-release
+						// with REJECT so the record is archived and the consumer thread continues.
+						TopicPartition tp = e.topicPartition();
+						long offset = e.offset();
+						this.logger.warn(e, () -> "RecordDeserializationException at "
+								+ tp + " offset " + offset + "; rejecting record and continuing");
+						try {
+							this.consumer.acknowledge(tp.topic(), tp.partition(), offset,
+									AcknowledgeType.REJECT);
+						}
+						catch (Exception ackEx) {
+							this.logger.error(ackEx, () -> "Failed to reject undeserializable record at "
+									+ tp + " offset " + offset);
+						}
+						continue;
+					}
+					catch (CorruptRecordException e) {
+						// CRC check failure. The client automatically rejects the corrupt batch.
+						this.logger.error(e, "CorruptRecordException during poll; "
+								+ "Kafka client has auto-rejected the corrupt batch");
+						continue;
+					}
 					catch (IllegalStateException e) {
-						// KIP-932: In explicit mode, poll() throws if unacknowledged records exist
+						// In explicit mode, poll() throws if unacknowledged records exist
 						if (this.isExplicitMode && !this.pendingAcknowledgments.isEmpty()) {
 							this.logger.trace(() -> "Poll blocked waiting for " + this.pendingAcknowledgments.size() +
 									" acknowledgments");
@@ -432,25 +458,53 @@ private void processRecords(ConsumerRecords<K, V> records) {
 
 		private void handleProcessingError(ConsumerRecord<K, V> record,
 				@Nullable ShareConsumerAcknowledgment acknowledgment, Exception e) {
-			this.logger.error(e, "Error processing record: " + record);
+
+			// Delegate to recoverer to decide ACCEPT, RELEASE, or REJECT
+			AcknowledgeType action;
+			try {
+				action = ShareKafkaMessageListenerContainer.this.getShareConsumerRecordRecoverer().recover(record, e);
+			}
+			catch (Exception recovererEx) {
+				// If the recoverer itself throws, fall back to REJECT
+				this.logger.error(recovererEx, () -> "ShareConsumerRecordRecoverer threw an exception; "
+						+ "falling back to REJECT for record from "
+						+ record.topic() + "-" + record.partition() + "@" + record.offset());
+				action = AcknowledgeType.REJECT;
+			}
+
+			// RENEW is not valid for error recovery (it extends lock during processing, not after failure)
+			if (action == AcknowledgeType.RENEW) {
+				this.logger.warn(() -> "ShareConsumerRecordRecoverer returned RENEW for record from "
+						+ record.topic() + "-" + record.partition() + "@" + record.offset()
+						+ "; RENEW is not valid for error recovery, using REJECT instead");
+				action = AcknowledgeType.REJECT;
+			}
+
+			final AcknowledgeType actionToLog = action;
 
 			if (this.isExplicitMode && acknowledgment != null) {
-				// Remove from pending and auto-reject on error
+				// Remove from pending and from timestamp tracking so timeout checker doesn't hold stale entries
 				this.pendingAcknowledgments.remove(record);
+				this.acknowledgmentTimestamps.remove(record);
 				try {
-					acknowledgment.reject();
+					switch (action) {
+						case ACCEPT -> acknowledgment.acknowledge();
+						case RELEASE -> acknowledgment.release();
+						case REJECT -> acknowledgment.reject();
+						default -> acknowledgment.reject();
+					}
 				}
 				catch (Exception ackEx) {
-					this.logger.error(ackEx, "Failed to reject record after processing error");
+					this.logger.error(ackEx, () -> "Failed to " + actionToLog + " record after processing error");
 				}
 			}
 			else {
-				// In implicit mode, auto-reject on error
+				// Implicit mode: apply recoverer's decision via consumer.acknowledge
 				try {
-					this.consumer.acknowledge(record, AcknowledgeType.REJECT);
+					this.consumer.acknowledge(record, action);
 				}
 				catch (Exception ackEx) {
-					this.logger.error(ackEx, "Failed to reject record after processing error");
+					this.logger.error(ackEx, () -> "Failed to " + actionToLog + " record after processing error");
 				}
 			}
 		}