From cda52fe966db7d21e904bf815ead18f70722effb Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 21 Jul 2016 16:58:59 -0400 Subject: [PATCH] GH-125: Introduce `sync` mode for KafkaProducerMH Fixes: GH-125 (https://github.com/spring-projects/spring-integration-kafka/issues/125) * Add `sync` mode for the `KafkaProducerMessageHandler` (`false` by default) to wait for result from the send `Future` * Add `sendTimeout` do not block `Future.get()` forever * Provide test-case based on the `MockProducer` to complete the record with an exception and verify that the `sync` mode works well * Some dependencies upgrade Wrap `Future.get()` `TimeoutException` into `MessageTimeoutException` Fix whitespace after comma --- .../outbound/KafkaProducerMessageHandler.java | 56 +++++++++++++++++-- .../xml/KafkaOutboundAdapterParserTests.java | 51 +++++++++++++++++ 2 files changed, 102 insertions(+), 5 deletions(-) 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 f082e2c161..d902b34ac7 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-2015 the original author or authors. + * Copyright 2013-2016 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,8 +16,12 @@ package org.springframework.integration.kafka.outbound; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; +import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.kafka.core.KafkaTemplate; @@ -25,6 +29,7 @@ import org.springframework.kafka.support.KafkaHeaders; import org.springframework.messaging.Message; import org.springframework.util.Assert; import org.springframework.util.StringUtils; +import org.springframework.util.concurrent.ListenableFuture; /** * Kafka Message Handler. @@ -40,6 +45,8 @@ import org.springframework.util.StringUtils; */ public class KafkaProducerMessageHandler extends AbstractMessageHandler { + private static final long DEFAULT_SEND_TIMEOUT = 10000; + private final KafkaTemplate kafkaTemplate; private EvaluationContext evaluationContext; @@ -50,6 +57,10 @@ public class KafkaProducerMessageHandler extends AbstractMessageHandler { private volatile Expression partitionIdExpression; + private boolean sync; + + private long sendTimeout = DEFAULT_SEND_TIMEOUT; + public KafkaProducerMessageHandler(final KafkaTemplate kafkaTemplate) { Assert.notNull(kafkaTemplate, "kafkaTemplate cannot be null"); this.kafkaTemplate = kafkaTemplate; @@ -71,6 +82,26 @@ public class KafkaProducerMessageHandler extends AbstractMessageHandler { return this.kafkaTemplate; } + /** + * The {@code boolean} indicated if {@link KafkaProducerMessageHandler} + * should wait for send operation results or not. Defaults to {@code false}. + * In {@code sync} mode a downstream send operation exception will be re-thrown. + * @param sync the send mode; async by default. + * @since 2.0.1 + */ + public void setSync(boolean sync) { + this.sync = sync; + } + + /** + * Specify a timeout in milliseconds how long {@link KafkaProducerMessageHandler} + * should wait wait for send operation results. Defaults to 10 seconds. + * @param sendTimeout the timeout to wait for result fo send operation. + */ + public void setSendTimeout(long sendTimeout) { + this.sendTimeout = sendTimeout; + } + @Override protected void onInit() throws Exception { super.onInit(); @@ -94,20 +125,35 @@ public class KafkaProducerMessageHandler extends AbstractMessageHandler { ? this.messageKeyExpression.getValue(this.evaluationContext, message) : message.getHeaders().get(KafkaHeaders.MESSAGE_KEY); + ListenableFuture future; + if (partitionId == null) { if (messageKey == null) { - this.kafkaTemplate.send(topic, (V) message.getPayload()); + future = this.kafkaTemplate.send(topic, (V) message.getPayload()); } else { - this.kafkaTemplate.send(topic, (K) messageKey, (V) message.getPayload()); + future = this.kafkaTemplate.send(topic, (K) messageKey, (V) message.getPayload()); } } else { if (messageKey == null) { - this.kafkaTemplate.send(topic, partitionId, (V) message.getPayload()); + future = this.kafkaTemplate.send(topic, partitionId, (V) message.getPayload()); } else { - this.kafkaTemplate.send(topic, partitionId, (K) messageKey, (V) message.getPayload()); + future = this.kafkaTemplate.send(topic, partitionId, (K) messageKey, (V) message.getPayload()); + } + } + if (this.sync) { + if (this.sendTimeout < 0) { + future.get(); + } + else { + try { + future.get(this.sendTimeout, TimeUnit.MILLISECONDS); + } + catch (TimeoutException te) { + throw new MessageTimeoutException(message, "Timeout waiting for response from KafkaProducer", te); + } } } } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java index 6d32da9126..a201eca256 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java @@ -17,14 +17,29 @@ package org.springframework.integration.kafka.config.xml; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.mockito.Mockito.mock; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeoutException; + +import org.apache.kafka.clients.producer.MockProducer; +import org.apache.kafka.common.serialization.IntegerSerializer; +import org.apache.kafka.common.serialization.StringSerializer; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; import org.springframework.integration.test.util.TestUtils; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -54,4 +69,40 @@ public class KafkaOutboundAdapterParserTests { assertThat(TestUtils.getPropertyValue(messageHandler, "partitionIdExpression.expression")).isEqualTo("2"); } + + @Test + public void testSyncMode() { + MockProducer mockProducer = + new MockProducer<>(false, new IntegerSerializer(), new StringSerializer()); + KafkaTemplate template = new KafkaTemplate<>(() -> mockProducer); + KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler<>(template); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.afterPropertiesSet(); + + handler.setSync(true); + handler.setTopicExpression(new LiteralExpression("foo")); + + Executors.newSingleThreadExecutor() + .submit(() -> { + RuntimeException exception = new RuntimeException("Async Producer Mock exception"); + while (!mockProducer.errorNext(exception)) { + Thread.sleep(100); + } + return null; + }); + + assertThatExceptionOfType(MessageHandlingException.class) + .isThrownBy(() -> handler.handleMessage(new GenericMessage<>("foo"))) + .withMessageContaining("Async Producer Mock exception") + .withCauseExactlyInstanceOf(ExecutionException.class) + .withRootCauseExactlyInstanceOf(RuntimeException.class); + + handler.setSendTimeout(1); + + assertThatExceptionOfType(MessageTimeoutException.class) + .isThrownBy(() -> handler.handleMessage(new GenericMessage<>("foo"))) + .withMessageContaining("Timeout waiting for response from KafkaProducer") + .withCauseExactlyInstanceOf(TimeoutException.class); + } + }