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

GH-143: Add sync to XML Namespace

Browse Source 
Resolves: https://github.com/spring-projects/spring-integration-kafka/issues/143
This commit is contained in:
Gary Russell
2016-09-16 09:28:53 -04:00
committed by Artem Bilan
parent cbfc6b0e73
commit adfd6ee49e
5 changed files with 19 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -66,7 +66,7 @@ public class KafkaOutboundChannelAdapterParser extends AbstractOutboundChannelAd
if (partitionIdExpressionDef != null) {
kafkaProducerMessageHandlerBuilder.addPropertyValue("partitionIdExpression", partitionIdExpressionDef);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(kafkaProducerMessageHandlerBuilder, element, "sync");
return kafkaProducerMessageHandlerBuilder.getBeanDefinition();
}

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

@@ -84,8 +84,8 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractMessageHandler {
}
/**
* The {@code boolean} indicated if {@link KafkaProducerMessageHandler}
* should wait for send operation results or not. Defaults to {@code false}.
* A {@code boolean} indicating if the {@link KafkaProducerMessageHandler}
* should wait for the 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

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

@@ -96,6 +96,18 @@
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<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.
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:int xsd:boolean"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="order">
<xsd:annotation>
<xsd:documentation>

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

@@ -15,6 +15,7 @@
auto-startup="false"
channel="inputToKafka"
order="3"
sync="true"
topic="foo"
message-key-expression="'bar'"
partition-id-expression="'2'">

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

@@ -67,16 +67,19 @@ public class KafkaOutboundAdapterParserTests {
assertThat(TestUtils.getPropertyValue(messageHandler, "topicExpression.literalValue")).isEqualTo("foo");
assertThat(TestUtils.getPropertyValue(messageHandler, "messageKeyExpression.expression")).isEqualTo("'bar'");
assertThat(TestUtils.getPropertyValue(messageHandler, "partitionIdExpression.expression")).isEqualTo("'2'");
assertThat(TestUtils.getPropertyValue(messageHandler, "sync", Boolean.class)).isTrue();
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();
}
@Test
public void testSyncMode() {
@SuppressWarnings("resource")
MockProducer<Integer, String> mockProducer =
new MockProducer<>(false, new IntegerSerializer(), new StringSerializer());
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(() -> mockProducer);