Configure dlq producer properties

Fixes #58
Resolves #257

- add the ability to configure the dlq producer properties
- new property on KafkaConsumerProperties for dlqProducerProperties
- dlq sender refactoring in Kafka binder
- make dlq type raw so that non byte[] key/payloads can be sent
- add new tests for verifying dlq producer properties

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java

@@ -22,6 +22,7 @@ import java.util.Map;
 /**
  * @author Marius Bogoevici
  * @author Ilayaperumal Gopinathan
+ * @author Soby Chacko
  *
  * <p>
  * Thanks to Laszlo Szabo for providing the initial patch for generic property support.
@@ -41,6 +42,8 @@ public class KafkaConsumerProperties {
 
 	private String dlqName;
 
+	private KafkaProducerProperties dlqProducerProperties = new KafkaProducerProperties();
+
 	private int recoveryInterval = 5000;
 
 	private String[] trustedPackages;
@@ -133,4 +136,12 @@ public class KafkaConsumerProperties {
 	public void setTrustedPackages(String[] trustedPackages) {
 		this.trustedPackages = trustedPackages;
 	}
+
+	public KafkaProducerProperties getDlqProducerProperties() {
+		return dlqProducerProperties;
+	}
+
+	public void setDlqProducerProperties(KafkaProducerProperties dlqProducerProperties) {
+		this.dlqProducerProperties = dlqProducerProperties;
+	}
 }

spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc

@@ -186,6 +186,11 @@ dlqName::
   The name of the DLQ topic to receive the error messages.
 +
 Default: null (If not specified, messages that result in errors will be forwarded to a topic named `error.<destination>.<group>`).
+dlqProducerProperties::
+  Using this, dlq specific producer properties can be set.
+  All the properties available through kafka producer properties can be set through this property.
++
+Default: Default Kafka producer properties.
 
 [[kafka-producer-properties]]
 === Kafka Producer Properties

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -18,7 +18,6 @@ package org.springframework.cloud.stream.binder.kafka;
 
 import java.io.PrintWriter;
 import java.io.StringWriter;
-import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -28,6 +27,7 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.function.Predicate;
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -41,7 +41,6 @@ import org.apache.kafka.common.header.internals.RecordHeader;
 import org.apache.kafka.common.header.internals.RecordHeaders;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
-import org.apache.kafka.common.utils.Utils;
 
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
@@ -422,21 +421,36 @@ public class KafkaMessageChannelBinder extends
 	protected MessageHandler getErrorMessageHandler(final ConsumerDestination destination, final String group,
 			final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties) {
 		if (extendedConsumerProperties.getExtension().isEnableDlq()) {
-			ProducerFactory<byte[], byte[]> producerFactory = this.transactionManager != null
+			ProducerFactory<?,?> producerFactory = this.transactionManager != null
 					? this.transactionManager.getProducerFactory()
-					: getProducerFactory(null, new ExtendedProducerProperties<>(new KafkaProducerProperties()));
-			final KafkaTemplate<byte[], byte[]> kafkaTemplate = new KafkaTemplate<>(producerFactory);
+					: getProducerFactory(null,
+						new ExtendedProducerProperties<>(extendedConsumerProperties.getExtension().getDlqProducerProperties()));
+			final KafkaTemplate<?,?> kafkaTemplate = new KafkaTemplate<>(producerFactory);
+			String dlqName = StringUtils.hasText(extendedConsumerProperties.getExtension().getDlqName())
+					? extendedConsumerProperties.getExtension().getDlqName()
+					: "error." + destination.getName() + "." + group;
+
+			@SuppressWarnings({"unchecked", "raw"})
+			DlqSender<?,?> dlqSender = new DlqSender(kafkaTemplate, dlqName);
+
 			return message -> {
 				final ConsumerRecord<?, ?> record = message.getHeaders()
 						.get(KafkaHeaders.RAW_DATA, ConsumerRecord.class);
-				final byte[] key = record.key() != null ? Utils.toArray(ByteBuffer.wrap((byte[]) record.key()))
-						: null;
-				final byte[] payload = record.value() != null
-						? Utils.toArray(ByteBuffer.wrap((byte[]) record.value()))
-						: null;
-				String dlqName = StringUtils.hasText(extendedConsumerProperties.getExtension().getDlqName())
-						? extendedConsumerProperties.getExtension().getDlqName()
-						: "error." + destination.getName() + "." + group;
+
+				if (extendedConsumerProperties.isUseNativeDecoding()) {
+					if (record != null) {
+						Map<String, String> configuration = extendedConsumerProperties.getExtension().getDlqProducerProperties().getConfiguration();
+						if (record.key() != null && !record.key().getClass().isInstance(byte[].class)) {
+							ensureDlqMessageCanBeProperlySerialized(
+									configuration,
+									(Map<String, String> config) -> !config.containsKey("key.serializer"), "Key");
+						}
+						if (record.value() != null && !record.value().getClass().isInstance(byte[].class)) {
+							ensureDlqMessageCanBeProperlySerialized(configuration,
+									(Map<String, String> config) -> !config.containsKey("value.serializer"), "Payload");
+						}
+					}
+				}
 
 				Headers kafkaHeaders = new RecordHeaders(record.headers().toArray());
 				kafkaHeaders.add(new RecordHeader(X_ORIGINAL_TOPIC,
@@ -448,37 +462,21 @@ public class KafkaMessageChannelBinder extends
 					kafkaHeaders.add(new RecordHeader(X_EXCEPTION_STACKTRACE,
 							getStackTraceAsString(throwable).getBytes(StandardCharsets.UTF_8)));
 				}
-				ProducerRecord<byte[], byte[]> producerRecord = new ProducerRecord<>(dlqName, record.partition(),
-						key, payload, kafkaHeaders);
-				ListenableFuture<SendResult<byte[], byte[]>> sentDlq = kafkaTemplate.send(producerRecord);
-				sentDlq.addCallback(new ListenableFutureCallback<SendResult<byte[], byte[]>>() {
-					StringBuilder sb = new StringBuilder().append(" a message with key='")
-							.append(toDisplayString(ObjectUtils.nullSafeToString(key), 50)).append("'")
-							.append(" and payload='")
-							.append(toDisplayString(ObjectUtils.nullSafeToString(payload), 50))
-							.append("'").append(" received from ")
-							.append(record.partition());
-
-					@Override
-					public void onFailure(Throwable ex) {
-						KafkaMessageChannelBinder.this.logger.error(
-								"Error sending to DLQ " + sb.toString(), ex);
-					}
-
-					@Override
-					public void onSuccess(SendResult<byte[], byte[]> result) {
-						if (KafkaMessageChannelBinder.this.logger.isDebugEnabled()) {
-							KafkaMessageChannelBinder.this.logger.debug(
-									"Sent to DLQ " + sb.toString());
-						}
-					}
-
-				});
+				dlqSender.sendToDlq(record, kafkaHeaders);
 			};
 		}
 		return null;
 	}
 
+	private static void ensureDlqMessageCanBeProperlySerialized(Map<String, String> configuration,
+																Predicate<Map<String, String>> configPredicate,
+																String dataType) {
+			if (CollectionUtils.isEmpty(configuration) || configPredicate.test(configuration)) {
+				throw new IllegalArgumentException("Native decoding is used on the consumer. " +
+						dataType + " is not byte[] and no serializer is set on the DLQ producer.");
+			}
+	}
+
 	private ConsumerFactory<?, ?> createKafkaConsumerFactory(boolean anonymous, String consumerGroup,
 			ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties) {
 		Map<String, Object> props = new HashMap<>();
@@ -613,4 +611,56 @@ public class KafkaMessageChannelBinder extends
 
 	}
 
+	private final class DlqSender<K,V> {
+
+		private final KafkaTemplate<K,V> kafkaTemplate;
+		private final String dlqName;
+
+		DlqSender(KafkaTemplate<K, V> kafkaTemplate, String dlqName) {
+			this.kafkaTemplate = kafkaTemplate;
+			this.dlqName = dlqName;
+		}
+
+		@SuppressWarnings("unchecked")
+		public void sendToDlq(ConsumerRecord<?,?> consumerRecord, Headers headers) {
+			K key = (K)consumerRecord.key();
+			V value = (V)consumerRecord.value();
+			ProducerRecord<K,V> producerRecord = new ProducerRecord<>(this.dlqName, consumerRecord.partition(),
+					key, value, headers);
+
+			StringBuilder sb = new StringBuilder().append(" a message with key='")
+					.append(toDisplayString(ObjectUtils.nullSafeToString(key), 50)).append("'")
+					.append(" and payload='")
+					.append(toDisplayString(ObjectUtils.nullSafeToString(value), 50))
+					.append("'").append(" received from ")
+					.append(consumerRecord.partition());
+			ListenableFuture<SendResult<K, V>> sentDlq = null;
+			try {
+				sentDlq = this.kafkaTemplate.send(producerRecord);
+				sentDlq.addCallback(new ListenableFutureCallback<SendResult<K, V>>() {
+
+					@Override
+					public void onFailure(Throwable ex) {
+						KafkaMessageChannelBinder.this.logger.error(
+								"Error sending to DLQ " + sb.toString(), ex);
+					}
+
+					@Override
+					public void onSuccess(SendResult<K, V> result) {
+						if (KafkaMessageChannelBinder.this.logger.isDebugEnabled()) {
+							KafkaMessageChannelBinder.this.logger.debug(
+									"Sent to DLQ " + sb.toString());
+						}
+					}
+				});
+			}
+			catch (Exception ex) {
+				if (sentDlq == null) {
+					KafkaMessageChannelBinder.this.logger.error(
+							"Error sending to DLQ " + sb.toString(), ex);
+				}
+			}
+
+		}
+	}
 }

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java

@@ -122,9 +122,6 @@ import static org.assertj.core.api.Assertions.fail;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
 
-
-
-
 /**
  * @author Soby Chacko
  * @author Ilayaperumal Gopinathan
@@ -395,6 +392,147 @@ public class KafkaBinderTests extends
 		consumerBinding.unbind();
 	}
 
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testDlqWithNativeSerializationEnabledOnDlqProducer() throws Exception {
+		Binder binder = getBinder();
+		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
+
+		//Native serialization for producer
+		producerProperties.setUseNativeEncoding(true);
+		Map<String, String> producerConfig = new HashMap<>();
+		producerConfig.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+		producerProperties.getExtension().setConfiguration(producerConfig);
+
+		BindingProperties outputBindingProperties = createProducerBindingProperties(
+				producerProperties);
+
+		DirectChannel moduleOutputChannel = createBindableChannel("output",
+				outputBindingProperties);
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+
+		//Native Deserialization for consumer
+		consumerProperties.setUseNativeDecoding(true);
+		Map<String, String> consumerConfig = new HashMap<>();
+		consumerConfig.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+		consumerProperties.getExtension().setConfiguration(consumerConfig);
+
+		//Setting dlq producer properties on the consumer
+		consumerProperties.getExtension().setDlqProducerProperties(producerProperties.getExtension());
+		consumerProperties.getExtension().setEnableDlq(true);
+
+		DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties));
+
+		Binding<MessageChannel> producerBinding = binder.bindProducer("foo.bar",
+				moduleOutputChannel, outputBindingProperties.getProducer());
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.bar",
+				"testDlqWithNativeEncoding-1", moduleInputChannel, consumerProperties);
+
+		// Let the consumer actually bind to the producer before sending a msg
+		binderBindUnbindLatency();
+
+		FailingInvocationCountingMessageHandler handler = new FailingInvocationCountingMessageHandler();
+		moduleInputChannel.subscribe(handler);
+
+		//Consumer for the DLQ destination
+		QueueChannel dlqChannel = new QueueChannel();
+		ExtendedConsumerProperties<KafkaConsumerProperties> dlqConsumerProperties = createConsumerProperties();
+		dlqConsumerProperties.setMaxAttempts(1);
+
+		Binding<MessageChannel> dlqConsumerBinding = binder.bindConsumer(
+				"error.foo.bar." + "testDlqWithNativeEncoding-1", null, dlqChannel, dlqConsumerProperties);
+		binderBindUnbindLatency();
+
+		Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload("foo")
+				.build();
+
+		moduleOutputChannel.send(message);
+
+		Message<?> receivedMessage = receive(dlqChannel, 5);
+		assertThat(receivedMessage).isNotNull();
+		assertThat(receivedMessage.getPayload()).isEqualTo("foo".getBytes());
+		assertThat(handler.getInvocationCount()).isEqualTo(consumerProperties.getMaxAttempts());
+		assertThat(receivedMessage.getHeaders().get(KafkaMessageChannelBinder.X_ORIGINAL_TOPIC))
+				.isEqualTo("foo.bar".getBytes(StandardCharsets.UTF_8));
+		assertThat(new String((byte[]) receivedMessage.getHeaders().get(KafkaMessageChannelBinder.X_EXCEPTION_MESSAGE)))
+				.startsWith("failed to send Message to channel 'input'");
+		assertThat(receivedMessage.getHeaders().get(KafkaMessageChannelBinder.X_EXCEPTION_STACKTRACE))
+				.isNotNull();
+		binderBindUnbindLatency();
+
+		dlqConsumerBinding.unbind();
+
+		producerBinding.unbind();
+		consumerBinding.unbind();
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testDlqWithNativeDecodingOnConsumerButMissingSerializerOnDlqProducer() throws Exception {
+		Binder binder = getBinder();
+		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
+		//Native serialization for producer
+		producerProperties.setUseNativeEncoding(true);
+		Map<String, String> producerConfig = new HashMap<>();
+		producerConfig.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+		producerProperties.getExtension().setConfiguration(producerConfig);
+		BindingProperties outputBindingProperties = createProducerBindingProperties(
+				producerProperties);
+		DirectChannel moduleOutputChannel = createBindableChannel("output",
+				outputBindingProperties);
+
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+		//Native Deserialization for consumer
+		consumerProperties.setUseNativeDecoding(true);
+
+		Map<String, String> consumerConfig = new HashMap<>();
+		consumerConfig.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+
+		//No Dlq producer properties set on the consumer with a native serializer. This should cause an error for DLQ sending.
+
+		consumerProperties.getExtension().setConfiguration(consumerConfig);
+		consumerProperties.getExtension().setEnableDlq(true);
+
+		DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties));
+
+		Binding<MessageChannel> producerBinding = binder.bindProducer("foo.bar",
+				moduleOutputChannel, outputBindingProperties.getProducer());
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.bar",
+				"testDlqWithNativeEncoding-2", moduleInputChannel, consumerProperties);
+
+		// Let the consumer actually bind to the producer before sending a msg
+		binderBindUnbindLatency();
+
+		FailingInvocationCountingMessageHandler handler = new FailingInvocationCountingMessageHandler();
+		moduleInputChannel.subscribe(handler);
+
+		//Consumer for the DLQ destination
+		QueueChannel dlqChannel = new QueueChannel();
+		ExtendedConsumerProperties<KafkaConsumerProperties> dlqConsumerProperties = createConsumerProperties();
+		dlqConsumerProperties.setMaxAttempts(1);
+
+		Binding<MessageChannel> dlqConsumerBinding = binder.bindConsumer(
+				"error.foo.bar." + "testDlqWithNativeEncoding-2", null, dlqChannel, dlqConsumerProperties);
+		binderBindUnbindLatency();
+
+		Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload("foo")
+				.build();
+
+		moduleOutputChannel.send(message);
+
+		Message<?> receivedMessage = receive(dlqChannel, 5);
+		//Ensure that we didn't receive anything on DLQ because of serializer config missing
+		//on dlq producer while native Decoding is enabled.
+		assertThat(receivedMessage).isNull();
+
+		binderBindUnbindLatency();
+
+		dlqConsumerBinding.unbind();
+
+		producerBinding.unbind();
+		consumerBinding.unbind();
+	}
+
 	@Test
 	public void testDlqAndRetry() throws Exception {
 		testDlqGuts(true);
@@ -2307,7 +2445,7 @@ public class KafkaBinderTests extends
 						return future;
 					}
 
-		});
+				});
 
 		moduleOutputChannel.send(message);
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();