Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Support flush after send

Browse Source 
* Add `flushExpression`.

* Add DSL, XML configuration.
This commit is contained in:
Gary Russell
2020-03-23 11:53:00 -04:00
committed by Artem Bilan
parent e0be544a40
commit 1e484d0fad
11 changed files with 189 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaParsingUtils.java
View File

@@ -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");
}

63
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java
View File

@@ -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<K, V, S extends KafkaProducerMessag
}
/**
* Configure a {@link Function} that will be invoked at run time to determine the topic to
* which a message will be sent. Typically used with a Java 8 Lambda expression:
* Configure a {@link Function} that will be invoked at runtime to determine the topic
* to which a message will be sent. Typically used with a Java 8 Lambda expression:
* <pre class="code">
* {@code
* .<Foo>topic(m -> m.getPayload().getTopic())
@@ -133,8 +133,9 @@ public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessag
}
/**
* Configure a {@link Function} that will be invoked at run time to determine the message key under
* which a message will be stored in the topic. Typically used with a Java 8 Lambda expression:
* Configure a {@link Function} that will be invoked at runtime to determine the
* message key under which a message will be stored in the topic. Typically used with
* a Java 8 Lambda expression:
* <pre class="code">
* {@code
* .<Foo>messageKey(m -> m.getPayload().getKey())
@@ -169,8 +170,9 @@ public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessag
}
/**
* Configure a {@link Function} that will be invoked at run time to determine the partition id under
* which a message will be stored in the topic. Typically used with a Java 8 Lambda expression:
* Configure a {@link Function} that will be invoked at runtime to determine the
* partition id under which a message will be stored in the topic. Typically used with
* a Java 8 Lambda expression:
* <pre class="code">
* {@code
* .partitionId(m -> m.getHeaders().get("partitionId", Integer.class))
@@ -206,14 +208,15 @@ public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessag
}
/**
* Configure a {@link Function} that will be invoked at run time to determine the Kafka record timestamp
* will be stored in the topic. Typically used with a Java 8 Lambda expression:
* Configure a {@link Function} that will be invoked at runtime to determine the Kafka
* record timestamp will be stored in the topic. Typically used with a Java 8 Lambda
* expression:
* <pre class="code">
* {@code
* .timestamp(m -> m.getHeaders().get("mytimestamp_header", Long.class))
* }
* </pre>
* @param timestampFunction the partitionId function.
* @param timestampFunction the timestamp function.
* @param <P> the expected payload type.
* @return the spec.
*/
@@ -232,6 +235,46 @@ public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessag
return _this();
}
/**
* Configure a SpEL 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(String flushExpression) {
return this.flushExpression(PARSER.parseExpression(flushExpression));
}
/**
* Configure a {@link Function} that will be invoked at runtime 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}. Typically used with a
* Java 8 Lambda expression:
* <pre class="code">
* {@code
* .flush(m -> m.getPayload().shouldFlush())
* }
* </pre>
* @param flushFunction the flush function.
* @param <P> the expected payload type.
* @return the spec.
*/
public <P> S flush(Function<Message<P>, 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}

25
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java
View File

@@ -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<K, V> extends AbstractReplyProducingMes
private Expression timestampExpression;
private Expression flushExpression = new FunctionExpression<Message<?>>(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<K, V> 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<K, V> extends AbstractReplyProducingMes
@Override
protected Object handleRequestMessage(final Message<?> message) {
final ProducerRecord<K, V> producerRecord;
boolean flush = this.flushExpression.getValue(this.evaluationContext, message, Boolean.class);
boolean preBuilt = message.getPayload() instanceof ProducerRecord;
if (preBuilt) {
producerRecord = (ProducerRecord<K, V>) message.getPayload();
}
else {
producerRecord = createProducerRecord(message);
if (flush) {
producerRecord.headers().remove(KafkaIntegrationHeaders.FLUSH);
}
}
ListenableFuture<SendResult<K, V>> sendFuture;
RequestReplyFuture<K, V, Object> gatewayFuture = null;
@@ -426,6 +448,9 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
catch (ExecutionException e) {
throw new MessageHandlingException(message, e.getCause());
}
if (flush) {
this.kafkaTemplate.flush();
}
return processReplyFuture(gatewayFuture);
}

38
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaIntegrationHeaders.java Normal file
View File

@@ -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";
}

9
spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka.xsd
View File

@@ -568,6 +568,15 @@
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="flush-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the expression to determine whether the producer should be
flushed after a send. By default the producer is flushed if a
header 'kafka_flush' has a value 'Boolean.TRUE'.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="sync">
<xsd:annotation>
<xsd:documentation><![CDATA[

3
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests-context.xml
View File

@@ -20,6 +20,7 @@
partition-id-expression="'2'"
send-timeout-expression="1000"
timestamp-expression="T(System).currentTimeMillis()"
flush-expression="headers['foo']"
error-message-strategy="ems"
send-failure-channel="failures"
send-success-channel="successes"
@@ -54,7 +55,7 @@
<int:channel id="failures" />
<int:channel id="successes" />
<bean id="customHeaderMapper" class="org.springframework.kafka.support.DefaultKafkaHeaderMapper" />
</beans>

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java
View File

@@ -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"));

1
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests-context.xml
View File

@@ -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'"

4
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests.java
View File

@@ -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"));

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java
View File

@@ -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();

46
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java
View File

@@ -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<ProducerRecord> 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();
}
}