How to: Set up the Protobuf key and value format (tech preview)

The Protocol Buffer format is available to use for the single-stream variant of the Kafka connector only - it’s currently in tech preview. The properties and setup you see below are subject to change.

Configuration

Property

Description

Required

Default

protobuf.schema

The Protobuf schema describing the input or output shape of your data. This may contain one or more proto message definitions, corresponding to your Kafka message key and/or message value schemas. The schema can’t contain circular dependencies between message fields or custom imports.

See Key/value message class name for more details.

Required when value.format = protobuf or key.format = protobuf

value.protobuf.message-class-name

The proto message in your protobuf schema describing the shape of the input/output Kafka message value.

Required when value.format = protobuf

key.protobuf.message-class-name

The proto message in your protobuf schema describing the shape of the input/output Kafka message key.

Required when key.format = protobuf

Create a connection with the Protobuf format

For this tech preview, a Kafka connection that uses the Protobuf format can only be created using the Decodable CLI.

Here is the Decodable CLI command to create a Kafka connection with the Protobuf key and value format:

The SASL properties can be replaced with properties for the auth type of your choice.

decodable connection create --connector kafka \
  --type <source/sink> \
  --name protobuf_test \
  --stream-id <your_stream_id> \
  --prop bootstrap.servers=<your_broker_url_1;your_broker_url_2> \
  --prop topic=<your_topic> \
  --prop value.format=protobuf \
  --prop key.format=protobuf \
  --prop protobuf.schema=$(cat <your_proto_schema>.proto) \
  --prop key.protobuf.message-class-name=<your_key_message> \
  --prop value.protobuf.message-class-name=<your_value_message> \
  --prop security.protocol=SASL_SSL \
  --prop sasl.mechanism=SCRAM-SHA-256 \
  --prop sasl.username=<your_sasl_username> \
  --prop sasl.password=<your_pwd_secret_id>

See the example below.

Update your connection

To update the protobuf properties on your connection, you may use either the Decodable web app or the CLI.

Using the web app

Navigate to the Configuration tab on the connection you created. There, you can update the schema and the key/value message names you configured.

Using the CLI

Run the command below to update the protobuf properties on the connection you created:

decodable connection update <your_connection_id>
  --prop key.protobuf.message-class-name=<your_updated_key_message>
  --prop value.protobuf.message-class-name=<your_updated_value_message>
  --prop protobuf.schema=$(cat <your_updated_schema>.proto)

Key/value message class name

The connector uses key.protobuf.message-class-name and value.protobuf.message-class-name to retrieve the java classes generated from your protobuf schema. The class names will be the same as the message names in your schema (don’t include the leading package name). The protobuf serializer/deserializer will use the classes to write/read the topic message keys and values accordingly.

As an example, consider the following protobuf schema, where:

MessageKey describes the schema of the message key in the input/output topic
MessageValue describes the schema of the message value in the input/output topic

> cat example.proto

  syntax = "proto3";

  package co.decodable;

  option java_multiple_files = true;
  option java_outer_classname = "ExampleProto";
  option java_package = "com.decodable.proto";


  message MessageKey {
    uint64 key_field1 = 1;
    string key_field2 = 2;
  }

  message MessageValue {
    boolean value_field1 = 1;
    repeated string value_field2 = 2;
  }

With this schema, the configuration would be:

decodable connection create --connector kafka \
  ...
  --field key_field1=bigint \
  --field key_field2=string \
  --field value_field1=boolean \
  --field value_field2=ARRAY<string> \
  --prop protobuf.schema=$(cat example.proto) \
  --prop key.protobuf.message-class-name=MessageKey \
  --prop value.protobuf.message-class-name=MessageValue \
  ...