Confluent Schema Registry with Avro Tools Example

SpecificRecord is used to avoid run time errors while sending Avro Data. We can generate a equivalent java class for an avro schema using avro tools.
Download Avro tool jar from below

https://repo1.maven.org/maven2/org/apache/avro/avro-tools/1.8.2/avro-tools-1.8.2.jar

Here is the Customer avro(customer.avsc).

{
"namespace":"com.kafka.training",
"type":"record",
"name":"Customer",
"fields":[
{"name":"id","type":"int"},
{"name":"name","type":"string"},
{"name":"email","type":"string"}
]
}

Run avro tool on top of avro scema and give target as src location of workspace.

java -jar avro-tools-1.8.2.jar compile schema /home/kafka/eclipse-workspace/AvroSchemaPrj/customer.avsc /home/kafka/eclipse-workspace/AvroSchemaPrj/src

Add below dependancy in pom.xml also.

<dependency<
    <groupId<org.apache.avro</groupId<
    <artifactId<avro-tools</artifactId<
    <version<1.8.2</version<
</dependency<

AvroSenderSpecRecord- Java to send Customer data to test-specific-topic topic using avro tool by avro schema generation.

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Properties;

import org.apache.avro.Schema;
import org.apache.avro.Schema.Parser;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import com.kafka.training.Customer;

import io.confluent.kafka.serializers.KafkaAvroSerializer;

public class AvroSenderSpecRecord {

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		

		Properties props=new Properties();
		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
		props.put("schema.registry.url", "http://localhost:8081");
		
		KafkaProducer producer=new KafkaProducer<>(props);
		Customer cust= Customer.newBuilder().setId(1001).setName("Amulya").setEmail("amulya.s@gmail.com").build();
		ProducerRecord record=new ProducerRecord<>("test-specific-topic","test-customer",
				cust);
		producer.send(record);
		producer.close();
		System.out.println("message sent");
	}

}

Run the AvroSenderSpecRecord.java and run below commands to check output
Here is the commands to verify topic equavalent schema and kafka avro console consumer command

[kafka@localhost bin]$ ./kafka-avro-console-consumer --topic test-specific-topic --bootstrap-server localhost:9092 --from-beginning
{"id":1001,"name":"Amulya","email":"amulya.s@gmail.com"}

[kafka@localhost bin]$ curl -X GET http://localhost:8081/subjects/
["test-specific-topic-value","test-topic-value","next-topic-key","test-topic-1-value","next-topic-value"]


[kafka@localhost bin]$ curl -X GET http://localhost:8081/subjects/test-specific-topic-value/versions/1
{"subject":"test-specific-topic-value","version":1,"id":6,"schema":"{\"type\":\"record\",\"name\":\"Customer\",\"namespace\":\"com.kafka.training\",\"fields\":[{\"name\":\"id\",\"type\":\"int\"},{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"email\",\"type\":\"string\"}]}"}


[kafka@localhost bin]$ curl -X GET http://localhost:8081/schemas/ids/6
{\"name\":\"email\",\"type\":\"string\"}