diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaConsumerProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaConsumerProperties.java index b9b78f3c0..e8cc7a741 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaConsumerProperties.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaConsumerProperties.java @@ -31,6 +31,8 @@ public class KafkaConsumerProperties { private KafkaMessageChannelBinder.StartOffset startOffset = null; + private boolean enableDlq = false; + public void setMinPartitionCount(int minPartitionCount) { this.minPartitionCount = minPartitionCount; } @@ -63,4 +65,12 @@ public class KafkaConsumerProperties { public void setStartOffset(KafkaMessageChannelBinder.StartOffset startOffset) { this.startOffset = startOffset; } + + public boolean isEnableDlq() { + return enableDlq; + } + + public void setEnableDlq(boolean enableDlq) { + this.enableDlq = enableDlq; + } } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 3b2a25282..0cf5e39ff 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -32,15 +32,14 @@ import java.util.concurrent.Executors; import java.util.concurrent.ThreadFactory; import java.util.concurrent.atomic.AtomicInteger; -import kafka.admin.AdminUtils; -import kafka.api.OffsetRequest; -import kafka.serializer.Decoder; -import kafka.serializer.DefaultDecoder; -import kafka.utils.ZKStringSerializer$; -import kafka.utils.ZkUtils; import org.I0Itec.zkclient.ZkClient; +import org.apache.kafka.clients.producer.Callback; +import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.serialization.ByteArraySerializer; +import org.apache.kafka.common.utils.Utils; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; @@ -63,12 +62,16 @@ import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.kafka.core.ConnectionFactory; import org.springframework.integration.kafka.core.DefaultConnectionFactory; +import org.springframework.integration.kafka.core.KafkaMessage; import org.springframework.integration.kafka.core.Partition; import org.springframework.integration.kafka.core.ZookeeperConfiguration; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; +import org.springframework.integration.kafka.listener.AcknowledgingMessageListener; import org.springframework.integration.kafka.listener.Acknowledgment; +import org.springframework.integration.kafka.listener.ErrorHandler; import org.springframework.integration.kafka.listener.KafkaMessageListenerContainer; import org.springframework.integration.kafka.listener.KafkaNativeOffsetManager; +import org.springframework.integration.kafka.listener.MessageListener; import org.springframework.integration.kafka.listener.OffsetManager; import org.springframework.integration.kafka.support.KafkaHeaders; import org.springframework.integration.kafka.support.ProducerConfiguration; @@ -90,8 +93,15 @@ import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; +import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; +import kafka.admin.AdminUtils; +import kafka.api.OffsetRequest; +import kafka.serializer.Decoder; +import kafka.serializer.DefaultDecoder; +import kafka.utils.ZKStringSerializer$; +import kafka.utils.ZkUtils; import scala.collection.Seq; /** @@ -105,7 +115,12 @@ import scala.collection.Seq; * @author Mark Fisher * @author Soby Chacko */ -public class KafkaMessageChannelBinder extends AbstractBinder, ExtendedProducerProperties> implements ExtendedPropertiesBinder { +public class KafkaMessageChannelBinder + extends + AbstractBinder, + ExtendedProducerProperties> + implements ExtendedPropertiesBinder, + DisposableBean { public static final ByteArraySerializer BYTE_ARRAY_SERIALIZER = new ByteArraySerializer(); @@ -153,19 +168,20 @@ public class KafkaMessageChannelBinder extends AbstractBinder dlqProducer; + private KafkaExtendedBindingProperties extendedBindingProperties = new KafkaExtendedBindingProperties(); - public KafkaMessageChannelBinder(ZookeeperConnect zookeeperConnect, String brokers, - String zkAddress, String... headersToMap) { + public KafkaMessageChannelBinder(ZookeeperConnect zookeeperConnect, String brokers, String zkAddress, + String... headersToMap) { this.zookeeperConnect = zookeeperConnect; this.brokers = brokers; this.zkAddress = zkAddress; if (headersToMap.length > 0) { - String[] combinedHeadersToMap = Arrays.copyOfRange( - BinderHeaders.STANDARD_HEADERS, 0, + String[] combinedHeadersToMap = Arrays.copyOfRange(BinderHeaders.STANDARD_HEADERS, 0, BinderHeaders.STANDARD_HEADERS.length + headersToMap.length); - System.arraycopy(headersToMap, 0, combinedHeadersToMap, - BinderHeaders.STANDARD_HEADERS.length, headersToMap.length); + System.arraycopy(headersToMap, 0, combinedHeadersToMap, BinderHeaders.STANDARD_HEADERS.length, + headersToMap.length); this.headersToMap = combinedHeadersToMap; } else { @@ -218,8 +234,7 @@ public class KafkaMessageChannelBinder extends AbstractBinder= 'a') && (b <= 'z') || (b >= 'A') && (b <= 'Z') || (b >= '0') && (b <= '9') || (b == '.') || (b == '-') || (b == '_'))) { - throw new IllegalArgumentException("Topic name can only have ASCII alphanumerics, '.', '_' and '-'"); + throw new IllegalArgumentException( + "Topic name can only have ASCII alphanumerics, '.', '_' and '-'"); } } } @@ -313,22 +337,29 @@ public class KafkaMessageChannelBinder extends AbstractBinder doBindConsumer(String name, String group, MessageChannel inputChannel, ExtendedConsumerProperties properties) { - // If the caller provides a consumer group, use it; otherwise an anonymous consumer group - // is generated each time, such that each anonymous binding will receive all messages. - // Consumers reset offsets at the latest time by default, which allows them to receive only + // If the caller provides a consumer group, use it; otherwise an anonymous + // consumer group + // is generated each time, such that each anonymous binding will receive all + // messages. + // Consumers reset offsets at the latest time by default, which allows them to + // receive only // messages sent after they've been bound. That behavior can be changed with the // "resetOffsets" and "startOffset" properties. boolean anonymous = !StringUtils.hasText(group); + Assert.isTrue(!anonymous || !properties.getExtension().isEnableDlq(), + "DLQ support is not available for anonymous subscriptions"); String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group; - // The reference point, if not set explicitly is the latest time for anonymous subscriptions and the - // earliest time for group subscriptions. This allows the latter to receive messages published before the group + // The reference point, if not set explicitly is the latest time for anonymous + // subscriptions and the + // earliest time for group subscriptions. This allows the latter to receive + // messages published before the group // has been created. - long referencePoint = properties.getExtension().getStartOffset() != null ? - properties.getExtension().getStartOffset().getReferencePoint() : (anonymous ? OffsetRequest.LatestTime() : OffsetRequest.EarliestTime()); + long referencePoint = properties.getExtension().getStartOffset() != null + ? properties.getExtension().getStartOffset().getReferencePoint() + : (anonymous ? OffsetRequest.LatestTime() : OffsetRequest.EarliestTime()); return createKafkaConsumer(name, inputChannel, properties, consumerGroup, referencePoint); } - @Override public Binding doBindProducer(String name, MessageChannel moduleOutputChannel, ExtendedProducerProperties properties) { @@ -352,8 +383,10 @@ public class KafkaMessageChannelBinder extends AbstractBinder producerFB = new ProducerFactoryBean<>(producerMetadata, brokers, additionalProps); + additionalProps.put(ProducerConfig.LINGER_MS_CONFIG, + String.valueOf(properties.getExtension().getBatchTimeout())); + ProducerFactoryBean producerFB = new ProducerFactoryBean<>(producerMetadata, brokers, + additionalProps); try { final ProducerConfiguration producerConfiguration = new ProducerConfiguration<>( @@ -361,12 +394,19 @@ public class KafkaMessageChannelBinder extends AbstractBinder producerBinding = new DefaultBinding<>(name, null, moduleOutputChannel, consumer); + DefaultBinding producerBinding = new DefaultBinding<>(name, null, moduleOutputChannel, + consumer); consumer.start(); return producerBinding; } @@ -376,7 +416,8 @@ public class KafkaMessageChannelBinder extends AbstractBinder ensureTopicCreated(final String topicName, final int numPartitions, int replicationFactor) { @@ -388,9 +429,8 @@ public class KafkaMessageChannelBinder extends AbstractBinder brokerList = ZkUtils.getSortedBrokerList(zkClient); - final scala.collection.Map> replicaAssignment = AdminUtils.assignReplicasToBrokers - (brokerList, - numPartitions, replicationFactor, -1, -1); + final scala.collection.Map> replicaAssignment = AdminUtils + .assignReplicasToBrokers(brokerList, numPartitions, replicationFactor, -1, -1); retryOperations.execute(new RetryCallback() { @Override @@ -401,21 +441,22 @@ public class KafkaMessageChannelBinder extends AbstractBinder partitions = retryOperations.execute(new RetryCallback, Exception>() { + Collection partitions = retryOperations + .execute(new RetryCallback, Exception>() { - @Override - public Collection doWithRetry(RetryContext context) throws Exception { - connectionFactory.refreshMetadata(Collections.singleton(topicName)); - Collection partitions = connectionFactory.getPartitions(topicName); - if (partitions.size() < numPartitions) { - throw new IllegalStateException("The number of expected partitions was: " + numPartitions - + ", but " + - partitions.size() + " have been found instead"); - } - connectionFactory.getLeaders(partitions); - return partitions; - } - }); + @Override + public Collection doWithRetry(RetryContext context) throws Exception { + connectionFactory.refreshMetadata(Collections.singleton(topicName)); + Collection partitions = connectionFactory.getPartitions(topicName); + if (partitions.size() < numPartitions) { + throw new IllegalStateException( + "The number of expected partitions was: " + numPartitions + ", but " + + partitions.size() + " have been found instead"); + } + connectionFactory.getLeaders(partitions); + return partitions; + } + }); return partitions; } catch (Exception e) { @@ -438,7 +479,7 @@ public class KafkaMessageChannelBinder extends AbstractBinder allPartitions = ensureTopicCreated(name, numPartitions, replicationFactor); Decoder valueDecoder = new DefaultDecoder(null); @@ -466,8 +507,8 @@ public class KafkaMessageChannelBinder extends AbstractBinder() { + @Override + public Object doWithRetry(RetryContext context) { + originalMessageListener.onMessage(message); + return null; + } + }); + } + catch (Throwable throwable) { + if (throwable instanceof RuntimeException) { + throw (RuntimeException) throwable; + } + else { + throw new RuntimeException(throwable); + } + } + } + }); + } + else { + messageListenerContainer.setMessageListener(new AcknowledgingMessageListener() { + final AcknowledgingMessageListener originalMessageListener = (AcknowledgingMessageListener) messageListenerContainer + .getMessageListener(); + @Override + public void onMessage(final KafkaMessage message, final Acknowledgment acknowledgment) { + retryTemplate.execute(new RetryCallback() { + @Override + public Object doWithRetry(RetryContext context) { + originalMessageListener.onMessage(message, acknowledgment); + return null; + } + }); + } + }); + } + } + + if (properties.getExtension().isEnableDlq()) { + final String dlqTopic = "error." + name + "." + group; + initDlqProducer(); + messageListenerContainer.setErrorHandler(new ErrorHandler() { + @Override + public void handle(Exception thrownException, final KafkaMessage message) { + final byte[] key = message.getMessage().key() != null ? Utils.toArray(message.getMessage().key()) + : null; + final byte[] payload = message.getMessage().payload() != null + ? Utils.toArray(message.getMessage().payload()) : null; + dlqProducer.send(new ProducerRecord<>(dlqTopic, key, payload), new Callback() { + @Override + public void onCompletion(RecordMetadata metadata, Exception exception) { + StringBuffer messageLog = new StringBuffer(); + messageLog.append(" a message with key='" + + toDisplayString(ObjectUtils.nullSafeToString(key), 50) + "'"); + messageLog.append(" and payload='" + + toDisplayString(ObjectUtils.nullSafeToString(payload), 50) + "'"); + messageLog.append(" received from " + message.getMetadata().getPartition()); + if (exception != null) { + logger.error("Error sending to DLQ" + messageLog.toString(), exception); + } + else { + if (logger.isDebugEnabled()) { + logger.debug("Sent to DLQ " + messageLog.toString()); + } + } + } + }); + } + }); + } + kafkaMessageDrivenChannelAdapter.start(); EventDrivenConsumer edc = new EventDrivenConsumer(bridge, rh) { @@ -529,12 +650,39 @@ public class KafkaMessageChannelBinder extends AbstractBinder producerMetadata = new ProducerMetadata<>("dlqKafkaProducer", + byte[].class, byte[].class, BYTE_ARRAY_SERIALIZER, BYTE_ARRAY_SERIALIZER); + producerMetadata.setSync(false); + producerMetadata.setCompressionType(ProducerMetadata.CompressionType.none); + producerMetadata.setBatchBytes(16384); + Properties additionalProps = new Properties(); + additionalProps.put(ProducerConfig.ACKS_CONFIG, String.valueOf(requiredAcks)); + additionalProps.put(ProducerConfig.LINGER_MS_CONFIG, + String.valueOf(0)); + ProducerFactoryBean producerFactoryBean = new ProducerFactoryBean<>( + producerMetadata, brokers, additionalProps); + dlqProducer = producerFactoryBean.getObject(); + } + } + } + } + catch (Exception e) { + throw new RuntimeException("Cannot initialize DLQ producer:", e); + } + } + private OffsetManager createOffsetManager(String group, long referencePoint) { try { - KafkaNativeOffsetManager kafkaOffsetManager = - new KafkaNativeOffsetManager(connectionFactory, zookeeperConnect, - Collections.emptyMap()); + KafkaNativeOffsetManager kafkaOffsetManager = new KafkaNativeOffsetManager(connectionFactory, + zookeeperConnect, Collections.emptyMap()); kafkaOffsetManager.setConsumerId(group); kafkaOffsetManager.setReferenceTimestamp(referencePoint); kafkaOffsetManager.afterPropertiesSet(); @@ -552,14 +700,22 @@ public class KafkaMessageChannelBinder extends AbstractBinder messageHeadersList) { Iterator iterator = messageHeadersList.iterator(); while (iterator.hasNext()) { MessageHeaders headers = iterator.next(); Acknowledgment acknowledgment = (Acknowledgment) headers.get(KafkaHeaders.ACKNOWLEDGMENT); - Assert.notNull(acknowledgment, "Acknowledgement shouldn't be null when acknowledging kafka message " + - "manually."); + Assert.notNull(acknowledgment, + "Acknowledgement shouldn't be null when acknowledging kafka message " + "manually."); acknowledgment.acknowledge(); } } @@ -576,7 +732,7 @@ public class KafkaMessageChannelBinder extends AbstractBinder consumerProperties; + private final ExtendedConsumerProperties consumerProperties; public ReceivingHandler(ExtendedConsumerProperties consumerProperties) { this.consumerProperties = consumerProperties; @@ -587,8 +743,7 @@ public class KafkaMessageChannelBinder extends AbstractBinder requestMessage) { if (HeaderMode.embeddedHeaders.equals(consumerProperties.getHeaderMode())) { MessageValues messageValues = extractMessageValues(requestMessage); - return MessageBuilder.createMessage(messageValues.getPayload(), new KafkaBinderHeaders( - messageValues)); + return MessageBuilder.createMessage(messageValues.getPayload(), new KafkaBinderHeaders(messageValues)); } else { return requestMessage; @@ -603,7 +758,6 @@ public class KafkaMessageChannelBinder extends AbstractBinder properties, - int numberOfPartitions, - ProducerConfiguration producerConfiguration) { + int numberOfPartitions, ProducerConfiguration producerConfiguration) { this.topicName = topicName; producerProperties = properties; this.numberOfKafkaPartitions = numberOfPartitions; ConfigurableListableBeanFactory beanFactory = KafkaMessageChannelBinder.this.getBeanFactory(); this.setBeanFactory(beanFactory); this.producerConfiguration = producerConfiguration; - this.partitionHandler = new PartitionHandler(beanFactory, evaluationContext, partitionSelector, - properties); + this.partitionHandler = new PartitionHandler(beanFactory, evaluationContext, partitionSelector, properties); } @Override @@ -655,8 +807,7 @@ public class KafkaMessageChannelBinder extends AbstractBinder producerProperties = createProducerProperties(); + producerProperties.setPartitionCount(10); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.setMaxAttempts(3); + consumerProperties.setBackOffInitialInterval(100); + consumerProperties.setBackOffMaxInterval(150); + consumerProperties.getExtension().setMinPartitionCount(10); + consumerProperties.getExtension().setEnableDlq(true); + long uniqueBindingId = System.currentTimeMillis(); + Binding producerBinding = binder.bindProducer("retryTest." + uniqueBindingId + ".0", + moduleOutputChannel, producerProperties); + Binding consumerBinding = binder.bindConsumer("retryTest." + uniqueBindingId + ".0", "testGroup", + moduleInputChannel, consumerProperties); + + ExtendedConsumerProperties dlqConsumerProperties = createConsumerProperties(); + dlqConsumerProperties.setMaxAttempts(1); + + Binding dlqConsumerBinding = binder.bindConsumer( + "error.retryTest." + uniqueBindingId + ".0.testGroup", null, dlqChannel, dlqConsumerProperties); + + String testMessagePayload = "test." + UUID.randomUUID().toString(); + Message testMessage = MessageBuilder.withPayload(testMessagePayload).build(); + moduleOutputChannel.send(testMessage); + + Message receivedMessage = receive(dlqChannel, 3); + assertNotNull(receivedMessage); + assertEquals(testMessagePayload, receivedMessage.getPayload()); + assertThat(handler.getInvocationCount(), equalTo(consumerProperties.getMaxAttempts())); + dlqConsumerBinding.unbind(); + consumerBinding.unbind(); + producerBinding.unbind(); + } + @Test(expected = IllegalArgumentException.class) public void testValidateKafkaTopicName() { KafkaMessageChannelBinder.validateTopicName("foo:bar"); @@ -400,4 +445,22 @@ public class KafkaBinderTests extends PartitionCapableBinderTests message) throws MessagingException { + invocationCount++; + throw new RuntimeException(); + } + + public int getInvocationCount() { + return invocationCount; + } + } } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java index 19b3c7c70..c0d913ce5 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java @@ -63,7 +63,17 @@ public abstract class AbstractBinderTests receive(PollableChannel channel) { - return channel.receive((int)(1000 * timeoutMultiplier)); + return receive(channel, 1); + } + + /** + * Attempt to receive a message on the given channel, + * waiting up to 1s * additionalMultiplier * {@link #timeoutMultiplier}). + * + * Allows accomodating tests which are slower than normal (e.g. retry). + */ + protected Message receive(PollableChannel channel, int additionalMultiplier) { + return channel.receive((int)(1000 * timeoutMultiplier * additionalMultiplier)); } @Test diff --git a/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc b/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc index 9a048e6c5..831241ab9 100644 --- a/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc +++ b/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc @@ -863,6 +863,12 @@ minPartitionCount:: The minimum number of partitions expected by the consumer if it creates the consumed topic automatically. + Default: `1`. +enableDlq:: + When set to true, it will send enable DLQ behavior for the consumer. + Messages that result in errors will be forwarded to a topic named `error..`. + This provides an alternative option to the more common Kafka replay scenario for the case when the number of errors is relatively small and replaying the entire original topic may be too cumbersome. ++ +Default: `false`. ==== Kafka Producer Properties