Se afișează postările cu eticheta AVRO JSON. Afișați toate postările

9/29/2025

Using Schema Registry in Quarkus app

Using Schema Registry in Quarkus Application Development

Schema Registry is a critical component when working with Apache Kafka and Avro in Quarkus applications. It ensures that the data structure (schema) is consistent and compatible across producers and consumers. Here's a concise guide to integrating Schema Registry in a Quarkus application using a Student object.

1. Add Dependencies

Include the necessary dependencies in your pom.xml for Kafka, Avro, and Schema Registry support:


<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-smallrye-reactive-messaging-kafka</artifactId>
</dependency>
<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-apicurio-registry-avro</artifactId>
</dependency>

2. Configure Application Properties

Set up the connection to your Schema Registry in the application.properties file:


# Kafka broker configuration
kafka.bootstrap.servers=localhost:9092

# Schema Registry configuration
mp.messaging.connector.smallrye-kafka.schema.registry.url=http://localhost:8081

# Avro serialization
mp.messaging.outgoing.my-topic.value.serializer=io.apicurio.registry.utils.serde.AvroKafkaSerializer
mp.messaging.incoming.my-topic.value.deserializer=io.apicurio.registry.utils.serde.AvroKafkaDeserializer

🔁 Replace http://localhost:8081 with the URL of your Schema Registry (e.g., Confluent or Apicurio).

3. Define Avro Schema

Create Avro schema files (e.g., student.avsc) and generate Java classes using the Avro Maven plugin:

Example: `student.avsc`


{
  "namespace": "com.example.avro",
  "type": "record",
  "name": "Student",
  "fields": [
    { "name": "name", "type": "string" },
    { "name": "email", "type": "string" },
    { "name": "grade", "type": "int" }
  ]
}

Add the Avro Maven Plugin to `pom.xml`:


<plugin>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <version>1.11.1</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <goals>
                <goal>schema</goal>
            </goals>
            <configuration>
                <sourceDirectory>${project.basedir}/src/main/avro</sourceDirectory>
                <outputDirectory>${project.build.directory}/generated-sources/avro</outputDirectory>
            </configuration>
        </execution>
    </executions>
</plugin>

Place your student.avsc file in src/main/avro.

4. Implement Kafka Producers and Consumers

Use Quarkus' reactive messaging to produce and consume messages with the Student Avro object.

✅ Producer Example:


import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;
import com.example.avro.Student;

@ApplicationScoped
public class KafkaStudentProducer {

    @Inject
    @Channel("my-topic")
    Emitter<Student> emitter;

    public void send(Student student) {
        emitter.send(student);
    }
}

✅ Consumer Example:


import org.eclipse.microprofile.reactive.messaging.Incoming;
import com.example.avro.Student;

public class KafkaStudentConsumer {

    @Incoming("my-topic")
    public void consume(Student student) {
        System.out.println("Received student: " + student);
    }
}

Here, Student is the Java class generated from your Avro schema (student.avsc).

5. Test Schema Compatibility

Ensure your producer and consumer schemas are compatible. The Schema Registry (e.g., Apicurio or Confluent) will validate compatibility automatically at runtime.

✅ Summary

By following these steps, you can seamlessly integrate Schema Registry into your Quarkus application, enabling robust, versioned, and schema-compliant Kafka messaging with Avro.

Using the Student object example, you’ve seen:

How to configure Schema Registry
Define and generate Avro classes
Produce and consume Kafka messages with reactive messaging
Automatically validate schema compatibility at runtime

💡 Tip: If you're using Apicurio Registry, you can also explore features like artifact versioning, API-based registration, and schema evolution rules (BACKWARD, FORWARD, FULL compatibility).

Avro ↔ JSON Conversion in Java

Full Maven Project Template

In this guide, you’ll learn how to:

Define an Avro schema for a Product (fields: name, code, price)
Read Avro-serialized data (generic or specific)
Convert Avro records to JSON via Jackson
Use a full Maven project setup to tie it all together

1. Project Structure

Here’s a suggested directory layout:


product-avro-json/
├── pom.xml
├── src
│   ├── main
│   │   ├── avro
│   │   │   └── product.avsc
│   │   └── java
│   │       └── com/example
│   │           ├── AvroReader.java
│   │           ├── AvroToJsonConverter.java
│   │           └── MainApp.java
│   └── test
│       └── java
│           └── com/example
│               └── AvroJsonTest.java

2. `pom.xml`


<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="
           http://maven.apache.org/POM/4.0.0
           http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.example</groupId>
  <artifactId>product-avro-json</artifactId>
  <version>1.0-SNAPSHOT</version>
  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>11</maven.compiler.source>
    <maven.compiler.target>11</maven.compiler.target>
    <avro.version>1.10.2</avro.version>
    <jackson.version>2.12.3</jackson.version>
  </properties>

  <dependencies>
    <!-- Avro runtime -->
    <dependency>
      <groupId>org.apache.avro</groupId>
      <artifactId>avro</artifactId>
      <version>${avro.version}</version>
    </dependency>
    <!-- Avro compiler (for code generation) -->
    <dependency>
      <groupId>org.apache.avro</groupId>
      <artifactId>avro-compiler</artifactId>
      <version>${avro.version}</version>
    </dependency>
    <!-- Jackson for JSON conversion -->
    <dependency>
      <groupId>com.fasterxml.jackson.core</groupId>
      <artifactId>jackson-databind</artifactId>
      <version>${jackson.version}</version>
    </dependency>
    <!-- JUnit (for testing) -->
    <dependency>
      <groupId>org.junit.jupiter</groupId>
      <artifactId>junit-jupiter-api</artifactId>
      <version>5.7.0</version>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>org.junit.jupiter</groupId>
      <artifactId>junit-jupiter-engine</artifactId>
      <version>5.7.0</version>
      <scope>test</scope>
    </dependency>
  </dependencies>

  <build>
    <plugins>
      <!-- Avro Maven Plugin: generate Java classes from .avsc -->
      <plugin>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro-maven-plugin</artifactId>
        <version>${avro.version}</version>
        <executions>
          <execution>
            <phase>generate-sources</phase>
            <goals>
              <goal>schema</goal>
            </goals>
            <configuration>
              <sourceDirectory>${project.basedir}/src/main/avro</sourceDirectory>
              <outputDirectory>${project.build.directory}/generated-sources/avro</outputDirectory>
            </configuration>
          </execution>
        </executions>
      </plugin>

      <!-- Ensure generated sources are compiled -->
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>3.8.1</version>
        <configuration>
          <source>11</source>
          <target>11</target>
          <annotationProcessorPaths>
            <!-- include avro plugin if needed -->
          </annotationProcessorPaths>
        </configuration>
        <executions>
          <execution>
            <id>default-compile</id>
            <goals><goal>compile</goal></goals>
          </execution>
          <execution>
            <id>default-testCompile</id>
            <goals><goal>testCompile</goal></goals>
          </execution>
        </executions>
      </plugin>
    </plugins>
  </build>
</project>

This pom.xml does:

Include avro and avro-compiler
Use the avro-maven-plugin to generate Java classes from .avsc
Include Jackson for JSON serialization
Setup compilation and test dependencies

3. Avro Schema: `product.avsc`

Place this in src/main/avro/product.avsc:


{
  "namespace": "com.example.avro",
  "type": "record",
  "name": "Product",
  "fields": [
    { "name": "name", "type": "string" },
    { "name": "code", "type": "string" },
    { "name": "price", "type": "double" }
  ]
}

After you build the project, the Avro plugin will produce a generated Java class com.example.avro.Product.

4. Java Source Files

`AvroReader.java`


package com.example;

import org.apache.avro.file.DataFileReader;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.io.DatumReader;

import java.io.File;
import java.io.IOException;

public class AvroReader {

    public static Iterable<GenericRecord> readGenericRecords(String avroFilePath) throws IOException {
        File file = new File(avroFilePath);
        DatumReader<GenericRecord> datumReader = new GenericDatumReader<>();
        DataFileReader<GenericRecord> dataFileReader = new DataFileReader<>(file, datumReader);
        return dataFileReader;
    }

    public static <T> Iterable<T> readSpecificRecords(String avroFilePath, Class<T> clazz) throws IOException {
        // Not full implementation; for specific, you’d use SpecificDatumReader
        throw new UnsupportedOperationException("Not implemented");
    }
}

`AvroToJsonConverter.java`


package com.example;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.avro.generic.GenericRecord;

import java.io.IOException;

public class AvroToJsonConverter {
    private static final ObjectMapper objectMapper = new ObjectMapper();

    public static String convertGenericRecordToJson(GenericRecord record) throws IOException {
        return objectMapper.writeValueAsString(record);
    }
}

`MainApp.java`


package com.example;

import org.apache.avro.generic.GenericRecord;

import java.io.IOException;

public class MainApp {

    public static void main(String[] args) {
        if (args.length != 1) {
            System.err.println("Usage: java -jar product-avro-json.jar <path-to-avro-file>");
            System.exit(1);
        }
        String avroFilePath = args[0];

        try {
            for (GenericRecord rec : AvroReader.readGenericRecords(avroFilePath)) {
                String json = AvroToJsonConverter.convertGenericRecordToJson(rec);
                System.out.println(json);
            }
        } catch (IOException e) {
            System.err.println("Error: " + e.getMessage());
            e.printStackTrace();
        }
    }
}

5. (Optional) Test Example: `AvroJsonTest.java`

You could put a test to check the conversion:


package com.example;

import org.apache.avro.generic.GenericRecord;
import org.junit.jupiter.api.Test;

import java.io.File;
import java.io.IOException;

import static org.junit.jupiter.api.Assertions.*;

public class AvroJsonTest {

    @Test
    public void testConversion() throws IOException {
        String avroFile = "src/test/resources/test-products.avro";
        Iterable<GenericRecord> recs = AvroReader.readGenericRecords(avroFile);
        boolean found = false;
        for (GenericRecord rec : recs) {
            String json = AvroToJsonConverter.convertGenericRecordToJson(rec);
            assertTrue(json.contains("\"name\""));
            assertTrue(json.contains("\"code\""));
            assertTrue(json.contains("\"price\""));
            found = true;
        }
        assertTrue(found, "No record found in test Avro file");
    }
}

You’d have to supply a test Avro file in src/test/resources/test-products.avro.

6. How to Build & Run

Compile & generate sources:
```
mvn clean compile
```
The Avro plugin will generate Java classes from product.avsc in the target/generated-sources/avro directory.
Package into a JAR:
```
mvn package
```
This produces product-avro-json-1.0-SNAPSHOT.jar in target/.
Run the application:
```
java -jar target/product-avro-json-1.0-SNAPSHOT.jar path/to/your/products.avro
```
It will print JSON lines corresponding to each record in the Avro file.

7. Explanation & Notes

The Avro Maven plugin reads .avsc files in src/main/avro and generates Java classes under target/generated-sources/avro.
In this example, we use generic records via GenericRecord to read data; this is flexible and schema-driven.
We use Jackson to convert Avro GenericRecord to JSON.
If you prefer specific records (i.e. use the generated Product class directly), you can replace generic reading with SpecificDatumReader<Product> and then do objectMapper.writeValueAsString(productInstance).
Be careful about null values or optional fields if you extend the schema.

Using Avro with a Quarkus Application + Kafka + Schema Registry

Using Avro with a Quarkus Application + Kafka + Schema Registry (Product Example)

This guide walks you through building a Quarkus application that:

Sends and receives Kafka messages
Uses Avro for message serialization
Integrates with a Schema Registry
Works with a custom Product object

Prerequisites

You'll need:

Java 17+
Maven
Docker (for Kafka & Schema Registry)
Basic understanding of Quarkus, Kafka, and Avro

Architecture Overview

We’ll build a system with:

A REST API to send Product data to a Kafka topic
A Kafka consumer that reads Product messages
A REST API to stream consumed products using SSE (Server-Sent Events)
Avro used for data serialization
Schema stored and retrieved from a Schema Registry

1. Create the Quarkus Project

Generate a project with required extensions:


quarkus create app org.acme:product-avro-kafka \
  --extension='rest-jackson,messaging-kafka,apicurio-registry-avro' \
  --no-code

cd product-avro-kafka

2. Define the Avro Schema

Create the Avro schema file product.avsc inside src/main/avro:


{
  "namespace": "org.acme.kafka.quarkus",
  "type": "record",
  "name": "Product",
  "fields": [
    { "name": "name", "type": "string" },
    { "name": "price", "type": "double" }
  ]
}

When you build the app, Quarkus automatically generates the Java class Product.java based on this schema.

3. Implement the Kafka Producer

Create ProductResource.java to expose a REST endpoint that sends product data to Kafka:


package org.acme.kafka;

import org.acme.kafka.quarkus.Product;
import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;
import org.jboss.logging.Logger;

import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.core.Response;

@Path("/products")
public class ProductResource {

    private static final Logger LOGGER = Logger.getLogger(ProductResource.class);

    @Channel("products")
    Emitter<Product> emitter;

    @POST
    public Response sendProduct(Product product) {
        LOGGER.infof("Sending product: %s - $%.2f", product.getName(), product.getPrice());
        emitter.send(product);
        return Response.accepted().build();
    }
}

4. Kafka Configuration

In src/main/resources/application.properties, add:


# Kafka producer config
mp.messaging.outgoing.products.connector=smallrye-kafka
mp.messaging.outgoing.products.topic=products
mp.messaging.outgoing.products.apicurio.registry.auto-register=true

If using Confluent Schema Registry, use quarkus-confluent-registry-avro extension instead and replace the apicurio property with:


mp.messaging.outgoing.products.auto.register.schemas=true

5. Implement the Kafka Consumer and SSE Streaming

Create a class ConsumedProductResource.java:


package org.acme.kafka;

import jakarta.enterprise.context.ApplicationScoped;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.MediaType;

import org.acme.kafka.quarkus.Product;
import org.eclipse.microprofile.reactive.messaging.Channel;
import io.smallrye.mutiny.Multi;
import org.jboss.resteasy.reactive.RestStreamElementType;

@ApplicationScoped
@Path("/consumed-products")
public class ConsumedProductResource {

    @Channel("products-from-kafka")
    Multi<Product> products;

    @GET
    @Produces(MediaType.SERVER_SENT_EVENTS)
    @RestStreamElementType(MediaType.TEXT_PLAIN)
    public Multi<String> stream() {
        return products.map(p -> String.format("Product: %s - $%.2f", p.getName(), p.getPrice()));
    }
}

6. Kafka Consumer Configuration

Add to application.properties:


# Kafka consumer config
mp.messaging.incoming.products-from-kafka.connector=smallrye-kafka
mp.messaging.incoming.products-from-kafka.topic=products
mp.messaging.incoming.products-from-kafka.auto.offset.reset=earliest
mp.messaging.incoming.products-from-kafka.enable.auto.commit=false

7. Running the Application in Dev Mode

Run the application:


./mvnw quarkus:dev

Thanks to Dev Services, Kafka and Apicurio Schema Registry are started automatically in dev mode. No extra Docker setup needed.

Test the endpoints:

Send a product:


curl -X POST -H "Content-Type: application/json" \
     -d '{"name":"Laptop","price":999.99}' \
     http://localhost:8080/products

View the streamed products:


curl -N http://localhost:8080/consumed-products

You’ll see output like:


Product: Laptop - $999.99

8. Running in Production (JVM or Native)

In production, you'll need to run Kafka and Schema Registry yourself (e.g., via Docker). Here's a minimal docker-compose.yml:


version: '2'
services:
  zookeeper:
    image: quay.io/strimzi/kafka:0.41.0-kafka-3.7.0
    ...

  kafka:
    image: quay.io/strimzi/kafka:0.41.0-kafka-3.7.0
    ...

  schema-registry:
    image: apicurio/apicurio-registry-mem:2.4.2.Final
    ports:
      - "8081:8080"
    environment:
      - REGISTRY_STORAGE=mem

You will also need to configure these properties:


kafka.bootstrap.servers=localhost:9092
mp.messaging.connector.smallrye-kafka.schema.registry.url=http://localhost:8081/apis/registry/v2

9. Building the Application

For JVM build:


./mvnw package
java -jar target/quarkus-app/quarkus-run.jar

For native build (requires GraalVM):


./mvnw package -Pnative
./target/product-avro-kafka-1.0.0-SNAPSHOT-runner

10. Summary

You've just built a Quarkus application that:

✅ Defines a Product Avro schema
✅ Generates Java code from .avsc files
✅ Produces and consumes Avro messages using Kafka
✅ Uses Apicurio (or Confluent) Schema Registry
✅ Leverages Quarkus Dev Services for quick setup
✅ Streams data with Server-Sent Events (SSE)

AVRO

Updated Guide: Apache Avro™ with Java

This is a short guide for getting started with Apache Avro using Java. This guide only covers using Avro for data serialization;

📥 Download

Avro implementations for C, C++, C#, Java, PHP, Python, and Ruby can be downloaded from the Apache Avro™ Download page. This guide uses Avro 1.12.0, the latest version at the time of writing. For the examples in this guide, download avro-1.12.0.jar and avro-tools-1.12.0.jar.

Alternatively, if you are using Maven, add the following dependency to your pom.xml:


<dependency>
  <groupId>org.apache.avro</groupId>
  <artifactId>avro</artifactId>
  <version>1.12.0</version>
</dependency>

As well as the Avro Maven plugin (for performing code generation):


<plugin>
  <groupId>org.apache.avro</groupId>
  <artifactId>avro-maven-plugin</artifactId>
  <version>1.12.0</version>
  <configuration>
    <sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
    <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
  </configuration>
  <executions>
    <execution>
      <phase>generate-sources</phase>
      <goals>
        <goal>schema</goal>
      </goals>
    </execution>
  </executions>
</plugin>
<plugin>
  <groupId>org.apache.maven.plugins</groupId>
  <artifactId>maven-compiler-plugin</artifactId>
  <configuration>
    <source>1.8</source>
    <target>1.8</target>
  </configuration>
</plugin>

You may also build the required Avro jars from source. Building Avro is beyond the scope of this guide;

🧾 Defining a Schema

Avro schemas are defined using JSON or IDL (the latter requires an extra dependency). Schemas are composed of primitive types and complex types.

Here’s a simple schema example, student.avsc:


{
 "namespace": "example.avro",
 "type": "record",
 "name": "Student",
 "fields": [
     {"name": "name", "type": "string"},
     {"name": "favorite_number",  "type": ["int", "null"]},
     {"name": "favorite_color", "type": ["string", "null"]}
 ]
}

This schema defines a record representing a hypothetical student. At minimum, a record definition must include its type ("record"), a name ("Student"), and its fields: name, favorite_number, and favorite_color.

⚙️ Serializing and Deserializing with Code Generation

Compiling the Schema

You can generate Java classes from the schema using avro-tools:


java -jar /path/to/avro-tools-1.12.0.jar compile schema student.avsc .

This will generate the appropriate source files in a package based on the schema’s namespace.

👩‍💻 Creating Students


Student student1 = new Student();
student1.setName("Alyssa");
student1.setFavoriteNumber(256);

// Alternate constructor
Student student2 = new Student("Ben", 7, "red");

// Construct via builder
Student student3 = Student.newBuilder()
             .setName("Charlie")
             .setFavoriteColor("blue")
             .setFavoriteNumber(null)
             .build();

Avro objects can be created by constructors or builders. Builders provide schema validation and handle default values, while constructors are faster.

💾 Serializing


DatumWriter<Student> studentDatumWriter = new SpecificDatumWriter<>(Student.class);
DataFileWriter<Student> dataFileWriter = new DataFileWriter<>(studentDatumWriter);

dataFileWriter.create(student1.getSchema(), new File("students.avro"));
dataFileWriter.append(student1);
dataFileWriter.append(student2);
dataFileWriter.append(student3);
dataFileWriter.close();

📂 Deserializing


DatumReader<Student> studentDatumReader = new SpecificDatumReader<>(Student.class);
DataFileReader<Student> dataFileReader = new DataFileReader<>(file, studentDatumReader);
Student student = null;
while (dataFileReader.hasNext()) {
    student = dataFileReader.next(student);
    System.out.println(student);
}

Output:


{"name": "Alyssa", "favorite_number": 256, "favorite_color": null}
{"name": "Ben", "favorite_number": 7, "favorite_color": "red"}
{"name": "Charlie", "favorite_number": null, "favorite_color": "blue"}

🧪 Serializing and Deserializing Without Code Generation

You can also use GenericRecords to avoid code generation.


Schema schema = new SchemaParser().parse(new File("student.avsc")).mainSchema();

GenericRecord student1 = new GenericData.Record(schema);
student1.put("name", "Alyssa");
student1.put("favorite_number", 256);

GenericRecord student2 = new GenericData.Record(schema);
student2.put("name", "Ben");
student2.put("favorite_number", 7);
student2.put("favorite_color", "red");

Serialize:


DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(schema);
DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<>(datumWriter);

dataFileWriter.create(schema, new File("students.avro"));
dataFileWriter.append(student1);
dataFileWriter.append(student2);
dataFileWriter.close();

Deserialize:


DatumReader<GenericRecord> datumReader = new GenericDatumReader<>(schema);
DataFileReader<GenericRecord> dataFileReader = new DataFileReader<>(file, datumReader);

GenericRecord student = null;
while (dataFileReader.hasNext()) {
    student = dataFileReader.next(student);
    System.out.println(student);
}

Output:


{"name": "Alyssa", "favorite_number": 256, "favorite_color": null}
{"name": "Ben", "favorite_number": 7, "favorite_color": "red"}

🚀 Compiling and Running the Example

Navigate to the project directory and run:


$ mvn compile
$ mvn -q exec:java -Dexec.mainClass=example.SpecificMain

For the generic version:


$ mvn compile
$ mvn -q exec:java -Dexec.mainClass=example.GenericMain

⚡ Beta Feature: Faster Code Generation

Enable faster encoding/decoding with:


$ mvn -q exec:java -Dexec.mainClass=example.SpecificMain \
  -Dorg.apache.avro.specific.use_custom_coders=true

No schema recompilation is required. This is a runtime toggle via a system property.

🧠 Summary

You’ve now learned how to:

Define a schema for a Student
Serialize/deserialize using Avro with and without code generation
Use Avro tools and Maven integration
Optimize performance with reuse patterns and optional beta features

9/29/2025