knative-extensions / knative-extensions/func-go

Kafka middleware: support for producing messages

Open
#176 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Go
Stars
7
Forks
12
PR merge metrics
No merged PRs in 30d

Description

The middleware is currently consume-only. Add support for producing messages. Several approaches to consider:

**Return value**: handler returns a `*Message`, middleware produces it to a configured output topic. Simple for pipelines but limited to one output topic and one message per input. Requires a new interface (e.g. `ProducingHandler`).
```go
func (f *MyFunction) Handle(ctx context.Context, msg kafka.Message) (*kafka.Message, error) {
result := transform(msg.Value)
return &kafka.Message{Value: result, Topic: "output-topic"}, nil
}
```

**Producer argument**: handler receives a `kafka.Producer` as an additional argument. Full flexibility (any topic, any number of messages) but also requires a new interface.
```go
func (f *MyFunction) Handle(ctx context.Context, msg kafka.Message, producer kafka.Producer) error {
producer.Send("topic-a", kafka.Message{Value: transformed})
producer.Send("topic-b", kafka.Message{Value: enriched})
return nil
}
```

**Producer via context**: `kafka.ProducerFromContext(ctx)` — doesn't change the `Handler` interface. Backwards compatible. Producer is available if configured, absent if not.
```go
func (f *MyFunction) Handle(ctx context.Context, msg kafka.Message) error {
producer := kafka.ProducerFromContext(ctx)
producer.Send("topic-a", kafka.Message{Value: result})
return nil
}
```

**Producer via Start hook**: user creates their own producer in `Start`, stores it on the struct, uses it in `Handle`. Only works for instanced functions.
```go
func (f *MyFunction) Start(ctx context.Context, cfg map[string]string) error {
f.producer = kafka.NewProducer(cfg)
return nil
}
func (f *MyFunction) Handle(ctx context.Context, msg kafka.Message) error {
f.producer.Send("output", kafka.Message{Value: result})
return nil
}
```

**Standalone producer utility**: provide `kafka.NewProducer()` as a utility, user manages lifecycle. Most flexible, least middleware involvement.

Contributor guide

Open the contributing guide

Research direction

Start by reviewing the existing Kafka middleware and its consume-only Handler path. Compare the five proposed producing approaches, then establish the chosen interface, producer lifecycle, and output behavior before implementing support. Done means the middleware can produce messages while preserving the intended handler behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
go, kafka
Domain
stream-processing
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Quiet
Clarity
Needs clarification
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.