Protobuf Serialization with Kafka Streams: A Complete Example

The choice of serialization format directly affects throughput, schema evolution, and operational overhead in a Kafka Streams application. Protocol Buffers (Protobuf) is a strong alternative to Avro for teams already invested in gRPC, or for anyone who wants a compact, statically-typed wire format with code-generated message classes. This guide builds a complete transaction-enrichment pipeline that consumes Protobuf-encoded records, performs a stateful join, and produces enriched Protobuf output, using the Confluent Schema Registry & Connect Ecosystem for schema management and compatibility enforcement.

Every snippet targets Apache Kafka 3.6 and Confluent Platform 7.6 (the Confluent release built on Kafka 3.6), and is written to compile and run as shown. Where a configuration choice has a non-obvious gotcha, it is called out explicitly.

Prerequisites and Dependency Configuration

Protobuf support in Kafka Streams needs three things: the Confluent Protobuf Serde, the Protobuf runtime, and the Java classes generated from your .proto files. The Confluent Serde sits between your application code and the Schema Registry, handling the wire-format framing (a magic byte plus a 4-byte schema ID) and the registry lookups.

Keep the Confluent and Apache Kafka versions aligned: Confluent Platform 7.6.x is built against Apache Kafka 3.6.x, so the io.confluent artifacts below are pinned to 7.6.0.

Gradle:

dependencies {
    implementation 'org.apache.kafka:kafka-streams:3.6.0'
    implementation 'io.confluent:kafka-streams-protobuf-serde:7.6.0'
    implementation 'com.google.protobuf:protobuf-java:3.24.0'
    // Optional: only needed if you use the JSON <-> Protobuf utilities (JsonFormat).
    // implementation 'com.google.protobuf:protobuf-java-util:3.24.0'
}

repositories {
    mavenCentral()
    maven { url 'https://packages.confluent.io/maven/' }
}

The Confluent artifacts are not on Maven Central; they are served from https://packages.confluent.io/maven/, so that repository must be declared.

Maven:

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-streams-protobuf-serde</artifactId>
    <version>7.6.0</version>
</dependency>
<dependency>
    <groupId>com.google.protobuf</groupId>
    <artifactId>protobuf-java</artifactId>
    <version>3.24.0</version>
</dependency>

kafka-streams-protobuf-serde transitively pulls in kafka-protobuf-serializer, so you do not need to declare the latter separately. Add the Confluent Maven repository (https://packages.confluent.io/maven/) to your <repositories> block as well.

The Serde needs to reach the Schema Registry. Two of the most important client properties are serializer-side settings:

schema.registry.url=http://schema-registry.example.com:8081
auto.register.schemas=false
use.latest.version=true

auto.register.schemas=false is the important production practice: registration becomes a deliberate, CI/CD-driven step rather than a side effect of application startup, so application code can never silently mutate a subject. use.latest.version=true only takes effect once auto.register.schemas is false; together they tell the serializer to look up the latest registered schema for the subject and serialize against it (with a backward-compatibility check, controlled by latest.compatibility.strict, which defaults to true). Neither property affects deserialization: the deserializer always resolves the schema ID embedded in each record against the registry, regardless of these settings. See Schema Evolution and Compatibility in Confluent Schema Registry.

When Protobuf, when Avro?

Both formats are first-class in Schema Registry, and the Streams code differs only by Serde class. The decision is about your existing toolchain and evolution model, not raw performance:

Reach for Protobuf when the same message definitions already back your gRPC or mobile clients; reach for Avro when your pipeline is Kafka-native and you want GenericRecord-style dynamic processing. Mixing both in one cluster is fine — the subject’s schema type is recorded per subject.

Defining the Protobuf Schema

The pipeline processes financial transactions and enriches them with a customer tier and a computed risk score. We define two message types — a raw transaction and an enriched transaction — so each contract can evolve independently.

Create src/main/proto/transaction.proto:

syntax = "proto3";

package com.example.kafka;

option java_package = "com.example.kafka.model";
option java_multiple_files = true;

message RawTransaction {
  string transaction_id = 1;
  string account_id = 2;
  double amount = 3;
  string currency = 4;
  int64 timestamp = 5;
  string merchant_category = 6;
}

message EnrichedTransaction {
  string transaction_id = 1;
  string account_id = 2;
  double amount = 3;
  string currency = 4;
  int64 timestamp = 5;
  string merchant_category = 6;
  string customer_tier = 7;
  double risk_score = 8;
  bool flagged = 9;
}

In proto3, scalar fields are not “optional” in the wire sense — they always have a type default (empty string, 0, false) and a reader cannot distinguish “unset” from “set to the default” unless you mark the field optional to get explicit presence. What matters for schema evolution is simpler: adding a new field with a new field number is backward compatible because old readers ignore unknown fields, and field numbers are part of the wire contract and must never be reused or reassigned once in production.

Generate the Java classes with the protobuf-gradle-plugin or the protobuf-maven-plugin as part of the build. With java_multiple_files = true, each message becomes its own top-level class (RawTransaction, EnrichedTransaction) in com.example.kafka.model.

Configuring Protobuf Serdes in Kafka Streams

A Serde built with the no-argument KafkaProtobufSerde<>() constructor deserializes into a generic DynamicMessage, not your generated class — calling txn.getAmount() on that would not compile. To get a typed RawTransaction/EnrichedTransaction, pass the target class to the constructor. Internally the Serde then sets specific.protobuf.value.type (or ...key.type) on the underlying deserializer so it materializes the concrete type.

package com.example.kafka;

import com.example.kafka.model.RawTransaction;
import com.example.kafka.model.EnrichedTransaction;
import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
import io.confluent.kafka.streams.serdes.protobuf.KafkaProtobufSerde;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class TransactionStreamsConfig {

    private static final String SCHEMA_REGISTRY_URL = "http://schema-registry.example.com:8081";

    public static Properties getStreamsProperties(String applicationId) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker-1:9092,kafka-broker-2:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName());
        props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, SCHEMA_REGISTRY_URL);
        props.put("auto.register.schemas", false);
        props.put("use.latest.version", true);
        props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
        return props;
    }

    public static Serde<RawTransaction> rawTransactionSerde() {
        KafkaProtobufSerde<RawTransaction> serde = new KafkaProtobufSerde<>(RawTransaction.class);
        Map<String, Object> config = new HashMap<>();
        config.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, SCHEMA_REGISTRY_URL);
        serde.configure(config, false); // false = value serde
        return serde;
    }

    public static Serde<EnrichedTransaction> enrichedTransactionSerde() {
        KafkaProtobufSerde<EnrichedTransaction> serde = new KafkaProtobufSerde<>(EnrichedTransaction.class);
        Map<String, Object> config = new HashMap<>();
        config.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, SCHEMA_REGISTRY_URL);
        serde.configure(config, false);
        return serde;
    }
}

DEFAULT_KEY_SERDE_CLASS_CONFIG takes a class name string, so use Serdes.StringSerde.class.getName() — passing Serdes.String().getClass() would hand the config a Class object for a wrapper instance, not the serde class name it expects. The boolean second argument to configure marks whether the Serde handles record keys (true) or values (false); here both are value serdes. In a real deployment, source SCHEMA_REGISTRY_URL and bootstrap servers from the environment rather than hardcoding them.

Building the Transaction Enrichment Topology

The topology reads raw transactions, joins them against a customer-tier KTable, computes a risk score, and writes enriched transactions. The KTable is backed by a compacted topic that holds the latest tier per customer.

Extracting topology construction into a static buildTopology() method keeps it unit-testable with TopologyTestDriver (see below).

package com.example.kafka;

import com.example.kafka.model.RawTransaction;
import com.example.kafka.model.EnrichedTransaction;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Joined;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Produced;

public class TransactionEnrichmentApp {

    public static Topology buildTopology() {
        StreamsBuilder builder = new StreamsBuilder();

        KStream<String, RawTransaction> rawTransactions = builder.stream(
                "transactions.raw",
                Consumed.with(Serdes.String(), TransactionStreamsConfig.rawTransactionSerde())
        );

        KTable<String, String> customerTiers = builder.table(
                "customers.tiers",
                Consumed.with(Serdes.String(), Serdes.String())
        );

        KStream<String, EnrichedTransaction> enriched = rawTransactions
                .filter((key, txn) -> txn.getAmount() > 0)
                .leftJoin(
                        customerTiers,
                        (txn, tier) -> {
                            double riskScore = calculateRiskScore(txn, tier);
                            return EnrichedTransaction.newBuilder()
                                    .setTransactionId(txn.getTransactionId())
                                    .setAccountId(txn.getAccountId())
                                    .setAmount(txn.getAmount())
                                    .setCurrency(txn.getCurrency())
                                    .setTimestamp(txn.getTimestamp())
                                    .setMerchantCategory(txn.getMerchantCategory())
                                    .setCustomerTier(tier != null ? tier : "UNKNOWN")
                                    .setRiskScore(riskScore)
                                    .setFlagged(riskScore > 0.7)
                                    .build();
                        },
                        Joined.with(Serdes.String(), TransactionStreamsConfig.rawTransactionSerde(), Serdes.String())
                );

        enriched.to(
                "transactions.enriched",
                Produced.with(Serdes.String(), TransactionStreamsConfig.enrichedTransactionSerde())
        );

        return builder.build();
    }

    public static void main(String[] args) {
        KafkaStreams streams = new KafkaStreams(
                buildTopology(),
                TransactionStreamsConfig.getStreamsProperties("transaction-enricher-v1"));

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
        streams.start();
    }

    private static double calculateRiskScore(RawTransaction txn, String tier) {
        double baseScore = txn.getAmount() > 10000 ? 0.4 : 0.1;
        if ("PREMIUM".equals(tier)) baseScore -= 0.2;
        if ("RISKY_MERCHANT".equals(txn.getMerchantCategory())) baseScore += 0.5;
        return Math.min(Math.max(baseScore, 0.0), 1.0);
    }
}

A few patterns worth highlighting. The filter drops zero- and negative-amount records before the join, so they never touch the state store. The KStream-KTable leftJoin keeps transactions whose account is missing from the tier table, defaulting them to "UNKNOWN" rather than dropping them. calculateRiskScore is a pure function, which keeps the topology deterministic and easy to assert on in tests. EXACTLY_ONCE_V2 gives transactional, read-process-write semantics across the consume and produce sides; it requires brokers on 2.5 or later and reduces the default commit.interval.ms from 30000 to 100 ms — which raises commit frequency and therefore the cost of small transactions, so size your transaction.timeout.ms and downstream consumers’ read_committed latency expectations accordingly.

Note that a KStream-KTable join is not strictly time-synchronized: the table reflects whatever tier value has been processed when the stream-side record arrives. For a compacted profile topic this is usually the intended behavior, but be aware of it if tier updates and transactions race.

Schema Registration and Evolution in Production

Registration belongs in your deployment pipeline, not in application startup. The Schema Registry exposes a REST API; the script below registers both Protobuf schemas under the {topic}-value subjects:

#!/usr/bin/env bash
set -euo pipefail
SCHEMA_REGISTRY="http://schema-registry.example.com:8081"

curl -fsS -X POST \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schemaType": "PROTOBUF", "schema": "syntax = \"proto3\";\npackage com.example.kafka;\n\noption java_package = \"com.example.kafka.model\";\noption java_multiple_files = true;\n\nmessage RawTransaction {\n  string transaction_id = 1;\n  string account_id = 2;\n  double amount = 3;\n  string currency = 4;\n  int64 timestamp = 5;\n  string merchant_category = 6;\n}\n"}' \
  "${SCHEMA_REGISTRY}/subjects/transactions.raw-value/versions"

curl -fsS -X POST \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schemaType": "PROTOBUF", "schema": "syntax = \"proto3\";\npackage com.example.kafka;\n\noption java_package = \"com.example.kafka.model\";\noption java_multiple_files = true;\n\nmessage EnrichedTransaction {\n  string transaction_id = 1;\n  string account_id = 2;\n  double amount = 3;\n  string currency = 4;\n  int64 timestamp = 5;\n  string merchant_category = 6;\n  string customer_tier = 7;\n  double risk_score = 8;\n  bool flagged = 9;\n}\n"}' \
  "${SCHEMA_REGISTRY}/subjects/transactions.enriched-value/versions"

The Registry accepts either application/vnd.schemaregistry.v1+json or plain application/json; the versioned media type is the documented form. The default subject-naming strategy (TopicNameStrategy) derives the subject as {topic}-value (or {topic}-key), which is exactly what the Serde uses at runtime. If you adopt RecordNameStrategy or TopicRecordNameStrategy, set the matching value.subject.name.strategy on the Serde so registration and runtime agree.

What actually counts as a compatible Protobuf change

The Registry enforces the configured compatibility mode on each new version. For Protobuf under the default BACKWARD mode, the changes the Registry treats as compatible are a specific, finite set:

Two non-obvious points fall out of this. First, “never change a field’s type” is slightly too strict: the documented scalar widenings are allowed, but anything else (including reusing a number for a different type) breaks the wire contract — so the safe operational rule remains “add new numbers, never repurpose old ones.” Second, best practice for Protobuf is BACKWARD_TRANSITIVE, not the default BACKWARD, because adding a new message type is not forward compatible, and transitive checks validate a new version against all prior versions rather than only the immediately preceding one. Set it once per subject:

curl -fsS -X PUT \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"compatibility": "BACKWARD_TRANSITIVE"}' \
  "${SCHEMA_REGISTRY}/config/transactions.raw-value"

See Schema Evolution and Compatibility in Confluent Schema Registry for how each mode maps onto Protobuf.

Monitoring and Error Handling

A poison-pill record — bytes that cannot be deserialized — will, by default, halt the stream thread. To route such records to a dead-letter topic and keep processing, implement a DeserializationExceptionHandler. In Kafka 3.6 this interface extends only Configurable; its sole abstract callbacks are configure(Map) and handle(ProcessorContext, ConsumerRecord, Exception). There is no close() in the contract on 3.6, so the handler must manage the lifecycle of any resource it creates (here, a producer) itself — for example, by closing it on JVM shutdown.

package com.example.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
import org.apache.kafka.streams.processor.ProcessorContext;

import java.util.HashMap;
import java.util.Map;

public class DeadLetterDeserializationHandler implements DeserializationExceptionHandler {

    private Producer<byte[], byte[]> dlqProducer;

    @Override
    public void configure(Map<String, ?> configs) {
        Map<String, Object> producerProps = new HashMap<>();
        producerProps.put("bootstrap.servers", configs.get("bootstrap.servers"));
        this.dlqProducer = new KafkaProducer<>(
                producerProps, new ByteArraySerializer(), new ByteArraySerializer());
        // The interface has no close() on 3.6; close the producer on shutdown.
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            if (dlqProducer != null) dlqProducer.close();
        }));
    }

    @Override
    public DeserializationHandlerResponse handle(
            ProcessorContext context, ConsumerRecord<byte[], byte[]> record, Exception exception) {
        dlqProducer.send(new ProducerRecord<>(
                "transactions.raw.dlq", record.key(), record.value()));
        return DeserializationHandlerResponse.CONTINUE;
    }
}

Three details matter here. First, the handler receives the raw bytes (ConsumerRecord<byte[], byte[]>), which is exactly why a byte[]-keyed/valued producer is the right tool — the original payload is forwarded to the DLQ untouched. Second, KafkaProducer needs explicit serializer instances when you do not set key/value.serializer in the props, so the three-argument constructor new KafkaProducer<>(props, keySerializer, valueSerializer) is used. Third, returning CONTINUE lets the stream skip the bad record and keep going; return FAIL if you would rather stop and page someone. Note that the DLQ send is fire-and-forget, so under load you should add a callback or periodic flush() to surface producer errors. Wire the handler in via:

default.deserialization.exception.handler=com.example.kafka.DeadLetterDeserializationHandler

On Kafka 3.9+ (KIP-1033/KIP-1036), the ProcessorContext-based handle method is deprecated in favor of an ErrorHandlerContext overload; on 3.6 the signature above is the current one.

Failure modes and how to detect them

The handler above catches the most common failure, but the Protobuf-plus-registry stack has a handful of distinct ones worth recognizing on sight:

The recurring lesson: the deserialization handler is a narrow tool. It catches bytes that will not decode; it does not catch a NullPointerException in calculateRiskScore. For those, use the separate processing exception handler introduced by KIP-1033 in Kafka 3.9 (config key processing.exception.handler, with LogAndContinue/LogAndFail built-ins) or defensive code in the mapper.

For metrics, Kafka Streams publishes a rich JMX tree. Stream-thread metrics live under the MBean kafka.streams:type=stream-thread-metrics,thread-id=[threadId] and include process-rate/process-total, poll-rate/poll-total, and commit-rate/commit-total. Most are recorded at the DEBUG recording level, so set metrics.recording.level=DEBUG if you need the per-node detail. The Prometheus JMX Exporter is a common way to scrape them:

java -javaagent:/opt/prometheus/jmx_prometheus_javaagent-0.20.0.jar=8080:/opt/prometheus/kafka-streams.yml \
     -cp target/transaction-enricher.jar com.example.kafka.TransactionEnrichmentApp

See the Confluent reference on monitoring Kafka Streams for the full metric catalog.

Testing the Pipeline

TopologyTestDriver from kafka-streams-test-utils runs the topology in-process, with no broker and no Schema Registry. The catch with Protobuf is the Serde: a KafkaProtobufSerde would try to reach a live registry. In a unit test, either inject a MockSchemaRegistryClient (the Serde has a constructor that accepts a SchemaRegistryClient) or point schema.registry.url at a mock:// pseudo-URL, which the Confluent client resolves to an in-memory MockSchemaRegistryClient keyed by the scope after mock://.

package com.example.kafka;

import com.example.kafka.model.EnrichedTransaction;
import com.example.kafka.model.RawTransaction;
import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
import io.confluent.kafka.streams.serdes.protobuf.KafkaProtobufSerde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.TopologyTestDriver;
import org.junit.jupiter.api.Test;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

class TransactionEnrichmentAppTest {

    private static final String MOCK_SR = "mock://test-scope";

    private static <T extends com.google.protobuf.Message> KafkaProtobufSerde<T> mockSerde(Class<T> type) {
        KafkaProtobufSerde<T> serde = new KafkaProtobufSerde<>(type);
        Map<String, Object> cfg = new HashMap<>();
        cfg.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, MOCK_SR);
        serde.configure(cfg, false);
        return serde;
    }

    @Test
    void shouldEnrichTransactionWithUnknownTier() {
        Properties props = TransactionStreamsConfig.getStreamsProperties("test-app");
        props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, MOCK_SR);

        try (TopologyTestDriver driver =
                     new TopologyTestDriver(TransactionEnrichmentApp.buildTopology(), props)) {

            KafkaProtobufSerde<RawTransaction> rawSerde = mockSerde(RawTransaction.class);
            KafkaProtobufSerde<EnrichedTransaction> enrichedSerde = mockSerde(EnrichedTransaction.class);

            TestInputTopic<String, RawTransaction> in = driver.createInputTopic(
                    "transactions.raw", Serdes.String().serializer(), rawSerde.serializer());
            TestOutputTopic<String, EnrichedTransaction> out = driver.createOutputTopic(
                    "transactions.enriched", Serdes.String().deserializer(), enrichedSerde.deserializer());

            RawTransaction raw = RawTransaction.newBuilder()
                    .setTransactionId("txn-123")
                    .setAccountId("acc-456")
                    .setAmount(500.0)
                    .setCurrency("USD")
                    .setTimestamp(System.currentTimeMillis())
                    .setMerchantCategory("GROCERY")
                    .build();

            in.pipeInput("txn-123", raw);

            EnrichedTransaction result = out.readValue();
            assertNotNull(result);
            assertEquals("txn-123", result.getTransactionId());
            assertEquals("UNKNOWN", result.getCustomerTier());
            assertTrue(result.getRiskScore() >= 0.0);
        }
    }
}

mock:// URLs are scoped: every Serde configured with the same mock://test-scope shares one in-memory registry, so the value written by the input Serde is readable by the output Serde. Wrap the driver in try-with-resources so its state directory is cleaned up. For an integration test against a real registry, use Testcontainers to run Kafka and Schema Registry, and remove the mock URL. Google’s Protobuf Java tutorial is a good reference for building message fixtures.

Operational Considerations

Schema Registry availability. Run Schema Registry as multiple instances; its leader election and storage are backed by Kafka, so its availability follows the cluster’s. The client accepts a comma-separated list and will fail over between them:

schema.registry.url=http://schema-registry-1:8081,http://schema-registry-2:8081,http://schema-registry-3:8081

State store tuning. The customer-tier KTable is backed by RocksDB and a compacted changelog. As the number of distinct customers grows, watch off-heap memory and disk. To tune RocksDB, implement org.apache.kafka.streams.state.RocksDBConfigSetter and register it via the rocksdb.config.setter config (constant StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG), passing the class — this is the supported way to cap the block cache and write buffers.

Schema IDs across environments. Each record carries a 4-byte schema ID that is resolved against the registry at read time. Those IDs are assigned per registry and are not portable, so when promoting data or mirroring across environments, migrate schemas through the registry’s import/export endpoints rather than assuming IDs match. The Schema Registry API reference documents these endpoints.

Throughput knobs. Protobuf decoding is CPU-light, but message objects still occupy heap. max.poll.records bounds how many records each consumer poll returns, and max.poll.interval.ms (a consumer config Streams passes through; default 300000 ms) bounds the time the application may spend between polls before the group coordinator evicts the member. Reasonable starting points for a moderate-throughput pipeline:

max.poll.records=500
max.poll.interval.ms=300000

The Schema Registry & Connect Ecosystem overview shows how Protobuf serialization fits alongside Kafka Connect for source and sink pipelines.

Deliberate registration, explicitly-typed Serdes, a dead-letter path for poison pills, and topology-level tests are what separate a Protobuf Kafka Streams application that survives production from one that merely compiles.