INTEXT-84 Kafka: Enhance Avro serialization support

samples/kafka/build.gradle

@@ -1,27 +1,62 @@
 description = 'Spring Integration Kafka Sample'
+
+buildscript {
+
+    repositories {
+            mavenCentral()
+        mavenLocal()
+    }
+        dependencies {
+            classpath 'org.apache.maven:maven-artifact:2.2.1' // 3.x won't work
+            classpath 'org.apache.avro:avro-compiler:1.7.3' // use Avro 1.7.4 to compile the Avro files
+            //classpath 'org.clojars.miguno:avro-gradle-plugin:1.7.2'
+            classpath "org.apache.avro:avro-gradle-plugin:1.7.2"
+        }
+
+}
+
+
 apply plugin: 'base'
 apply plugin: 'java'
 apply plugin: 'eclipse'
 apply plugin: 'application'
 apply plugin: 'idea'
 apply plugin: 'maven'
+apply plugin: 'avro-gradle-plugin'
+
+ext {
+    avroTaskGroup = "Avro"
+    avroSource = "schemas"
+    avroDest = "target/generated-avro-sources/main/java"
+}
 
 repositories {
-	mavenLocal()
-	mavenCentral()
-	maven { url "http://repo.springsource.org/libs-snapshot" }
-        maven { url 'http://repo.springsource.org/plugins-release' }
+
+mavenLocal()
+maven {
+                url 'https://repository.apache.org/content/groups/public'
+        }
+        maven { url 'https://repo.springsource.org/libs-milestone' }
+
 }
 
 dependencies {
+    compile([
+            "org.apache.avro:avro:1.7.3",
+            "org.apache.avro:avro-compiler:1.7.3"
+    ])
 	  compile "org.springframework:spring-beans:3.1.3.RELEASE"
         compile "org.springframework:spring-context:3.1.3.RELEASE"
         compile "org.springframework:spring-expression:3.1.3.RELEASE"
 	   compile "org.springframework.integration:spring-integration-stream:2.2.0.RELEASE"
 	compile("org.springframework.integration:spring-integration-kafka:0.5.0.BUILD-SNAPSHOT") {
    		exclude module: 'log4j'
+                exclude module: 'jms'
+                exclude module: 'jmxtools'
+                exclude module: 'jmxri'
 	}
 compile("log4j:log4j:1.2.15") {
+		exclude module: 'mail'
         exclude module: 'jms'
         exclude module: 'jmx'
         exclude module: 'jmxtools'
@@ -32,6 +67,30 @@ compile("log4j:log4j:1.2.15") {
         compile "commons-logging:commons-logging:1.1.1"
 }
 
+compileAvro.group = avroTaskGroup
+compileAvro.description = "Generates Java code from avro schema"
+compileAvro.source = avroSource
+compileAvro.destinationDir = file(avroDest)
+
+task cleanAvro(type: Delete) {
+    group = avroTaskGroup
+    description = "deletes generated avro code"
+    delete avroDest
+}
+
+compileJava.dependsOn compileAvro
+
+sourceSets {
+    main {
+        java {
+            srcDir avroDest
+        }
+        resources {
+            srcDir avroSource
+        }
+    }
+}
+
 task wrapper(type: Wrapper) {
 	description = 'Generates gradlew[.bat] scripts'
 	gradleVersion = '1.3'

samples/kafka/schemas/user.avdl

@@ -0,0 +1,7 @@
+@namespace("org.springframework.integration.samples.kafka.user")
+protocol UserProtocol{
+record User {
+	string firstName;
+	string lastName;
+}
+}

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

@@ -19,6 +19,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.MessageChannel;
+import org.springframework.integration.samples.kafka.user.User;
 import org.springframework.integration.support.MessageBuilder;
 
 public class OutboundRunner {
@@ -34,8 +35,11 @@ public class OutboundRunner {
 
         //sending 100,000 messages to Kafka server for topic test1
         for (int i = 0; i < 500; i++) {
+			final User user = new User();
+			user.setFirstName("fname" + i);
+			user.setLastName("lname" + i);
             channel.send(
-                    MessageBuilder.withPayload("hello Fom ob adapter test1 -  " + i)
+                    MessageBuilder.withPayload(user)
                             .setHeader("messageKey", String.valueOf(i))
                             .setHeader("topic", "test1").build());
 

samples/kafka/src/main/resources/org/springframework/integration/samples/kafka/inbound/kafkaInboundAdapterParserTests-context.xml

@@ -27,22 +27,26 @@
         <int:poller fixed-delay="1" time-unit="MILLISECONDS"/>
     </int-kafka:inbound-channel-adapter>
 
-    <bean id="kafkaDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroBackedKafkaDecoder">
+    <bean id="kafkaReflectionDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaDecoder">
         <constructor-arg type="java.lang.Class" value="java.lang.String"/>
     </bean>
 
+    <bean id="kafkaSpecificDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroSpecificDatumBackedKafkaDecoder">
+                <constructor-arg value="org.springframework.integration.samples.kafka.user.User" />
+    </bean>
+
     <int-kafka:consumer-context id="consumerContext"
                                 consumer-timeout="1000"
                                 zookeeper-connect="zookeeperConnect">
         <int-kafka:consumer-configurations>
             <int-kafka:consumer-configuration group-id="default"
-                                              value-decoder="kafkaDecoder"
-                                              key-decoder="kafkaDecoder"
+                                              value-decoder="kafkaSpecificDecoder"
+                                              key-decoder="kafkaReflectionDecoder"
                                               max-messages="5000">
                 <int-kafka:topic id="test1" streams="4"/>
             </int-kafka:consumer-configuration>
             <int-kafka:consumer-configuration group-id="default1"
-                                              max-messages="5">
+                                              max-messages="50">
                 <int-kafka:topic id="test2" streams="4"/>
             </int-kafka:consumer-configuration>
         </int-kafka:consumer-configurations>

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

@@ -21,20 +21,24 @@
     
     <task:executor id="taskExecutor" pool-size="5" keep-alive="120" queue-capacity="500"/>
 
-    <bean id="kafkaEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroBackedKafkaEncoder">
+    <bean id="kafkaReflectionEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder">
         <constructor-arg value="java.lang.String" />
     </bean>
 
+    <bean id="kafkaSpecificEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroSpecificDatumBackedKafkaEncoder">
+            <constructor-arg value="org.springframework.integration.samples.kafka.user.User" />
+    </bean>
+
     <bean id="customPartitioner" class="org.springframework.integration.samples.kafka.outbound.CustomPartitioner"/>
 
     <int-kafka:producer-context id="kafkaProducerContext">
         <int-kafka:producer-configurations>
             <int-kafka:producer-configuration broker-list="localhost:9092"
                                               key-class-type="java.lang.String"
-                                              value-class-type="java.lang.String"
+                                              value-class-type="org.springframework.integration.samples.kafka.user.User"
                                               topic="test1"
-                                              value-encoder="kafkaEncoder"
-                                              key-encoder="kafkaEncoder"
+                                              value-encoder="kafkaSpecificEncoder"
+                                              key-encoder="kafkaReflectionEncoder"
                                               compression-codec="default"
                                               partitioner="customPartitioner"/>
             <int-kafka:producer-configuration broker-list="localhost:9092"