INTEXT-104: Kafka: order for o-channel-adapter

JIRA: https://jira.spring.io/browse/INTEXT-104

`order` attribute now honored if kafka outbound adapter is connected to a subscribable channel
unit tests and samples are updated

Polishing: use `<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>` for adapter tags to cover `SmartLifecycle` options.

samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/UserTransformer.java

@@ -0,0 +1,27 @@
+package org.springframework.integration.samples.kafka.outbound;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.integration.samples.kafka.user.User;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.transformer.Transformer;
+import org.springframework.messaging.Message;
+
+/**
+ * @author Soby Chacko
+ */
+public class UserTransformer implements Transformer {
+
+    Log logger = LogFactory.getLog(getClass());
+
+    @Override
+    public Message<?> transform(Message<?> message) {
+        if(message.getPayload().getClass().isAssignableFrom(User.class)) {
+            User user = (User) message.getPayload();
+            user.setFirstName(user.getFirstName().toString()+user.getFirstName());
+            logger.info("user confirmed " + user.getFirstName());
+            return MessageBuilder.withPayload(user).copyHeaders(message.getHeaders()).build();
+        }
+        return message;
+    }
+}

samples/kafka/src/main/resources/org/springframework/integration/samples/kafka/outbound/kafkaOutboundAdapterParserTests-context.xml

@@ -1,24 +1,31 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <beans xmlns="http://www.springframework.org/schema/beans"
-	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:task="http://www.springframework.org/schema/task"
-	xsi:schemaLocation="http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
+       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:task="http://www.springframework.org/schema/task"
+       xmlns:int-stream="http://www.springframework.org/schema/integration/stream"
+       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/task http://www.springframework.org/schema/task/spring-task.xsd">
+	http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
+	http://www.springframework.org/schema/integration/stream
+      http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd">
 
-	<int:channel id="inputToKafka">
-		<int:queue/>
-	</int:channel>
+	<int:publish-subscribe-channel id="inputToKafka"/>
 
 	<int-kafka:outbound-channel-adapter kafka-producer-context-ref="kafkaProducerContext"
 			auto-startup="false"
-			channel="inputToKafka">
-		<int:poller fixed-delay="1000" time-unit="MILLISECONDS" receive-timeout="0" task-executor="taskExecutor"/>
+			channel="inputToKafka"
+            order="1">
 	</int-kafka:outbound-channel-adapter>
 
+    <int-stream:stdout-channel-adapter id="stdout" append-newline="true"/>
+
+    <int:transformer order="0" ref="userTransformer" input-channel="inputToKafka" output-channel="stdout"/>
+
+    <bean id="userTransformer" class="org.springframework.integration.samples.kafka.outbound.UserTransformer"/>
+
 	<task:executor id="taskExecutor" pool-size="5" keep-alive="120" queue-capacity="500"/>
 
 	<bean id="kafkaReflectionEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder">

spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd

@@ -9,7 +9,7 @@
     <xsd:import namespace="http://www.springframework.org/schema/beans"/>
     <xsd:import namespace="http://www.springframework.org/schema/tool"/>
     <xsd:import namespace="http://www.springframework.org/schema/integration"
-                schemaLocation="http://www.springframework.org/schema/integration/spring-integration.xsd"/>
+                schemaLocation="http://www.springframework.org/schema/integration/spring-integration-4.0.xsd"/>
 
     <xsd:annotation>
         <xsd:documentation><![CDATA[
@@ -380,16 +380,7 @@
             <xsd:sequence>
                 <xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
             </xsd:sequence>
-            <xsd:attributeGroup ref="coreKafkaComponentAttributes"/>
-            <xsd:attribute name="channel" type="xsd:string">
-                <xsd:annotation>
-                    <xsd:appinfo>
-                        <tool:annotation kind="ref">
-                            <tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
-                        </tool:annotation>
-                    </xsd:appinfo>
-                </xsd:annotation>
-            </xsd:attribute>
+            <xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
             <xsd:attribute name="send-timeout" type="xsd:string">
                 <xsd:annotation>
                     <xsd:documentation><![CDATA[
@@ -419,8 +410,7 @@
                     </xsd:appinfo>
                 </xsd:annotation>
             </xsd:attribute>
-            <xsd:attribute name="kafka-consumer-context-ref" use="required"
-                           type="xsd:string">
+            <xsd:attribute name="kafka-consumer-context-ref" use="required" type="xsd:string">
                 <xsd:annotation>
                     <xsd:documentation>
                         Kafka Server Bean Name
@@ -430,30 +420,6 @@
         </xsd:complexType>
     </xsd:element>
 
-    <xsd:attributeGroup name="coreKafkaComponentAttributes">
-        <xsd:attribute name="id" type="xsd:string" use="optional">
-            <xsd:annotation>
-                <xsd:documentation>
-                    Identifies the underlying Spring bean definition, which is an
-                    instance of either 'EventDrivenConsumer' or 'PollingConsumer',
-                    depending on whether the component's input channel is a
-                    'SubscribableChannel' or 'PollableChannel'.
-                </xsd:documentation>
-            </xsd:annotation>
-        </xsd:attribute>
-        <xsd:attribute name="auto-startup" default="true" use="optional">
-            <xsd:annotation>
-                <xsd:documentation>
-                    Flag to indicate that the component should start automatically
-                    on startup (default true).
-                </xsd:documentation>
-            </xsd:annotation>
-            <xsd:simpleType>
-                <xsd:union memberTypes="xsd:boolean xsd:string"/>
-            </xsd:simpleType>
-        </xsd:attribute>
-    </xsd:attributeGroup>
-
     <xsd:element name="outbound-channel-adapter">
         <xsd:annotation>
             <xsd:documentation>
@@ -466,42 +432,20 @@
             <xsd:sequence>
                 <xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
             </xsd:sequence>
-            <xsd:attribute name="id" type="xsd:string" use="optional">
-                <xsd:annotation>
-                    <xsd:documentation>
-                        Identifies the underlying Spring bean definition, which is an
-                        instance of either 'EventDrivenConsumer' or 'PollingConsumer',
-                        depending on whether the component's input channel is a
-                        'SubscribableChannel' or 'PollableChannel'.
-                    </xsd:documentation>
-                </xsd:annotation>
-            </xsd:attribute>
-            <xsd:attribute name="auto-startup" default="true" use="optional">
-                <xsd:annotation>
-                    <xsd:documentation>
-                        Flag to indicate that the component should start automatically
-                        on startup (default true).
-                    </xsd:documentation>
-                </xsd:annotation>
-                <xsd:simpleType>
-                    <xsd:union memberTypes="xsd:boolean xsd:string"/>
-                </xsd:simpleType>
-            </xsd:attribute>
-            <xsd:attribute name="kafka-producer-context-ref" use="required"
-                           type="xsd:string">
+			<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
+            <xsd:attribute name="kafka-producer-context-ref" use="required" type="xsd:string">
                 <xsd:annotation>
                     <xsd:documentation>
                         Kafka producer context reference.
                     </xsd:documentation>
                 </xsd:annotation>
             </xsd:attribute>
-            <xsd:attribute name="channel" type="xsd:string">
+            <xsd:attribute name="order">
                 <xsd:annotation>
-                    <xsd:appinfo>
-                        <tool:annotation kind="ref">
-                            <tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
-                        </tool:annotation>
-                    </xsd:appinfo>
+                    <xsd:documentation>
+                        Specifies the order for invocation when this endpoint is connected as a
+                        subscriber to a SubscribableChannel.
+                    </xsd:documentation>
                 </xsd:annotation>
             </xsd:attribute>
         </xsd:complexType>

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

@@ -17,6 +17,7 @@
                                         kafka-producer-context-ref="kafkaProducerContext"
                                         auto-startup="false"
                                         channel="inputToKafka"
+                                        order="3"
             >
         <int:poller fixed-delay="1000" time-unit="MILLISECONDS" receive-timeout="0" task-executor="taskExecutor"/>
     </int-kafka:outbound-channel-adapter>

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

@@ -18,6 +18,7 @@ package org.springframework.integration.kafka.config.xml;
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.integration.endpoint.PollingConsumer;
@@ -32,20 +33,23 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  */
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration
-public class KafkaOutboundAdapterParserTests<K,V> {
+public class KafkaOutboundAdapterParserTests<K, V> {
 
 	@Autowired
 	private ApplicationContext appContext;
 
 	@Test
 	@SuppressWarnings("unchecked")
-	public void testOutboundAdapterConfiguration(){
-		final PollingConsumer pollingConsumer = appContext.getBean("kafkaOutboundChannelAdapter", PollingConsumer.class);
-		final KafkaProducerMessageHandler<K,V> messageHandler = appContext.getBean(KafkaProducerMessageHandler.class);
+	public void testOutboundAdapterConfiguration() {
+		final PollingConsumer pollingConsumer =
+				appContext.getBean("kafkaOutboundChannelAdapter", PollingConsumer.class);
+		final KafkaProducerMessageHandler<K, V> messageHandler = appContext.getBean(KafkaProducerMessageHandler.class);
 		Assert.assertNotNull(pollingConsumer);
 		Assert.assertNotNull(messageHandler);
-		final KafkaProducerContext<K,V> producerContext = messageHandler.getKafkaProducerContext();
+		Assert.assertEquals(messageHandler.getOrder(), 3);
+		final KafkaProducerContext<K, V> producerContext = messageHandler.getKafkaProducerContext();
 		Assert.assertNotNull(producerContext);
 		Assert.assertEquals(producerContext.getTopicsConfiguration().size(), 2);
 	}
+
 }