Produce Avro messages with the schema registry

Set up a client VM

Create a Linux virtual machine (VM) instance in Compute Engine that can access the Kafka cluster. When you configure the VM, set the following options:

• Region. Create the VM in the same region as your Kafka cluster.

• Subnet. Create the VM in the same VPC network as the subnet that you used in your Kafka cluster configuration. For more information, see View a cluster's subnets.

• Access scopes. Assign the https://www.googleapis.com/auth/cloud-platform access scope to the VM. This scope authorizes the VM to send requests to the Managed Kafka API.

The following steps show how to set these options.

gcloud compute instances create VM_NAME \
  --scopes=https://www.googleapis.com/auth/cloud-platform \
  --subnet=projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET \
  --zone=ZONE

For more information about creating a VM, see Create a VM instance in a specific subnet.

Grant IAM roles

Grant the following Identity and Access Management (IAM) roles to the Compute Engine default service account:

• Managed Kafka Client (roles/managedkafka.client)
• Schema Registry Admin (roles/managedkafka.schemaRegistryAdmin)
• Service Account Token Creator (roles/iam.serviceAccountTokenCreator)

• Service Account OpenID Token Creator (roles/iam.serviceAccountOpenIdTokenCreator)

gcloud projects add-iam-policy-binding PROJECT_ID \
  --member="serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com" \
  --role=roles/managedkafka.client

gcloud projects add-iam-policy-binding PROJECT_ID\
  --member="serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com" \
  --role=roles/managedkafka.schemaRegistryAdmin

gcloud projects add-iam-policy-binding PROJECT_ID\
  --member="serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com" \
  --role=roles/iam.serviceAccountTokenCreator

gcloud projects add-iam-policy-binding PROJECT_ID \
  --member="serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com" \
  --role=roles/iam.serviceAccountOpenIdTokenCreator

gcloud projects describe PROJECT_ID

Connect to the VM

Use SSH to connect to the VM instance.

gcloud compute ssh VM_NAME \
  --project=PROJECT_ID \
  --zone=ZONE

Additional configuration might be required for first time SSH usage. For more information, see About SSH connections.

Set up an Apache Maven project

From your SSH session, run the following commands to set up a Maven project.

1. Install Java and Maven with the command:

   sudo apt-get install maven openjdk-17-jdk
   

2. Set up an Apache Maven project.

   Use the following command to create a package com.google.example in a directory called demo.

   mvn archetype:generate -DartifactId=demo -DgroupId=com.google.example\
      -DarchetypeArtifactId=maven-archetype-quickstart\
      -DarchetypeVersion=1.5 -DinteractiveMode=false
   

Define the schema and its Java implementation

In this example, a message represents a "user" that has a name and an optional ID. The corresponds to an Avro schema with two fields: a required field name of type string and an optional integer id. To use this schema in a Java program, you will also need to generate an Java implementation of an object corresponding to this schema.

1. Change into the project directory with cd demo.

2. Create the folders for storing schema files in your code:

   mkdir -p src/main/avro
   

3. Create the Avro schema definition by pasting the following code into a file called src/main/avro/User.avsc:

   {
     "namespace": "com.google.example",
     "type": "record",
     "name": "User",
     "fields": [
       {"name": "name", "type": "string"},
       {"name": "id",  "type": ["int", "null"]}
     ]
   }
   

4. Configure your Maven project to use an Avro Java code generation plugin by adding the following to the build node of your pom.xml. Note that the pom.xml may have other plugins nodes inside the pluginManagement node. Don't change the pluginManagement node in this step. The plugins node needs to be at the same level level as pluginManagement.

   <plugins>
   <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.basedir}/src/main/java/</outputDirectory>
         </configuration>
       </execution>
     </executions>
   </plugin>
   </plugins>
   

5. Add Avro as a dependency by adding the following to the end of the project/dependencies node of pom.xml. Note that the pom.xml already has a dependencies node inside dependencyManagement tag. Don't change the dependencyManagement node in this step.

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

6. Generate Java sources

     mvn generate-sources
   

7. Run the following command to check that the implementation source file was created. The source is a Java class file that implements constructors, accessors, serializers and de-serializers for User objects. You will use this class in the producer code.

   cat src/main/java/com/google/example/User.java
   

For more information about Apache Avro, see Apache Avro getting started guide.

Create a producer client

This section walks through the steps of writing, building, and running a producer client.

Implement the producer

The producer uses KafkaAvroSerializer.java to encode messages and manage their schemas. The serializer automatically connects to the schema registry, registers the schema under a subject, retrieves its ID, and then serializes the message using Avro. You still need to configure the producer and serializer.

1. Create the producer client class by pasting the following code into a new file called src/main/java/com/google/example/UserProducer.java

   
   package com.google.example;
   import java.util.Properties;
   import org.apache.kafka.clients.producer.KafkaProducer;
   import org.apache.kafka.clients.producer.ProducerRecord;
   import org.apache.kafka.clients.producer.RecordMetadata;
   import org.apache.kafka.clients.producer.ProducerConfig;
   import org.apache.kafka.common.serialization.StringSerializer;
   import io.confluent.kafka.serializers.KafkaAvroSerializer;
   
     public class UserProducer {
   
       private static Properties configure() throws Exception {
           Properties p = new Properties();
           p.load(new java.io.FileReader("client.properties"));
           p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,