feat(kafka): add franz-go protocol binding

docs/index.md

@@ -126,6 +126,7 @@ err := json.Unmarshal(bytes, &event)
 | [JSON Event Format](event_data_structure.md#marshalunmarshal-event-to-json)             | :heavy_check_mark:                                    | :heavy_check_mark:                                    |
 | [Sarama Kafka Protocol Binding](https://github.com/cloudevents/sdk-go/tree/main/samples/kafka) | :heavy_check_mark:                                    | :heavy_check_mark:                                    |
 | [Confluent Kafka Protocol Binding](https://github.com/cloudevents/sdk-go/tree/main/samples/kafka_confluent) | :heavy_check_mark:                                    | :heavy_check_mark:                                    |
+| [Franz-go Kafka Protocol Binding](https://github.com/cloudevents/sdk-go/tree/main/protocol/kafka_franz) | :heavy_check_mark:                                    | :heavy_check_mark:                                    |
 | [MQTT Protocol Binding](https://github.com/cloudevents/sdk-go/tree/main/samples/mqtt)   | :heavy_check_mark:                                    | :heavy_check_mark:                                    |
 | [NATS Protocol Binding](https://github.com/cloudevents/sdk-go/tree/main/samples/nats)   | :heavy_check_mark:                                    | :heavy_check_mark:                                    |
 | [STAN Protocol Binding](https://github.com/cloudevents/sdk-go/tree/main/samples/stan)   | :heavy_check_mark:                                    | :heavy_check_mark:                                    |

docs/protocol_implementations.md

@@ -26,6 +26,7 @@ the `Write<DataStructure>` functions, while the latter is done implementing spec
 * [Kafka Protocol Binding](https://github.com/cloudevents/sdk-go/tree/main/protocol/kafka_sarama) using [Sarama](https://github.com/Shopify/sarama)
 * [Kafka Protocol Binding](https://github.com/cloudevents/sdk-go/tree/main/protocol/kafka_confluent) using [confluent-kafka-go](https://github.com/confluentinc/confluent-kafka-go)
   > It bring some new features compared the above [sarama](https://github.com/Shopify/sarama) binding. Like [pattern subscription](https://github.com/confluentinc/confluent-kafka-go/issues/96), [async message confirmation](https://github.com/cloudevents/sdk-go/issues/846) and other enhancements.
+* [Kafka Protocol Binding](https://github.com/cloudevents/sdk-go/tree/main/protocol/kafka_franz) using [franz-go](https://github.com/twmb/franz-go)
 * [MQTT Protocol Binding](https://github.com/cloudevents/sdk-go/tree/main/protocol/mqtt_paho) using [eclipse/paho.golang](https://github.com/eclipse/paho.golang)
 * [NATS Protocol Binding](https://github.com/cloudevents/sdk-go/tree/main/protocol/nats) using [nats.go](https://github.com/nats-io/nats.go)
 * [STAN Protocol Binding](https://github.com/cloudevents/sdk-go/tree/main/protocol/stan) using [stan.go](https://github.com/nats-io/stan.go)

protocol/kafka_franz/v2/doc.go

@@ -0,0 +1,7 @@
+/*
+ Copyright 2026 The CloudEvents Authors
+ SPDX-License-Identifier: Apache-2.0
+*/
+
+// Package kafka_franz implements a Kafka binding using github.com/twmb/franz-go/pkg/kgo.
+package kafka_franz

protocol/kafka_franz/v2/go.mod

@@ -0,0 +1,27 @@
+module github.com/cloudevents/sdk-go/protocol/kafka_franz/v2
+
+go 1.25.0
+
+replace github.com/cloudevents/sdk-go/v2 => ../../../v2
+
+require (
+	github.com/cloudevents/sdk-go/v2 v2.16.2
+	github.com/stretchr/testify v1.11.1
+	github.com/twmb/franz-go v1.20.3
+)
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/google/go-cmp v0.7.0 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/klauspost/compress v1.18.0 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/pierrec/lz4/v4 v4.1.22 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/twmb/franz-go/pkg/kmsg v1.12.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.27.1 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)

protocol/kafka_franz/v2/go.sum

@@ -0,0 +1,50 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
+github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
+github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
+github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
+github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
+github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
+github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs=
+github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
+github.com/pierrec/lz4/v4 v4.1.22 h1:cKFw6uJDK+/gfw5BcDL0JL5aBsAFdsIT18eRtLj7VIU=
+github.com/pierrec/lz4/v4 v4.1.22/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+github.com/twmb/franz-go v1.20.3 h1:gjwZwZmmvo/t7mxyj6frxDORVxsqrycXPnDrpkXldfY=
+github.com/twmb/franz-go v1.20.3/go.mod h1:YCnepDd4gl6vdzG03I5Wa57RnCTIC6DVEyMpDX/J8UA=
+github.com/twmb/franz-go/pkg/kmsg v1.12.0 h1:CbatD7ers1KzDNgJqPbKOq0Bz/WLBdsTH75wgzeVaPc=
+github.com/twmb/franz-go/pkg/kmsg v1.12.0/go.mod h1:+DPt4NC8RmI6hqb8G09+3giKObE6uD2Eya6CfqBpeJY=
+github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
+github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
+go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
+go.uber.org/zap v1.27.1 h1:08RqriUEv8+ArZRYSTXy1LeBScaMpVSTBhCeaZYfMYc=
+go.uber.org/zap v1.27.1/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
+golang.org/x/crypto v0.43.0 h1:dduJYIi3A3KOfdGOHX8AVZ/jGiyPa3IbBozJ5kNuE04=
+golang.org/x/crypto v0.43.0/go.mod h1:BFbav4mRNlXJL4wNeejLpWxB7wMbc79PdRGhWKncxR0=
+golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
+golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f h1:BLraFXnmrev5lT+xlilqcH8XK9/i0At2xKjWk4p6zsU=
+gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

protocol/kafka_franz/v2/message.go

@@ -0,0 +1,183 @@
+/*
+ Copyright 2026 The CloudEvents Authors
+ SPDX-License-Identifier: Apache-2.0
+*/
+
+package kafka_franz
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"strconv"
+	"strings"
+	"sync"
+
+	"github.com/twmb/franz-go/pkg/kgo"
+
+	"github.com/cloudevents/sdk-go/v2/binding"
+	"github.com/cloudevents/sdk-go/v2/binding/format"
+	"github.com/cloudevents/sdk-go/v2/binding/spec"
+)
+
+const (
+	prefix            = "ce_"
+	contentTypeHeader = "content-type"
+)
+
+const (
+	KafkaOffsetKey    = "kafkaoffset"
+	KafkaPartitionKey = "kafkapartition"
+	KafkaTopicKey     = "kafkatopic"
+	KafkaMessageKey   = "kafkamessagekey"
+)
+
+var specs = spec.WithPrefix(prefix)
+
+// Message represents a Kafka message.
+// This message can be read several times safely.
+type Message struct {
+	record      *kgo.Record
+	properties  map[string][]byte
+	format      format.Format
+	version     spec.Version
+	contentType string
+}
+
+var (
+	_ binding.Message               = (*Message)(nil)
+	_ binding.MessageMetadataReader = (*Message)(nil)
+)
+
+// NewMessage returns a binding.Message that holds the provided kgo.Record.
+// The returned binding.Message can be read several times safely.
+func NewMessage(record *kgo.Record) *Message {
+	if record == nil {
+		panic("the kgo.Record must not be nil")
+	}
+
+	var contentType, contentVersion string
+	properties := make(map[string][]byte, len(record.Headers)+4)
+	for _, header := range record.Headers {
+		key := strings.ToLower(header.Key)
+		if key == contentTypeHeader {
+			contentType = string(header.Value)
+		}
+		if key == specs.PrefixedSpecVersionName() {
+			contentVersion = string(header.Value)
+		}
+		properties[key] = append([]byte(nil), header.Value...)
+	}
+
+	properties[prefix+KafkaOffsetKey] = []byte(strconv.FormatInt(record.Offset, 10))
+	properties[prefix+KafkaPartitionKey] = []byte(strconv.FormatInt(int64(record.Partition), 10))
+	properties[prefix+KafkaTopicKey] = []byte(record.Topic)
+	if len(record.Key) > 0 {
+		properties[prefix+KafkaMessageKey] = append([]byte(nil), record.Key...)
+	}
+
+	message := &Message{
+		record:      record,
+		properties:  properties,
+		contentType: contentType,
+	}
+	if ft := format.Lookup(contentType); ft != nil {
+		message.format = ft
+	} else if v := specs.Version(contentVersion); v != nil {
+		message.version = v
+	}
+
+	return message
+}
+
+func (m *Message) ReadEncoding() binding.Encoding {
+	if m.version != nil {
+		return binding.EncodingBinary
+	}
+	if m.format != nil {
+		return binding.EncodingStructured
+	}
+	return binding.EncodingUnknown
+}
+
+func (m *Message) ReadStructured(ctx context.Context, encoder binding.StructuredWriter) error {
+	if m.format == nil {
+		return binding.ErrNotStructured
+	}
+	return encoder.SetStructuredEvent(ctx, m.format, bytes.NewReader(m.record.Value))
+}
+
+func (m *Message) ReadBinary(ctx context.Context, encoder binding.BinaryWriter) error {
+	if m.version == nil {
+		return binding.ErrNotBinary
+	}
+
+	var err error
+	for key, value := range m.properties {
+		switch {
+		case strings.HasPrefix(key, prefix):
+			attr := m.version.Attribute(key)
+			if attr != nil {
+				err = encoder.SetAttribute(attr, string(value))
+			} else {
+				err = encoder.SetExtension(strings.TrimPrefix(key, prefix), string(value))
+			}
+		case key == contentTypeHeader:
+			err = encoder.SetAttribute(m.version.AttributeFromKind(spec.DataContentType), string(value))
+		}
+		if err != nil {
+			return err
+		}
+	}
+
+	if m.record.Value != nil {
+		err = encoder.SetData(bytes.NewBuffer(m.record.Value))
+	}
+	return err
+}
+
+func (m *Message) Finish(error) error {
+	return nil
+}
+
+func (m *Message) GetAttribute(k spec.Kind) (spec.Attribute, interface{}) {
+	if m.version == nil {
+		return nil, nil
+	}
+	attr := m.version.AttributeFromKind(k)
+	if attr == nil {
+		return nil, nil
+	}
+	return attr, string(m.properties[attr.PrefixedName()])
+}
+
+func (m *Message) GetExtension(name string) interface{} {
+	value, ok := m.properties[prefix+name]
+	if !ok {
+		return nil
+	}
+	return string(value)
+}
+
+type receivedMessage struct {
+	*Message
+	finish     func(error) error
+	finishOnce sync.Once
+	finishErr  error
+}
+
+var _ binding.MessageWrapper = (*receivedMessage)(nil)
+
+func (m *receivedMessage) Finish(err error) error {
+	m.finishOnce.Do(func() {
+		m.finishErr = m.Message.Finish(err)
+		if m.finish != nil {
+			m.finishErr = errors.Join(m.finishErr, m.finish(err))
+		}
+	})
+	return m.finishErr
+}
+
+func (m *receivedMessage) GetWrappedMessage() binding.Message {
+	return m.Message
+}