GH-1388: DeadLetterPR: Don't start a Tx if allowed

Resolves #1388 (comment)

Don't start a new transaction in the `DeadLetterPublishingRecoverer` if the
`KafkaTemplate` allows it.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-2020 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.
@@ -217,6 +217,15 @@ public interface KafkaOperations<K, V> {
 	 */
 	boolean isTransactional();
 
+	/**
+	 * Return true if this template, when transactional, allows non-transactional operations.
+	 * @return true to allow.
+	 * @since 2.4.3
+	 */
+	default boolean isAllowNonTransactional() {
+		return false;
+	}
+
 	/**
 	 * A callback for executing arbitrary operations on the {@link Producer}.
 	 * @param <K> the key type.

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -192,6 +192,11 @@ public void setAllowNonTransactional(boolean allowNonTransactional) {
 		this.allowNonTransactional = allowNonTransactional;
 	}
 
+	@Override
+	public boolean isAllowNonTransactional() {
+		return this.allowNonTransactional;
+	}
+
 	/**
 	 * Return the producer factory used by this template.
 	 * @return the factory.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -149,7 +149,7 @@ public void accept(ConsumerRecord<?, ?> record, Exception exception) {
 		ProducerRecord<Object, Object> outRecord = createProducerRecord(record, tp, headers,
 				deserEx == null ? null : deserEx.getData(), isKey);
 		KafkaTemplate<Object, Object> kafkaTemplate = findTemplateForValue(outRecord.value());
-		if (this.transactional && !kafkaTemplate.inTransaction()) {
+		if (this.transactional && !kafkaTemplate.inTransaction() && !kafkaTemplate.isAllowNonTransactional()) {
 			kafkaTemplate.executeInTransaction(t -> {
 				publish(outRecord, t);
 				return null;
@@ -235,8 +235,11 @@ private void enhanceHeaders(RecordHeaders kafkaHeaders, ConsumerRecord<?, ?> rec
 				record.timestampType().toString().getBytes(StandardCharsets.UTF_8)));
 		kafkaHeaders.add(new RecordHeader(KafkaHeaders.DLT_EXCEPTION_FQCN,
 				exception.getClass().getName().getBytes(StandardCharsets.UTF_8)));
-		kafkaHeaders.add(new RecordHeader(KafkaHeaders.DLT_EXCEPTION_MESSAGE,
-				exception.getMessage().getBytes(StandardCharsets.UTF_8)));
+		String message = exception.getMessage();
+		if (message != null) {
+			kafkaHeaders.add(new RecordHeader(KafkaHeaders.DLT_EXCEPTION_MESSAGE,
+					exception.getMessage().getBytes(StandardCharsets.UTF_8)));
+		}
 		kafkaHeaders.add(new RecordHeader(KafkaHeaders.DLT_EXCEPTION_STACKTRACE,
 				getStackTraceAsString(exception).getBytes(StandardCharsets.UTF_8)));
 	}

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

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2020 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 static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.willAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.kafka.core.KafkaOperations.OperationsCallback;
+import org.springframework.kafka.core.KafkaTemplate;
+
+/**
+ * @author Gary Russell
+ * @since 2.4.3
+ *
+ */
+public class DeadLetterPublishingRecovererTests {
+
+	@SuppressWarnings("unchecked")
+	@Test
+	void testTxNoTx() {
+		KafkaTemplate<?, ?> template = mock(KafkaTemplate.class);
+		given(template.isTransactional()).willReturn(true);
+		given(template.inTransaction()).willReturn(false);
+		given(template.isAllowNonTransactional()).willReturn(true);
+		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
+		ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 0, 0L, "bar", "baz");
+		recoverer.accept(record, new RuntimeException());
+		verify(template, never()).executeInTransaction(any());
+		verify(template).send(any(ProducerRecord.class));
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	void testTxExisting() {
+		KafkaTemplate<?, ?> template = mock(KafkaTemplate.class);
+		given(template.isTransactional()).willReturn(true);
+		given(template.inTransaction()).willReturn(true);
+		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
+		ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 0, 0L, "bar", "baz");
+		recoverer.accept(record, new RuntimeException());
+		verify(template, never()).executeInTransaction(any());
+		verify(template).send(any(ProducerRecord.class));
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	void testNonTx() {
+		KafkaTemplate<?, ?> template = mock(KafkaTemplate.class);
+		given(template.isTransactional()).willReturn(false);
+		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
+		ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 0, 0L, "bar", "baz");
+		recoverer.accept(record, new RuntimeException());
+		verify(template, never()).inTransaction();
+		verify(template, never()).executeInTransaction(any());
+		verify(template).send(any(ProducerRecord.class));
+	}
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	void testTxNewTx() {
+		KafkaTemplate<?, ?> template = mock(KafkaTemplate.class);
+		given(template.isTransactional()).willReturn(true);
+		given(template.inTransaction()).willReturn(false);
+		given(template.isAllowNonTransactional()).willReturn(false);
+		willAnswer(inv -> {
+			((OperationsCallback) inv.getArgument(0)).doInOperations(template);
+			return null;
+		}).given(template).executeInTransaction(any());
+		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
+		ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 0, 0L, "bar", "baz");
+		recoverer.accept(record, new RuntimeException());
+		verify(template).executeInTransaction(any());
+		verify(template).send(any(ProducerRecord.class));
+	}
+
+}