feat: Added an implementation of Kafka Binding (#114)

Signed-off-by: Daniel Azuma <dazuma@gmail.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: dazuma

CLAUDE.md

@@ -59,6 +59,19 @@ Handles encoding/decoding CloudEvents to/from HTTP (Rack env hashes). Supports:
 
 `HttpBinding.default` returns a singleton with `JsonFormat` and `TextFormat` pre-registered. Custom formatters are registered via `register_formatter`.
 
+### Kafka Binding (`KafkaBinding`)
+
+Handles encoding/decoding CloudEvents to/from Kafka messages. CloudEvents 1.0 only (no V0.3). Supports:
+- **Binary content mode** — event attributes as `ce_*` headers (plain UTF-8, no percent-encoding), data in value
+- **Structured content mode** — entire event serialized in value (e.g., JSON)
+- **No batch mode** (per the Kafka spec)
+- **Tombstone support** — `nil` value represents an event with no data
+- **Key mapping** — configurable callables to map between Kafka record keys and event attributes (default: `partitionkey` extension)
+
+Kafka messages are represented as plain `{ key:, value:, headers: }` Hashes, decoupled from any specific Kafka client library.
+
+`KafkaBinding.default` returns a singleton with `JsonFormat` and `TextFormat` pre-registered. Key mappers are configurable at construction and overridable per-call via `key_mapper:` / `reverse_key_mapper:` keyword arguments.
+
 ### Format Layer
 
 - **`JsonFormat`** — Encodes/decodes `application/cloudevents+json` and batch format. Also handles JSON data encoding/decoding for binary mode.
@@ -77,6 +90,7 @@ All errors inherit from `CloudEventsError`: `NotCloudEventError`, `UnsupportedFo
 
 ## Contributing
 
+- Use red-green test-driven development when making changes, unless instructed otherwise.
 - Conventional Commits format required (`fix:`, `feat:`, `docs:`, etc.)
 - Commits must be signed off (`git commit --signoff`)
 - Run `toys ci` before submitting PRs

features/step_definitions/steps.rb

@@ -50,17 +50,21 @@
 end
 
 Given "Kafka Protocol Binding is supported" do
-  pending "Kafka Protocol Binding is not yet implemented"
+  @kafka_binding = CloudEvents::KafkaBinding.default
 end
 
-Given "a Kafka message with payload:" do |_str|
-  pending
+Given "a Kafka message with payload:" do |str|
+  @kafka_value = str
 end
 
-Given "Kafka headers:" do |_table|
-  pending
+Given "Kafka headers:" do |table|
+  @kafka_headers = {}
+  table.hashes.each do |hash|
+    @kafka_headers[hash["key"].strip] = hash["value"]
+  end
 end
 
 When "parsed as Kafka message" do
-  pending
+  message = { key: nil, value: @kafka_value, headers: @kafka_headers }
+  @event = @kafka_binding.decode_event(message, reverse_key_mapper: nil)
 end

lib/cloud_events.rb

@@ -5,6 +5,7 @@
 require "cloud_events/event"
 require "cloud_events/format"
 require "cloud_events/http_binding"
+require "cloud_events/kafka_binding"
 require "cloud_events/json_format"
 require "cloud_events/text_format"
 

lib/cloud_events/errors.rb

@@ -51,6 +51,14 @@ class SpecVersionError < CloudEventsError
   class AttributeError < CloudEventsError
   end
 
+  ##
+  # An error raised when a protocol binding that does not support batch
+  # content mode receives a batch. For example, the Kafka protocol binding
+  # does not support batches per the CloudEvents spec.
+  #
+  class BatchNotSupportedError < CloudEventsError
+  end
+
   ##
   # Alias of UnsupportedFormatError, for backward compatibility.
   #