From 7fae571c2ddccff8f35cff4c94d4156bedbd01b7 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 26 Feb 2019 12:36:25 -0500 Subject: [PATCH] Changes for compatibility with Spring Kafka 2.3 Also remove test for bad partition in DSL tests - the send now gives an error about the topic not being present in metadata instead of "not in range". Use sync for publishing in DSL tests, to fail fast. --- .../kafka/inbound/KafkaMessageSource.java | 49 ++++++++++++++++--- .../outbound/KafkaProducerMessageHandler.java | 3 +- .../integration/kafka/dsl/KafkaDslTests.java | 19 ++++--- .../inbound/MessageDrivenAdapterTests.java | 5 +- .../kafka/inbound/MessageSourceTests.java | 9 ++-- .../kafka/dsl/KafkaDslKotlinTests.kt | 20 +++++--- 6 files changed, 74 insertions(+), 31 deletions(-) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java index edaaff9310..00793befa3 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-2019 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. @@ -42,7 +42,6 @@ import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.WakeupException; -import org.springframework.beans.factory.DisposableBean; import org.springframework.context.Lifecycle; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.acks.AcknowledgmentCallback; @@ -56,6 +55,7 @@ import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.converter.KafkaMessageHeaders; import org.springframework.kafka.support.converter.MessagingMessageConverter; import org.springframework.kafka.support.converter.RecordMessageConverter; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -80,8 +80,7 @@ import org.springframework.util.Assert; * @since 3.0.1 * */ -public class KafkaMessageSource extends AbstractMessageSource - implements DisposableBean, Lifecycle { +public class KafkaMessageSource extends AbstractMessageSource implements Lifecycle { private static final long DEFAULT_POLL_TIMEOUT = 50L; @@ -116,6 +115,8 @@ public class KafkaMessageSource extends AbstractMessageSource private boolean rawMessageHeader; + private Duration commitTimeout; + private volatile Consumer consumer; private boolean running; @@ -232,6 +233,20 @@ public class KafkaMessageSource extends AbstractMessageSource this.rawMessageHeader = rawMessageHeader; } + protected Duration getCommitTimeout() { + return this.commitTimeout; + } + + /** + * Set the timeout for commits. + * @param commitTimeout the timeout. + * @since 3.2 + */ + public void setCommitTimeout(Duration commitTimeout) { + this.commitTimeout = commitTimeout; + this.ackCallbackFactory.setCommitTimeout(commitTimeout); + } + private ConsumerFactory fixOrRejectConsumerFactory(ConsumerFactory suppliedConsumerFactory) { Object maxPoll = suppliedConsumerFactory.getConfigurationProperties() .get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG); @@ -372,9 +387,15 @@ public class KafkaMessageSource extends AbstractMessageSource */ public static class KafkaAckCallbackFactory implements AcknowledgmentCallbackFactory> { + private Duration commitTimeout; + + public void setCommitTimeout(Duration commitTimeout) { + this.commitTimeout = commitTimeout; + } + @Override public AcknowledgmentCallback createCallback(KafkaAckInfo info) { - return new KafkaAckCallback<>(info); + return new KafkaAckCallback<>(info, this.commitTimeout); } } @@ -391,13 +412,20 @@ public class KafkaMessageSource extends AbstractMessageSource private final KafkaAckInfo ackInfo; + private final Duration commitTimeout; + private volatile boolean acknowledged; private boolean autoAckEnabled = true; public KafkaAckCallback(KafkaAckInfo ackInfo) { + this(ackInfo, null); + } + + public KafkaAckCallback(KafkaAckInfo ackInfo, @Nullable Duration commitTimeout) { Assert.notNull(ackInfo, "'ackInfo' cannot be null"); this.ackInfo = ackInfo; + this.commitTimeout = commitTimeout; } @Override @@ -494,8 +522,15 @@ public class KafkaMessageSource extends AbstractMessageSource this.ackInfo.setAckDeferred(true); } if (ackInfo != null) { - ackInfo.getConsumer().commitSync(Collections.singletonMap(ackInfo.getTopicPartition(), - new OffsetAndMetadata(ackInfo.getRecord().offset() + 1))); + Map offset = + Collections.singletonMap(ackInfo.getTopicPartition(), + new OffsetAndMetadata(ackInfo.getRecord().offset() + 1)); + if (this.commitTimeout == null) { + ackInfo.getConsumer().commitSync(offset); + } + else { + ackInfo.getConsumer().commitSync(offset, this.commitTimeout); + } } else { if (this.logger.isDebugEnabled()) { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java index 28ae87729a..b66b5641d2 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2018 the original author or authors. + * Copyright 2013-2019 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. @@ -460,6 +460,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes public void processSendResult(final Message message, final ProducerRecord producerRecord, ListenableFuture> future, MessageChannel metadataChannel) throws InterruptedException, ExecutionException { + if (getSendFailureChannel() != null || metadataChannel != null) { future.addCallback(new ListenableFutureCallback>() { diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java index d161ac1540..ab93f61d9a 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2018 the original author or authors. + * Copyright 2015-2019 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. @@ -17,7 +17,6 @@ package org.springframework.integration.kafka.dsl; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.util.Collection; import java.util.Collections; @@ -28,6 +27,7 @@ import java.util.stream.Stream; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.TopicPartition; import org.junit.ClassRule; import org.junit.Test; @@ -44,7 +44,6 @@ import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.Pollers; -import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; @@ -150,11 +149,6 @@ public class KafkaDslTests { @Test public void testKafkaAdapters() throws Exception { - assertThatThrownBy(() -> this.sendToKafkaFlowInput.send(new GenericMessage<>("foo"))) - .hasMessageContaining("10 is not in the range"); - - this.kafkaProducer1.setPartitionIdExpression(new ValueExpression<>(0)); - this.kafkaProducer2.setPartitionIdExpression(new ValueExpression<>(0)); this.sendToKafkaFlowInput.send(new GenericMessage<>("foo", Collections.singletonMap("foo", "bar"))); assertThat(TestUtils.getPropertyValue(this.kafkaProducer1, "headerMapper")).isSameAs(this.mapper); @@ -298,7 +292,9 @@ public class KafkaDslTests { @Bean public ProducerFactory producerFactory() { - return new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka())); + Map props = KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka()); + props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000); + return new DefaultKafkaProducerFactory<>(props); } @Bean @@ -324,13 +320,15 @@ public class KafkaDslTests { private KafkaProducerMessageHandlerSpec kafkaMessageHandler( ProducerFactory producerFactory, String topic) { + return Kafka .outboundChannelAdapter(producerFactory) + .sync(true) .messageKey(m -> m .getHeaders() .get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)) .headerMapper(mapper()) - .partitionId(m -> 10) + .partitionId(m -> 0) .topicExpression("headers[kafka_topic] ?: '" + topic + "'") .configureKafkaTemplate(t -> t.id("kafkaTemplate:" + topic)); } @@ -353,6 +351,7 @@ public class KafkaDslTests { public IntegrationFlow outboundGateFlow() { return IntegrationFlows.from(Gate.class) .handle(Kafka.outboundGateway(producerFactory(), replyContainer()) + .sync(true) .configureKafkaTemplate(t -> t.replyTimeout(30_000))) .get(); } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java index cdeba539cd..142508849e 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java @@ -18,6 +18,7 @@ package org.springframework.integration.kafka.inbound; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyMap; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.BDDMockito.given; @@ -461,7 +462,7 @@ public class MessageDrivenAdapterTests { willAnswer(i -> { commitLatch.countDown(); return null; - }).given(consumer).commitSync(any(Map.class)); + }).given(consumer).commitSync(anyMap(), any()); given(consumer.assignment()).willReturn(records.keySet()); final CountDownLatch pauseLatch = new CountDownLatch(1); willAnswer(i -> { @@ -488,7 +489,7 @@ public class MessageDrivenAdapterTests { adapter.afterPropertiesSet(); adapter.start(); assertThat(commitLatch.await(10, TimeUnit.SECONDS)).isTrue(); - verify(consumer, times(2)).commitSync(any(Map.class)); + verify(consumer, times(2)).commitSync(anyMap(), any()); assertThat(outputChannel.getQueueSize()).isEqualTo(2); adapter.pause(); assertThat(pauseLatch.await(10, TimeUnit.SECONDS)).isTrue(); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java index c807bab1ef..f5f73c6437 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-2019 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. @@ -268,6 +268,7 @@ public class MessageSourceTests { .getConfigurationProperties(); given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer); KafkaMessageSource source = new KafkaMessageSource(consumerFactory, "foo"); + source.setCommitTimeout(Duration.ofSeconds(30)); Message received = source.receive(); assertThat(received.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(0L); @@ -292,11 +293,13 @@ public class MessageSourceTests { inOrder.verify(consumer).poll(any(Duration.class)); inOrder.verify(consumer).seek(topicPartition, 0L); // rollback inOrder.verify(consumer).poll(any(Duration.class)); - inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(1L))); + inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(1L)), + Duration.ofSeconds(30)); inOrder.verify(consumer).poll(any(Duration.class)); inOrder.verify(consumer).seek(topicPartition, 1L); // rollback inOrder.verify(consumer).poll(any(Duration.class)); - inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(2L))); + inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(2L)), + Duration.ofSeconds(30)); inOrder.verify(consumer).poll(any(Duration.class)); inOrder.verify(consumer).close(); inOrder.verifyNoMoreInteractions(); diff --git a/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt b/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt index 4d18c31e27..efbf18320d 100644 --- a/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt +++ b/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt @@ -25,8 +25,10 @@ import assertk.assertions.isNull import assertk.assertions.isSameAs import assertk.assertions.isTrue import assertk.catch +import kafka.tools.ConsoleProducer import org.apache.kafka.clients.consumer.ConsumerConfig import org.apache.kafka.clients.consumer.ConsumerRebalanceListener +import org.apache.kafka.clients.producer.ProducerConfig import org.apache.kafka.common.TopicPartition import org.junit.ClassRule import org.junit.Test @@ -80,6 +82,7 @@ import java.util.stream.Stream /** * @author Artem Bilan + * @author Gary Russell * * @since 3.0.3 */ @@ -152,11 +155,6 @@ class KafkaDslKotlinTests { @Test fun testKafkaAdapters() { - val exception = catch { this.sendToKafkaFlowInput.send(GenericMessage("foo")) } - assertThat(exception!!.message).isNotNull().contains("10 is not in the range") - - this.kafkaProducer1.setPartitionIdExpression(ValueExpression(0)) - this.kafkaProducer2.setPartitionIdExpression(ValueExpression(0)) this.sendToKafkaFlowInput.send(GenericMessage("foo", hashMapOf("foo" to "bar"))) assertThat(TestUtils.getPropertyValue(this.kafkaProducer1, "headerMapper")).isSameAs(this.mapper) @@ -280,7 +278,11 @@ class KafkaDslKotlinTests { .get() @Bean - fun producerFactory() = DefaultKafkaProducerFactory(KafkaTestUtils.producerProps(embeddedKafka.embeddedKafka)) + fun producerFactory(): DefaultKafkaProducerFactory { + val props = KafkaTestUtils.producerProps(embeddedKafka.embeddedKafka) + props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "10000") + return DefaultKafkaProducerFactory(props) + } @Bean fun sendToKafkaFlow() = @@ -309,7 +311,8 @@ class KafkaDslKotlinTests { Kafka.outboundChannelAdapter(producerFactory) .messageKey { m -> m.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] } .headerMapper(mapper()) - .partitionId { _ -> 10 } + .sync(true) + .partitionId { _ -> 0 } .topicExpression("headers[kafka_topic] ?: '$topic'") .configureKafkaTemplate { t -> t.id("kafkaTemplate:$topic") } @@ -334,7 +337,8 @@ class KafkaDslKotlinTests { @Bean fun outboundGateFlow() = IntegrationFlows.from(Gate::class.java) - .handle(Kafka.outboundGateway(replyingKafkaTemplate())) + .handle(Kafka.outboundGateway(replyingKafkaTemplate()) + .sync(true)) .get() private fun replyContainer(): GenericMessageListenerContainer {