diff --git a/kafka-avro/README.adoc b/kafka-avro/README.adoc index e7cbded29..86b3d05e2 100644 --- a/kafka-avro/README.adoc +++ b/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! - diff --git a/kafka-avro/pom.xml b/kafka-avro/pom.xml index 0fe7aa6a1..4e029ca69 100644 --- a/kafka-avro/pom.xml +++ b/kafka-avro/pom.xml @@ -34,16 +34,11 @@ Messaging UTF-8 UTF-8 + + 1.12.2 - - - - confluent - https://packages.confluent.io/maven/ - - - @@ -62,13 +57,6 @@ pom import - - - - org.apache.avro - avro - ${avro-version} - @@ -87,11 +75,11 @@ org.apache.camel.springboot - camel-spring-boot-xml-starter + camel-spring-boot-starter org.apache.camel.springboot - camel-stream-starter + camel-timer-starter org.apache.camel.springboot @@ -101,17 +89,12 @@ org.apache.camel.springboot camel-kafka-starter - - io.confluent - kafka-avro-serializer - ${kafka-avro-serializer-version} - - - + org.apache.avro avro-maven-plugin ${avro-version} @@ -120,12 +103,10 @@ generate-sources schema - protocol - idl-protocol ${project.basedir}/src/main/resources/avro - ${project.basedir}/src/main/java/ + ${project.build.directory}/generated-sources/avro String false true @@ -133,7 +114,7 @@ - --> + org.codehaus.mojo @@ -148,7 +129,7 @@ - target/generated-sources/avro + ${project.build.directory}/generated-sources/avro diff --git a/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/Application.java b/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/Application.java index de96d48c7..bf644bc4c 100644 --- a/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/Application.java +++ b/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); diff --git a/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/AvroRouteBuilder.java b/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/AvroRouteBuilder.java index 05b34c5e5..9cbb2dc35 100644 --- a/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/AvroRouteBuilder.java +++ b/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()); } } diff --git a/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/CustomKafkaAvroDeserializer.java b/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/CustomKafkaAvroDeserializer.java deleted file mode 100644 index c2f24c382..000000000 --- a/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/CustomKafkaAvroDeserializer.java +++ /dev/null @@ -1,61 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.camel.example.kafka.avro; - -import java.util.Collections; -import java.util.List; -import java.util.Map; - -import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient; -import io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer; -import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig; -import org.apache.kafka.common.serialization.Deserializer; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -public class CustomKafkaAvroDeserializer extends AbstractKafkaAvroDeserializer implements Deserializer { - private static final Logger LOG = LoggerFactory.getLogger(CustomKafkaAvroDeserializer.class); - private static final String SCHEMA_REGISTRY_URL = "http://localhost:8081"; - - @Override - public void configure(KafkaAvroDeserializerConfig config) { - LOG.info("ENTER CustomKafkaAvroDeserializer : configure method "); - LOG.info("ENTER CustomKafkaAvroDeserializer : SCHEMA_REGISTRY_URL " + SCHEMA_REGISTRY_URL); - - final List schemas = Collections.singletonList(SCHEMA_REGISTRY_URL); - this.schemaRegistry = new CachedSchemaRegistryClient(schemas, Integer.MAX_VALUE); - this.useSpecificAvroReader = true; - - LOG.info("EXIT CustomKafkaAvroDeserializer : configure method "); - - } - - @Override - public void configure(Map configs, boolean isKey) { - configure(null); - } - - @Override - public Object deserialize(String s, byte[] bytes) { - LOG.info("ENTER CustomKafkaAvroDeserializer : deserialize method "); - return deserialize(bytes).toString(); - } - - @Override - public void close() { - } -} diff --git a/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/CustomKafkaAvroSerializer.java b/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/CustomKafkaAvroSerializer.java deleted file mode 100644 index 816e34a27..000000000 --- a/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/CustomKafkaAvroSerializer.java +++ /dev/null @@ -1,77 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.camel.example.kafka.avro; - -import java.util.Collections; -import java.util.List; -import java.util.Map; - -import io.confluent.kafka.schemaregistry.avro.AvroSchema; -import io.confluent.kafka.schemaregistry.avro.AvroSchemaUtils; -import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient; -import io.confluent.kafka.serializers.AbstractKafkaAvroSerializer; -import io.confluent.kafka.serializers.KafkaAvroSerializerConfig; -import org.apache.kafka.common.config.ConfigException; -import org.apache.kafka.common.serialization.Serializer; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -public class CustomKafkaAvroSerializer extends AbstractKafkaAvroSerializer implements Serializer { - - private static final Logger LOG = LoggerFactory.getLogger(CustomKafkaAvroSerializer.class); - private static final String SCHEMA_REGISTRY_URL = "http://localhost:8081"; - private boolean useSpecificAvroReader = true; - private boolean isKey; - - @Override - public void close() { - } - - - @Override - public byte[] serialize(String topic, Object record) { - LOG.info("****************serialize*******************************"); - LOG.info("Serialize method: topic " + topic); - LOG.info("Serialize method: byte " + record); - AvroSchema schema = new AvroSchema(AvroSchemaUtils.getSchema(record)); - return serializeImpl( - getSubjectName(topic, isKey, record, schema), record, schema); - } - - @Override - public void configure(KafkaAvroSerializerConfig config) { - LOG.info("ENTER CustomKafkaAvroDeserializer : configure method "); - LOG.info("ENTER CustomKafkaAvroDeserializer : SCHEMA_REGISTRY_URL " + SCHEMA_REGISTRY_URL); - - try { - final List schemas = Collections.singletonList(SCHEMA_REGISTRY_URL); - this.schemaRegistry = new CachedSchemaRegistryClient(schemas, Integer.MAX_VALUE); - this.useSpecificAvroReader = true; - - } catch (ConfigException e) { - e.printStackTrace(); - throw new org.apache.kafka.common.config.ConfigException(e.getMessage()); - } - LOG.info("EXIT CustomKafkaAvroserializer : configure method "); - } - - - @Override - public void configure(Map arg0, boolean arg1) { - configure(null); - } -} diff --git a/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/KafkaAvroMessageConsumerProcessor.java b/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/KafkaAvroMessageConsumerProcessor.java index fd0609813..f6ef04e23 100644 --- a/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/KafkaAvroMessageConsumerProcessor.java +++ b/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/KafkaAvroMessageConsumerProcessor.java @@ -21,12 +21,13 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public class KafkaAvroMessageConsumerProcessor implements Processor { +public class KafkaAvroMessageConsumerProcessor implements Processor { private static final Logger LOG = LoggerFactory.getLogger(KafkaAvroMessageConsumerProcessor.class); + @Override public void process(Exchange exchange) throws Exception { - String body = exchange.getIn().getBody(String.class); - LOG.info("KafkaAvroMessageConsumerProcessor:" + body); + Employee employee = exchange.getIn().getBody(Employee.class); + LOG.info("Consumed employee: {}", employee); } } diff --git a/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/KafkaAvroMessageProcessor.java b/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/KafkaAvroMessageProcessor.java index a4bb64a45..e2f963998 100644 --- a/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/KafkaAvroMessageProcessor.java +++ b/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/KafkaAvroMessageProcessor.java @@ -21,16 +21,19 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public class KafkaAvroMessageProcessor implements Processor { - private static final Logger LOG = LoggerFactory.getLogger(KafkaAvroProcessor.class); - public void process(Exchange exc) throws Exception { - //un-comment this after build - /* Employee emp = Employee.newBuilder() - .setFirstName("kakarla") - .setLastName("Ranjith") - .setBirthDate(new java.util.Date().getTime()) - .build(); - exc.getOut().setBody(emp);*/ +public class KafkaAvroMessageProcessor implements Processor { + private static final Logger LOG = LoggerFactory.getLogger(KafkaAvroMessageProcessor.class); + + @Override + public void process(Exchange exchange) throws Exception { + Employee employee = Employee.newBuilder() + .setFirstName("Kakarla") + .setLastName("Ranjith") + .setBirthDate(System.currentTimeMillis()) + .build(); + + LOG.info("Producing employee: {}", employee); + exchange.getIn().setBody(employee); } } diff --git a/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/KafkaAvroProcessor.java b/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/KafkaAvroProcessor.java index 9baa4d857..8dab9033a 100644 --- a/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/KafkaAvroProcessor.java +++ b/kafka-avro/src/main/java/org/apache/camel/example/kafka/avro/KafkaAvroProcessor.java @@ -27,14 +27,13 @@ public class KafkaAvroProcessor implements Processor { private static final Logger LOG = LoggerFactory.getLogger(KafkaAvroProcessor.class); - + @Override - public void process(Exchange exc) throws Exception { + public void process(Exchange exchange) throws Exception { @SuppressWarnings("unchecked") - List recordMetaData1 = (List) exc.getIn().getHeader(KafkaConstants.KAFKA_RECORD_META); - for (RecordMetadata rd: recordMetaData1) { - LOG.info("producer partition is:" + rd.partition()); - LOG.info("producer partition message is:" + rd.toString()); + List recordMetadataList = (List) exchange.getIn().getHeader(KafkaConstants.KAFKA_RECORD_META); + for (RecordMetadata recordMetadata : recordMetadataList) { + LOG.info("Producer sent record to partition {}: {}", recordMetadata.partition(), recordMetadata); } } } diff --git a/kafka-avro/src/main/resources/application.properties b/kafka-avro/src/main/resources/application.properties index 55209479a..8c11403e7 100644 --- a/kafka-avro/src/main/resources/application.properties +++ b/kafka-avro/src/main/resources/application.properties @@ -15,18 +15,14 @@ ## limitations under the License. ## --------------------------------------------------------------------------- -## Modify value of kafka.host and kafka.port before running application - -kafka.bootstrap.url=localhost:9092,localhost:9093,localhost:9094 +## Started with: camel infra run kafka +kafka.bootstrap.url=localhost:9092 # Producer properties producer.topic=employees-avro - -# Consumer properties +# Consumer properties consumer.topic=employees-avro consumer.group=kafkaGroup -consumer.maxPollRecords=5000 -partitionValue=0 period=1000 diff --git a/kafka-avro/src/main/resources/avro/employee.avsc b/kafka-avro/src/main/resources/avro/employee.avsc index cd34a963b..14c40dd24 100644 --- a/kafka-avro/src/main/resources/avro/employee.avsc +++ b/kafka-avro/src/main/resources/avro/employee.avsc @@ -1,7 +1,7 @@ { "type": "record", "name": "Employee", - "namespace": "com.example.kafkatutorials", + "namespace": "org.apache.camel.example.kafka.avro", "fields": [ { "name": "firstName", diff --git a/kafka-avro/src/main/resources/register_schema.py b/kafka-avro/src/main/resources/register_schema.py deleted file mode 100644 index f8e122bcd..000000000 --- a/kafka-avro/src/main/resources/register_schema.py +++ /dev/null @@ -1,48 +0,0 @@ -# -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You under the Apache License, Version 2.0 -# (the "License"); you may not use this file except in compliance with -# the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -import os -import sys - -import requests - -schema_registry_url = sys.argv[1] -topic = sys.argv[2] -schema_file = sys.argv[3] - -aboslute_path_to_schema = os.path.join(os.getcwd(), schema_file) - -print("Schema Registry URL: " + schema_registry_url) -print("Topic: " + topic) -print("Schema file: " + schema_file) -print - -with open(aboslute_path_to_schema, 'r') as content_file: - schema = content_file.read() - -payload = "{ \"schema\": \"" \ - + schema.replace("\"", "\\\"").replace("\t", "").replace("\n", "") \ - + "\" }" - -url = schema_registry_url + "/subjects/" + topic + "-value/versions" -headers = {"Content-Type": "application/vnd.schemaregistry.v1+json"} - -r = requests.post(url, headers=headers, data=payload) -if r.status_code == requests.codes.ok: - print("Success") -else: - r.raise_for_status() diff --git a/kafka-avro/src/main/resources/spring/camel-context.xml b/kafka-avro/src/main/resources/spring/camel-context.xml deleted file mode 100644 index 886674f78..000000000 --- a/kafka-avro/src/main/resources/spring/camel-context.xml +++ /dev/null @@ -1,34 +0,0 @@ - - - - - - - - - - - - - - -