From c9c80a204f30e0926221c0e28cc90ae74b41c526 Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Thu, 25 Jun 2026 17:21:04 +0300 Subject: [PATCH 01/18] Add Kafka event sourcing middleware package --- cloudevents/instance.go | 2 +- cloudevents/service.go | 2 +- cmd/fkafka/main.go | 46 ++++++++ go.mod | 16 ++- go.sum | 78 +++++++++++- kafka/consumer.go | 243 ++++++++++++++++++++++++++++++++++++++ kafka/consumer_test.go | 163 +++++++++++++++++++++++++ kafka/handler_test.go | 144 ++++++++++++++++++++++ kafka/instance.go | 96 +++++++++++++++ kafka/message.go | 20 ++++ kafka/service.go | 256 ++++++++++++++++++++++++++++++++++++++++ 11 files changed, 1060 insertions(+), 6 deletions(-) create mode 100644 cmd/fkafka/main.go create mode 100644 kafka/consumer.go create mode 100644 kafka/consumer_test.go create mode 100644 kafka/handler_test.go create mode 100644 kafka/instance.go create mode 100644 kafka/message.go create mode 100644 kafka/service.go diff --git a/cloudevents/instance.go b/cloudevents/instance.go index 40904e3..2f10123 100644 --- a/cloudevents/instance.go +++ b/cloudevents/instance.go @@ -162,7 +162,7 @@ type handlerCtxEvtEvtErr interface { Handle(context.Context, event.Event) (*event.Event, error) } -func getReceiverFn(f any) any { +func GetReceiverFn(f any) any { switch h := f.(type) { case handler: return h.Handle diff --git a/cloudevents/service.go b/cloudevents/service.go index 19c0742..6c1ef62 100644 --- a/cloudevents/service.go +++ b/cloudevents/service.go @@ -160,7 +160,7 @@ func newCloudeventHandler(f any) http.Handler { h = dh.Handler } else { // Instanced Functions implement one of the defined interfaces. - h = getReceiverFn(f) + h = GetReceiverFn(f) } protocol, err := cloudevents.NewHTTP() diff --git a/cmd/fkafka/main.go b/cmd/fkafka/main.go new file mode 100644 index 0000000..1ae46f9 --- /dev/null +++ b/cmd/fkafka/main.go @@ -0,0 +1,46 @@ +package main + +import ( + "context" + "fmt" + "os" + + cloudevents "github.com/cloudevents/sdk-go/v2/event" + + "knative.dev/func-go/kafka" +) + +// Main illustrates how scaffolding works to wrap a user's function +// when the event source is Kafka. +// +// The function uses a standard CloudEvents handler signature. +// Kafka messages are automatically wrapped as CloudEvents by the runtime. +// +// Required environment variables: +// +// KAFKA_BROKERS — comma-separated broker addresses +// KAFKA_TOPIC — topic to consume from +// KAFKA_CONSUMER_GROUP — consumer group ID +func main() { + if err := kafka.Start(New()); err != nil { + fmt.Fprintln(os.Stderr, err.Error()) + os.Exit(1) + } +} + +// MyFunction is an example instanced CloudEvents function that consumes +// from Kafka. +type MyFunction struct{} + +func New() *MyFunction { + return &MyFunction{} +} + +func (f *MyFunction) Handle(ctx context.Context, e cloudevents.Event) error { + fmt.Printf("Received event: type=%s source=%s id=%s\n", + e.Type(), e.Source(), e.ID()) + if e.Data() != nil { + fmt.Printf("Data: %s\n", string(e.Data())) + } + return nil +} \ No newline at end of file diff --git a/go.mod b/go.mod index 343d999..e48e3fd 100644 --- a/go.mod +++ b/go.mod @@ -3,19 +3,33 @@ module knative.dev/func-go go 1.25.0 require ( + github.com/IBM/sarama v1.50.3 github.com/cloudevents/sdk-go/v2 v2.15.2 github.com/rs/zerolog v1.32.0 knative.dev/hack v0.0.0-20260428014158-b2a37f1b6e7b ) require ( + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/eapache/go-resiliency v1.7.0 // indirect github.com/google/uuid v1.6.0 // indirect + github.com/hashicorp/go-uuid v1.0.3 // indirect + github.com/jcmturner/aescts/v2 v2.0.0 // indirect + github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect + github.com/jcmturner/gofork v1.7.6 // indirect + github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect + github.com/jcmturner/rpc/v2 v2.0.3 // indirect github.com/json-iterator/go v1.1.12 // indirect + github.com/klauspost/compress v1.18.6 // indirect github.com/mattn/go-colorable v0.1.13 // indirect github.com/mattn/go-isatty v0.0.20 // 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.27 // indirect + github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 // indirect go.uber.org/multierr v1.11.0 // indirect go.uber.org/zap v1.27.0 // indirect - golang.org/x/sys v0.18.0 // indirect + golang.org/x/crypto v0.53.0 // indirect + golang.org/x/net v0.56.0 // indirect + golang.org/x/sys v0.46.0 // indirect ) diff --git a/go.sum b/go.sum index d18bdd9..f716086 100644 --- a/go.sum +++ b/go.sum @@ -1,17 +1,40 @@ +github.com/IBM/sarama v1.50.3 h1:zpY2iZYmt+z+0Bo3aYF+cD48OBt2hIgiDPZUuZKTXcc= +github.com/IBM/sarama v1.50.3/go.mod h1:Jo4MSfdDT3ycmQj7/ab8eLZwnvwCKZm/8H7SCbtyo8U= github.com/cloudevents/sdk-go/v2 v2.15.2 h1:54+I5xQEnI73RBhWHxbI1XJcqOFOVJN85vb41+8mHUc= github.com/cloudevents/sdk-go/v2 v2.15.2/go.mod h1:lL7kSWAE/V8VI4Wh0jbL2v/jvqsm6tjmaQBSvxcv4uE= github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc= 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/eapache/go-resiliency v1.7.0 h1:n3NRTnBn5N0Cbi/IeOHuQn9s2UwVUH7Ga0ZWcP+9JTA= +github.com/eapache/go-resiliency v1.7.0/go.mod h1:5yPzW0MIvSe0JDsv0v+DvcjEv2FyD6iZYSs1ZI+iQho= github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= github.com/google/go-cmp v0.5.0 h1:/QaMHBdZ26BB3SSst0Iwl10Epc+xhTquomWX0oZEB6w= github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= 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/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4= +github.com/gorilla/sessions v1.2.1/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM= +github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= +github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/Co8= +github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= +github.com/jcmturner/aescts/v2 v2.0.0 h1:9YKLH6ey7H4eDBXW8khjYslgyqG2xZikXP0EQFKrle8= +github.com/jcmturner/aescts/v2 v2.0.0/go.mod h1:AiaICIRyfYg35RUkr8yESTqvSy7csK90qZ5xfvvsoNs= +github.com/jcmturner/dnsutils/v2 v2.0.0 h1:lltnkeZGL0wILNvrNiVCR6Ro5PGU/SeBvVO/8c/iPbo= +github.com/jcmturner/dnsutils/v2 v2.0.0/go.mod h1:b0TnjGOvI/n42bZa+hmXL+kFJZsFT7G4t3HTlQ184QM= +github.com/jcmturner/gofork v1.7.6 h1:QH0l3hzAU1tfT3rZCnW5zXl+orbkNMMRGJfdJjHVETg= +github.com/jcmturner/gofork v1.7.6/go.mod h1:1622LH6i/EZqLloHfE7IeZ0uEJwMSUyQ/nDd82IeqRo= +github.com/jcmturner/goidentity/v6 v6.0.1 h1:VKnZd2oEIMorCTsFBnJWbExfNN7yZr3EhJAxwOkZg6o= +github.com/jcmturner/goidentity/v6 v6.0.1/go.mod h1:X1YW3bgtvwAXju7V3LCIMpY0Gbxyjn/mY9zx4tFonSg= +github.com/jcmturner/gokrb5/v8 v8.4.4 h1:x1Sv4HaTpepFkXbt2IkL29DXRf8sOfZXo8eRKh687T8= +github.com/jcmturner/gokrb5/v8 v8.4.4/go.mod h1:1btQEpgT6k+unzCwX1KdWMEwPPkkgBtP+F6aCACiMrs= +github.com/jcmturner/rpc/v2 v2.0.3 h1:7FXXj8Ti1IaVFpSAziCZWNzbNuZmnvw/i6CqLNdWfZY= +github.com/jcmturner/rpc/v2 v2.0.3/go.mod h1:VUJYCIDm3PVOEHw8sgt091/20OJjskO/YJki3ELg/Hc= 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.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= +github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA= github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg= github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM= @@ -23,33 +46,82 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w 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/pierrec/lz4/v4 v4.1.27 h1:+PhzhWDrjRj89TH2sw43nE3+4+W8lSxIuQadEHZyjUk= +github.com/pierrec/lz4/v4 v4.1.27/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= 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/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 h1:bsUq1dX0N8AOIL7EB/X911+m4EHsnWEHeJ0c+3TTBrg= +github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4= github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg= github.com/rs/zerolog v1.32.0 h1:keLypqrlIjaFsbmJOBdB/qvyF8KEtCWHwobLp5l/mQ0= github.com/rs/zerolog v1.32.0/go.mod h1:/7mN4D5sKwJLZQ2b/znpjC3/GQWY/xaDXUM0kKWRHss= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= -github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= +github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= +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/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw= github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc= +github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= 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.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8= go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= +golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58= +golang.org/x/crypto v0.53.0 h1:QZ4Muo8THX6CizN2vPPd5fBGHyogrdK9fG4wLPFUsto= +golang.org/x/crypto v0.53.0/go.mod h1:DNLU434OwVakk9PzuwV8w62mAJpRJL3vsgcfp4Qnsio= +golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= +golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= +golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= +golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= +golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o= +golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= +golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4= -golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw= +golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= +golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac h1:7zkz7BUtwNFFqcowJ+RIgu2MaV/MapERkDIy+mwPyjs= golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= knative.dev/hack v0.0.0-20260428014158-b2a37f1b6e7b h1:MvbV2F2BdI8qKrYYUhDwbUZbX0BAYRSIpXM2TOtTvs0= diff --git a/kafka/consumer.go b/kafka/consumer.go new file mode 100644 index 0000000..74967d9 --- /dev/null +++ b/kafka/consumer.go @@ -0,0 +1,243 @@ +package kafka + +import ( + "context" + "fmt" + "os" + "strings" + "sync/atomic" + "time" + + "github.com/IBM/sarama" + "github.com/cloudevents/sdk-go/v2/event" + "github.com/rs/zerolog/log" +) + +func kafkaBrokers() []string { + v := os.Getenv("KAFKA_BROKERS") + if v == "" { + return nil + } + return splitAndTrim(v) +} + +func kafkaTopic() string { + return strings.TrimSpace(os.Getenv("KAFKA_TOPIC")) +} + +func kafkaConsumerGroup() string { + return strings.TrimSpace(os.Getenv("KAFKA_CONSUMER_GROUP")) +} + +func splitAndTrim(s string) []string { + parts := strings.Split(s, ",") + var result []string + for _, p := range parts { + p = strings.TrimSpace(p) + if p != "" { + result = append(result, p) + } + } + return result +} + +func consumeLoop(ctx context.Context, f any, ready *atomic.Bool) error { + defer ready.Store(false) + + brokers := kafkaBrokers() + topic := kafkaTopic() + group := kafkaConsumerGroup() + + if len(brokers) == 0 { + return fmt.Errorf("KAFKA_BROKERS environment variable is required") + } + if topic == "" { + return fmt.Errorf("KAFKA_TOPIC environment variable is required") + } + if group == "" { + return fmt.Errorf("KAFKA_CONSUMER_GROUP environment variable is required") + } + + log.Info(). + Strs("brokers", brokers). + Str("topic", topic). + Str("group", group). + Msg("connecting to kafka") + + config := sarama.NewConfig() + config.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{ + sarama.NewBalanceStrategyRoundRobin(), + } + config.Consumer.Offsets.Initial = sarama.OffsetNewest + + client, err := sarama.NewConsumerGroup(brokers, group, config) + if err != nil { + return fmt.Errorf("creating consumer group: %w", err) + } + defer func() { + if err := client.Close(); err != nil { + log.Error().Err(err).Msg("error closing kafka consumer group") + } + }() + + handler := &consumerGroupHandler{ + f: f, + ready: ready, + brokers: strings.Join(brokers, ","), + } + + for { + if err := client.Consume(ctx, []string{topic}, handler); err != nil { + if ctx.Err() != nil { + return nil + } + return fmt.Errorf("consumer error: %w", err) + } + if ctx.Err() != nil { + return nil + } + ready.Store(false) + } +} + +type consumerGroupHandler struct { + f any + ready *atomic.Bool + brokers string +} + +func (h *consumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error { + h.ready.Store(true) + log.Info().Msg("kafka consumer ready (partitions assigned)") + return nil +} + +func (h *consumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { + h.ready.Store(false) + log.Info().Msg("kafka consumer partitions revoked") + return nil +} + +func (h *consumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { + for { + select { + case msg, ok := <-claim.Messages(): + if !ok { + return nil + } + m := Message{ + Key: msg.Key, + Value: msg.Value, + Topic: msg.Topic, + Partition: msg.Partition, + Offset: msg.Offset, + Timestamp: msg.Timestamp, + } + for _, rh := range msg.Headers { + if rh != nil { + m.Headers = append(m.Headers, Header{ + Key: string(rh.Key), + Value: rh.Value, + }) + } + } + + e := kafkaMessageToEvent(m, h.brokers) + + if err := invokeHandler(h.f, session.Context(), e); err != nil { + log.Error().Err(err). + Str("topic", msg.Topic). + Int32("partition", msg.Partition). + Int64("offset", msg.Offset). + Msg("error handling kafka message") + continue + } + session.MarkMessage(msg, "") + case <-session.Context().Done(): + return nil + } + } +} + +// kafkaMessageToEvent converts a Kafka message to a CloudEvent. +// If the message already contains CloudEvent headers (ce_ prefix per the +// CloudEvents Kafka Protocol Binding), it is parsed as an existing CloudEvent +// rather than re-wrapped. +func kafkaMessageToEvent(msg Message, brokers string) event.Event { + if e, ok := parseCEFromHeaders(msg); ok { + return e + } + + e := event.New() + e.SetSpecVersion("1.0") + e.SetID(fmt.Sprintf("partition:%d/offset:%d", msg.Partition, msg.Offset)) + e.SetSource(fmt.Sprintf("kafka://%s/%s", brokers, msg.Topic)) + e.SetType("dev.knative.kafka.event") + e.SetTime(msg.Timestamp) + if len(msg.Key) > 0 { + e.SetSubject(string(msg.Key)) + } + _ = e.SetData("application/octet-stream", msg.Value) + e.SetExtension("kafkatopic", msg.Topic) + e.SetExtension("kafkapartition", msg.Partition) + e.SetExtension("kafkaoffset", msg.Offset) + if len(msg.Key) > 0 { + e.SetExtension("kafkakey", string(msg.Key)) + } + return e +} + +// parseCEFromHeaders checks if the Kafka message is already a CloudEvent +// (binary content mode with ce_ prefixed headers) and parses it. +func parseCEFromHeaders(msg Message) (event.Event, bool) { + headers := make(map[string]string) + for _, h := range msg.Headers { + headers[strings.ToLower(h.Key)] = string(h.Value) + } + + if _, ok := headers["ce_specversion"]; !ok { + return event.Event{}, false + } + + e := event.New() + e.SetSpecVersion(headers["ce_specversion"]) + if v, ok := headers["ce_id"]; ok { + e.SetID(v) + } + if v, ok := headers["ce_source"]; ok { + e.SetSource(v) + } + if v, ok := headers["ce_type"]; ok { + e.SetType(v) + } + if v, ok := headers["ce_subject"]; ok { + e.SetSubject(v) + } + if v, ok := headers["ce_time"]; ok { + t, err := time.Parse(time.RFC3339Nano, v) + if err == nil { + e.SetTime(t) + } + } + + contentType := "application/json" + if v, ok := headers["ce_datacontenttype"]; ok { + contentType = v + } + _ = e.SetData(contentType, msg.Value) + + // Set non-standard ce_ headers as extensions + for k, v := range headers { + if strings.HasPrefix(k, "ce_") { + attr := strings.TrimPrefix(k, "ce_") + switch attr { + case "specversion", "id", "source", "type", "subject", "time", "datacontenttype": + continue + default: + e.SetExtension(attr, v) + } + } + } + + return e, true +} \ No newline at end of file diff --git a/kafka/consumer_test.go b/kafka/consumer_test.go new file mode 100644 index 0000000..6277fb4 --- /dev/null +++ b/kafka/consumer_test.go @@ -0,0 +1,163 @@ +package kafka + +import ( + "fmt" + "testing" + "time" +) + +func TestKafkaMessageToEvent(t *testing.T) { + ts := time.Date(2025, 6, 15, 12, 0, 0, 0, time.UTC) + msg := Message{ + Key: []byte("my-key"), + Value: []byte(`{"hello":"world"}`), + Topic: "my-topic", + Partition: 2, + Offset: 42, + Timestamp: ts, + } + + e := kafkaMessageToEvent(msg, "broker1:9092,broker2:9092") + + if e.SpecVersion() != "1.0" { + t.Errorf("specversion = %q, want 1.0", e.SpecVersion()) + } + if e.ID() != "partition:2/offset:42" { + t.Errorf("id = %q, want partition:2/offset:42", e.ID()) + } + if e.Source() != "kafka://broker1:9092,broker2:9092/my-topic" { + t.Errorf("source = %q", e.Source()) + } + if e.Type() != "dev.knative.kafka.event" { + t.Errorf("type = %q", e.Type()) + } + if e.Subject() != "my-key" { + t.Errorf("subject = %q, want my-key", e.Subject()) + } + if !e.Time().Equal(ts) { + t.Errorf("time = %v, want %v", e.Time(), ts) + } + if string(e.Data()) != `{"hello":"world"}` { + t.Errorf("data = %q", string(e.Data())) + } + + exts := e.Extensions() + if exts["kafkatopic"] != "my-topic" { + t.Errorf("kafkatopic = %v", exts["kafkatopic"]) + } + if fmt.Sprintf("%v", exts["kafkapartition"]) != "2" { + t.Errorf("kafkapartition = %v", exts["kafkapartition"]) + } + if fmt.Sprintf("%v", exts["kafkaoffset"]) != "42" { + t.Errorf("kafkaoffset = %v", exts["kafkaoffset"]) + } + if exts["kafkakey"] != "my-key" { + t.Errorf("kafkakey = %v", exts["kafkakey"]) + } +} + +func TestKafkaMessageToEvent_NoKey(t *testing.T) { + msg := Message{ + Value: []byte("data"), + Topic: "t", + Partition: 0, + Offset: 0, + Timestamp: time.Now(), + } + + e := kafkaMessageToEvent(msg, "b:9092") + + if e.Subject() != "" { + t.Errorf("subject should be empty when key is nil, got %q", e.Subject()) + } + if _, ok := e.Extensions()["kafkakey"]; ok { + t.Error("kafkakey extension should not be set when key is nil") + } +} + +func TestKafkaMessageToEvent_CEPassThrough(t *testing.T) { + msg := Message{ + Value: []byte(`{"temperature":22}`), + Topic: "events", + Headers: []Header{ + {Key: "ce_specversion", Value: []byte("1.0")}, + {Key: "ce_id", Value: []byte("abc-123")}, + {Key: "ce_source", Value: []byte("//my-sensor")}, + {Key: "ce_type", Value: []byte("sensor.reading")}, + {Key: "ce_subject", Value: []byte("temp")}, + {Key: "ce_time", Value: []byte("2025-06-15T12:00:00Z")}, + {Key: "ce_datacontenttype", Value: []byte("application/json")}, + {Key: "ce_customext", Value: []byte("custom-value")}, + }, + Partition: 1, + Offset: 99, + } + + e := kafkaMessageToEvent(msg, "b:9092") + + if e.SpecVersion() != "1.0" { + t.Errorf("specversion = %q", e.SpecVersion()) + } + if e.ID() != "abc-123" { + t.Errorf("id = %q, want abc-123", e.ID()) + } + if e.Source() != "//my-sensor" { + t.Errorf("source = %q", e.Source()) + } + if e.Type() != "sensor.reading" { + t.Errorf("type = %q", e.Type()) + } + if e.Subject() != "temp" { + t.Errorf("subject = %q", e.Subject()) + } + expectedTime := time.Date(2025, 6, 15, 12, 0, 0, 0, time.UTC) + if !e.Time().Equal(expectedTime) { + t.Errorf("time = %v, want %v", e.Time(), expectedTime) + } + if e.DataContentType() != "application/json" { + t.Errorf("datacontenttype = %q", e.DataContentType()) + } + if string(e.Data()) != `{"temperature":22}` { + t.Errorf("data = %q", string(e.Data())) + } + if v, ok := e.Extensions()["customext"]; !ok || v != "custom-value" { + t.Errorf("customext = %v", v) + } +} + +func TestKafkaMessageToEvent_CEPassThrough_DefaultContentType(t *testing.T) { + msg := Message{ + Value: []byte(`{}`), + Headers: []Header{ + {Key: "ce_specversion", Value: []byte("1.0")}, + {Key: "ce_id", Value: []byte("x")}, + {Key: "ce_source", Value: []byte("s")}, + {Key: "ce_type", Value: []byte("t")}, + }, + } + + e := kafkaMessageToEvent(msg, "b:9092") + + if e.DataContentType() != "application/json" { + t.Errorf("default content type = %q, want application/json", e.DataContentType()) + } +} + +func TestKafkaMessageToEvent_NotCE(t *testing.T) { + msg := Message{ + Value: []byte("plain data"), + Headers: []Header{ + {Key: "x-custom", Value: []byte("val")}, + }, + Topic: "t", + Partition: 0, + Offset: 0, + Timestamp: time.Now(), + } + + e := kafkaMessageToEvent(msg, "b:9092") + + if e.Type() != "dev.knative.kafka.event" { + t.Errorf("non-CE message should get type dev.knative.kafka.event, got %q", e.Type()) + } +} diff --git a/kafka/handler_test.go b/kafka/handler_test.go new file mode 100644 index 0000000..f5d1d31 --- /dev/null +++ b/kafka/handler_test.go @@ -0,0 +1,144 @@ +package kafka + +import ( + "context" + "testing" + + "github.com/cloudevents/sdk-go/v2/event" + + ce "knative.dev/func-go/cloudevents" +) + +func TestInvokeHandler_AllSignatures(t *testing.T) { + tests := []struct { + name string + f any + }{ + {"Handle()", &hNoArgs{}}, + {"Handle() error", &hErr{}}, + {"Handle(ctx)", &hCtx{}}, + {"Handle(ctx) error", &hCtxErr{}}, + {"Handle(event)", &hEvt{}}, + {"Handle(event) error", &hEvtErr{}}, + {"Handle(ctx, event)", &hCtxEvt{}}, + {"Handle(ctx, event) error", &hCtxEvtErr{}}, + {"Handle(event) *event", &hEvtEvt{}}, + {"Handle(event) (*event, error)", &hEvtEvtErr{}}, + {"Handle(ctx, event) *event", &hCtxEvtEvt{}}, + {"Handle(ctx, event) (*event, error)", &hCtxEvtEvtErr{}}, + } + + e := event.New() + e.SetType("test") + e.SetSource("test") + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := invokeHandler(tt.f, context.Background(), e) + if err != nil { + t.Errorf("unexpected error: %v", err) + } + }) + } +} + +func TestInvokeHandler_DefaultHandler(t *testing.T) { + called := false + dh := ce.DefaultHandler{ + Handler: func(ctx context.Context, e event.Event) error { + called = true + return nil + }, + } + + e := event.New() + e.SetType("test") + e.SetSource("test") + + err := invokeHandler(dh, context.Background(), e) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !called { + t.Fatal("DefaultHandler's function was not called") + } +} + +func TestInvokeHandler_ResponseEventIgnored(t *testing.T) { + responseEventWarned = false + + resp := event.New() + resp.SetType("response") + resp.SetSource("test") + + f := &hCtxEvtEvtErrReturnsEvent{resp: &resp} + + e := event.New() + e.SetType("test") + e.SetSource("test") + + err := invokeHandler(f, context.Background(), e) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !responseEventWarned { + t.Fatal("expected response event warning to be set") + } +} + +// Test handler types for all 12 supported CloudEvents signatures. +type hNoArgs struct{} + +func (h *hNoArgs) Handle() {} + +type hErr struct{} + +func (h *hErr) Handle() error { return nil } + +type hCtx struct{} + +func (h *hCtx) Handle(context.Context) {} + +type hCtxErr struct{} + +func (h *hCtxErr) Handle(context.Context) error { return nil } + +type hEvt struct{} + +func (h *hEvt) Handle(event.Event) {} + +type hEvtErr struct{} + +func (h *hEvtErr) Handle(event.Event) error { return nil } + +type hCtxEvt struct{} + +func (h *hCtxEvt) Handle(context.Context, event.Event) {} + +type hCtxEvtErr struct{} + +func (h *hCtxEvtErr) Handle(context.Context, event.Event) error { return nil } + +type hEvtEvt struct{} + +func (h *hEvtEvt) Handle(event.Event) *event.Event { return nil } + +type hEvtEvtErr struct{} + +func (h *hEvtEvtErr) Handle(event.Event) (*event.Event, error) { return nil, nil } + +type hCtxEvtEvt struct{} + +func (h *hCtxEvtEvt) Handle(context.Context, event.Event) *event.Event { return nil } + +type hCtxEvtEvtErr struct{} + +func (h *hCtxEvtEvtErr) Handle(context.Context, event.Event) (*event.Event, error) { return nil, nil } + +type hCtxEvtEvtErrReturnsEvent struct { + resp *event.Event +} + +func (h *hCtxEvtEvtErrReturnsEvent) Handle(_ context.Context, _ event.Event) (*event.Event, error) { + return h.resp, nil +} diff --git a/kafka/instance.go b/kafka/instance.go new file mode 100644 index 0000000..569d75d --- /dev/null +++ b/kafka/instance.go @@ -0,0 +1,96 @@ +package kafka + +import ( + "context" + + "github.com/cloudevents/sdk-go/v2/event" + "github.com/rs/zerolog/log" + + ce "knative.dev/func-go/cloudevents" +) + +// Starter is a function which defines a method to be called on function start. +type Starter interface { + Start(context.Context, map[string]string) error +} + +// Stopper is a function which defines a method to be called on function stop. +type Stopper interface { + Stop(context.Context) error +} + +// ReadinessReporter is a function which defines a method to be used to +// determine readiness. +type ReadinessReporter interface { + Ready(context.Context) (bool, error) +} + +// LivenessReporter is a function which defines a method to be used to +// determine liveness. +type LivenessReporter interface { + Alive(context.Context) (bool, error) +} + +var responseEventWarned bool + +// invokeHandler calls a CloudEvents handler directly with a constructed event. +func invokeHandler(f any, ctx context.Context, e event.Event) error { + var fn any + if dh, ok := f.(ce.DefaultHandler); ok { + fn = dh.Handler + } else { + fn = ce.GetReceiverFn(f) + } + return invokeHandlerFn(fn, ctx, e) +} + +func invokeHandlerFn(fn any, ctx context.Context, e event.Event) error { + switch h := fn.(type) { + case func(): + h() + return nil + case func() error: + return h() + case func(context.Context): + h(ctx) + return nil + case func(context.Context) error: + return h(ctx) + case func(event.Event): + h(e) + return nil + case func(event.Event) error: + return h(e) + case func(context.Context, event.Event): + h(ctx, e) + return nil + case func(context.Context, event.Event) error: + return h(ctx, e) + case func(event.Event) *event.Event: + resp := h(e) + warnResponseEvent(resp) + return nil + case func(event.Event) (*event.Event, error): + resp, err := h(e) + warnResponseEvent(resp) + return err + case func(context.Context, event.Event) *event.Event: + resp := h(ctx, e) + warnResponseEvent(resp) + return nil + case func(context.Context, event.Event) (*event.Event, error): + resp, err := h(ctx, e) + warnResponseEvent(resp) + return err + default: + panic("handler function does not match any supported CloudEvents signature") + } +} + +func warnResponseEvent(resp *event.Event) { + if resp != nil && !responseEventWarned { + responseEventWarned = true + log.Warn().Msg("handler returned a response event, but response events are ignored when consuming from Kafka") + } +} + diff --git a/kafka/message.go b/kafka/message.go new file mode 100644 index 0000000..39c0905 --- /dev/null +++ b/kafka/message.go @@ -0,0 +1,20 @@ +package kafka + +import "time" + +// Message represents a Kafka message delivered to the function's handler. +type Message struct { + Key []byte + Value []byte + Headers []Header + Topic string + Partition int32 + Offset int64 + Timestamp time.Time +} + +// Header is a key-value pair attached to a Kafka message. +type Header struct { + Key string + Value []byte +} diff --git a/kafka/service.go b/kafka/service.go new file mode 100644 index 0000000..6a82a3a --- /dev/null +++ b/kafka/service.go @@ -0,0 +1,256 @@ +package kafka + +import ( + "bufio" + "context" + "fmt" + "net" + "net/http" + "os" + "os/signal" + "runtime" + "strings" + "sync/atomic" + "syscall" + "time" + + "github.com/rs/zerolog/log" +) + +const ( + DefaultListenAddress = "[::]:8080" + ServerShutdownTimeout = 30 * time.Second + InstanceStopTimeout = 30 * time.Second +) + +// Start a Kafka consumer service using the given CloudEvents handler. +// The handler must implement one of the supported CloudEvents Handle signatures. +func Start(f any) error { + log.Debug().Msg("func runtime creating kafka consumer instance") + return New(f).Start(context.Background()) +} + +// Service runs a Kafka consumer that delivers messages as CloudEvents to +// the function handler. An HTTP server runs alongside for health probes only. +type Service struct { + http.Server + listener net.Listener + f any + stop chan error + ready atomic.Bool +} + +// New creates a Service for the given handler. +func New(f any) *Service { + svc := &Service{ + f: f, + stop: make(chan error), + Server: http.Server{ + ReadTimeout: 30 * time.Second, + WriteTimeout: 30 * time.Second, + IdleTimeout: 30 * time.Second, + MaxHeaderBytes: 1 << 20, + ReadHeaderTimeout: 2 * time.Second, + }, + } + mux := http.NewServeMux() + mux.HandleFunc("/health/readiness", svc.Ready) + mux.HandleFunc("/health/liveness", svc.Alive) + svc.Handler = mux + return svc +} + +// Start the Kafka consumer and health HTTP server. +func (s *Service) Start(ctx context.Context) (err error) { + addr := listenAddress() + log.Debug().Str("address", addr).Msg("kafka service starting") + + if s.listener, err = net.Listen("tcp", addr); err != nil { + return + } + + if err = s.startInstance(ctx); err != nil { + return + } + + s.handleSignals() + + go func() { + if err := s.Serve(s.listener); err != http.ErrServerClosed { + log.Error().Err(err).Msg("http server exited with unexpected error") + s.stop <- err + } + }() + + go func() { + if err := consumeLoop(ctx, s.f, &s.ready); err != nil { + log.Error().Err(err).Msg("kafka consumer exited with error") + s.stop <- err + } + }() + + log.Debug().Msg("waiting for stop signals or errors") + select { + case err = <-s.stop: + if err != nil { + log.Error().Err(err).Msg("function error") + } + case <-ctx.Done(): + log.Debug().Msg("function canceled") + } + return s.shutdown(err) +} + +// Addr returns the address upon which the health server is listening. +func (s *Service) Addr() net.Addr { + if s.listener == nil { + return nil + } + return s.listener.Addr() +} + +// Ready handles readiness checks. +func (s *Service) Ready(w http.ResponseWriter, r *http.Request) { + if !s.ready.Load() { + w.WriteHeader(http.StatusServiceUnavailable) + fmt.Fprintln(w, "kafka consumer not yet ready") + return + } + if i, ok := s.f.(ReadinessReporter); ok { + ready, err := i.Ready(r.Context()) + if err != nil { + log.Debug().Err(err).Msg("error checking readiness") + w.WriteHeader(http.StatusInternalServerError) + fmt.Fprint(w, "error checking readiness: ", err.Error()) + return + } + if !ready { + w.WriteHeader(http.StatusServiceUnavailable) + fmt.Fprintln(w, "function not yet ready") + return + } + } + fmt.Fprintf(w, "READY") +} + +// Alive handles liveness checks. +func (s *Service) Alive(w http.ResponseWriter, r *http.Request) { + if i, ok := s.f.(LivenessReporter); ok { + alive, err := i.Alive(r.Context()) + if err != nil { + log.Err(err).Msg("error checking liveness") + w.WriteHeader(http.StatusInternalServerError) + fmt.Fprint(w, "error checking liveness: ", err.Error()) + return + } + if !alive { + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = w.Write([]byte("function not alive")) + return + } + } + fmt.Fprintf(w, "ALIVE") +} + +func (s *Service) startInstance(ctx context.Context) error { + if i, ok := s.f.(Starter); ok { + cfg, err := newCfg() + if err != nil { + return err + } + go func() { + if err := i.Start(ctx, cfg); err != nil { + s.stop <- err + } + }() + } else { + log.Debug().Msg("function does not implement Start. Skipping") + } + return nil +} + +func (s *Service) handleSignals() { + sigs := make(chan os.Signal, 2) + signal.Notify(sigs) + go func() { + for { + sig := <-sigs + if sig == syscall.SIGINT || sig == syscall.SIGTERM { + log.Debug().Any("signal", sig).Msg("signal received") + s.stop <- nil + } else if runtime.GOOS == "linux" && sig == syscall.Signal(0x17) { + // Ignore SIGURG + } + } + }() +} + +func (s *Service) shutdown(sourceErr error) (err error) { + log.Debug().Msg("function stopping") + var runtimeErr, instanceErr error + + ctx, cancel := context.WithTimeout(context.Background(), ServerShutdownTimeout) + defer cancel() + runtimeErr = s.Shutdown(ctx) + + if i, ok := s.f.(Stopper); ok { + ctx, cancel = context.WithTimeout(context.Background(), InstanceStopTimeout) + defer cancel() + instanceErr = i.Stop(ctx) + } + + return collapseErrors("shutdown error", sourceErr, instanceErr, runtimeErr) +} + +func listenAddress() string { + if v := os.Getenv("LISTEN_ADDRESS"); v != "" { + return v + } + return DefaultListenAddress +} + +func readCfg() (map[string]string, error) { + cfg := map[string]string{} + f, err := os.Open("cfg") + if err != nil { + return cfg, nil + } + defer f.Close() + + scanner := bufio.NewScanner(f) + i := 0 + for scanner.Scan() { + i++ + line := scanner.Text() + parts := strings.SplitN(line, "=", 2) + if len(parts) != 2 { + return cfg, fmt.Errorf("config line %v invalid: %v", i, line) + } + cfg[strings.TrimSpace(parts[0])] = strings.Trim(strings.TrimSpace(parts[1]), "\"") + } + return cfg, scanner.Err() +} + +func newCfg() (cfg map[string]string, err error) { + if cfg, err = readCfg(); err != nil { + return + } + for _, e := range os.Environ() { + pair := strings.SplitN(e, "=", 2) + cfg[pair[0]] = pair[1] + } + return +} + +func collapseErrors(msg string, ee ...error) (err error) { + for _, e := range ee { + if e != nil { + if err == nil { + err = e + } else { + log.Error().Err(e).Msg(msg) + } + } + } + return +} \ No newline at end of file From 922daccf47bbae0fa36bfd0f91faa91b01bbe45c Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Thu, 16 Jul 2026 15:16:53 +0300 Subject: [PATCH 02/18] Fix data race on responseEventWarned using sync.Once --- kafka/handler_test.go | 6 ++---- kafka/instance.go | 10 ++++++---- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/kafka/handler_test.go b/kafka/handler_test.go index f5d1d31..921cfad 100644 --- a/kafka/handler_test.go +++ b/kafka/handler_test.go @@ -2,6 +2,7 @@ package kafka import ( "context" + "sync" "testing" "github.com/cloudevents/sdk-go/v2/event" @@ -65,7 +66,7 @@ func TestInvokeHandler_DefaultHandler(t *testing.T) { } func TestInvokeHandler_ResponseEventIgnored(t *testing.T) { - responseEventWarned = false + responseEventOnce = sync.Once{} resp := event.New() resp.SetType("response") @@ -81,9 +82,6 @@ func TestInvokeHandler_ResponseEventIgnored(t *testing.T) { if err != nil { t.Fatalf("unexpected error: %v", err) } - if !responseEventWarned { - t.Fatal("expected response event warning to be set") - } } // Test handler types for all 12 supported CloudEvents signatures. diff --git a/kafka/instance.go b/kafka/instance.go index 569d75d..43485de 100644 --- a/kafka/instance.go +++ b/kafka/instance.go @@ -2,6 +2,7 @@ package kafka import ( "context" + "sync" "github.com/cloudevents/sdk-go/v2/event" "github.com/rs/zerolog/log" @@ -31,7 +32,7 @@ type LivenessReporter interface { Alive(context.Context) (bool, error) } -var responseEventWarned bool +var responseEventOnce sync.Once // invokeHandler calls a CloudEvents handler directly with a constructed event. func invokeHandler(f any, ctx context.Context, e event.Event) error { @@ -88,9 +89,10 @@ func invokeHandlerFn(fn any, ctx context.Context, e event.Event) error { } func warnResponseEvent(resp *event.Event) { - if resp != nil && !responseEventWarned { - responseEventWarned = true - log.Warn().Msg("handler returned a response event, but response events are ignored when consuming from Kafka") + if resp != nil { + responseEventOnce.Do(func() { + log.Warn().Msg("handler returned a response event, but response events are ignored when consuming from Kafka") + }) } } From 0d6421bec6a7cb8ab054fca61bb350ba0399d29a Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Thu, 16 Jul 2026 15:29:42 +0300 Subject: [PATCH 03/18] Parse ce_dataschema as standard CE attribute in pass-through --- kafka/consumer.go | 5 ++++- kafka/consumer_test.go | 4 ++++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/kafka/consumer.go b/kafka/consumer.go index 74967d9..711bfcb 100644 --- a/kafka/consumer.go +++ b/kafka/consumer.go @@ -219,6 +219,9 @@ func parseCEFromHeaders(msg Message) (event.Event, bool) { e.SetTime(t) } } + if v, ok := headers["ce_dataschema"]; ok { + e.SetDataSchema(v) + } contentType := "application/json" if v, ok := headers["ce_datacontenttype"]; ok { @@ -231,7 +234,7 @@ func parseCEFromHeaders(msg Message) (event.Event, bool) { if strings.HasPrefix(k, "ce_") { attr := strings.TrimPrefix(k, "ce_") switch attr { - case "specversion", "id", "source", "type", "subject", "time", "datacontenttype": + case "specversion", "id", "source", "type", "subject", "time", "datacontenttype", "dataschema": continue default: e.SetExtension(attr, v) diff --git a/kafka/consumer_test.go b/kafka/consumer_test.go index 6277fb4..43c33df 100644 --- a/kafka/consumer_test.go +++ b/kafka/consumer_test.go @@ -87,6 +87,7 @@ func TestKafkaMessageToEvent_CEPassThrough(t *testing.T) { {Key: "ce_subject", Value: []byte("temp")}, {Key: "ce_time", Value: []byte("2025-06-15T12:00:00Z")}, {Key: "ce_datacontenttype", Value: []byte("application/json")}, + {Key: "ce_dataschema", Value: []byte("https://example.com/schema/sensor.json")}, {Key: "ce_customext", Value: []byte("custom-value")}, }, Partition: 1, @@ -120,6 +121,9 @@ func TestKafkaMessageToEvent_CEPassThrough(t *testing.T) { if string(e.Data()) != `{"temperature":22}` { t.Errorf("data = %q", string(e.Data())) } + if e.DataSchema() != "https://example.com/schema/sensor.json" { + t.Errorf("dataschema = %q, want https://example.com/schema/sensor.json", e.DataSchema()) + } if v, ok := e.Extensions()["customext"]; !ok || v != "custom-value" { t.Errorf("customext = %v", v) } From d50f078b6163fd830e810d8bedad2acbd3dcc86f Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Thu, 16 Jul 2026 15:33:03 +0300 Subject: [PATCH 04/18] Handle *DefaultHandler pointer in invokeHandler --- kafka/handler_test.go | 22 ++++++++++++++++++++++ kafka/instance.go | 7 +++++-- 2 files changed, 27 insertions(+), 2 deletions(-) diff --git a/kafka/handler_test.go b/kafka/handler_test.go index 921cfad..c6c66bc 100644 --- a/kafka/handler_test.go +++ b/kafka/handler_test.go @@ -65,6 +65,28 @@ func TestInvokeHandler_DefaultHandler(t *testing.T) { } } +func TestInvokeHandler_DefaultHandlerPointer(t *testing.T) { + called := false + dh := &ce.DefaultHandler{ + Handler: func(ctx context.Context, e event.Event) error { + called = true + return nil + }, + } + + e := event.New() + e.SetType("test") + e.SetSource("test") + + err := invokeHandler(dh, context.Background(), e) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !called { + t.Fatal("*DefaultHandler's function was not called") + } +} + func TestInvokeHandler_ResponseEventIgnored(t *testing.T) { responseEventOnce = sync.Once{} diff --git a/kafka/instance.go b/kafka/instance.go index 43485de..2fa5749 100644 --- a/kafka/instance.go +++ b/kafka/instance.go @@ -37,9 +37,12 @@ var responseEventOnce sync.Once // invokeHandler calls a CloudEvents handler directly with a constructed event. func invokeHandler(f any, ctx context.Context, e event.Event) error { var fn any - if dh, ok := f.(ce.DefaultHandler); ok { + switch dh := f.(type) { + case ce.DefaultHandler: fn = dh.Handler - } else { + case *ce.DefaultHandler: + fn = dh.Handler + default: fn = ce.GetReceiverFn(f) } return invokeHandlerFn(fn, ctx, e) From 608c28c2eea80fcfd754859a7e379dbe4ebc2f4d Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Thu, 16 Jul 2026 15:39:05 +0300 Subject: [PATCH 05/18] Validate handler signature at startup instead of first message --- kafka/instance.go | 30 ++++++++++++++++++++++++++++++ kafka/service.go | 4 ++++ 2 files changed, 34 insertions(+) diff --git a/kafka/instance.go b/kafka/instance.go index 2fa5749..ea968ab 100644 --- a/kafka/instance.go +++ b/kafka/instance.go @@ -2,6 +2,7 @@ package kafka import ( "context" + "fmt" "sync" "github.com/cloudevents/sdk-go/v2/event" @@ -34,6 +35,35 @@ type LivenessReporter interface { var responseEventOnce sync.Once +func validateHandler(f any) error { + var fn any + switch dh := f.(type) { + case ce.DefaultHandler: + fn = dh.Handler + case *ce.DefaultHandler: + fn = dh.Handler + default: + fn = ce.GetReceiverFn(f) + } + switch fn.(type) { + case func(), + func() error, + func(context.Context), + func(context.Context) error, + func(event.Event), + func(event.Event) error, + func(context.Context, event.Event), + func(context.Context, event.Event) error, + func(event.Event) *event.Event, + func(event.Event) (*event.Event, error), + func(context.Context, event.Event) *event.Event, + func(context.Context, event.Event) (*event.Event, error): + return nil + default: + return fmt.Errorf("handler function does not match any supported CloudEvents signature") + } +} + // invokeHandler calls a CloudEvents handler directly with a constructed event. func invokeHandler(f any, ctx context.Context, e event.Event) error { var fn any diff --git a/kafka/service.go b/kafka/service.go index 6a82a3a..c045a38 100644 --- a/kafka/service.go +++ b/kafka/service.go @@ -62,6 +62,10 @@ func New(f any) *Service { // Start the Kafka consumer and health HTTP server. func (s *Service) Start(ctx context.Context) (err error) { + if err = validateHandler(s.f); err != nil { + return + } + addr := listenAddress() log.Debug().Str("address", addr).Msg("kafka service starting") From 15d7f2673f15d48cf2b5b476bbfb3fe0384e2549 Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Thu, 16 Jul 2026 16:32:10 +0300 Subject: [PATCH 06/18] Move context.Context to first parameter in invokeHandler --- kafka/consumer.go | 2 +- kafka/handler_test.go | 8 ++++---- kafka/instance.go | 6 +++--- 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/kafka/consumer.go b/kafka/consumer.go index 711bfcb..ceed52a 100644 --- a/kafka/consumer.go +++ b/kafka/consumer.go @@ -144,7 +144,7 @@ func (h *consumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, e := kafkaMessageToEvent(m, h.brokers) - if err := invokeHandler(h.f, session.Context(), e); err != nil { + if err := invokeHandler(session.Context(), h.f, e); err != nil { log.Error().Err(err). Str("topic", msg.Topic). Int32("partition", msg.Partition). diff --git a/kafka/handler_test.go b/kafka/handler_test.go index c6c66bc..bb9f1bd 100644 --- a/kafka/handler_test.go +++ b/kafka/handler_test.go @@ -35,7 +35,7 @@ func TestInvokeHandler_AllSignatures(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - err := invokeHandler(tt.f, context.Background(), e) + err := invokeHandler(context.Background(), tt.f, e) if err != nil { t.Errorf("unexpected error: %v", err) } @@ -56,7 +56,7 @@ func TestInvokeHandler_DefaultHandler(t *testing.T) { e.SetType("test") e.SetSource("test") - err := invokeHandler(dh, context.Background(), e) + err := invokeHandler(context.Background(), dh, e) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -78,7 +78,7 @@ func TestInvokeHandler_DefaultHandlerPointer(t *testing.T) { e.SetType("test") e.SetSource("test") - err := invokeHandler(dh, context.Background(), e) + err := invokeHandler(context.Background(), dh, e) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -100,7 +100,7 @@ func TestInvokeHandler_ResponseEventIgnored(t *testing.T) { e.SetType("test") e.SetSource("test") - err := invokeHandler(f, context.Background(), e) + err := invokeHandler(context.Background(), f, e) if err != nil { t.Fatalf("unexpected error: %v", err) } diff --git a/kafka/instance.go b/kafka/instance.go index ea968ab..aa07955 100644 --- a/kafka/instance.go +++ b/kafka/instance.go @@ -65,7 +65,7 @@ func validateHandler(f any) error { } // invokeHandler calls a CloudEvents handler directly with a constructed event. -func invokeHandler(f any, ctx context.Context, e event.Event) error { +func invokeHandler(ctx context.Context, f any, e event.Event) error { var fn any switch dh := f.(type) { case ce.DefaultHandler: @@ -75,10 +75,10 @@ func invokeHandler(f any, ctx context.Context, e event.Event) error { default: fn = ce.GetReceiverFn(f) } - return invokeHandlerFn(fn, ctx, e) + return invokeHandlerFn(ctx, fn, e) } -func invokeHandlerFn(fn any, ctx context.Context, e event.Event) error { +func invokeHandlerFn(ctx context.Context, fn any, e event.Event) error { switch h := fn.(type) { case func(): h() From f06959a664e876f4be76859f5dc37e04e88a7de8 Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Thu, 16 Jul 2026 16:39:46 +0300 Subject: [PATCH 07/18] Add unit tests for health handlers, CE pass-through, and consumeLoop --- kafka/consumer_test.go | 96 +++++++++++++++++++++++++++++ kafka/service_test.go | 134 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 230 insertions(+) create mode 100644 kafka/service_test.go diff --git a/kafka/consumer_test.go b/kafka/consumer_test.go index 43c33df..da5f1fa 100644 --- a/kafka/consumer_test.go +++ b/kafka/consumer_test.go @@ -1,7 +1,10 @@ package kafka import ( + "context" "fmt" + "strings" + "sync/atomic" "testing" "time" ) @@ -165,3 +168,96 @@ func TestKafkaMessageToEvent_NotCE(t *testing.T) { t.Errorf("non-CE message should get type dev.knative.kafka.event, got %q", e.Type()) } } + +func TestKafkaMessageToEvent_CEPassThrough_MissingAttributes(t *testing.T) { + // Message has ce_specversion but is missing ce_id, ce_source, ce_type. + // It should still be treated as a CE pass-through (not wrapped), and the + // missing required attributes should be empty strings. + msg := Message{ + Value: []byte(`{"data":"value"}`), + Topic: "events", + Headers: []Header{ + {Key: "ce_specversion", Value: []byte("1.0")}, + }, + Partition: 0, + Offset: 10, + } + + e := kafkaMessageToEvent(msg, "b:9092") + + // Should be treated as CE pass-through, not the generated wrapper. + if e.SpecVersion() != "1.0" { + t.Errorf("specversion = %q, want 1.0", e.SpecVersion()) + } + // The wrapper would set type to "dev.knative.kafka.event", so verify it is + // empty (CE pass-through with missing ce_type). + if e.Type() != "" { + t.Errorf("type = %q, want empty string", e.Type()) + } + if e.ID() != "" { + t.Errorf("id = %q, want empty string", e.ID()) + } + if e.Source() != "" { + t.Errorf("source = %q, want empty string", e.Source()) + } + // Data should still be set. + if string(e.Data()) != `{"data":"value"}` { + t.Errorf("data = %q", string(e.Data())) + } +} + +// dummyHandler is a minimal handler used in consumeLoop tests. +// consumeLoop fails on env-var validation before it tries to invoke the handler. +type dummyHandler struct{} + +func (h *dummyHandler) Handle() {} + +func TestConsumeLoop_MissingEnvVars(t *testing.T) { + tests := []struct { + name string + envVars map[string]string + wantErr string + }{ + { + name: "no KAFKA_BROKERS", + envVars: map[string]string{}, + wantErr: "KAFKA_BROKERS", + }, + { + name: "no KAFKA_TOPIC", + envVars: map[string]string{ + "KAFKA_BROKERS": "localhost:9092", + }, + wantErr: "KAFKA_TOPIC", + }, + { + name: "no KAFKA_CONSUMER_GROUP", + envVars: map[string]string{ + "KAFKA_BROKERS": "localhost:9092", + "KAFKA_TOPIC": "my-topic", + }, + wantErr: "KAFKA_CONSUMER_GROUP", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // Clear all three env vars, then set only what the subtest provides. + t.Setenv("KAFKA_BROKERS", "") + t.Setenv("KAFKA_TOPIC", "") + t.Setenv("KAFKA_CONSUMER_GROUP", "") + for k, v := range tt.envVars { + t.Setenv(k, v) + } + + var ready atomic.Bool + err := consumeLoop(context.Background(), &dummyHandler{}, &ready) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), tt.wantErr) { + t.Errorf("error = %q, want it to contain %q", err.Error(), tt.wantErr) + } + }) + } +} diff --git a/kafka/service_test.go b/kafka/service_test.go new file mode 100644 index 0000000..5d3bb3d --- /dev/null +++ b/kafka/service_test.go @@ -0,0 +1,134 @@ +package kafka + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "testing" +) + +// readyHandler implements ReadinessReporter. +type readyHandler struct { + readyVal bool + readyErr error +} + +func (h *readyHandler) Handle() {} +func (h *readyHandler) Ready(_ context.Context) (bool, error) { + return h.readyVal, h.readyErr +} + +// aliveHandler implements LivenessReporter. +type aliveHandler struct { + aliveVal bool + aliveErr error +} + +func (h *aliveHandler) Handle() {} +func (h *aliveHandler) Alive(_ context.Context) (bool, error) { + return h.aliveVal, h.aliveErr +} + +// plainHandler implements neither ReadinessReporter nor LivenessReporter. +type plainHandler struct{} + +func (h *plainHandler) Handle() {} + +func TestReady_NotReady(t *testing.T) { + svc := New(&plainHandler{}) + // ready defaults to false + + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/health/readiness", nil) + svc.Ready(w, r) + + if w.Code != http.StatusServiceUnavailable { + t.Errorf("expected status %d, got %d", http.StatusServiceUnavailable, w.Code) + } +} + +func TestReady_Ready(t *testing.T) { + svc := New(&plainHandler{}) + svc.ready.Store(true) + + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/health/readiness", nil) + svc.Ready(w, r) + + if w.Code != http.StatusOK { + t.Errorf("expected status %d, got %d", http.StatusOK, w.Code) + } + if w.Body.String() != "READY" { + t.Errorf("expected body %q, got %q", "READY", w.Body.String()) + } +} + +func TestReady_UserReporterNotReady(t *testing.T) { + h := &readyHandler{readyVal: false, readyErr: nil} + svc := New(h) + svc.ready.Store(true) + + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/health/readiness", nil) + svc.Ready(w, r) + + if w.Code != http.StatusServiceUnavailable { + t.Errorf("expected status %d, got %d", http.StatusServiceUnavailable, w.Code) + } +} + +func TestReady_UserReporterError(t *testing.T) { + h := &readyHandler{readyVal: false, readyErr: fmt.Errorf("db down")} + svc := New(h) + svc.ready.Store(true) + + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/health/readiness", nil) + svc.Ready(w, r) + + if w.Code != http.StatusInternalServerError { + t.Errorf("expected status %d, got %d", http.StatusInternalServerError, w.Code) + } +} + +func TestAlive_Default(t *testing.T) { + svc := New(&plainHandler{}) + + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/health/liveness", nil) + svc.Alive(w, r) + + if w.Code != http.StatusOK { + t.Errorf("expected status %d, got %d", http.StatusOK, w.Code) + } + if w.Body.String() != "ALIVE" { + t.Errorf("expected body %q, got %q", "ALIVE", w.Body.String()) + } +} + +func TestAlive_NotAlive(t *testing.T) { + h := &aliveHandler{aliveVal: false, aliveErr: nil} + svc := New(h) + + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/health/liveness", nil) + svc.Alive(w, r) + + if w.Code != http.StatusServiceUnavailable { + t.Errorf("expected status %d, got %d", http.StatusServiceUnavailable, w.Code) + } +} + +func TestAlive_Error(t *testing.T) { + h := &aliveHandler{aliveVal: false, aliveErr: fmt.Errorf("check failed")} + svc := New(h) + + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/health/liveness", nil) + svc.Alive(w, r) + + if w.Code != http.StatusInternalServerError { + t.Errorf("expected status %d, got %d", http.StatusInternalServerError, w.Code) + } +} From b9d48bd71ebac39a4a6910acbc51d2829f79ae7c Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Thu, 16 Jul 2026 17:13:45 +0300 Subject: [PATCH 08/18] Cancel consumer context on shutdown so Sarama unblocks promptly --- kafka/service.go | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/kafka/service.go b/kafka/service.go index c045a38..02dfb90 100644 --- a/kafka/service.go +++ b/kafka/service.go @@ -34,10 +34,11 @@ func Start(f any) error { // the function handler. An HTTP server runs alongside for health probes only. type Service struct { http.Server - listener net.Listener - f any - stop chan error - ready atomic.Bool + listener net.Listener + f any + stop chan error + ready atomic.Bool + cancelConsume context.CancelFunc } // New creates a Service for the given handler. @@ -86,8 +87,10 @@ func (s *Service) Start(ctx context.Context) (err error) { } }() + consumerCtx, cancelConsume := context.WithCancel(ctx) + s.cancelConsume = cancelConsume go func() { - if err := consumeLoop(ctx, s.f, &s.ready); err != nil { + if err := consumeLoop(consumerCtx, s.f, &s.ready); err != nil { log.Error().Err(err).Msg("kafka consumer exited with error") s.stop <- err } @@ -191,6 +194,9 @@ func (s *Service) handleSignals() { func (s *Service) shutdown(sourceErr error) (err error) { log.Debug().Msg("function stopping") + if s.cancelConsume != nil { + s.cancelConsume() + } var runtimeErr, instanceErr error ctx, cancel := context.WithTimeout(context.Background(), ServerShutdownTimeout) From d035e92e3282dc03188fa72ac3014b8f8d74841f Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Thu, 16 Jul 2026 23:04:35 +0300 Subject: [PATCH 09/18] Recover from GetReceiverFn panic in validateHandler --- kafka/handler_test.go | 8 ++++++++ kafka/instance.go | 14 ++++++++++++-- 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/kafka/handler_test.go b/kafka/handler_test.go index bb9f1bd..d022199 100644 --- a/kafka/handler_test.go +++ b/kafka/handler_test.go @@ -87,6 +87,14 @@ func TestInvokeHandler_DefaultHandlerPointer(t *testing.T) { } } +func TestValidateHandler_InvalidSignature(t *testing.T) { + type badHandler struct{} + err := validateHandler(&badHandler{}) + if err == nil { + t.Fatal("expected error for handler without Handle method") + } +} + func TestInvokeHandler_ResponseEventIgnored(t *testing.T) { responseEventOnce = sync.Once{} diff --git a/kafka/instance.go b/kafka/instance.go index aa07955..1f7c396 100644 --- a/kafka/instance.go +++ b/kafka/instance.go @@ -35,7 +35,7 @@ type LivenessReporter interface { var responseEventOnce sync.Once -func validateHandler(f any) error { +func validateHandler(f any) (err error) { var fn any switch dh := f.(type) { case ce.DefaultHandler: @@ -43,7 +43,17 @@ func validateHandler(f any) error { case *ce.DefaultHandler: fn = dh.Handler default: - fn = ce.GetReceiverFn(f) + func() { + defer func() { + if r := recover(); r != nil { + err = fmt.Errorf("handler function does not match any supported CloudEvents signature") + } + }() + fn = ce.GetReceiverFn(f) + }() + if err != nil { + return + } } switch fn.(type) { case func(), From 446dd5d3310a411249a75d0b23d35d461a58acbe Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Thu, 16 Jul 2026 23:08:48 +0300 Subject: [PATCH 10/18] Support content-type header for CE datacontenttype in Kafka pass-through --- kafka/consumer.go | 3 +++ kafka/consumer_test.go | 39 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+) diff --git a/kafka/consumer.go b/kafka/consumer.go index ceed52a..bcd9b15 100644 --- a/kafka/consumer.go +++ b/kafka/consumer.go @@ -224,6 +224,9 @@ func parseCEFromHeaders(msg Message) (event.Event, bool) { } contentType := "application/json" + if v, ok := headers["content-type"]; ok { + contentType = v + } if v, ok := headers["ce_datacontenttype"]; ok { contentType = v } diff --git a/kafka/consumer_test.go b/kafka/consumer_test.go index da5f1fa..16817dd 100644 --- a/kafka/consumer_test.go +++ b/kafka/consumer_test.go @@ -150,6 +150,45 @@ func TestKafkaMessageToEvent_CEPassThrough_DefaultContentType(t *testing.T) { } } +func TestKafkaMessageToEvent_CEPassThrough_ContentTypeHeader(t *testing.T) { + msg := Message{ + Value: []byte(``), + Headers: []Header{ + {Key: "ce_specversion", Value: []byte("1.0")}, + {Key: "ce_id", Value: []byte("x")}, + {Key: "ce_source", Value: []byte("s")}, + {Key: "ce_type", Value: []byte("t")}, + {Key: "content-type", Value: []byte("application/xml")}, + }, + } + + e := kafkaMessageToEvent(msg, "b:9092") + + if e.DataContentType() != "application/xml" { + t.Errorf("content type = %q, want application/xml", e.DataContentType()) + } +} + +func TestKafkaMessageToEvent_CEPassThrough_CEDataContentTypeOverridesContentType(t *testing.T) { + msg := Message{ + Value: []byte(`{}`), + Headers: []Header{ + {Key: "ce_specversion", Value: []byte("1.0")}, + {Key: "ce_id", Value: []byte("x")}, + {Key: "ce_source", Value: []byte("s")}, + {Key: "ce_type", Value: []byte("t")}, + {Key: "content-type", Value: []byte("application/xml")}, + {Key: "ce_datacontenttype", Value: []byte("text/plain")}, + }, + } + + e := kafkaMessageToEvent(msg, "b:9092") + + if e.DataContentType() != "text/plain" { + t.Errorf("content type = %q, want text/plain (ce_datacontenttype should override content-type)", e.DataContentType()) + } +} + func TestKafkaMessageToEvent_NotCE(t *testing.T) { msg := Message{ Value: []byte("plain data"), From 8c6cded4cb2007f11b7d12d417a00d38347b2b7d Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Thu, 16 Jul 2026 23:35:33 +0300 Subject: [PATCH 11/18] Use buffered stop channel with non-blocking sends --- kafka/service.go | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/kafka/service.go b/kafka/service.go index 02dfb90..631048d 100644 --- a/kafka/service.go +++ b/kafka/service.go @@ -45,7 +45,7 @@ type Service struct { func New(f any) *Service { svc := &Service{ f: f, - stop: make(chan error), + stop: make(chan error, 1), Server: http.Server{ ReadTimeout: 30 * time.Second, WriteTimeout: 30 * time.Second, @@ -83,7 +83,7 @@ func (s *Service) Start(ctx context.Context) (err error) { go func() { if err := s.Serve(s.listener); err != http.ErrServerClosed { log.Error().Err(err).Msg("http server exited with unexpected error") - s.stop <- err + s.sendStop(err) } }() @@ -92,7 +92,7 @@ func (s *Service) Start(ctx context.Context) (err error) { go func() { if err := consumeLoop(consumerCtx, s.f, &s.ready); err != nil { log.Error().Err(err).Msg("kafka consumer exited with error") - s.stop <- err + s.sendStop(err) } }() @@ -167,7 +167,7 @@ func (s *Service) startInstance(ctx context.Context) error { } go func() { if err := i.Start(ctx, cfg); err != nil { - s.stop <- err + s.sendStop(err) } }() } else { @@ -184,7 +184,7 @@ func (s *Service) handleSignals() { sig := <-sigs if sig == syscall.SIGINT || sig == syscall.SIGTERM { log.Debug().Any("signal", sig).Msg("signal received") - s.stop <- nil + s.sendStop(nil) } else if runtime.GOOS == "linux" && sig == syscall.Signal(0x17) { // Ignore SIGURG } @@ -192,6 +192,13 @@ func (s *Service) handleSignals() { }() } +func (s *Service) sendStop(err error) { + select { + case s.stop <- err: + default: + } +} + func (s *Service) shutdown(sourceErr error) (err error) { log.Debug().Msg("function stopping") if s.cancelConsume != nil { From e50d677c314af43deb83c765569b0c14bd63823b Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Thu, 16 Jul 2026 23:41:24 +0300 Subject: [PATCH 12/18] Add GoDoc comment to exported GetReceiverFn --- cloudevents/instance.go | 1 + 1 file changed, 1 insertion(+) diff --git a/cloudevents/instance.go b/cloudevents/instance.go index 2f10123..84d3e54 100644 --- a/cloudevents/instance.go +++ b/cloudevents/instance.go @@ -162,6 +162,7 @@ type handlerCtxEvtEvtErr interface { Handle(context.Context, event.Event) (*event.Event, error) } +// GetReceiverFn extracts the bound Handle method from a CloudEvents handler as a raw function value. func GetReceiverFn(f any) any { switch h := f.(type) { case handler: From 6dfc99939ae6d61eb14fadb182e66c55fdb4147c Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Thu, 16 Jul 2026 23:59:29 +0300 Subject: [PATCH 13/18] Close listener when startInstance fails to prevent port leak --- kafka/service.go | 1 + 1 file changed, 1 insertion(+) diff --git a/kafka/service.go b/kafka/service.go index 631048d..5e03918 100644 --- a/kafka/service.go +++ b/kafka/service.go @@ -75,6 +75,7 @@ func (s *Service) Start(ctx context.Context) (err error) { } if err = s.startInstance(ctx); err != nil { + s.listener.Close() return } From eb2e72d7d2e6231a2e97a1ab8d171b801693f440 Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Fri, 17 Jul 2026 00:06:01 +0300 Subject: [PATCH 14/18] Log debug message when static config file is absent --- kafka/service.go | 1 + 1 file changed, 1 insertion(+) diff --git a/kafka/service.go b/kafka/service.go index 5e03918..3c113ab 100644 --- a/kafka/service.go +++ b/kafka/service.go @@ -231,6 +231,7 @@ func readCfg() (map[string]string, error) { cfg := map[string]string{} f, err := os.Open("cfg") if err != nil { + log.Debug().Msg("no static config") return cfg, nil } defer f.Close() From 52ebda9a986053e1cc6a94a6be8436ee0e90f7be Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Fri, 17 Jul 2026 11:01:14 +0300 Subject: [PATCH 15/18] Reject nil handler function in validateHandler --- kafka/handler_test.go | 8 ++++++++ kafka/instance.go | 3 +++ 2 files changed, 11 insertions(+) diff --git a/kafka/handler_test.go b/kafka/handler_test.go index d022199..47cf844 100644 --- a/kafka/handler_test.go +++ b/kafka/handler_test.go @@ -95,6 +95,14 @@ func TestValidateHandler_InvalidSignature(t *testing.T) { } } +func TestValidateHandler_NilDefaultHandler(t *testing.T) { + dh := ce.DefaultHandler{Handler: nil} + err := validateHandler(dh) + if err == nil { + t.Fatal("expected error for DefaultHandler with nil Handler") + } +} + func TestInvokeHandler_ResponseEventIgnored(t *testing.T) { responseEventOnce = sync.Once{} diff --git a/kafka/instance.go b/kafka/instance.go index 1f7c396..81c2755 100644 --- a/kafka/instance.go +++ b/kafka/instance.go @@ -55,6 +55,9 @@ func validateHandler(f any) (err error) { return } } + if fn == nil { + return fmt.Errorf("handler function is nil") + } switch fn.(type) { case func(), func() error, From c446904fbd0f98f102412b740fefdace77fe361a Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Fri, 17 Jul 2026 11:51:03 +0300 Subject: [PATCH 16/18] Set explicit Sarama Kafka version to V2_0_0_0 --- kafka/consumer.go | 1 + 1 file changed, 1 insertion(+) diff --git a/kafka/consumer.go b/kafka/consumer.go index bcd9b15..d38ffd1 100644 --- a/kafka/consumer.go +++ b/kafka/consumer.go @@ -65,6 +65,7 @@ func consumeLoop(ctx context.Context, f any, ready *atomic.Bool) error { Msg("connecting to kafka") config := sarama.NewConfig() + config.Version = sarama.V2_0_0_0 config.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{ sarama.NewBalanceStrategyRoundRobin(), } From 612143596d3d5292f74539b1be5aa3a6fbc61f74 Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Fri, 17 Jul 2026 12:08:22 +0300 Subject: [PATCH 17/18] Fix goimports and rename dummy to placeholder in tests --- kafka/consumer.go | 2 +- kafka/consumer_test.go | 8 ++++---- kafka/instance.go | 1 - kafka/service.go | 2 +- 4 files changed, 6 insertions(+), 7 deletions(-) diff --git a/kafka/consumer.go b/kafka/consumer.go index d38ffd1..b5edf0b 100644 --- a/kafka/consumer.go +++ b/kafka/consumer.go @@ -247,4 +247,4 @@ func parseCEFromHeaders(msg Message) (event.Event, bool) { } return e, true -} \ No newline at end of file +} diff --git a/kafka/consumer_test.go b/kafka/consumer_test.go index 16817dd..c945a2a 100644 --- a/kafka/consumer_test.go +++ b/kafka/consumer_test.go @@ -245,11 +245,11 @@ func TestKafkaMessageToEvent_CEPassThrough_MissingAttributes(t *testing.T) { } } -// dummyHandler is a minimal handler used in consumeLoop tests. +// placeholderHandler is a minimal handler used in consumeLoop tests. // consumeLoop fails on env-var validation before it tries to invoke the handler. -type dummyHandler struct{} +type placeholderHandler struct{} -func (h *dummyHandler) Handle() {} +func (h *placeholderHandler) Handle() {} func TestConsumeLoop_MissingEnvVars(t *testing.T) { tests := []struct { @@ -290,7 +290,7 @@ func TestConsumeLoop_MissingEnvVars(t *testing.T) { } var ready atomic.Bool - err := consumeLoop(context.Background(), &dummyHandler{}, &ready) + err := consumeLoop(context.Background(), &placeholderHandler{}, &ready) if err == nil { t.Fatal("expected error, got nil") } diff --git a/kafka/instance.go b/kafka/instance.go index 81c2755..f5677d2 100644 --- a/kafka/instance.go +++ b/kafka/instance.go @@ -141,4 +141,3 @@ func warnResponseEvent(resp *event.Event) { }) } } - diff --git a/kafka/service.go b/kafka/service.go index 3c113ab..8026148 100644 --- a/kafka/service.go +++ b/kafka/service.go @@ -272,4 +272,4 @@ func collapseErrors(msg string, ee ...error) (err error) { } } return -} \ No newline at end of file +} From ae16f932d070c1a266e6985a804de1a0a7a9ab39 Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Fri, 17 Jul 2026 12:23:10 +0300 Subject: [PATCH 18/18] Fix missing newline at end of cmd/fkafka/main.go --- cmd/fkafka/main.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmd/fkafka/main.go b/cmd/fkafka/main.go index 1ae46f9..fe34227 100644 --- a/cmd/fkafka/main.go +++ b/cmd/fkafka/main.go @@ -43,4 +43,4 @@ func (f *MyFunction) Handle(ctx context.Context, e cloudevents.Event) error { fmt.Printf("Data: %s\n", string(e.Data())) } return nil -} \ No newline at end of file +}