CloudEvents over Kafka

The Kafka protocol binding defines two ways of carrying a CloudEvent on a Kafka record:

1. Structured mode. The whole event is encoded — typically as JSON — in the record value, with content-type set to application/cloudevents+json; charset=UTF-8.
2. Binary mode. The context attributes are mapped to record headers prefixed ce_, and the event data alone occupies the record value.

The two modes below are implemented in one class so the difference is visible in the producer code rather than described in prose.

import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.internals.RecordHeader;
import com.google.gson.Gson;
import java.nio.charset.StandardCharsets;
import java.time.OffsetDateTime;
import java.util.UUID;

public class CloudEventsKafkaExample {
    // CloudEvent data class
    public static class CloudEvent<T> {
        String id;
        String source;
        String specversion;
        String type;
        OffsetDateTime time;
        T data;

        public CloudEvent(String source, String type, T data) {
            this.id = UUID.randomUUID().toString();
            this.source = source;
            this.specversion = "1.0";
            this.type = type;
            this.time = OffsetDateTime.now();
            this.data = data;
        }
    }

    // Example event data
    public static class OrderCreated {
        String orderId;
        double amount;

        public OrderCreated(String orderId, double amount) {
            this.orderId = orderId;
            this.amount = amount;
        }
    }

    // Structured mode - entire CloudEvent as JSON in the record value
    public static ProducerRecord<String, String> createStructuredCloudEvent(
            String topic, OrderCreated orderData) {

        CloudEvent<OrderCreated> event = new CloudEvent<>(
            "/orders",
            "com.example.order.created",
            orderData
        );

        String jsonValue = new Gson().toJson(event);

        return new ProducerRecord<>(
            topic,
            null,  // key
            jsonValue
        );
    }

    // Binary mode - CloudEvent attributes in headers, data as the value
    public static ProducerRecord<String, String> createBinaryCloudEvent(
            String topic, OrderCreated orderData) {

        CloudEvent<OrderCreated> event = new CloudEvent<>(
            "/orders",
            "com.example.order.created",
            orderData
        );

        String jsonData = new Gson().toJson(orderData);

        ProducerRecord<String, String> record = new ProducerRecord<>(
            topic,
            null,  // key
            jsonData
        );

        // Add CloudEvents attributes as headers
        record.headers().add(
            new RecordHeader("ce_id", event.id.getBytes(StandardCharsets.UTF_8)));
        record.headers().add(
            new RecordHeader("ce_source", event.source.getBytes(StandardCharsets.UTF_8)));
        record.headers().add(
            new RecordHeader("ce_specversion", event.specversion.getBytes(StandardCharsets.UTF_8)));
        record.headers().add(
            new RecordHeader("ce_type", event.type.getBytes(StandardCharsets.UTF_8)));
        record.headers().add(
            new RecordHeader("ce_time", event.time.toString().getBytes(StandardCharsets.UTF_8)));

        return record;
    }

    // Example usage
    public static void main(String[] args) {
        OrderCreated orderData = new OrderCreated("12345", 99.99);

        ProducerRecord<String, String> structuredRecord =
            createStructuredCloudEvent("orders", orderData);

        ProducerRecord<String, String> binaryRecord =
            createBinaryCloudEvent("orders", orderData);

        // Use these records with a Kafka producer...
    }
}

Points worth holding on to:

1. In structured mode the only obligations are that the CloudEvent JSON is valid and that content-type identifies it as application/cloudevents+json. The record survives being bridged to another transport unchanged.
2. In binary mode the ce_ prefix makes the context attributes easy to identify, and a consumer or a stream processor can route and filter on them without deserialising the payload.
3. Binary mode is usually the more efficient of the two, because the event data is not encoded twice.

An official CloudEvents Java SDK provides a Kafka module that performs this mapping, which is preferable to hand-rolling the headers in production code; the example above is written out longhand to show what the binding actually does.