Rework kafka-avro example: drop Confluent Schema Registry, fix OOM

kafka-avro/README.adoc

@@ -1,55 +1,38 @@
-== Camel Kafka example
+== Camel Kafka Avro example
 
 === Introduction
 
-An example which shows how to integrate Camel with Kafka avro  to make use of avro serialize/deserializer
-
-First a kafka server should be started:
-----
-$ ./bin/schema-registry-start etc/schema-registry/schema-registry.properties
-----
-
-Then start confluent schema registry:
-register_schema.py is simple custom python script to register the employee.avsc schema in confluent schema registry:
-----
-$ python3.6 src/main/resources/register_schema.py http://localhost:8081 employees-avro src/main/resources/avro/employee.avsc
-Schema Registry URL: http://localhost:8081
-Topic: employees-avro
-Schema file: src/main/resources/avro/employee.avsc
-Success
-----
-
-----
-$ curl --noproxy '*' http://localhost:8081/subjects/employees-avro-value/versions/1
-{"subject":"employees-avro-value","version":1,"id":2,"schema":"{\"type\":\"record\",\"name\":\"Employee\",\"namespace\":\"com.example.kafkatutorials\",\"fields\":[{\"name\":\"firstName\",\"type\":\"string\"},{\"name\":\"lastName\",\"type\":\"string\"},{\"name\":\"birthDate\",\"type\":\"long\"}]}"}
-----
-
-To delete:
-----
-$ curl -X DELETE http://localhost:8081/subjects/topic-value/versions/version-no
-----
+An example which shows how to integrate Camel with Kafka using Avro serialization.
+
+A timer triggers a producer route that builds an `Employee` Avro record (generated at
+build time from `src/main/resources/avro/employee.avsc`), marshals it to Avro binary with
+Camel's `avro` data format, and sends it to Kafka. A consumer route reads the same topic,
+unmarshals the bytes back into an `Employee` object and logs it.
+
+The Avro schema is shared between producer and consumer through the application's
+classpath, so this example needs only a plain Kafka broker - it does not use, and does not
+require, a Confluent Schema Registry.
 
 === Preparing Kafka
 
-This example requires that Kafka Server is up and running.
+This example requires that a Kafka broker is up and running.
 
-You can use the Camel CLI to start a Kafka broker:
+You can use the Camel CLI to start one:
 
     $ camel infra run kafka
 
 === Build
 
-You will need to compile this example first:
+You will need to compile this example first, which also generates the `Employee` Avro class:
 
     $ mvn compile
 
 === Run
 
-Run the consumer first in separate shell 
-
     $ mvn spring-boot:run
 
-camel-context.xml file has both kafka-producer and kafka-consumer routes defined to produce/consume messages to topic my-topic.
+The application starts both routes: every second it produces a new `Employee` record to the
+`employees-avro` topic, and its consumer logs each `Employee` it reads back.
 
 Press `Ctrl-C` to exit.
 
@@ -63,11 +46,10 @@ You can enable verbose logging by adjusting the `src/main/resources/log4j2.prope
 
 === Help and contributions
 
-If you hit any problem using Camel or have some feedback, 
+If you hit any problem using Camel or have some feedback,
 then please https://camel.apache.org/community/support/[let us know].
 
-We also love contributors, 
+We also love contributors,
 so https://camel.apache.org/community/contributing/[get involved] :-)
 
 The Camel riders!
-

kafka-avro/pom.xml

@@ -34,16 +34,11 @@
         <category>Messaging</category>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+        <!-- must match the Avro version pulled in by camel-avro-starter so generated
+             classes stay binary compatible with the runtime Avro classes -->
+        <avro-version>1.12.2</avro-version>
     </properties>
 
-    <!--necessary to resolve confluent dependencies-->
-    <repositories>
-        <repository>
-            <id>confluent</id>
-            <url>https://packages.confluent.io/maven/</url>
-        </repository>
-    </repositories>
-
     <dependencyManagement>
          <dependencies>
             <!-- Camel BOM -->
@@ -62,13 +57,6 @@
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
-
-            <!-- manages conflict forcing CSB/Camel version -->
-            <dependency>
-                <groupId>org.apache.avro</groupId>
-                <artifactId>avro</artifactId>
-                <version>${avro-version}</version>
-            </dependency>
         </dependencies>
     </dependencyManagement>
 
@@ -87,11 +75,11 @@
         <!-- Camel -->
         <dependency>
             <groupId>org.apache.camel.springboot</groupId>
-            <artifactId>camel-spring-boot-xml-starter</artifactId>
+            <artifactId>camel-spring-boot-starter</artifactId>
         </dependency>
         <dependency>
             <groupId>org.apache.camel.springboot</groupId>
-            <artifactId>camel-stream-starter</artifactId>
+            <artifactId>camel-timer-starter</artifactId>
         </dependency>
         <dependency>
             <groupId>org.apache.camel.springboot</groupId>
@@ -101,17 +89,12 @@
             <groupId>org.apache.camel.springboot</groupId>
             <artifactId>camel-kafka-starter</artifactId>
         </dependency>
-        <dependency>
-            <groupId>io.confluent</groupId>
-            <artifactId>kafka-avro-serializer</artifactId>
-            <version>${kafka-avro-serializer-version}</version>
-        </dependency>
     </dependencies>
 
     <build>
         <plugins>
-            <!--Avro plugin to  generated classes from schema-->
-            <!-- <plugin>
+            <!-- Generates the Employee Avro class from src/main/resources/avro/employee.avsc -->
+            <plugin>
                 <groupId>org.apache.avro</groupId>
                 <artifactId>avro-maven-plugin</artifactId>
                 <version>${avro-version}</version>
@@ -120,20 +103,18 @@
                         <phase>generate-sources</phase>
                         <goals>
                             <goal>schema</goal>
-                            <goal>protocol</goal>
-                            <goal>idl-protocol</goal>
                         </goals>
                         <configuration>
                             <sourceDirectory>${project.basedir}/src/main/resources/avro</sourceDirectory>
-                            <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
+                            <outputDirectory>${project.build.directory}/generated-sources/avro</outputDirectory>
                             <stringType>String</stringType>
                             <createSetters>false</createSetters>
                             <enableDecimalLogicalType>true</enableDecimalLogicalType>
                             <fieldVisibility>private</fieldVisibility>
                         </configuration>
                     </execution>
                 </executions>
-            </plugin>  -->
+            </plugin>
             <!--force discovery of generated classes-->
             <plugin>
                 <groupId>org.codehaus.mojo</groupId>
@@ -148,7 +129,7 @@
                         </goals>
                         <configuration>
                             <sources>
-                                <source>target/generated-sources/avro</source>
+                                <source>${project.build.directory}/generated-sources/avro</source>
                             </sources>
                         </configuration>
                     </execution>

kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/Application.java

@@ -18,16 +18,14 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.context.annotation.ImportResource;
 
 //CHECKSTYLE:OFF
 /**
  * A sample Spring Boot application that starts the Camel routes.
  */
 @SpringBootApplication
-@ImportResource({"classpath:spring/camel-context.xml"})
 public class Application {
-                                                                                                                                                                                          
+
     // must have a main method spring-boot can run
     public static void main(String[] args) {
         SpringApplication.run(Application.class, args);

kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/AvroRouteBuilder.java

@@ -17,19 +17,38 @@
 package org.apache.camel.example.kafka.avro;
 
 import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.dataformat.avro.AvroDataFormat;
+import org.springframework.stereotype.Component;
 
+/**
+ * Produces {@link Employee} records to Kafka and consumes them back, using Camel's
+ * {@link AvroDataFormat} to marshal/unmarshal Avro binary. No Confluent Schema Registry is
+ * involved: both routes share the same generated Avro schema on the classpath, so a plain
+ * Kafka broker (e.g. started with {@code camel infra run kafka}) is all that is required.
+ */
+@Component
 public class AvroRouteBuilder extends RouteBuilder {
 
     @Override
     public void configure() throws Exception {
 
+        AvroDataFormat employeeAvroFormat = new AvroDataFormat(Employee.getClassSchema());
+        employeeAvroFormat.setInstanceClassName(Employee.class.getName());
+
         from("timer://foo?period={{period}}")
-        .setBody(constant("Hi This is Avro example"))
-        .process(new KafkaAvroMessageProcessor())
-            .to("kafka:{{producer.topic}}?brokers={{kafka.bootstrap.url}}&keySerializer=org.apache.kafka.common.serialization.StringSerializer&valueSerializer=org.apache.camel.example.kafka.avro.CustomKafkaAvroSerializer");
+            .process(new KafkaAvroMessageProcessor())
+            .marshal(employeeAvroFormat)
+            .to("kafka:{{producer.topic}}?brokers={{kafka.bootstrap.url}}"
+                + "&keySerializer=org.apache.kafka.common.serialization.StringSerializer"
+                + "&valueSerializer=org.apache.kafka.common.serialization.ByteArraySerializer"
+                + "&recordMetadata=true")
+            .process(new KafkaAvroProcessor());
 
-        from("kafka:{{consumer.topic}}?brokers={{kafka.bootstrap.url}}&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer&valueDeserializer=org.apache.camel.example.kafka.avro.CustomKafkaAvroDeserializer")
-         .process(new KafkaAvroMessageConsumerProcessor())
-            .log("${body}");
+        from("kafka:{{consumer.topic}}?brokers={{kafka.bootstrap.url}}"
+                + "&groupId={{consumer.group}}"
+                + "&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+                + "&valueDeserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer")
+            .unmarshal(employeeAvroFormat)
+            .process(new KafkaAvroMessageConsumerProcessor());
     }
 }