Enhance your career, get your certificate as a Data Streaming Engineer | Get your Certificate

Tutorial

How to handle multiple event types in a Kafka topic with Avro

Kafka
‹ Back to Tutorials

How to handle multiple event types in a Kafka topic with Avro

It's sometimes advantageous to produce distinct but related event types to the same topic, e.g., to guarantee the exact order of different events for the same key. For example, consider pageview and purchase records associated with the same customer ID. In order to properly attribute purchases to preceding pageviews, these distinct events must be sent to the same topic so that the order is preserved in one Kafka topic partition. But, let's say we also need to maintain the topic-name subject constraints with Schema Registry.

To accomplish this with Avro-formatted events, we can use a top-level union schema whose branches are the distinct record types.

Example Avro schema with a union of event types

The example in this tutorial uses a top-level Avro schema specifying that a record is either a purchase or a pageview:

[
  {
    "type":"record",
    "namespace": "io.confluent.developer.avro",
    "name":"Purchase",
    "fields": [
      {"name": "item", "type":"string"},
      {"name": "amount", "type": "double"},
      {"name": "customer_id", "type": "string"}
    ]
  },
  {
    "type":"record",
    "namespace": "io.confluent.developer.avro",
    "name":"Pageview",
    "fields": [
      {"name": "url", "type":"string"},
      {"name": "is_special", "type": "boolean"},
      {"name": "customer_id", "type":  "string"}
    ]
  }
]

Now, if you use the top-level schema for a topic, then you can produce either io.confluent.developer.avro.Purchase or io.confluent.developer.avro.Pageview records to the topic. Only this top-level union schema is registered in Schema Registry (as the avro-events-value subject); the individual Purchase and Pageview types are not registered as separate subjects.

Running the example

In order to run this example, first clone the confluentinc/tutorials GitHub repository (if you haven't already) and navigate to the tutorials directory:

git clone git@github.com:confluentinc/tutorials.git
cd tutorials

Now you can either execute the unit test included with the example, or run the example in Docker or in Confluent Cloud.

Execute the unit tests

To run the unit tests, use the provided Gradle Wrapper:

./gradlew clean :multiple-event-types-avro:kafka:test --info
Run in Docker

Prerequisites

Start Kafka and Schema Registry

Start Kafka by running:

docker compose -f ./docker/docker-compose-kafka-sr.yml up -d

Create topic

Create the avro-events topic:

docker exec -t broker kafka-topics --create --topic avro-events --bootstrap-server broker:9092

Register schema

Run the following task to register the schemas in Schema Registry:

./gradlew :multiple-event-types-avro:kafka:registerSchemasTask

Build the application

Build the application uberjar:

./gradlew :multiple-event-types-avro:kafka:shadowJar

Run the application

Run the application, which produces and consumes pageview and purchase events, with the following command:

java -jar multiple-event-types-avro/kafka/build/libs/multiple-event-types-avro-standalone-0.0.1.jar \
    multiple-event-types-avro/kafka/local.properties

Cleanup

Stop Kafka and Schema Registry:

docker compose -f ./docker/docker-compose-kafka-sr.yml down
Run in Confluent Cloud

Prerequisites

  • Java 17
  • A Confluent Cloud account
  • The Confluent CLI installed on your machine
  • Clone the confluentinc/tutorials repository and navigate into its top-level directory: 
    git clone git@github.com:confluentinc/tutorials.git
cd tutorials

Create Confluent Cloud resources

Log in to your Confluent Cloud account:

confluent login --prompt --save

Install a CLI plugin that will streamline the creation of resources in Confluent Cloud:

confluent plugin install confluent-quickstart

Run the plugin from the top-level directory of the tutorials repository to create the Confluent Cloud resources needed for this tutorial.

Note: You may specify a different cloud provider (gcp or azure) or region. You can find supported regions in a given cloud provider by running confluent kafka region list --cloud <CLOUD>.

confluent quickstart \
  --environment-name kafka-multiple-event-types-env \
  --kafka-cluster-name kafka-multiple-event-types-cluster \
  --create-kafka-key \
  --create-sr-key \
  --kafka-java-properties-file multiple-event-types-avro/kafka/cloud.properties

The plugin should complete in under a minute.

Create topic

Create the topic for the application:

confluent kafka topic create avro-events

Register schemas

Before registering the schemas, update the schemaRegistry block in multiple-event-types-avro/kafka/build.gradle to point at your Confluent Cloud Schema Registry cluster and authenticate with a clientConfig. Populate the placeholders below with the corresponding schema.registry.url and basic.auth.user.info values from multiple-event-types-avro/kafka/cloud.properties:

schemaRegistry {
    url = '<SCHEMA_REGISTRY_URL>'
    clientConfig = [
            'basic.auth.credentials.source': 'USER_INFO',
            'basic.auth.user.info'         : '<SCHEMA_REGISTRY_API_KEY>:<SCHEMA_REGISTRY_API_SECRET>'
    ]
    register {
        subject('avro-events-value', 'multiple-event-types-avro/kafka/src/main/avro/all-events.avsc', "AVRO")
    }
}

Then, run the following task to register the schemas in Schema Registry:

./gradlew :multiple-event-types-avro:kafka:registerSchemasTask

In the Confluent Cloud Console, navigate to Topics in the left-hand navigation, select the avro-events topic, and click Data contract. Validate that a Value schema has been set.

Build the application

Build the application uberjar:

./gradlew :multiple-event-types-avro:kafka:shadowJar

Run the application

Run the application, which produces and consumes pageview and purchase events, with the following command. Note that we are passing the client configuration as an argument:

java -jar multiple-event-types-avro/kafka/build/libs/multiple-event-types-avro-standalone-0.0.1.jar \
    multiple-event-types-avro/kafka/cloud.properties

In the Confluent Cloud Console, select the Messages tab for the avro-events topic and view the messages that are produced.

Clean up

When you are finished, delete the kafka-multiple-event-types-env environment by first getting the environment ID of the form env-123456 corresponding to it:

confluent environment list

Delete the environment, including all resources created for this tutorial:

confluent environment delete <ENVIRONMENT ID>
Do you have questions or comments? Join us in the #developer-confluent-io community Slack channel to engage in discussions with the creators of this content.