From 8662c10232a1309235703a7db84437a0324ebfbd Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Thu, 25 Aug 2016 16:58:07 -0400
Subject: [PATCH] Fix XSD type for `partition-id`

Even if target ``partitionId` value has to be `Integer`, we want to be able to configure the attribute via property-placeholder or Bean initialization SpEL

* Change `partition-id` XSD type from `integer` -> `string`
* Ensure that fix is correct with the test-case
---
 .../kafka/config/spring-integration-kafka-2.0.xsd        | 2 +-
 .../xml/KafkaOutboundAdapterParserTests-context.xml      | 9 ++++++++-
 .../config/xml/KafkaOutboundAdapterParserTests.java      | 5 +++++
 3 files changed, 14 insertions(+), 2 deletions(-)

diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-2.0.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-2.0.xsd
index a74655c18b..5294d79a96 100644
--- a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-2.0.xsd
+++ b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-2.0.xsd
@@ -79,7 +79,7 @@
 							]]></xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
-			<xsd:attribute name="partition-id" type="xsd:integer">
+			<xsd:attribute name="partition-id" type="xsd:string">
 				<xsd:annotation>
 					<xsd:documentation><![CDATA[
 								Specifies the target partition for the Kafka message.
diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests-context.xml
index 429addf5a9..ed7187411f 100644
--- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests-context.xml
+++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests-context.xml
@@ -3,9 +3,10 @@
 	   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/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">
 
 	<int:channel id="inputToKafka" />
 
@@ -34,4 +35,10 @@
 		</constructor-arg>
 	</bean>
 
+	<int-kafka:outbound-channel-adapter id="kafkaOutboundChannelAdapter2"
+										kafka-template="template"
+										auto-startup="false"
+										channel="inputToKafka"
+										partition-id="#{0}" />
+
 </beans>
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 6787ea59e2..ff59bf3328 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
@@ -67,6 +67,11 @@ 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'");
+
+		messageHandler
+				= this.appContext.getBean("kafkaOutboundChannelAdapter2.handler", KafkaProducerMessageHandler.class);
+		assertThat(messageHandler).isNotNull();
+		assertThat(TestUtils.getPropertyValue(messageHandler, "partitionIdExpression.literalValue")).isEqualTo("0");
 	}