diff --git a/pom.xml b/pom.xml index 2c83b127b..5fdc147b2 100644 --- a/pom.xml +++ b/pom.xml @@ -13,7 +13,7 @@ 1.8 2.1.1.BUILD-SNAPSHOT - 3.0.0.RELEASE + 3.0.1.BUILD-SNAPSHOT 1.0.0 2.0.0.BUILD-SNAPSHOT diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index eddc886e9..4bee676d4 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2017 the original author or authors. + * Copyright 2014-2018 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. @@ -27,15 +27,18 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; import java.util.function.Predicate; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.ConsumerRecord; 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.common.PartitionInfo; +import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.header.internals.RecordHeader; import org.apache.kafka.common.header.internals.RecordHeaders; @@ -46,6 +49,7 @@ import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.NoSuchBeanDefinitionException; import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; import org.springframework.cloud.stream.binder.BinderHeaders; +import org.springframework.cloud.stream.binder.DefaultPollableMessageSource; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; @@ -63,9 +67,13 @@ import org.springframework.expression.common.LiteralExpression; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; +import org.springframework.integration.kafka.inbound.KafkaMessageSource; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy; +import org.springframework.integration.support.AcknowledgmentCallback; +import org.springframework.integration.support.AcknowledgmentCallback.Status; import org.springframework.integration.support.ErrorMessageStrategy; +import org.springframework.integration.support.StaticMessageHeaderAccessor; import org.springframework.kafka.KafkaException; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; @@ -87,6 +95,8 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.support.ErrorMessage; import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.transaction.support.TransactionTemplate; import org.springframework.util.Assert; @@ -122,7 +132,7 @@ public class KafkaMessageChannelBinder extends private final KafkaBinderConfigurationProperties configurationProperties; - private final Map topicsInUse = new HashMap<>(); + private final Map topicsInUse = new ConcurrentHashMap<>(); private final KafkaTransactionManager transactionManager; @@ -318,14 +328,8 @@ public class KafkaMessageChannelBinder extends int partitionCount = extendedConsumerProperties.getInstanceCount() * extendedConsumerProperties.getConcurrency(); - Collection allPartitions = provisioningProvider.getPartitionsForTopic(partitionCount, - extendedConsumerProperties.getExtension().isAutoRebalanceEnabled(), - () -> { - Consumer consumer = consumerFactory.createConsumer(); - List partitionsFor = consumer.partitionsFor(destination.getName()); - consumer.close(); - return partitionsFor; - }); + Collection allPartitions = getPartitionInfo(destination, extendedConsumerProperties, + consumerFactory, partitionCount); Collection listenedPartitions; @@ -387,6 +391,68 @@ public class KafkaMessageChannelBinder extends } final KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter = new KafkaMessageDrivenChannelAdapter<>(messageListenerContainer); + kafkaMessageDrivenChannelAdapter.setMessageConverter(getMessageConverter(extendedConsumerProperties)); + kafkaMessageDrivenChannelAdapter.setBeanFactory(this.getBeanFactory()); + ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination, consumerGroup, + extendedConsumerProperties); + if (extendedConsumerProperties.getMaxAttempts() > 1) { + kafkaMessageDrivenChannelAdapter.setRetryTemplate(buildRetryTemplate(extendedConsumerProperties)); + kafkaMessageDrivenChannelAdapter.setRecoveryCallback(errorInfrastructure.getRecoverer()); + } + else { + kafkaMessageDrivenChannelAdapter.setErrorChannel(errorInfrastructure.getErrorChannel()); + } + return kafkaMessageDrivenChannelAdapter; + } + + @Override + protected PolledConsumerResources createPolledConsumerResources(String name, String group, + ConsumerDestination destination, ExtendedConsumerProperties consumerProperties) { + boolean anonymous = !StringUtils.hasText(group); + Assert.isTrue(!anonymous || !consumerProperties.getExtension().isEnableDlq(), + "DLQ support is not available for anonymous subscriptions"); + String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group; + final ConsumerFactory consumerFactory = createKafkaConsumerFactory(anonymous, consumerGroup, + consumerProperties); + KafkaMessageSource source = new KafkaMessageSource<>(consumerFactory, destination.getName()); + source.setMessageConverter(getMessageConverter(consumerProperties)); + source.setRawMessageHeader(consumerProperties.getExtension().isEnableDlq()); + + // I copied this from the regular consumer - it looks bogus to me - includes all partitions + // not just the ones this binding is listening to; doesn't seem right for a health check. + Collection partitionInfos = getPartitionInfo(destination, consumerProperties, consumerFactory, + -1); + this.topicsInUse.put(destination.getName(), new TopicInformation(group, partitionInfos)); + + source.setRebalanceListener(new ConsumerRebalanceListener() { + + @Override + public void onPartitionsRevoked(Collection partitions) { + KafkaMessageChannelBinder.this.logger.info("Revoked: " + partitions); + } + + @Override + public void onPartitionsAssigned(Collection partitions) { + KafkaMessageChannelBinder.this.logger.info("Assigned: " + partitions); + } + + }); + return new PolledConsumerResources(source, + registerErrorInfrastructure(destination, group, consumerProperties, true)); + } + + @Override + protected void postProcessPollableSource(DefaultPollableMessageSource bindingTarget) { + bindingTarget.setAttributesProvider((accessor, message) -> { + Object rawMessage = message.getHeaders().get(KafkaHeaders.RAW_DATA); + if (rawMessage != null) { + accessor.setAttribute(KafkaHeaders.RAW_DATA, rawMessage); + } + }); + } + + private MessagingMessageConverter getMessageConverter( + final ExtendedConsumerProperties extendedConsumerProperties) { MessagingMessageConverter messageConverter; if (extendedConsumerProperties.getExtension().getConverterBeanName() == null) { messageConverter = new MessagingMessageConverter(); @@ -406,6 +472,12 @@ public class KafkaMessageChannelBinder extends throw new IllegalStateException("Converter bean not present in application context", e); } } + messageConverter.setHeaderMapper(getHeaderMapper(extendedConsumerProperties)); + return messageConverter; + } + + private KafkaHeaderMapper getHeaderMapper( + final ExtendedConsumerProperties extendedConsumerProperties) { KafkaHeaderMapper mapper = null; if (this.configurationProperties.getHeaderMapperBeanName() != null) { mapper = getApplicationContext().getBean(this.configurationProperties.getHeaderMapperBeanName(), @@ -429,19 +501,21 @@ public class KafkaMessageChannelBinder extends } mapper = headerMapper; } - messageConverter.setHeaderMapper(mapper); - kafkaMessageDrivenChannelAdapter.setMessageConverter(messageConverter); - kafkaMessageDrivenChannelAdapter.setBeanFactory(this.getBeanFactory()); - ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination, consumerGroup, - extendedConsumerProperties); - if (extendedConsumerProperties.getMaxAttempts() > 1) { - kafkaMessageDrivenChannelAdapter.setRetryTemplate(buildRetryTemplate(extendedConsumerProperties)); - kafkaMessageDrivenChannelAdapter.setRecoveryCallback(errorInfrastructure.getRecoverer()); - } - else { - kafkaMessageDrivenChannelAdapter.setErrorChannel(errorInfrastructure.getErrorChannel()); - } - return kafkaMessageDrivenChannelAdapter; + return mapper; + } + + private Collection getPartitionInfo(final ConsumerDestination destination, + final ExtendedConsumerProperties extendedConsumerProperties, + final ConsumerFactory consumerFactory, int partitionCount) { + Collection allPartitions = provisioningProvider.getPartitionsForTopic(partitionCount, + extendedConsumerProperties.getExtension().isAutoRebalanceEnabled(), + () -> { + Consumer consumer = consumerFactory.createConsumer(); + List partitionsFor = consumer.partitionsFor(destination.getName()); + consumer.close(); + return partitionsFor; + }); + return allPartitions; } @Override @@ -451,8 +525,8 @@ public class KafkaMessageChannelBinder extends @Override protected MessageHandler getErrorMessageHandler(final ConsumerDestination destination, final String group, - final ExtendedConsumerProperties extendedConsumerProperties) { - KafkaConsumerProperties kafkaConsumerProperties = extendedConsumerProperties.getExtension(); + final ExtendedConsumerProperties properties) { + KafkaConsumerProperties kafkaConsumerProperties = properties.getExtension(); if (kafkaConsumerProperties.isEnableDlq()) { KafkaProducerProperties dlqProducerProperties = kafkaConsumerProperties.getDlqProducerProperties(); ProducerFactory producerFactory = this.transactionManager != null @@ -471,7 +545,7 @@ public class KafkaMessageChannelBinder extends final ConsumerRecord record = message.getHeaders() .get(KafkaHeaders.RAW_DATA, ConsumerRecord.class); - if (extendedConsumerProperties.isUseNativeDecoding()) { + if (properties.isUseNativeDecoding()) { if (record != null) { Map configuration = this.transactionManager == null ? dlqProducerProperties.getConfiguration() : this.configurationProperties.getTransaction().getProducer().getConfiguration(); @@ -487,6 +561,10 @@ public class KafkaMessageChannelBinder extends } } + if (record == null) { + this.logger.error("No raw record; cannot send to DLQ: " + message); + return; + } Headers kafkaHeaders = new RecordHeaders(record.headers().toArray()); kafkaHeaders.add(new RecordHeader(X_ORIGINAL_TOPIC, record.topic().getBytes(StandardCharsets.UTF_8))); @@ -503,6 +581,41 @@ public class KafkaMessageChannelBinder extends return null; } + @Override + protected MessageHandler getPolledConsumerErrorMessageHandler(ConsumerDestination destination, String group, + ExtendedConsumerProperties properties) { + if (properties.getExtension().isEnableDlq()) { + return getErrorMessageHandler(destination, group, properties); + } + final MessageHandler superHandler = super.getErrorMessageHandler(destination, group, properties); + return message -> { + ConsumerRecord record = (ConsumerRecord) message.getHeaders().get(KafkaHeaders.RAW_DATA); + if (!(message instanceof ErrorMessage)) { + logger.error("Expected an ErrorMessage, not a " + message.getClass().toString() + " for: " + + message); + } + else if (record == null) { + if (superHandler != null) { + superHandler.handleMessage(message); + } + } + else { + if (message.getPayload() instanceof MessagingException) { + AcknowledgmentCallback ack = StaticMessageHeaderAccessor.getAcknowledgmentCallback( + ((MessagingException) message.getPayload()).getFailedMessage()); + if (ack != null) { + if (isAutoCommitOnError(properties)) { + ack.acknowledge(Status.REJECT); + } + else { + ack.acknowledge(Status.REQUEUE); + } + } + } + } + }; + } + private static void ensureDlqMessageCanBeProperlySerialized(Map configuration, Predicate> configPredicate, String dataType) { diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaTestBinder.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaTestBinder.java index dd25bd4c6..1f83a73f9 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaTestBinder.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaTestBinder.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2017 the original author or authors. + * Copyright 2014-2018 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. @@ -16,7 +16,7 @@ package org.springframework.cloud.stream.binder.kafka; -import org.springframework.cloud.stream.binder.AbstractTestBinder; +import org.springframework.cloud.stream.binder.AbstractPollableConsumerTestBinder; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; @@ -28,7 +28,8 @@ import org.springframework.context.ApplicationContext; * @author Gary Russell */ public abstract class AbstractKafkaTestBinder extends - AbstractTestBinder, ExtendedProducerProperties> { + AbstractPollableConsumerTestBinder, ExtendedProducerProperties> { private ApplicationContext applicationContext; diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index bcd202a16..e5f0e3cfc 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2018 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. @@ -67,11 +67,13 @@ import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.DefaultPollableMessageSource; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.HeaderMode; import org.springframework.cloud.stream.binder.PartitionCapableBinderTests; import org.springframework.cloud.stream.binder.PartitionTestSupport; +import org.springframework.cloud.stream.binder.PollableSource; import org.springframework.cloud.stream.binder.Spy; import org.springframework.cloud.stream.binder.TestUtils; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; @@ -95,6 +97,7 @@ import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandle import org.springframework.integration.kafka.support.KafkaSendFailureException; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; @@ -109,6 +112,7 @@ import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.messaging.SubscribableChannel; @@ -143,7 +147,7 @@ public class KafkaBinderTests extends private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName(); @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10); + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10, "error.pollableDlq.group"); private KafkaTestBinder binder; @@ -2408,6 +2412,62 @@ public class KafkaBinderTests extends consumerBinding.unbind(); } + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Test + public void testPolledConsumer() throws Exception { + KafkaTestBinder binder = getBinder(); + PollableSource inboundBindTarget = new DefaultPollableMessageSource(); + binder.bindPollableConsumer("pollable", "group", inboundBindTarget, createConsumerProperties()); + Map producerProps = KafkaTestUtils.producerProps(embeddedKafka); + KafkaTemplate template = new KafkaTemplate(new DefaultKafkaProducerFactory<>(producerProps)); + template.send("pollable", "testPollable"); + boolean polled = inboundBindTarget.poll(m -> { + assertThat(m.getPayload()).isEqualTo("testPollable"); + }); + int n = 0; + while (n++ < 100 && !polled) { + polled = inboundBindTarget.poll(m -> { + assertThat(m.getPayload()).isEqualTo("testPollable".getBytes()); + }); + } + assertThat(polled).isTrue(); + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Test + public void testPolledConsumerWithDlq() throws Exception { + KafkaTestBinder binder = getBinder(); + PollableSource inboundBindTarget = new DefaultPollableMessageSource(); + ExtendedConsumerProperties properties = createConsumerProperties(); + properties.setMaxAttempts(2); + properties.setBackOffInitialInterval(0); + properties.getExtension().setEnableDlq(true); + Map producerProps = KafkaTestUtils.producerProps(embeddedKafka); + binder.bindPollableConsumer("pollableDlq", "group", inboundBindTarget, properties); + KafkaTemplate template = new KafkaTemplate(new DefaultKafkaProducerFactory<>(producerProps)); + template.send("pollableDlq", "testPollableDLQ"); + try { + int n = 0; + while (n++ < 100) { + inboundBindTarget.poll(m -> { + throw new RuntimeException("test DLQ"); + }); + Thread.sleep(100); + } + } + catch (MessageHandlingException e) { + assertThat(e.getCause().getMessage()).isEqualTo("test DLQ"); + } + Map consumerProps = KafkaTestUtils.consumerProps("dlq", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + ConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + Consumer consumer = cf.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "error.pollableDlq.group"); + ConsumerRecord deadLetter = KafkaTestUtils.getSingleRecord(consumer, "error.pollableDlq.group"); + assertThat(deadLetter).isNotNull(); + assertThat(deadLetter.value()).isEqualTo("testPollableDLQ"); + } + private final class FailingInvocationCountingMessageHandler implements MessageHandler { private int invocationCount; diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java index f89abdbb0..20b6fd572 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2017 the original author or authors. + * Copyright 2015-2018 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. @@ -63,7 +63,7 @@ public class KafkaTestBinder extends AbstractKafkaTestBinder { setApplicationContext(context); binder.setApplicationContext(context); binder.afterPropertiesSet(); - this.setBinder(binder); + this.setPollableConsumerBinder(binder); } catch (Exception e) { throw new RuntimeException(e);