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

GH-145 send-timeout(-expression) and XSD for them

Browse Source 
Fixes GH-145 (https://github.com/spring-projects/spring-integration-kafka/issues/145)
This commit is contained in:
Artem Bilan
2016-10-06 12:13:21 -04:00
committed by Artem Bilan
parent 171dc9832a
commit 8cd471532d
5 changed files with 65 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java
View File

@@ -67,6 +67,14 @@ public class KafkaOutboundChannelAdapterParser extends AbstractOutboundChannelAd
kafkaProducerMessageHandlerBuilder.addPropertyValue("partitionIdExpression", partitionIdExpressionDef);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(kafkaProducerMessageHandlerBuilder, element, "sync");
BeanDefinition sendTimeoutExpressionDef =
IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("send-timeout",
"send-timeout-expression", parserContext, element, false);
if (sendTimeoutExpressionDef != null) {
kafkaProducerMessageHandlerBuilder.addPropertyValue("sendTimeoutExpression", sendTimeoutExpressionDef);
}
return kafkaProducerMessageHandlerBuilder.getBeanDefinition();
}

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

@@ -23,6 +23,7 @@ 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.ValueExpression;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
@@ -60,7 +61,7 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractMessageHandler {
private boolean sync;
private long sendTimeout = DEFAULT_SEND_TIMEOUT;
private Expression sendTimeoutExpression = new ValueExpression<>(DEFAULT_SEND_TIMEOUT);
public KafkaProducerMessageHandler(final KafkaTemplate<K, V> kafkaTemplate) {
Assert.notNull(kafkaTemplate, "kafkaTemplate cannot be null");
@@ -95,12 +96,27 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractMessageHandler {
}
/**
* Specify a timeout in milliseconds how long {@link KafkaProducerMessageHandler}
* should wait wait for send operation results. Defaults to 10 seconds.
* Specify a timeout in milliseconds for how long this
* {@link KafkaProducerMessageHandler} should wait wait for send operation
* results. Defaults to 10 seconds. The timeout is applied only in {@link #sync} mode.
* @param sendTimeout the timeout to wait for result fo send operation.
* @since 2.0.1
*/
public void setSendTimeout(long sendTimeout) {
this.sendTimeout = sendTimeout;
setSendTimeoutExpression(new ValueExpression<>(sendTimeout));
}
/**
* Specify a SpEL expression to evaluate a timeout in milliseconds for how long this
* {@link KafkaProducerMessageHandler} should wait wait for send operation
* results. Defaults to 10 seconds. The timeout is applied only in {@link #sync} mode.
* @param sendTimeoutExpression the {@link Expression} for timeout to wait for result
* fo send operation.
* @since 2.1.1
*/
public void setSendTimeoutExpression(Expression sendTimeoutExpression) {
Assert.notNull(sendTimeoutExpression, "'sendTimeoutExpression' must not be null");
this.sendTimeoutExpression = sendTimeoutExpression;
}
@Override
@@ -149,12 +165,13 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractMessageHandler {
}
}
if (this.sync) {
if (this.sendTimeout < 0) {
Long sendTimeout = this.sendTimeoutExpression.getValue(this.evaluationContext, message, Long.class);
if (sendTimeout == null || sendTimeout < 0) {
future.get();
}
else {
try {
future.get(this.sendTimeout, TimeUnit.MILLISECONDS);
future.get(sendTimeout, TimeUnit.MILLISECONDS);
}
catch (TimeoutException te) {
throw new MessageTimeoutException(message, "Timeout waiting for response from KafkaProducer", te);

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

@@ -99,15 +99,37 @@
<xsd:attribute name="sync">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies whether to block the sending thread until the producer
callback has been invoked, indicating the broker has accepted the
message (or an exception thrown if the send fails). Default: false.
Specifies whether to block the sending thread until the producer
callback has been invoked, indicating the broker has accepted the
message (or an exception thrown if the send fails). Default: false.
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:int xsd:boolean"/>
<xsd:union memberTypes="xsd:boolean xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="send-timeout">
<xsd:annotation>
<xsd:documentation>
Specifies a timeout in milliseconds for how long the 'KafkaProducerMessageHandler'
should wait wait for send operation results. Defaults to 10 seconds.
The timeout is applied only in 'sync' mode.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:int xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="send-timeout-expression">
<xsd:annotation>
<xsd:documentation>
Specifies an expression that is evaluated to determine a timeout in milliseconds
for how long the 'KafkaProducerMessageHandler'
should wait wait for send operation results. Defaults to 10 seconds.
The timeout is applied only in 'sync' mode.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order">
<xsd:annotation>
<xsd:documentation>

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

@@ -3,10 +3,9 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<int:channel id="inputToKafka" />
@@ -18,7 +17,8 @@
sync="true"
topic="foo"
message-key-expression="'bar'"
partition-id-expression="'2'">
partition-id-expression="'2'"
send-timeout-expression="1000">
<int-kafka:request-handler-advice-chain>
<bean class="org.springframework.integration.handler.advice.RequestHandlerCircuitBreakerAdvice" />
</int-kafka:request-handler-advice-chain>
@@ -40,6 +40,7 @@
kafka-template="template"
auto-startup="false"
channel="inputToKafka"
partition-id="#{0}" />
partition-id="#{0}"
send-timeout="500" />
</beans>

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

@@ -68,12 +68,15 @@ public class KafkaOutboundAdapterParserTests {
assertThat(TestUtils.getPropertyValue(messageHandler, "messageKeyExpression.expression")).isEqualTo("'bar'");
assertThat(TestUtils.getPropertyValue(messageHandler, "partitionIdExpression.expression")).isEqualTo("'2'");
assertThat(TestUtils.getPropertyValue(messageHandler, "sync", Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(messageHandler, "sendTimeoutExpression.expression")).isEqualTo("1000");
messageHandler
= this.appContext.getBean("kafkaOutboundChannelAdapter2.handler", KafkaProducerMessageHandler.class);
assertThat(messageHandler).isNotNull();
assertThat(TestUtils.getPropertyValue(messageHandler, "partitionIdExpression.literalValue")).isEqualTo("0");
assertThat(TestUtils.getPropertyValue(messageHandler, "sync", Boolean.class)).isFalse();
assertThat(TestUtils.getPropertyValue(messageHandler, "sendTimeoutExpression.literalValue")).isEqualTo("500");
}