Manage schemas with Schema Registry - ApsaraMQ for Kafka

Prerequisites

You have an Alibaba Cloud MQ for Confluent instance. For more information, see Purchase and deploy an instance.
You have access permissions for the Kafka and Schema Registry clusters. For more information, see RBAC authorization.
Java 8 or 11 is installed. For information about Java version compatibility, see Java version support in Confluent Platform.
Maven 3.8 or later is installed. For more information, see Install Maven.

Step 1: Prepare the sample code

Run the following commands to clone the sample code and switch to the 7.9.0-post branch.

git clone https://github.com/confluentinc/examples.git
cd examples/clients/avro
git checkout 7.9.0-post

Create a client configuration file named java.config in the $HOME/.confluent/ directory, where $HOME is your user's home directory. Add the following settings to the file.

# Required connection configs for Kafka producer, consumer, and admin
bootstrap.servers={{ BROKER_ENDPOINT }}
security.protocol=SASL_SSL
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='{{ CLUSTER_API_KEY }}' password='{{ CLUSTER_API_SECRET }}';
sasl.mechanism=PLAIN
# Required for correctness in Apache Kafka clients prior to 2.6
client.dns.lookup=use_all_dns_ips
# Best practice for higher availability in Apache Kafka clients prior to 3.0
session.timeout.ms=45000
# Best practice for Kafka producer to prevent data loss
acks=all
# Required connection configs for Confluent Cloud Schema Registry
schema.registry.url=https://{{ SR_ENDPOINT }}
basic.auth.credentials.source=USER_INFO
basic.auth.user.info={{ SR_API_KEY }}:{{ SR_API_SECRET }}

Parameter	Description	Example value
BROKER_ENDPOINT	The endpoint of the KAFKA service. Obtain the endpoint from the Access Links and Ports page of the ApsaraMQ for Confluent console. To use the public endpoint, you must enable public network access. For details on other security configurations, see Network access and security settings.	pub-kafka-xxxxxxxxxxx.csp.aliyuncs.com:9092
CLUSTER_API_KEY	The username and password of an LDAP user from the Users page in the ApsaraMQ for Confluent console. During testing, you can use the root account and its password. To use other users, create them in the ApsaraMQ for Confluent console and grant them the required permissions for the Kafka cluster. For details, see Manage users and grant permissions.	root
CLUSTER_API_SECRET		**
SR_ENDPOINT	The endpoint of the SCHEMA_REGISTRY service. Obtain the endpoint from the Access Links and Ports page of the ApsaraMQ for Confluent console. To use the public endpoint, you must enable public network access. For details on other security configurations, see Network access and security settings.	pub-schemaregistry-xxxxxxxxxxx.csp.aliyuncs.com:443
SR_API_KEY	The username and password of an LDAP user from the Users page in the ApsaraMQ for Confluent console. During testing, you can use the root account and its password. To use other users, create them in the ApsaraMQ for Confluent console and grant them the required permissions for Schema Registry. For details, see Manage users and grant permissions.	root
SR_API_SECRET		**

Step 2: Create a topic

Note

For this tutorial, create a topic named transactions. If you use a different topic name, you must update the parameter in the code.

Log on to Control Center. On the Home page, click the controlcenter.clusterk card to go to the Cluster overview page.

The left-side navigation menu contains options such as Cluster overview, Brokers, Topics, and Consumers. The right side of the page displays an overview of the cluster's brokers and topics.
In the left-side navigation pane, click Topics. On the topic list page, click + Add topic.
On the New topic page, set a topic name and the number of partitions, and then click Create with defaults.

For example, set Topic name to transactions and Number of partitions to 1.
After you create the topic, you are redirected to its details page.

The page includes four tabs: Overview, Messages, Schema, and Configuration. The Overview tab shows metrics such as Production (production rate), Consumption (consumption rate), and Availability (availability).

Step 3: Enable schema validation

On the topic details page, click the Configuration tab, and then click Edit settings.
Then, click Switch to expert mode.
Set the confluent_value_schema_validation field to true and then click Save changes to enable schema validation for message content. After this feature is enabled, message formats are validated during production and consumption.

Step 4: Create a schema

Navigate to the examples/clients/avro directory of your project and run the following command to view the contents of the Payment.avsc file.

cat src/main/resources/avro/io/confluent/examples/clients/basicavro/Payment.avsc

The command returns the following output:

{
 "namespace": "io.confluent.examples.clients.basicavro",
 "type": "record",
 "name": "Payment",
 "fields": [
     {"name": "id", "type": "string"},
     {"name": "amount", "type": "double"}
 ]
}

On the topic details page in the Control Center console, click the Schema tab, and then click Set a schema.
On the Schema tab, click Avro, paste the content from the Payment.avsc file into the editor, and click Create.

Step 5: Send and consume messages

Send messages

If your schema uses the Avro validation format, you must specify the message serialization method as the KafkaAvroSerializer class and configure the message value class as the Payment class when you send messages.

The following code shows an example:

Producer example code

import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
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 io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.kafka.common.errors.SerializationException;
import java.util.Properties;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.io.FileInputStream;
import java.io.InputStream;
public class ProducerExample {
    private static final String TOPIC = "transactions";
    private static final Properties props = new Properties();
    private static String configFile;
    @SuppressWarnings("InfiniteLoopStatement")
    public static void main(final String[] args) throws IOException {
        if (args.length < 1) {
          // Backwards compatibility, assume localhost
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
          props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
        } else {
          // Load properties from a local configuration file
          // Create the configuration file (e.g. at '$HOME/.confluent/java.config') with configuration parameters
          // to connect to your Kafka cluster, which can be on your local host, Confluent Cloud, or any other cluster.
          // Documentation at https://docs.confluent.io/platform/current/tutorials/examples/clients/docs/java.html
          configFile = args[0];
          if (!Files.exists(Paths.get(configFile))) {
            throw new IOException(configFile + " not found.");
          } else {
            try (InputStream inputStream = new FileInputStream(configFile)) {
              props.load(inputStream);
            }
          }
        }
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
        try (KafkaProducer<String, Payment> producer = new KafkaProducer<String, Payment>(props)) {
            for (long i = 0; i < 10; i++) {
                final String orderId = "id" + Long.toString(i);
                final Payment payment = new Payment(orderId, 1000.00d);
                final ProducerRecord<String, Payment> record = new ProducerRecord<String, Payment>(TOPIC, payment.getId().toString(), payment);
                producer.send(record);
                Thread.sleep(1000L);
            }
            producer.flush();
            System.out.printf("Successfully produced 10 messages to a topic called %s%n", TOPIC);
        } catch (final SerializationException e) {
            e.printStackTrace();
        } catch (final InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Follow these steps to send messages:

Go to the examples/clients/avro project directory and run the following command to compile the project.
```
mvn clean compile package
```

After the compilation completes, run the following command to produce messages.

mvn exec:java -Dexec.mainClass=io.confluent.examples.clients.basicavro.ProducerExample \
  -Dexec.args="$HOME/.confluent/java.config"

The following output indicates that the messages were produced successfully.

...
Successfully produced 10 messages to a topic called transactions
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
...

You can view the produced messages in the Control Center console.

On the Topics page, select the transactions topic and click the Messages tab. You will see 10 messages with keys ranging from "id0" to "id9", and the value is in JSON format, for example, {"id":"id9","amount":1000}.

Consume messages

If the schema's validation format is Avro, when you consume messages, you must specify the message deserialization method as the KafkaAvroDeSerializer class and configure the message value class as the Payment class.

The following code shows an example:

Consumer example code

import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.io.FileInputStream;
import java.io.InputStream;
public class ConsumerExample {
    private static final String TOPIC = "transactions";
    private static final Properties props = new Properties();
    private static String configFile;
    @SuppressWarnings("InfiniteLoopStatement")
    public static void main(final String[] args) throws IOException {
        if (args.length < 1) {
          // Backwards compatibility, assume localhost
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
          props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
        } else {
          // Load properties from a local configuration file
          // Create the configuration file (e.g. at '$HOME/.confluent/java.config') with configuration parameters
          // to connect to your Kafka cluster, which can be on your local host, Confluent Cloud, or any other cluster.
          // Documentation at https://docs.confluent.io/platform/current/tutorials/examples/clients/docs/java.html
          configFile = args[0];
          if (!Files.exists(Paths.get(configFile))) {
            throw new IOException(configFile + " not found.");
          } else {
            try (InputStream inputStream = new FileInputStream(configFile)) {
              props.load(inputStream);
            }
          }
        }
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-payments");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
        props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true); 
        try (final KafkaConsumer<String, Payment> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList(TOPIC));
            while (true) {
                final ConsumerRecords<String, Payment> records = consumer.poll(Duration.ofMillis(100));
                for (final ConsumerRecord<String, Payment> record : records) {
                    final String key = record.key();
                    final Payment value = record.value();
                    System.out.printf("key = %s, value = %s%n", key, value);
                }
            }
        }
    }
}

Follow these steps to consume messages:

If you have not yet compiled the project, go to the examples/clients/avro project directory and run the following command:
```
mvn clean compile package
```

Run the following command to consume messages.

mvn exec:java -Dexec.mainClass=io.confluent.examples.clients.basicavro.ConsumerExample \
  -Dexec.args="$HOME/.confluent/java.config"

The following output indicates that the messages were consumed successfully.

...
key = id0, value = {"id": "id0", "amount": 1000.0}
key = id1, value = {"id": "id1", "amount": 1000.0}
key = id2, value = {"id": "id2", "amount": 1000.0}
key = id3, value = {"id": "id3", "amount": 1000.0}
key = id4, value = {"id": "id4", "amount": 1000.0}
key = id5, value = {"id": "id5", "amount": 1000.0}
key = id6, value = {"id": "id6", "amount": 1000.0}
key = id7, value = {"id": "id7", "amount": 1000.0}
key = id8, value = {"id": "id8", "amount": 1000.0}
key = id9, value = {"id": "id9", "amount": 1000.0}
...

References

For more information about Schema Registry, see Schema Registry Overview.