diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaParsingUtils.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaParsingUtils.java index 8c022f00f1..c8d6a2f3c2 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaParsingUtils.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaParsingUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 2019-2020 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. @@ -83,6 +83,13 @@ public final class KafkaParsingUtils { builder.addPropertyValue("timestampExpression", timestampExpressionDef); } + BeanDefinition flushExpressionDef = + IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("flush-expression", element); + + if (flushExpressionDef != null) { + builder.addPropertyValue("flushExpression", flushExpressionDef); + } + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "header-mapper"); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java index 67d543f842..13fcddd19a 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2020 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. @@ -86,8 +86,8 @@ public class KafkaProducerMessageHandlerSpec * {@code * .topic(m -> m.getPayload().getTopic()) @@ -133,8 +133,9 @@ public class KafkaProducerMessageHandlerSpec * {@code * .messageKey(m -> m.getPayload().getKey()) @@ -169,8 +170,9 @@ public class KafkaProducerMessageHandlerSpec * {@code * .partitionId(m -> m.getHeaders().get("partitionId", Integer.class)) @@ -206,14 +208,15 @@ public class KafkaProducerMessageHandlerSpec * {@code * .timestamp(m -> m.getHeaders().get("mytimestamp_header", Long.class)) * } * - * @param timestampFunction the partitionId function. + * @param timestampFunction the timestamp function. * @param

the expected payload type. * @return the spec. */ @@ -232,6 +235,46 @@ public class KafkaProducerMessageHandlerSpec + * {@code + * .flush(m -> m.getPayload().shouldFlush()) + * } + * + * @param flushFunction the flush function. + * @param

the expected payload type. + * @return the spec. + */ + public

S flush(Function, Boolean> flushFunction) { + return flushExpression(new FunctionExpression<>(flushFunction)); + } + + /** + * Configure an {@link Expression} to determine whether or not to flush the producer + * after a send. By default the producer is flushed if a header {@code kafka_flush} + * has a value {@link Boolean#TRUE}. + * @param flushExpression the timestamp expression to use. + * @return the spec. + */ + public S flushExpression(Expression flushExpression) { + this.target.setFlushExpression(flushExpression); + return _this(); + } /** * A {@code boolean} indicating if the {@link KafkaProducerMessageHandler} 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 f53f31ed00..c31bdee2f6 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 @@ -41,8 +41,10 @@ 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.expression.FunctionExpression; import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.kafka.support.KafkaIntegrationHeaders; import org.springframework.integration.kafka.support.KafkaSendFailureException; import org.springframework.integration.support.DefaultErrorMessageStrategy; import org.springframework.integration.support.ErrorMessageStrategy; @@ -124,6 +126,9 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes private Expression timestampExpression; + private Expression flushExpression = new FunctionExpression>(message -> + Boolean.TRUE.equals(message.getHeaders().get(KafkaIntegrationHeaders.FLUSH))); + private boolean sync; private Expression sendTimeoutExpression = new ValueExpression<>(DEFAULT_SEND_TIMEOUT); @@ -196,6 +201,19 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes this.timestampExpression = timestampExpression; } + /** + * Specify a SpEL expression that evaluates to a {@link Boolean} to determine whether + * the producer should be flushed after the send. Defaults to looking for a + * {@link Boolean} value in a {@link KafkaIntegrationHeaders#FLUSH} header; false if + * absent. + * @param flushExpression the {@link Expression}. + * @since 3.3 + */ + public void setFlushExpression(Expression flushExpression) { + Assert.notNull(flushExpression, "'flushExpression' cannot be null"); + this.flushExpression = flushExpression; + } + /** * Set the header mapper to use. * @param headerMapper the mapper; can be null to disable header mapping. @@ -390,12 +408,16 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes @Override protected Object handleRequestMessage(final Message message) { final ProducerRecord producerRecord; + boolean flush = this.flushExpression.getValue(this.evaluationContext, message, Boolean.class); boolean preBuilt = message.getPayload() instanceof ProducerRecord; if (preBuilt) { producerRecord = (ProducerRecord) message.getPayload(); } else { producerRecord = createProducerRecord(message); + if (flush) { + producerRecord.headers().remove(KafkaIntegrationHeaders.FLUSH); + } } ListenableFuture> sendFuture; RequestReplyFuture gatewayFuture = null; @@ -426,6 +448,9 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes catch (ExecutionException e) { throw new MessageHandlingException(message, e.getCause()); } + if (flush) { + this.kafkaTemplate.flush(); + } return processReplyFuture(gatewayFuture); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaIntegrationHeaders.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaIntegrationHeaders.java new file mode 100644 index 0000000000..557705f7d8 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaIntegrationHeaders.java @@ -0,0 +1,38 @@ +/* + * Copyright 2020 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.support; + +import org.springframework.kafka.support.KafkaHeaders; + +/** + * Headers specifically for Spring Integration components. + * + * @author Gary Russell + * @since 3.3 + * + */ +public final class KafkaIntegrationHeaders { + + private KafkaIntegrationHeaders() { + } + + /** + * Set to {@link Boolean#TRUE} to flush after sending. + */ + public static final String FLUSH = KafkaHeaders.PREFIX + "flush"; + +} diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka.xsd index 70c7224a24..699c4003f2 100644 --- a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka.xsd +++ b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka.xsd @@ -568,6 +568,15 @@ ]]> + + + + + - + 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 997cceaddd..00229017f0 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 @@ -74,6 +74,8 @@ class KafkaOutboundAdapterParserTests { assertThat(TestUtils.getPropertyValue(messageHandler, "sendTimeoutExpression.expression")).isEqualTo("1000"); assertThat(TestUtils.getPropertyValue(messageHandler, "timestampExpression.expression")) .isEqualTo("T(System).currentTimeMillis()"); + assertThat(TestUtils.getPropertyValue(messageHandler, "flushExpression.expression")) + .isEqualTo("headers['foo']"); assertThat(TestUtils.getPropertyValue(messageHandler, "errorMessageStrategy")) .isSameAs(this.appContext.getBean("ems")); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests-context.xml index 31123d4839..f6484acec5 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests-context.xml +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests-context.xml @@ -21,6 +21,7 @@ send-success-channel="successes" send-failure-channel="failures" send-timeout-expression="44" + flush-expression="headers['foo']" sync="true" timestamp-expression="T(System).currentTimeMillis()" topic-expression="'topic'" diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests.java index 9804875ee7..d1eb64be7e 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 2019-2020 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. @@ -60,6 +60,8 @@ public class KafkaOutboundGatewayParserTests { assertThat(TestUtils.getPropertyValue(this.messageHandler, "sendTimeoutExpression.expression")).isEqualTo("44"); assertThat(TestUtils.getPropertyValue(this.messageHandler, "timestampExpression.expression")) .isEqualTo("T(System).currentTimeMillis()"); + assertThat(TestUtils.getPropertyValue(this.messageHandler, "flushExpression.expression")) + .isEqualTo("headers['foo']"); assertThat(TestUtils.getPropertyValue(this.messageHandler, "errorMessageStrategy")) .isSameAs(this.context.getBean("ems")); 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 e3ab95f6a1..fc7786bed5 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 @@ -338,6 +338,7 @@ public class KafkaDslTests { e -> e.id("kafkaProducer1"))) .subscribe(sf -> sf.handle( kafkaMessageHandler(producerFactory(), TEST_TOPIC2) + .flush(msg -> true) .timestamp(m -> 1487694048644L), e -> e.id("kafkaProducer2"))) ); @@ -381,6 +382,7 @@ public class KafkaDslTests { public IntegrationFlow outboundGateFlow() { return IntegrationFlows.from(Gate.class) .handle(Kafka.outboundGateway(producerFactory(), replyContainer()) + .flushExpression("true") .sync(true) .configureKafkaTemplate(t -> t.defaultReplyTimeout(Duration.ofSeconds(30)))) .get(); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java index 533b1b6a0c..669083e23d 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java @@ -71,6 +71,7 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.expression.FunctionExpression; import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; +import org.springframework.integration.kafka.support.KafkaIntegrationHeaders; import org.springframework.integration.kafka.support.KafkaSendFailureException; import org.springframework.integration.support.MessageBuilder; import org.springframework.kafka.core.ConsumerFactory; @@ -688,4 +689,49 @@ class KafkaProducerMessageHandlerTests { verify(template).send(any(ProducerRecord.class)); } + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Test + void testFlush() { + ProducerFactory pf = mock(ProducerFactory.class); + Producer producer = mock(Producer.class); + given(pf.createProducer()).willReturn(producer); + ListenableFuture future = mock(ListenableFuture.class); + willReturn(future).given(producer).send(any(ProducerRecord.class), any(Callback.class)); + KafkaTemplate template = new KafkaTemplate(pf); + KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(template); + handler.setTopicExpression(new LiteralExpression("bar")); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.afterPropertiesSet(); + handler.start(); + handler.handleMessage( + new GenericMessage<>("foo", Collections.singletonMap(KafkaIntegrationHeaders.FLUSH, Boolean.TRUE))); + InOrder inOrder = inOrder(producer); + ArgumentCaptor captor = ArgumentCaptor.forClass(ProducerRecord.class); + inOrder.verify(producer).send(captor.capture(), any(Callback.class)); + inOrder.verify(producer).flush(); + handler.stop(); + assertThat(captor.getValue().headers().lastHeader(KafkaIntegrationHeaders.FLUSH)).isNull(); + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Test + void testNoFlush() { + ProducerFactory pf = mock(ProducerFactory.class); + Producer producer = mock(Producer.class); + given(pf.createProducer()).willReturn(producer); + ListenableFuture future = mock(ListenableFuture.class); + willReturn(future).given(producer).send(any(ProducerRecord.class), any(Callback.class)); + KafkaTemplate template = new KafkaTemplate(pf); + KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(template); + handler.setTopicExpression(new LiteralExpression("bar")); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.afterPropertiesSet(); + handler.start(); + handler.handleMessage(new GenericMessage<>("foo")); + InOrder inOrder = inOrder(producer); + inOrder.verify(producer).send(any(ProducerRecord.class), any(Callback.class)); + inOrder.verify(producer, never()).flush(); + handler.stop(); + } + }