Kuestenlogik.Bowline.Protocol.Kafka 0.9.4

Bowline.Protocol.Kafka

Apache Kafka protocol plugin for the Bowline workbench. Browse topics, consume live traffic, produce test messages, and replay recorded Kafka streams through the Bowline mock server. Built on Confluent.Kafka.

URL shapes

kafka://broker.example:9092                                   # single broker
kafka://b1:9092,b2:9092,b3:9092                               # bootstrap servers CSV
broker.example:9094                                           # bare host:port also accepted
kafka://broker:9092?schema-registry=http://sr:8081            # with Confluent Schema Registry
kafka://broker:9092?sr=http%3A%2F%2Fsr%3A8081                 # short alias + URL-encoded

Discovery

A short-lived IAdminClient queries broker metadata. Every non-internal topic surfaces as a service with two methods:

A synthetic Cluster service surfaces broker / topic counts so the sidebar shows "what's there" even before you pick a topic. Internal topics (__consumer_offsets, _schemas, _confluent*) are hidden unless the workbench's internal-services toggle is on.

Consume envelope

Each message arrives as:

{
  "topic": "orders",
  "partition": 0,
  "offset": 1234,
  "timestamp": 1745500800123,
  "key": "order-42",
  "keyBase64": "b3JkZXItNDI=",
  "value": "{\"id\":42,\"total\":19.95}",
  "valueBase64": "eyJpZCI6NDIsInRvdGFsIjoxOS45NX0=",
  "encoding": "avro",
  "bytes": 23
}

key / value are the UTF-8 decode when the bytes map cleanly; null for binary payloads. keyBase64 / valueBase64 are always present so the workbench can hex-dump arbitrary payloads (Avro, Protobuf, MessagePack, …). The encoding field is always present (nullable) — it's set when the body was decoded through the schema registry, otherwise null.

When the URL carries ?schema-registry=… and the message body is framed in the Confluent wire format (0x00 magic byte + 4-byte big-endian schema id + payload), the plugin decodes it on the fly and the value field carries the JSON projection. Coverage at v0:

Plain UTF-8 / opaque binary stays in the original fallback path — the registry isn't consulted unless the framing prefix is present.

Security

Two markers feed the Kafka security knobs from Bowline's environment auth helpers:

Both markers are stripped from the metadata dict before it's forwarded as Kafka message headers, so secrets never reach the wire.

Produce

The produce method publishes a single message taken from the first invocation payload:

Mock replay

KafkaMockEmitter plugs into Bowline's mock server via IBowlineMockEmitter. Recordings tagged protocol: "kafka" get re-published at the original cadence:

Payload source: responseBinary (base64) is preferred so Avro / Protobuf round-trips byte-for-byte; text-only recordings fall back to UTF-8 encoding of body.

Install

dotnet tool install -g Kuestenlogik.Bowline.Tool
bowline plugin install Kuestenlogik.Bowline.Protocol.Kafka

Then start the workbench and enter a Kafka broker URL in the sidebar.

Tests

The unit-test suite is hermetic and runs anywhere dotnet test does:

dotnet test --filter "Category!=Docker"

Real-broker round-trips (KafkaRoundTripE2ETests, KafkaSchemaRegistryE2ETests) sit behind the Docker xUnit trait. They use Testcontainers to spin up Confluent Kafka 7.7 (KRaft mode, no Zookeeper) and Schema Registry 7.7 in a shared bridge network, then drive the plugin against the live processes:

docker info       # confirm Docker Desktop / engine is reachable
dotnet test       # all tests, Docker ones included

Relationship to Bowline.Protocol.Storm

This plugin is the generic Kafka baseline: schema-registry-agnostic, serializer-agnostic, usable against any cluster. The Bowline.Protocol.Storm plugin layers Storm-specific conventions (schema-registry URL resolution, typed payload decoders, Storm topic-naming awareness) on top for users inside the Storm ecosystem. Run either alone or install both side by side — each registers under its own protocol id (kafka vs storm).