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 Protobuf

Kafka
‹ Back to Tutorials

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

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 Protobuf-formatted events, we can use schema references, where a schema contains a field whose type is a reference to another schema.

Example Protobuf schema with references

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

message CustomerEvent {
  oneof action {
    Purchase purchase = 1;
    Pageview pageview = 2;
  }
  string id = 3;
}

Where these references are defined as follows:

message Pageview {
  string url = 1;
  bool is_special = 2;
  string customer_id = 3;
}
message Purchase {
  string item = 1;
  double amount = 2;
  string customer_id = 3;
}

Now, if you use the top-level schema for a topic, then you can produce either PageviewProto.Pageview or PurchaseProto.Purchase records to the topic.

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-protobuf: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 proto-events topic:

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

Build the application

Build the application uberjar:

./gradlew :multiple-event-types-protobuf: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-protobuf/kafka/build/libs/multiple-event-types-protobuf-standalone-0.0.1.jar \
    multiple-event-types-protobuf/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-protobuf/kafka/cloud.properties

The plugin should complete in under a minute.

Create topic

Create the topic for the application:

confluent kafka topic create proto-events

Build the application

Build the application uberjar:

./gradlew :multiple-event-types-protobuf: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-protobuf/kafka/build/libs/multiple-event-types-protobuf-standalone-0.0.1.jar \
    multiple-event-types-protobuf/kafka/cloud.properties

In the Confluent Cloud Console, select the Messages tab for the proto-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.