GH-1882 DLPR: Add Exception Cause Header

Resolves #1882

* Checkstyle fixes.

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -5321,7 +5321,8 @@ ErrorHandler errorHandler = new DefaultErrorHandler(recoverer, new FixedBackOff(
 
 The record sent to the dead-letter topic is enhanced with the following headers:
 
-* `KafkaHeaders.DLT_EXCEPTION_FQCN`: The Exception class name.
+* `KafkaHeaders.DLT_EXCEPTION_FQCN`: The Exception class name (generally a `ListenerExecutionFailedException`, but can be others).
+* `KafkaHeaders.DLT_EXCEPTION_CAUSE_FQCN`: The Exception cause class name, if present (since version 2.8).
 * `KafkaHeaders.DLT_EXCEPTION_STACKTRACE`: The Exception stack trace.
 * `KafkaHeaders.DLT_EXCEPTION_MESSAGE`: The  Exception message.
 * `KafkaHeaders.DLT_KEY_EXCEPTION_FQCN`: The Exception class name (key deserialization errors only).
@@ -5332,6 +5333,9 @@ The record sent to the dead-letter topic is enhanced with the following headers:
 * `KafkaHeaders.DLT_ORIGINAL_OFFSET`: The original offset.
 * `KafkaHeaders.DLT_ORIGINAL_TIMESTAMP`: The original timestamp.
 * `KafkaHeaders.DLT_ORIGINAL_TIMESTAMP_TYPE`: The original timestamp type.
+* `KafkaHeaders.DLT_ORIGINAL_CONSUMER_GROUP`: The original consumer group that failed to process the record (since version 2.8).
+
+Key exceptions are only caused by `DeserializationException` s so there is no `DLT_KEY_EXCEPTION_CAUSE_FQCN`.
 
 There are two mechanisms to add more headers.
 

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

@@ -514,14 +514,14 @@ private Duration determineSendTimeout(KafkaOperations<?, ?> template) {
 	}
 
 	private void enhanceHeaders(Headers kafkaHeaders, ConsumerRecord<?, ?> record, Exception exception) {
-		maybeAddOriginalHeaders(kafkaHeaders, record);
+		maybeAddOriginalHeaders(kafkaHeaders, record, exception);
 		Headers headers = this.headersFunction.apply(record, exception);
 		if (headers != null) {
 			headers.forEach(kafkaHeaders::add);
 		}
 	}
 
-	private void maybeAddOriginalHeaders(Headers kafkaHeaders, ConsumerRecord<?, ?> record) {
+	private void maybeAddOriginalHeaders(Headers kafkaHeaders, ConsumerRecord<?, ?> record, Exception ex) {
 		maybeAddHeader(kafkaHeaders, this.headerNames.original.topicHeader,
 				record.topic().getBytes(StandardCharsets.UTF_8));
 		maybeAddHeader(kafkaHeaders, this.headerNames.original.partitionHeader,
@@ -532,6 +532,13 @@ private void maybeAddOriginalHeaders(Headers kafkaHeaders, ConsumerRecord<?, ?>
 				ByteBuffer.allocate(Long.BYTES).putLong(record.timestamp()).array());
 		maybeAddHeader(kafkaHeaders, this.headerNames.original.timestampTypeHeader,
 				record.timestampType().toString().getBytes(StandardCharsets.UTF_8));
+		if (ex instanceof ListenerExecutionFailedException) {
+			String consumerGroup = ((ListenerExecutionFailedException) ex).getGroupId();
+			if (consumerGroup != null) {
+				maybeAddHeader(kafkaHeaders, this.headerNames.original.consumerGroup,
+						consumerGroup.getBytes(StandardCharsets.UTF_8));
+			}
+		}
 	}
 
 	private void maybeAddHeader(Headers kafkaHeaders, String header, byte[] value) {
@@ -540,11 +547,14 @@ private void maybeAddHeader(Headers kafkaHeaders, String header, byte[] value) {
 		}
 	}
 
-	void addExceptionInfoHeaders(Headers kafkaHeaders, Exception exception,
-								boolean isKey) {
+	void addExceptionInfoHeaders(Headers kafkaHeaders, Exception exception, boolean isKey) {
 		kafkaHeaders.add(new RecordHeader(isKey ? this.headerNames.exceptionInfo.keyExceptionFqcn
 				: this.headerNames.exceptionInfo.exceptionFqcn,
 				exception.getClass().getName().getBytes(StandardCharsets.UTF_8)));
+		if (!isKey && exception.getCause() != null) {
+			kafkaHeaders.add(new RecordHeader(this.headerNames.exceptionInfo.exceptionCauseFqcn,
+					exception.getCause().getClass().getName().getBytes(StandardCharsets.UTF_8)));
+		}
 		String message = exception.getMessage();
 		if (message != null) {
 			kafkaHeaders.add(new RecordHeader(isKey
@@ -579,9 +589,11 @@ protected HeaderNames getHeaderNames() {
 					.timestampTypeHeader(KafkaHeaders.DLT_ORIGINAL_TIMESTAMP_TYPE)
 					.topicHeader(KafkaHeaders.DLT_ORIGINAL_TOPIC)
 					.partitionHeader(KafkaHeaders.DLT_ORIGINAL_PARTITION)
+					.consumerGroupHeader(KafkaHeaders.DLT_ORIGINAL_CONSUMER_GROUP)
 				.exception()
 					.keyExceptionFqcn(KafkaHeaders.DLT_KEY_EXCEPTION_FQCN)
 					.exceptionFqcn(KafkaHeaders.DLT_EXCEPTION_FQCN)
+					.exceptionCauseFqcn(KafkaHeaders.DLT_EXCEPTION_CAUSE_FQCN)
 					.keyExceptionMessage(KafkaHeaders.DLT_KEY_EXCEPTION_MESSAGE)
 					.exceptionMessage(KafkaHeaders.DLT_EXCEPTION_MESSAGE)
 					.keyExceptionStacktrace(KafkaHeaders.DLT_KEY_EXCEPTION_STACKTRACE)
@@ -605,50 +617,67 @@ public static class HeaderNames {
 		}
 
 		static class Original {
-			private final String offsetHeader;
-			private final String timestampHeader;
-			private final String timestampTypeHeader;
-			private final String topicHeader;
-			private final String partitionHeader;
+
+			final String offsetHeader;
+
+			final String timestampHeader;
+
+			final String timestampTypeHeader;
+
+			final String topicHeader;
+
+			final String partitionHeader;
+
+			final String consumerGroup;
 
 			Original(String offsetHeader,
 					String timestampHeader,
 					String timestampTypeHeader,
 					String topicHeader,
-					String partitionHeader) {
+					String partitionHeader,
+					String consumerGroup) {
 				this.offsetHeader = offsetHeader;
 				this.timestampHeader = timestampHeader;
 				this.timestampTypeHeader = timestampTypeHeader;
 				this.topicHeader = topicHeader;
 				this.partitionHeader = partitionHeader;
+				this.consumerGroup = consumerGroup;
 			}
 		}
 
 		static class ExceptionInfo {
 
-			private final String keyExceptionFqcn;
-			private final String exceptionFqcn;
-			private final String keyExceptionMessage;
-			private final String exceptionMessage;
-			private final String keyExceptionStacktrace;
-			private final String exceptionStacktrace;
+			final String keyExceptionFqcn;
+
+			final String exceptionFqcn;
+
+			final String exceptionCauseFqcn;
+
+			final String keyExceptionMessage;
+
+			final String exceptionMessage;
+
+			final String keyExceptionStacktrace;
+
+			final String exceptionStacktrace;
 
 			ExceptionInfo(String keyExceptionFqcn,
-						String exceptionFqcn,
-						String keyExceptionMessage,
-						String exceptionMessage,
-						String keyExceptionStacktrace,
-						String exceptionStacktrace) {
+					String exceptionFqcn,
+					String exceptionCauseFqcn,
+					String keyExceptionMessage,
+					String exceptionMessage,
+					String keyExceptionStacktrace,
+					String exceptionStacktrace) {
 				this.keyExceptionFqcn = keyExceptionFqcn;
 				this.exceptionFqcn = exceptionFqcn;
+				this.exceptionCauseFqcn = exceptionCauseFqcn;
 				this.keyExceptionMessage = keyExceptionMessage;
 				this.exceptionMessage = exceptionMessage;
 				this.keyExceptionStacktrace = keyExceptionStacktrace;
 				this.exceptionStacktrace = exceptionStacktrace;
 			}
 		}
 
-
 		/**
 		 * Provides a convenient API for creating
 		 * {@link DeadLetterPublishingRecoverer.HeaderNames}.
@@ -685,6 +714,8 @@ public class Original {
 
 				private String partitionHeader;
 
+				private String consumerGroupHeader;
+
 				/**
 				 * Sets the name of the header that will be used to store the offset
 				 * of the original record.
@@ -745,6 +776,18 @@ public Builder.Original partitionHeader(String partitionHeader) {
 					return this;
 				}
 
+				/**
+				 * Sets the name of the header that will be used to store the consumer
+				 * group that failed to consume the original record.
+				 * @param consumerGroupHeader the consumer group header name.
+				 * @return the Original builder instance
+				 * @since 2.8
+				 */
+				public Builder.Original consumerGroupHeader(String consumerGroupHeader) {
+					this.consumerGroupHeader = consumerGroupHeader;
+					return this;
+				}
+
 				/**
 				 * Returns the exception builder.
 				 * @return the exception builder.
@@ -760,16 +803,18 @@ public ExceptionInfo exception() {
 				 * @since 2.7
 				 */
 				private DeadLetterPublishingRecoverer.HeaderNames.Original build() {
-					Assert.notNull(this.offsetHeader, "offsetHeader header cannot be null");
-					Assert.notNull(this.timestampHeader, "timestampHeader header cannot be null");
-					Assert.notNull(this.timestampTypeHeader, "timestampTypeHeader header cannot be null");
-					Assert.notNull(this.topicHeader, "topicHeader header cannot be null");
-					Assert.notNull(this.partitionHeader, "partitionHeader header cannot be null");
+					Assert.notNull(this.offsetHeader, "offsetHeader cannot be null");
+					Assert.notNull(this.timestampHeader, "timestampHeader cannot be null");
+					Assert.notNull(this.timestampTypeHeader, "timestampTypeHeader cannot be null");
+					Assert.notNull(this.topicHeader, "topicHeader cannot be null");
+					Assert.notNull(this.partitionHeader, "partitionHeader cannot be null");
+					Assert.notNull(this.consumerGroupHeader, "consumerGroupHeader cannot be null");
 					return new DeadLetterPublishingRecoverer.HeaderNames.Original(this.offsetHeader,
 							this.timestampHeader,
 							this.timestampTypeHeader,
 							this.topicHeader,
-							this.partitionHeader);
+							this.partitionHeader,
+							this.consumerGroupHeader);
 				}
 			}
 
@@ -785,6 +830,8 @@ public class ExceptionInfo {
 
 				private String exceptionFqcn;
 
+				private String exceptionCauseFqcn;
+
 				private String keyExceptionMessage;
 
 				private String exceptionMessage;
@@ -817,6 +864,17 @@ public ExceptionInfo exceptionFqcn(String exceptionFqcn) {
 					return this;
 				}
 
+				/**
+				 * Sets the name of the header that will be used to store the exceptionCauseFqcn
+				 * of the original record.
+				 * @param exceptionCauseFqcn the exceptionFqcn header name.
+				 * @return the Exception builder instance
+				 * @since 2.8
+				 */
+				public ExceptionInfo exceptionCauseFqcn(String exceptionCauseFqcn) {
+					this.exceptionCauseFqcn = exceptionCauseFqcn;
+					return this;
+				}
 				/**
 				 * Sets the name of the header that will be used to store the keyExceptionMessage
 				 * of the original record.
@@ -873,13 +931,15 @@ public ExceptionInfo exceptionStacktrace(String exceptionStacktrace) {
 				public DeadLetterPublishingRecoverer.HeaderNames build() {
 					Assert.notNull(this.keyExceptionFqcn, "keyExceptionFqcn header cannot be null");
 					Assert.notNull(this.exceptionFqcn, "exceptionFqcn header cannot be null");
+					Assert.notNull(this.exceptionCauseFqcn, "exceptionCauseFqcn header cannot be null");
 					Assert.notNull(this.keyExceptionMessage, "keyExceptionMessage header cannot be null");
 					Assert.notNull(this.exceptionMessage, "exceptionMessage header cannot be null");
 					Assert.notNull(this.keyExceptionStacktrace, "keyExceptionStacktrace header cannot be null");
 					Assert.notNull(this.exceptionStacktrace, "exceptionStacktrace header cannot be null");
 					return new DeadLetterPublishingRecoverer.HeaderNames(Builder.this.original.build(),
 							new HeaderNames.ExceptionInfo(this.keyExceptionFqcn,
 									this.exceptionFqcn,
+									this.exceptionCauseFqcn,
 									this.keyExceptionMessage,
 									this.exceptionMessage,
 									this.keyExceptionStacktrace,

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DeadLetterPublishingRecovererFactory.java

@@ -72,9 +72,11 @@ protected DeadLetterPublishingRecoverer.HeaderNames getHeaderNames() {
 							.timestampTypeHeader(KafkaHeaders.ORIGINAL_TIMESTAMP_TYPE)
 							.topicHeader(KafkaHeaders.ORIGINAL_TOPIC)
 							.partitionHeader(KafkaHeaders.ORIGINAL_PARTITION)
+							.consumerGroupHeader(KafkaHeaders.ORIGINAL_CONSUMER_GROUP)
 						.exception()
 							.keyExceptionFqcn(KafkaHeaders.KEY_EXCEPTION_FQCN)
 							.exceptionFqcn(KafkaHeaders.EXCEPTION_FQCN)
+							.exceptionCauseFqcn(KafkaHeaders.EXCEPTION_CAUSE_FQCN)
 							.keyExceptionMessage(KafkaHeaders.KEY_EXCEPTION_MESSAGE)
 							.exceptionMessage(KafkaHeaders.EXCEPTION_MESSAGE)
 							.keyExceptionStacktrace(KafkaHeaders.KEY_EXCEPTION_STACKTRACE)

spring-kafka/src/main/java/org/springframework/kafka/support/KafkaHeaders.java

@@ -147,6 +147,12 @@ public abstract class KafkaHeaders {
 	 */
 	public static final String DLT_EXCEPTION_FQCN = PREFIX + "dlt-exception-fqcn";
 
+	/**
+	 * Exception cause class name for a record published sent to a dead-letter topic.
+	 * @since 2.8
+	 */
+	public static final String DLT_EXCEPTION_CAUSE_FQCN = PREFIX + "dlt-exception-cause-fqcn";
+
 	/**
 	 * Exception stack trace for a record published to a dead-letter topic.
 	 * @since 2.2
@@ -198,6 +204,12 @@ public abstract class KafkaHeaders {
 	 */
 	public static final String DLT_ORIGINAL_OFFSET = PREFIX + "dlt-original-offset";
 
+	/**
+	 * Consumer group that failed to consumer a record published to a dead-letter topic.
+	 * @since 2.8
+	 */
+	public static final String DLT_ORIGINAL_CONSUMER_GROUP = PREFIX + "dlt-original-consumer-group";
+
 	/**
 	 * Original timestamp for a record published to a dead-letter topic.
 	 * @since 2.2
@@ -228,6 +240,12 @@ public abstract class KafkaHeaders {
 	 */
 	public static final String EXCEPTION_FQCN = PREFIX + "exception-fqcn";
 
+	/**
+	 * Exception class name for a record published sent to another topic.
+	 * @since 2.8
+	 */
+	public static final String EXCEPTION_CAUSE_FQCN = PREFIX + "exception-fqcn";
+
 	/**
 	 * Exception stack trace for a record published to another topic.
 	 * @since 2.2
@@ -279,6 +297,12 @@ public abstract class KafkaHeaders {
 	 */
 	public static final String ORIGINAL_OFFSET = PREFIX + "original-offset";
 
+	/**
+	 * Consumer group that failed to consumer a record published to another topic.
+	 * @since 2.8
+	 */
+	public static final String ORIGINAL_CONSUMER_GROUP = PREFIX + "dlt-original-consumer-group";
+
 	/**
 	 * Original timestamp for a record published to another topic.
 	 * @since 2.2

spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java

@@ -694,7 +694,8 @@ private void testFixLagGuts(String topic, int whichTm) throws InterruptedExcepti
 	public void testMaxFailures() throws Exception {
 		logger.info("Start testMaxFailures");
 		Map<String, Object> props = KafkaTestUtils.consumerProps("txTestMaxFailures", "false", embeddedKafka);
-		props.put(ConsumerConfig.GROUP_ID_CONFIG, "groupInARBP");
+		String group = "groupInARBP";
+		props.put(ConsumerConfig.GROUP_ID_CONFIG, group);
 		props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
 		ContainerProperties containerProps = new ContainerProperties(topic3);
@@ -769,6 +770,8 @@ public void accept(ConsumerRecord<?, ?> record, Consumer<?, ?> consumer, Excepti
 		MessageHeaders headers = new MessageHeaders(map);
 		assertThat(new String(headers.get(KafkaHeaders.DLT_EXCEPTION_FQCN, byte[].class)))
 				.contains("ListenerExecutionFailedException");
+		assertThat(new String(headers.get(KafkaHeaders.DLT_EXCEPTION_CAUSE_FQCN, byte[].class)))
+				.isEqualTo("java.lang.RuntimeException");
 		assertThat(headers.get(KafkaHeaders.DLT_EXCEPTION_MESSAGE, byte[].class))
 				.contains("fail for max failures".getBytes());
 		assertThat(headers.get(KafkaHeaders.DLT_EXCEPTION_STACKTRACE)).isNotNull();
@@ -777,6 +780,7 @@ public void accept(ConsumerRecord<?, ?> record, Consumer<?, ?> consumer, Excepti
 		assertThat(headers.get(KafkaHeaders.DLT_ORIGINAL_TIMESTAMP, byte[].class)).isNotNull();
 		assertThat(headers.get(KafkaHeaders.DLT_ORIGINAL_TIMESTAMP_TYPE, byte[].class)).isNotNull();
 		assertThat(headers.get(KafkaHeaders.DLT_ORIGINAL_TOPIC, byte[].class)).isEqualTo(topic3.getBytes());
+		assertThat(headers.get(KafkaHeaders.DLT_ORIGINAL_CONSUMER_GROUP)).isEqualTo(group.getBytes());
 		assertThat(headers.get("baz")).isEqualTo("qux".getBytes());
 		pf.destroy();
 		assertThat(stopLatch.await(10, TimeUnit.SECONDS)).isTrue();