Proposal: queue processing and the Kafka runtime

[Zaba505]

Note

Part of the humus re-evaluation; see Proposal: humus on the bedrock process model, wave 2. It builds on Proposal: the humus App lifecycle, Proposal: configuration for humus apps, Proposal: OpenTelemetry without globals and Proposal: humustest. It depends on no other wave-2 proposal. Proposal: exposing health probes (wave 3) adds probes to the App proposed here. It answers the open question Proposal: humustest handed it: unit tests run on kfake, listening on humustest's Host.

Summary

• queue keeps Processor and gains Delivery. It is the contract between the end user's code and any broker: what a message's processing returns, and when the message is acknowledged. Consumer, Acknowledger, Runtime, App and ErrEndOfQueue are removed.
• queue/kafka is a nested module with a Service and an App. The Service joins a consumer group with franz-go, dials the brokers through the Host, and processes each partition's records one at a time, in offset order. The App runs one Service on humus.App.
• The delivery semantics are fixed and tested. Under at-least-once, a record is committed only once it and every earlier record of its partition were processed, and a processing error ends the Service, so the App exits 1 and the record is redelivered. Under at-most-once, a batch is committed before its records are processed, and a processing error is reported and processing goes on.
• Shutdown drains. On Serve's context the Service stops fetching, lets the record in flight on each partition finish, commits, and leaves the group, all within the grace period.
• Telemetry is injected. One consumer span per record continues the producer's trace through the record's headers, and the metrics follow OpenTelemetry's messaging conventions. kotel is not used.
• Tests run on kfake over humustest's Host, inside synctest. kfake takes a listen function and was built for synctest; a prototype's whole-App tests pass under -race.

func main() {
	bedrock.Main(kafka.App{
		Host:      humus.OSHost(),
		Telemetry: buildTelemetry,
		Init:      app.Init, // returns kafka.Settings and a queue.Processor[kafka.Message]
	})
}

Background

The packages before the reset

queue had three phases and a runtime (queue/queue.go):

• Consumer[T], Processor[T] and Acknowledger[T], each with a Func adapter (L18-L57);
• ErrEndOfQueue, for a finite queue (L13-L16);
• Runtime and App, on bedrock v0.21.0's Run(context.Context) error shape (L59-L85).

The end user only ever wrote a Processor[kafka.Message]. Consumer and Acknowledger were the seams of queue/kafka's own orchestrators (Consumer[fetch], Acknowledger[[]*kgo.Record], #365), exported from queue because they had to live somewhere (event_loop.go#L28-L30).

queue/kafka built a franz-go client in ProcessQueue, ran a goroutine per assigned partition, and offered AtLeastOnce(topic, p) and AtMostOnce(topic, p) per topic (kafka.go, at_least_once.go, at_most_once.go).

What was wrong with it

• At-least-once committed records that failed. recordProcessor.process logged a processing error and returned nothing (runtime.go#L92-L104), and processFetches sent every record on to be committed whatever happened (at_least_once.go#L94-L107). A failed record was never redelivered, which is at-most-once.
• Commit failures were ignored. Both modes logged them and carried on, under // TODO: how should this be handled? (at_least_once.go#L118-L131, at_most_once.go#L112-L122). At-least-once also committed one record at a time, a round trip per record.
• At-most-once lost the order within a partition. It processed each record of a batch in its own goroutine (at_most_once.go#L148-L161).
• Revocation didn't wait. A revoked partition's channel was closed and the callback returned at once, without waiting for the record in flight or committing it (event_loop.go#L265-L288). The new owner reprocessed it, and a late commit could rewind the new owner's.
• Shutdown didn't drain. Cancelling ProcessQueue's context cancelled the record in flight and every loop at once, and nothing committed the final offsets (kafka.go#L213-L223). The package doc said it did (doc.go#L164-L171).
• Globals and the real network. Spans, metrics and the propagator came from otel.Get*, and logs from humus.Logger (kafka.go#L178-L191, otel.go#L17-L27). The brokers were dialled with franz-go's own dialer.
• The doc described another package. It documented AtMostOnceRuntime, NewAtLeastOnceRuntime, decoder functions, WithKafkaOptions and a messaging.client.processing.failures metric, none of which existed (doc.go#L15, #L122, #L201, #L225). The two metrics that did exist, messaging.client.messages.processed and messaging.client.messages.committed (story(queue/kafka): add otel metrics for message processing and commits #359), aren't in OpenTelemetry's conventions (otel.go#L37-L52).
• Slow tests. The behaviour was tested against Kafka in a container, behind the testcontainers build tag, which took about two minutes (story(queue/kafka): investigate testcontainers peformance #368). story(kafka): add unit tests using kfake #431 asked for kfake instead.

What wave 1 settled

• A humus.Service gets two signals: Serve's context is done as shutdown starts, before the drain delay, and Shutdown is called after it, with the grace period as its deadline. A queue consumer stops on the first (#630).
• Each runtime package provides Services for composition and an App built on humus.App with its own defaults.
• Packages take plain Settings, read no environment variables, and export a check per field (#631). Proposal: configuration for humus apps #631's migration note that Kafka's brokers, group ID and TLS would get HUMUS_KAFKA_* variables predates that revision and is superseded: this package reads no variables.
• Telemetry comes from s.Telemetry, never the globals (#632).
• A whole App is tested inside synctest on bedrocktest's process and humustest's Host (#633).
• No wave-2 App serves health probes (charter, revision 3).

Proposed API

queue

queue is in the core module and depends on nothing but the standard library.

package queue // github.com/z5labs/humus/queue

// Processor processes one message. Under AtLeastOnce it must be idempotent:
// a message whose processing failed, or was cut short by a crash or a
// rebalance, is delivered again.
type Processor[T any] interface {
	Process(ctx context.Context, msg T) error
}

// ProcessorFunc adapts a function to a Processor.
type ProcessorFunc[T any] func(ctx context.Context, msg T) error

func (f ProcessorFunc[T]) Process(ctx context.Context, msg T) error

// Delivery is when a message is acknowledged, relative to processing it.
type Delivery string

const (
	// AtLeastOnce acknowledges a message once Process has returned nil for
	// it. A message is never lost, and may be processed more than once.
	AtLeastOnce Delivery = "at-least-once"

	// AtMostOnce acknowledges a message before Process is called. A
	// message is never processed twice, and is lost if processing fails or
	// the process stops first.
	AtMostOnce Delivery = "at-most-once"
)

// CheckDelivery rejects a Delivery other than AtLeastOnce and AtMostOnce.
func CheckDelivery(d Delivery) error

The package doc is the contract every broker package implements:

• A processing error under AtLeastOnce ends the Service. The message is not acknowledged, the Service stops as it would on a signal, and its Serve returns the error, so the App exits 1. The message is delivered again when the App restarts, to it or to another replica.
• A processing error under AtMostOnce is reported, and processing goes on. The message was already acknowledged, so stopping wouldn't bring it back.
• No panic is recovered. A panicking Processor crashes the process.
• Shutdown drains. On Serve's context a Service stops taking messages, lets the ones in flight finish, acknowledges them, and returns. Process's context is not cancelled when shutdown starts, only when the grace period runs out.
• Retries are the Processor's. A Processor that wants to retry a transient failure does so itself, for example with bedrock's retry module, before it returns an error.

queue/kafka

queue/kafka is a nested module, github.com/z5labs/humus/queue/kafka, so franz-go stays out of the module graph of users who don't consume Kafka.

package kafka // github.com/z5labs/humus/queue/kafka

// Message is a Kafka record as a Processor sees it.
type Message struct {
	Topic     string
	Partition int32
	Offset    int64
	Key       []byte
	Value     []byte
	Headers   []Header // in the record's order; a key may repeat
	Timestamp time.Time
}

type Header struct {
	Key   string
	Value []byte
}

// Settings configures a Service. Every field is a plain value, and a field's
// zero value means zero, never "use the default".
type Settings struct {
	Brokers          []string
	GroupID          string
	Topics           []string
	Delivery         queue.Delivery
	TLS              *tls.Config
	SASL             sasl.Mechanism // github.com/twmb/franz-go/pkg/sasl
	SessionTimeout   time.Duration
	RebalanceTimeout time.Duration
	FetchMaxBytes    int32
	MaxPollRecords   int
}

func DefaultSettings() Settings

func CheckBrokers(brokers []string) error
func CheckGroupID(id string) error
func CheckTopics(topics []string) error
func CheckSessionTimeout(d time.Duration) error
func CheckRebalanceTimeout(d time.Duration) error
func CheckFetchMaxBytes(n int32) error
func CheckMaxPollRecords(n int) error

// Service consumes Settings.Topics as a member of the consumer group
// Settings.GroupID, and hands each record to a Processor.
type Service struct{ /* unexported */ }

// NewService checks settings and returns a Service. It opens no connection:
// Serve builds the client and joins the group.
func NewService(s *humus.Setup, settings Settings, p queue.Processor[Message]) (*Service, error)

func (s *Service) Serve(ctx context.Context) error
func (s *Service) Shutdown(ctx context.Context) error

// App is a bedrock.App that runs one Service on humus.App. Host, Telemetry
// and Init are required.
type App struct {
	Host      humus.Host
	Telemetry telemetry.Builder
	Init      func(ctx context.Context, s *humus.Setup) (Settings, queue.Processor[Message], error)
}

func (a App) Run(ctx context.Context, p bedrock.Process) bedrock.ExitStatus

Settings

Field	Default	Valid	Check
Brokers	none	at least one; each host:port, with a host and a port from 1 to 65535	CheckBrokers
GroupID	""	not empty	CheckGroupID
Topics	none	at least one; none empty; no topic twice	CheckTopics
Delivery	queue.AtLeastOnce	at-least-once, at-most-once	queue.CheckDelivery
TLS	nil, which means plaintext	any
SASL	nil, which means no authentication	any
SessionTimeout	45s	positive	CheckSessionTimeout
RebalanceTimeout	60s	positive	CheckRebalanceTimeout
FetchMaxBytes	50 MiB	positive	CheckFetchMaxBytes
MaxPollRecords	500	positive	CheckMaxPollRecords

The answer to #631's open question 2, optional settings, for these fields:

• Brokers, GroupID and Topics are required. They have no default, and their zero values fail their checks. An end user who sources them with config gets exit 78 naming the variable when one is missing.
• TLS and SASL are optional. nil turns them off, as telemetry.OTLPSettings.TLS does. TLS covers mTLS (story(kafka): add mTLS authentication support #352): the end user puts the client certificate in Certificates.
• SASL is franz-go's own sasl.Mechanism, from franz-go's own module, so the end user picks plain, scram, oauth or aws and supplies the credentials from wherever they keep them. humus decides nothing about secrets.
• Delivery is a setting rather than a constructor argument, so DefaultSettings gives the safe one and the check covers it. It is the end user's design decision, not an operator's, so most Inits set it in code rather than sourcing it.
• SessionTimeout and FetchMaxBytes keep franz-go's defaults. RebalanceTimeout goes from the pre-reset 30s to franz-go's 60s, because a revoked partition's record in flight now has to finish within it. MaxPollRecords is new: it bounds how many records a poll hands out, which bounds how much an at-most-once batch can lose and how long a revoke waits.

What Serve does

1. Builds a franz-go client with kgo.ConsumerGroup, kgo.ConsumeTopics, kgo.DisableAutoCommit, kgo.BlockRebalanceOnPoll and the partition callbacks below. It dials through s.Host.Dial with kgo.Dialer. franz-go refuses kgo.Dialer together with kgo.DialTLSConfig ("cannot set both Dialer and DialTLSConfig"), so with TLS set, an unexported dialer runs the TLS handshake over the Host's connection, setting ServerName from the broker's host when the config leaves it empty, as DialTLSConfig does.
2. Polls with PollRecords(ctx, MaxPollRecords), hands each partition's records to that partition's goroutine, and allows rebalancing again.
3. Each assigned partition has one goroutine, which processes its records one at a time, in offset order. Partitions are processed concurrently, so the concurrency is the number of assigned partitions.
   • Under AtLeastOnce, after each batch, it commits the last record that was processed, and every record before it, with one CommitRecords.
   • Under AtMostOnce, it commits the batch before it processes any of it, and then processes every record of the batch.
4. On revoke, franz-go calls the callback before the partition moves. The callback stops each revoked partition, waits for its record in flight, commits, and returns. All of this fits in RebalanceTimeout.
5. On loss, after a fatal group error such as being fenced out, the callback stops the partitions without committing, because a commit would fail. Their records are processed again by their new owner.
6. On shutdown, when Serve's context is done or Shutdown is called, it stops polling, stops every partition after its record in flight, commits, leaves the group with LeaveGroupContext, closes the client, and returns nil. Leaving the group rebalances at once, so the other replicas take the partitions without waiting for SessionTimeout.

Process is given a context that keeps the values of Serve's context, carries the record's span, and is cancelled only when Shutdown's context is done. So the record in flight isn't cut short when shutdown starts, and a slow one is cut short when the grace period runs out. Then Shutdown stops waiting: it cancels that context and the leave, and returns an error wrapping its context's error, which the lifecycle turns into a *humus.GracePeriodError. Records whose processing was cut short are not committed.

Failures

	AtLeastOnce	AtMostOnce
Process returns an error	not committed; the Service stops as on shutdown, and Serve returns the error; exit 1	already committed; recorded on the span and the metric and logged at ERROR; processing goes on
Process panics	the process crashes; nothing after the last commit is committed	the process crashes; the batch was committed
a commit fails because the member was fenced out of the group	not an error: the partition is lost, and its new owner processes the records again	the same
any other commit fails	the Service stops, and Serve returns the error; exit 1	the same
a fetch returns an error franz-go doesn't retry, such as an authorization failure	the Service stops, and Serve returns the error; exit 1	the same
the brokers are unreachable	franz-go keeps retrying and logs each failure; the Service keeps serving	the same

An at-least-once failure stops the whole Service, not just the partition. The other partitions finish their record in flight and commit, so only the failed record and those after it are processed again.

Telemetry

Everything comes from s.Telemetry. The instrumentation scope is github.com/z5labs/humus/queue/kafka.

• A span per record. It is named process <topic>, of kind CONSUMER, and its parent is the context s.Telemetry.Propagator extracts from the record's headers, so a producer's trace continues through the consumer. Its attributes are messaging.system=kafka, messaging.operation.type=process, messaging.operation.name=process, messaging.destination.name, messaging.destination.partition.id, messaging.consumer.group.name and messaging.kafka.offset. A processing error is recorded on it, with error.type and an error status. The record's key is not recorded, because it often holds a customer's identifier.

• Metrics, from semconv v1.43.0's messagingconv:

  • messaging.client.consumed.messages, a counter of the records handed to the Processor;
  • messaging.process.duration, a histogram of Process, with error.type when it fails;
  • messaging.client.operation.duration, a histogram of each commit, with messaging.operation.name=commit and error.type when it fails.

  Each has messaging.system, messaging.consumer.group.name, messaging.destination.name and messaging.destination.partition.id. These replace story(queue/kafka): add otel metrics for message processing and commits #359's two counters, which weren't conventions. The messaging conventions are still in development, so humus pins the semconv version it uses, and changing it is a breaking change.

• Logs. The Service logs, at INFO, the partitions it is assigned, has revoked and loses, and at ERROR, a processing error under AtMostOnce. An error that ends the Service is returned, and the lifecycle logs it. franz-go's own logs go through kgo.WithLogger and an unexported adapter onto the OpenTelemetry Logs API, with the scope github.com/twmb/franz-go/pkg/kgo. Its Level is the lowest of franz-go's levels the LoggerProvider is Enabled for, so franz-go builds no record that would be dropped.

The App

func (a App) Run(ctx context.Context, p bedrock.Process) bedrock.ExitStatus {
	return humus.App{
		Host:                a.Host,
		Telemetry:           a.Telemetry,
		ShutdownGracePeriod: 20 * time.Second,
		DrainDelay:          0,
		Init: func(ctx context.Context, s *humus.Setup) ([]humus.Service, error) {
			settings, p, err := a.Init(ctx, s)
			if err != nil {
				return nil, err
			}
			svc, err := NewService(s, settings, p)
			if err != nil {
				return nil, err
			}
			return []humus.Service{svc}, nil
		},
	}.Run(ctx, p)
}

• Drain delay: none. The drain delay gives a load balancer time to stop sending traffic, and nothing load-balances a consumer. Its traffic stops when it leaves the group, which it does as soon as it has drained.
• Grace period: 20s. With telemetry's default ShutdownTimeout of 5s, the whole shutdown takes at most 25s, inside Kubernetes' default terminationGracePeriodSeconds of 30s with a margin for the process to exit, as for a job. A Processor that can't finish a record within it is cut short, and the record is processed again.
• An operator changes either with HUMUS_SHUTDOWN_GRACE_PERIOD and HUMUS_SHUTDOWN_DRAIN_DELAY, which humus.App.Run reads. App has no fields for them: a zero field would mean zero, and an end user who wants other defaults in code uses humus.App.
• Telemetry is required, as on humus.App. A default Builder would name the service unknown_service, and a consumer in the middle of a traced pipeline needs to choose its exporter. Tests pass telemetry.NoopBuilder or humustest.Telemetry's Builder.
• One Service, no health. The App runs exactly the Service its Init describes. Proposal: exposing health probes, in wave 3, decides how a pod with no HTTP server of its own serves probes, and whether this App does by default. It adds that with a new field on App, whose Run appends one more Service; Init's shape doesn't change.
• Anything more is humus.App. Two topics with different Delivery, a consumer next to an HTTP API, or a health Service today: the end user returns kafka.NewService's Service from their own humus.App Init, next to the others. That is the advanced path the lifecycle documents.

In use

func Init(ctx context.Context, s *humus.Setup) (kafka.Settings, queue.Processor[kafka.Message], error) {
	d := kafka.DefaultSettings()
	env := func(name string) config.Value[string] { return config.Env(s.Process.Environment, name) }
	list := func(v string) []string { return strings.Split(v, ",") }

	settings, err := config.Struct(func(r *config.Resolver) kafka.Settings {
		k := d
		k.Brokers = r.Get(env("KAFKA_BROKERS").Map(list).Validate(kafka.CheckBrokers))
		k.GroupID = r.Get(env("KAFKA_GROUP_ID").Validate(kafka.CheckGroupID))
		k.Topics = r.Get(env("KAFKA_TOPICS").Map(list).Validate(kafka.CheckTopics))
		k.SessionTimeout = r.Get(env("KAFKA_SESSION_TIMEOUT").Parse(time.ParseDuration).Default(d.SessionTimeout).Validate(kafka.CheckSessionTimeout))
		return k
	}).Read(ctx)
	if err != nil {
		return kafka.Settings{}, nil, err // exit 78, naming every bad variable
	}

	pool, err := newPool(ctx, s) // dials through s.Host.Dial; closed in s.OnStop
	if err != nil {
		return kafka.Settings{}, nil, err
	}
	orders := orders.NewHandler(pool) // idempotent: an order already stored is skipped
	return settings, queue.ProcessorFunc[kafka.Message](orders.Handle), nil
}

Design decisions

• queue is the contract, not a framework. It holds what the end user's code depends on, Processor, and what it must know, Delivery and what each does with an error. A broker package implements both against its own client.
• Consumer and Acknowledger are removed. They were queue/kafka's internal seams, and no end user implemented them. A shared acknowledger would also hide the difference that matters most between brokers: Kafka acknowledges a partition's offset, so a commit covers every earlier record, while RabbitMQ acknowledges one delivery.
• How RabbitMQ (story(queue): implement a runtime for rabbitmq #310) would fit. A later queue/rabbitmq nested module provides a Service that takes Settings with a Delivery and a queue.Processor[rabbitmq.Message]. Under AtLeastOnce it acks a delivery after Process returns nil, and on an error stops, so the broker requeues the unacked delivery when the channel closes. Under AtMostOnce it consumes with auto-ack. It follows queue's contract and shares no code with queue/kafka.
• ErrEndOfQueue is removed. A topic doesn't end. A finite queue is a job, and the lifecycle already ends an App whose Service returns nil.
• One Processor and one Delivery per Service. The pre-reset per-topic choice let one consumer group mix semantics, and its revoke and commit paths differed per topic. A Message carries its topic, so one Processor can dispatch on it, and two semantics are two Services with two group IDs.
• A processing error under at-least-once stops the Service. Retrying in place would block the partition with no limit, skipping would lose the record, and pausing the partition would hide the failure while the lag grows. Stopping is loud: the App exits 1, the pod restarts, and a record that always fails shows up as a crash loop with its error in the logs. Kafka Streams makes the same default choice for records it can't deserialize. An end user who would rather skip a record returns nil, after storing it wherever they keep dead letters.
• Retries belong to the Processor. A transient failure is the Processor's to retry, and bedrock's retry module (v0.24.0) does that with a budget and backoff. Neither queue nor queue/kafka imports it, because a retry policy is the end user's decision and a retry inside the Service would block the partition just the same.
• No panic is recovered. This is the lifecycle's rule, and it has no per-message exception here. Under at-least-once, a crash loses nothing that a recovered panic would have saved: every record after the last commit is processed again. Under at-most-once, a recovered panic would continue in a process whose state the panic may have broken.
• Ordered within a partition, concurrent across partitions. Kafka orders a partition, and most consumers depend on that, for example to apply a customer's events in order. One goroutine per partition keeps the order. The pre-reset at-most-once broke it.
• BlockRebalanceOnPoll, with rebalancing allowed right after each poll. Records are handed out only to partitions this member owns, and a revoke can't run while records are being dispatched. Processing happens after rebalancing is allowed again, so a slow record never holds up a rebalance beyond the revoke callback, which waits only for the record in flight.
• Commit per batch, not per record. franz-go warns against committing every record, and a commit covers every earlier offset of its partition anyway. A revoke and a shutdown commit too, so only the records in flight at a crash are processed again.
• Shutdown stops on Serve's context. A consumer has no load balancer in front of it, so there is nothing to wait for during the drain delay: as the lifecycle decided, it stops taking work at once. Shutdown waits for that drain, bounded by the grace period.
• Process's context isn't cancelled when shutdown starts. If it were, a Processor honouring it would fail the record in flight, and under at-least-once every SIGTERM would end in exit 1. Its context is cancelled when the grace period runs out.
• The client is built in Serve, not NewService. franz-go's client starts a metadata goroutine as soon as it is built, and joining the group before Serve would claim partitions before anything processes them. So NewService acquires nothing and registers no OnStop function, and an Init that fails after it leaks nothing.
• The brokers are dialled through the Host. That puts the Service on humustest's network in tests, as telemetry does with the OTLP exporters.
• No kotel. franz-go's OpenTelemetry plugin uses semconv v1.18.0, whose names have since changed, and names its process span <topic> process. It starts a receive span for every buffered record, doubling the spans, and it falls back to the global providers for any option it isn't given. The process span, the propagation through headers, and three instruments are about a hundred lines that humus controls. The cost is franz-go's broker-level metrics, such as bytes read and connection errors; open question 4 asks whether to add kotel's Meter for those alone.
• No conc, kslog or kadm. The pre-reset package used sourcegraph/conc for goroutine pools, which sync covers, and kslog to bridge to slog, which humus no longer has. kadm stays only in the integration tests.
• Dependencies. github.com/twmb/franz-go is the client, as before: pure Go, maintained, with the group, commit and dialler hooks this design needs, and its pkg/sasl mechanisms live in the same module. github.com/twmb/franz-go/pkg/kfake is a test dependency. Both are in the nested module.
• Settings takes decoded values. TLS is a *tls.Config and SASL a mechanism, so how certificates and credentials are stored is the end user's, as Proposal: configuration for humus apps #631 decided.

Testing

kfake on humustest's Host

#633 left this proposal one question: can kfake listen on humustest's Host, and if not, what do the unit tests run against? It can.

• kfake.ListenFn(fn func(network, address string) (net.Listener, error)) overrides net.Listen. Passing h.Listen puts the fake cluster on humustest's network.
• kfake was built for synctest. It also exports kfake.VirtualNetwork, whose doc says it replaces TCP with net.Pipe connections because "a goroutine blocked on real TCP I/O is not durably blocked". humustest's network is the same idea, so humustest's Host is used, and the App reaches the cluster through s.Host.Dial with no test-only seam.
• Checked with franz-go v1.22.2 and kfake v0.0.0-20261013022759-7b6cac30a57a: a cluster built with kfake.ListenFn(h.Listen) inside synctest.Test listens on 127.0.0.1:<port> of humustest's network, and a client dialling with kgo.Dialer(h.Dial) produces, joins a group, consumes, commits and leaves, under -race, in no fake time. cluster.GroupInfo(group) gives the committed offsets and members to assert on.
• kfake.TLS(cfg) serves TLS, so mTLS is tested with certificates from humustest.NewCA, never fixed ones.

A test builds the cluster inside the bubble, with the bubble's t, and closes it before the bubble ends, as the rules for humustest's Host require.

Whole-App tests

These tests pass against the prototype below, under go test -race -count=3:

func TestApp_at_least_once_stops_on_a_failure_and_redelivers(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		h := humustest.NewHost(t)
		c := cluster(t, h.Listen, 1) // kfake.NewCluster(kfake.ListenFn(h.Listen), kfake.SeedTopics(1, "orders"))
		defer c.Close()
		produce(t, h, c, 5) // offsets 0 to 4

		bad := errors.New("cannot process")
		run := func(fail bool) (status, []int64) {
			var seen []int64
			app := kafka.App{
				Host:      h.Host(),
				Telemetry: telemetry.NoopBuilder,
				Init: func(ctx context.Context, s *humus.Setup) (kafka.Settings, queue.Processor[kafka.Message], error) {
					return settings(c), queue.ProcessorFunc[kafka.Message](func(ctx context.Context, m kafka.Message) error {
						seen = append(seen, m.Offset)
						if fail && m.Offset == 3 {
							return bad
						}
						return nil
					}), nil
				},
			}
			p, st := bedrocktest.Start(t, app)
			if !fail {
				synctest.Wait() // every record is processed and committed
				p.Signal(syscall.SIGTERM)
			}
			s := <-st
			return status{s.Code(), s.Cause()}, seen
		}

		st, seen := run(true)
		require.Equal(t, humus.ExitFailure, st.code)
		require.ErrorIs(t, st.cause, bad)
		require.Equal(t, []int64{0, 1, 2, 3}, seen)
		require.Equal(t, int64(3), committed(c), "0 to 2 are committed, 3 is not")

		st, seen = run(false)
		require.Zero(t, st.code, st.cause)
		require.Equal(t, []int64{3, 4}, seen, "3 is delivered again")
		require.Equal(t, int64(5), committed(c))
	})
}

func TestApp_SIGTERM_lets_the_record_in_flight_finish_and_commits_it(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		h := humustest.NewHost(t)
		c := cluster(t, h.Listen, 1)
		defer c.Close()
		produce(t, h, c, 5)

		entered := make(chan struct{}, 5)
		app := kafka.App{
			Host:      h.Host(),
			Telemetry: telemetry.NoopBuilder,
			Init: func(ctx context.Context, s *humus.Setup) (kafka.Settings, queue.Processor[kafka.Message], error) {
				return settings(c), queue.ProcessorFunc[kafka.Message](func(ctx context.Context, m kafka.Message) error {
					entered <- struct{}{}
					time.Sleep(3 * time.Second)
					return nil
				}), nil
			},
		}
		p, status := bedrocktest.Start(t, app)
		<-entered // offset 0 is in flight
		p.Signal(syscall.SIGTERM)

		st := <-status
		require.Zero(t, st.Code(), st.Cause())
		require.Equal(t, int64(1), committed(c), "0 finished and is committed; 1 to 4 never started")
	})
}

The helpers are a few lines each: produce writes records with a franz-go client that dials h.Dial, settings is DefaultSettings() with the cluster's ListenAddrs(), a group ID and the topic, and committed sums the group's committed offsets from c.GroupInfo.

The prototype's other tests show the following:

• Ten records over two partitions are processed and committed, SIGTERM exits 0, the group has no members afterwards, and humustest.Telemetry recorded ten process orders spans.
• With Settings.TLS set and a kfake cluster that requires a client certificate, both issued by humustest.NewCA, the App processes and commits as over plaintext.
• kafka.Service passes humustest.TestService, including "Shutdown is safe after Serve has returned on its own" and the leak check. newService builds a kfake cluster on s.Host.Listen and closes it with s.OnStop.
• NewService reports every bad field in one error, naming each.

The stories add tests for what the prototype leaves out:

• Each check accepts an in-range value and rejects an out-of-range one without naming the field, and DefaultSettings with Brokers, GroupID and Topics set passes NewService (Proposal: configuration for humus apps #631's list).
• Under AtMostOnce, a processing error is logged and the next record is processed, and the batch is committed before its first record is processed.
• A partition revoked by a second member joining finishes its record in flight and commits before the callback returns, and the new owner starts after it (two Apps on one Host).
• A grace period that runs out during a slow record exits 1 with a *humus.GracePeriodError, and the record is not committed.
• A producer's span context in the record's headers is the parent of the process span, and the three instruments are recorded with their attributes.
• franz-go's logs reach humustest.Telemetry's Logs(), and its Level follows the LoggerProvider.
• A runnable Example of kafka.App, tested, as the charter requires.

Integration tests

The tests against Kafka in a container stay, behind the testcontainers build tag (kafka_testcontainer.go), for what a fake can't show: a real broker's cooperative-sticky rebalances and its fencing. They are the only tests in the package that open a real socket, run in a CI job of their own, and don't count toward the coverage gate. Their main use is to catch where kfake and Kafka differ.

Migration from today

Before the reset	After
queue.Processor[T], queue.ProcessorFunc[T]	unchanged
queue.Consumer, queue.Acknowledger, their Func adapters, queue.Runtime, queue.RuntimeFunc	removed
queue.ErrEndOfQueue	removed; a finite queue is a job
queue.NewApp(runtime)	kafka.App, or humus.App with kafka.NewService
kafka.NewRuntime(brokers, groupID, opts...)	kafka.Settings{Brokers, GroupID, Topics} and kafka.NewService(s, settings, p)
kafka.AtLeastOnce(topic, p), kafka.AtMostOnce(topic, p)	Settings.Topics and Settings.Delivery; one Processor per Service
kafka.SessionTimeout, RebalanceTimeout, FetchMaxBytes	Settings fields with checks; RebalanceTimeout defaults to 60s
kafka.MaxConcurrentFetches	removed: its zero meant unlimited, and franz-go's default is unlimited
kafka.WithTLS(cfg)	Settings.TLS, dialled through Host.Dial
kafka.Message.Attrs	removed
kafka.GroupIDAttr, TopicAttr, PartitionAttr, OffsetAttr	removed; humus has no slog in its API, and the span and metric attributes follow the conventions
messaging.client.messages.processed, messaging.client.messages.committed	messaging.client.consumed.messages, messaging.process.duration, messaging.client.operation.duration
otel.Get*, humus.Logger, kotel, kslog	s.Telemetry
HUMUS_KAFKA_* variables, planned in #631's first post	none; the end user names and sources every setting
unit tests on testcontainers (#368)	unit tests on kfake over humustest's Host (#431); containers only behind the build tag

Two behaviours change for an existing processor:

• Under at-least-once, a processing error now stops the App and the record is processed again, where before it was committed and lost.
• Under at-most-once, a partition's records are processed in order, one at a time, where before a batch was processed concurrently.

Open questions

1. A queue package, or only queue/kafka? queue holds two small types until a second broker arrives. Folding them into queue/kafka is simpler today, but RabbitMQ would then define its own Processor and repeat the contract. Recommendation: keep queue, as the charter's inventory has it.
2. At-most-once processing errors. Proposed: logged and counted, and processing goes on, because the record was committed and stopping can't bring it back. The alternative stops the Service as at-least-once does, which is louder but restarts the pod for nothing.
3. Which other franz-go options become Settings fields? The proposal keeps the pre-reset set plus SASL and MaxPollRecords. Candidates are ConsumeResetOffset (where a new group starts; franz-go and this proposal start at the earliest offset, Kafka's Java client at the latest), InstanceID for static membership, FetchMaxWait and ClientID. A field can be added later without breaking anyone.
4. franz-go's broker-level metrics. Add kotel's Meter alone, given s.Telemetry.MeterProvider, for its connection, byte and request counters, or leave them out? They aren't conventions, and they add kotel as a dependency.
5. Brokers unreachable at startup. Proposed: franz-go keeps retrying, logging each failure, and the Service keeps serving. The alternative fails Serve if the group isn't joined within a bound. Readiness is the usual answer, and it belongs to Proposal: exposing health probes in wave 3.

Out of scope

• Producing, and an outbound Kafka client: the charter's "Later" list.
• RabbitMQ (story(queue): implement a runtime for rabbitmq #310): "Later". The design decisions above only show that queue fits it.
• Dead-letter topics, retry topics and skip policies: the Processor's.
• Exactly-once processing and transactions.
• Schema registries and decoding: a Processor decodes Message.Value itself.
• Health probes, and a monitor of group membership: Proposal: exposing health probes, in wave 3.

Prototype

The prototype runs against humus at c197f923, bedrock v0.23.1, franz-go v1.22.2 and kfake v0.0.0-20261013022759-7b6cac30a57a. go test -race -count=3 passes, covering 84.8% of queue/kafka. It omits the metrics, franz-go's log adapter, Settings' godoc, and the special case for commits by a fenced-out member. Its span attributes are plain strings rather than semconv. The stories implement this proposal rather than copying the prototype. Its tests are the ones under Testing.

queue/queue.go

// Package queue is a prototype of the queue package proposed in
// "Proposal: queue processing and the Kafka runtime".
package queue

import (
	"context"
	"errors"
)

// Processor processes one message.
type Processor[T any] interface {
	Process(ctx context.Context, msg T) error
}

// ProcessorFunc adapts a function to a Processor.
type ProcessorFunc[T any] func(ctx context.Context, msg T) error

// Process implements Processor by calling f(ctx, msg).
func (f ProcessorFunc[T]) Process(ctx context.Context, msg T) error { return f(ctx, msg) }

// Delivery is when a message is acknowledged, relative to processing it.
type Delivery string

const (
	AtLeastOnce Delivery = "at-least-once"
	AtMostOnce  Delivery = "at-most-once"
)

// CheckDelivery rejects a Delivery other than AtLeastOnce and AtMostOnce.
func CheckDelivery(d Delivery) error {
	switch d {
	case AtLeastOnce, AtMostOnce:
		return nil
	}
	return errors.New("must be at-least-once or at-most-once")
}

kafka/settings.go

package kafka

import (
	"crypto/tls"
	"errors"
	"fmt"
	"net"
	"strconv"
	"time"

	"github.com/twmb/franz-go/pkg/sasl"

	"github.com/z5labs/humus/queue"
)

// Settings configures a Service. Every field is a plain value, and a
// field's zero value means zero, never "use the default".
type Settings struct {
	Brokers          []string
	GroupID          string
	Topics           []string
	Delivery         queue.Delivery
	TLS              *tls.Config
	SASL             sasl.Mechanism
	SessionTimeout   time.Duration
	RebalanceTimeout time.Duration
	FetchMaxBytes    int32
	MaxPollRecords   int
}

// DefaultSettings returns the defaults of a Service.
func DefaultSettings() Settings {
	return Settings{
		Delivery:         queue.AtLeastOnce,
		SessionTimeout:   45 * time.Second,
		RebalanceTimeout: 60 * time.Second,
		FetchMaxBytes:    50 << 20,
		MaxPollRecords:   500,
	}
}

func CheckBrokers(bs []string) error {
	if len(bs) == 0 {
		return errors.New("must name at least one broker")
	}
	var errs []error
	for _, b := range bs {
		host, port, err := net.SplitHostPort(b)
		if err != nil {
			errs = append(errs, fmt.Errorf("%q: %w", b, err))
			continue
		}
		n, err := strconv.Atoi(port)
		if host == "" || err != nil || n < 1 || n > 65535 {
			errs = append(errs, fmt.Errorf("%q: must be host:port", b))
		}
	}
	return errors.Join(errs...)
}

func CheckGroupID(id string) error {
	if id == "" {
		return errors.New("must not be empty")
	}
	return nil
}

func CheckTopics(ts []string) error {
	if len(ts) == 0 {
		return errors.New("must name at least one topic")
	}
	seen := map[string]bool{}
	for _, t := range ts {
		if t == "" {
			return errors.New("must not contain an empty topic")
		}
		if seen[t] {
			return fmt.Errorf("names %q twice", t)
		}
		seen[t] = true
	}
	return nil
}

func checkPositive[T int | int32 | time.Duration](v T) error {
	if v <= 0 {
		return errors.New("must be positive")
	}
	return nil
}

func CheckSessionTimeout(d time.Duration) error   { return checkPositive(d) }
func CheckRebalanceTimeout(d time.Duration) error { return checkPositive(d) }
func CheckFetchMaxBytes(n int32) error            { return checkPositive(n) }
func CheckMaxPollRecords(n int) error             { return checkPositive(n) }

func (s Settings) check() error {
	var errs []error
	add := func(field string, err error) {
		if err != nil {
			errs = append(errs, fmt.Errorf("%s: %w", field, err))
		}
	}
	add("Brokers", CheckBrokers(s.Brokers))
	add("GroupID", CheckGroupID(s.GroupID))
	add("Topics", CheckTopics(s.Topics))
	add("Delivery", queue.CheckDelivery(s.Delivery))
	add("SessionTimeout", CheckSessionTimeout(s.SessionTimeout))
	add("RebalanceTimeout", CheckRebalanceTimeout(s.RebalanceTimeout))
	add("FetchMaxBytes", CheckFetchMaxBytes(s.FetchMaxBytes))
	add("MaxPollRecords", CheckMaxPollRecords(s.MaxPollRecords))
	return errors.Join(errs...)
}

kafka/service.go

// Package kafka is a prototype of the queue/kafka package proposed in
// "Proposal: queue processing and the Kafka runtime".
package kafka

import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"net"
	"sync"
	"sync/atomic"
	"time"

	"github.com/twmb/franz-go/pkg/kgo"
	"github.com/z5labs/humus"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/propagation"
	"go.opentelemetry.io/otel/trace"

	"github.com/z5labs/humus/queue"
)

const scope = "github.com/z5labs/humus/queue/kafka"

// Header is a record header.
type Header struct {
	Key   string
	Value []byte
}

// Message is a Kafka record, as a Processor sees it.
type Message struct {
	Topic     string
	Partition int32
	Offset    int64
	Key       []byte
	Value     []byte
	Headers   []Header
	Timestamp time.Time
}

// Service consumes Settings.Topics in the consumer group Settings.GroupID
// and hands each record to a Processor.
type Service struct {
	settings Settings
	proc     queue.Processor[Message]
	dial     func(ctx context.Context, network, address string) (net.Conn, error)
	tracer   trace.Tracer
	prop     propagation.TextMapPropagator

	serving  atomic.Pointer[run]
	stop     chan struct{}
	stopOnce sync.Once
}

type run struct {
	served chan struct{}
	abort  context.CancelFunc
}

var notServed = &run{}

// NewService checks settings and returns a Service. It opens nothing.
func NewService(s *humus.Setup, settings Settings, p queue.Processor[Message]) (*Service, error) {
	if err := settings.check(); err != nil {
		return nil, err
	}
	if p == nil {
		return nil, errors.New("kafka: nil Processor")
	}
	return &Service{
		settings: settings,
		proc:     p,
		dial:     tlsDialer(s.Host.Dial, settings.TLS),
		tracer:   s.Telemetry.TracerProvider.Tracer(scope),
		prop:     s.Telemetry.Propagator,
		stop:     make(chan struct{}),
	}, nil
}

func tlsDialer(dial func(context.Context, string, string) (net.Conn, error), cfg *tls.Config) func(context.Context, string, string) (net.Conn, error) {
	if cfg == nil {
		return dial
	}
	return func(ctx context.Context, network, addr string) (net.Conn, error) {
		conn, err := dial(ctx, network, addr)
		if err != nil {
			return nil, err
		}
		c := cfg.Clone()
		if c.ServerName == "" {
			c.ServerName, _, _ = net.SplitHostPort(addr)
		}
		tc := tls.Client(conn, c)
		if err := tc.HandshakeContext(ctx); err != nil {
			_ = conn.Close()
			return nil, err
		}
		return tc, nil
	}
}

// Serve joins the group and processes records until ctx is done, Shutdown
// is called, or processing fails. It then finishes the record in flight on
// each partition, commits, and leaves the group.
func (s *Service) Serve(ctx context.Context) error {
	// work is the context of processing, committing and leaving: it keeps
	// ctx's values but not its cancellation, and only Shutdown's deadline
	// cancels it.
	work, abort := context.WithCancel(context.WithoutCancel(ctx))
	defer abort()
	r := &run{served: make(chan struct{}), abort: abort}
	if !s.serving.CompareAndSwap(nil, r) {
		return nil
	}
	defer close(r.served)

	polling, stopPolling := context.WithCancel(ctx)
	defer stopPolling()
	go func() {
		select {
		case <-s.stop:
			stopPolling()
		case <-polling.Done():
		}
	}()

	g := &group{svc: s, work: work, fail: stopPolling, parts: map[tp]*partition{}}
	opts := []kgo.Opt{
		kgo.SeedBrokers(s.settings.Brokers...),
		kgo.Dialer(s.dial),
		kgo.ConsumerGroup(s.settings.GroupID),
		kgo.ConsumeTopics(s.settings.Topics...),
		kgo.DisableAutoCommit(),
		kgo.BlockRebalanceOnPoll(),
		kgo.SessionTimeout(s.settings.SessionTimeout),
		kgo.RebalanceTimeout(s.settings.RebalanceTimeout),
		kgo.FetchMaxBytes(s.settings.FetchMaxBytes),
		kgo.OnPartitionsAssigned(g.assigned),
		kgo.OnPartitionsRevoked(g.revoked),
		kgo.OnPartitionsLost(g.lost),
	}
	if s.settings.SASL != nil {
		opts = append(opts, kgo.SASL(s.settings.SASL))
	}
	cl, err := kgo.NewClient(opts...)
	if err != nil {
		return fmt.Errorf("kafka: %w", err)
	}
	g.cl = cl

	for polling.Err() == nil {
		fs := cl.PollRecords(polling, s.settings.MaxPollRecords)
		if polling.Err() != nil {
			break
		}
		fs.EachError(func(topic string, p int32, err error) {
			g.setErr(fmt.Errorf("kafka: fetching %s/%d: %w", topic, p, err))
		})
		fs.EachPartition(func(p kgo.FetchTopicPartition) {
			g.dispatch(polling, tp{p.Topic, p.Partition}, p.Records)
		})
		cl.AllowRebalance()
	}

	g.stopAll(false)
	cl.AllowRebalance()
	leaveErr := cl.LeaveGroupContext(work)
	cl.Close()
	if err := g.err(); err != nil {
		return err
	}
	if leaveErr != nil && work.Err() != nil {
		return fmt.Errorf("kafka: leaving the group: %w", leaveErr)
	}
	return nil
}

// Shutdown stops Serve and waits for it to return. If ctx is done first,
// it cancels the context of the records in flight and of leaving the group.
func (s *Service) Shutdown(ctx context.Context) error {
	s.serving.CompareAndSwap(nil, notServed)
	r := s.serving.Load()
	s.stopOnce.Do(func() { close(s.stop) })
	if r == notServed {
		return nil
	}
	select {
	case <-r.served:
		return nil
	case <-ctx.Done():
		r.abort()
		<-r.served
		return fmt.Errorf("kafka: shutting down the Service: %w", ctx.Err())
	}
}

type tp struct {
	topic     string
	partition int32
}

type partition struct {
	records chan []*kgo.Record
	quit    chan struct{}
	done    chan struct{}
	lost    atomic.Bool
}

type group struct {
	svc  *Service
	cl   *kgo.Client
	work context.Context
	fail context.CancelFunc

	mu    sync.Mutex
	parts map[tp]*partition
	first error
}

func (g *group) setErr(err error) {
	g.mu.Lock()
	if g.first == nil {
		g.first = err
	}
	g.mu.Unlock()
	g.fail()
}

func (g *group) err() error {
	g.mu.Lock()
	defer g.mu.Unlock()
	return g.first
}

func (g *group) assigned(_ context.Context, _ *kgo.Client, assigned map[string][]int32) {
	g.mu.Lock()
	defer g.mu.Unlock()
	for topic, ps := range assigned {
		for _, p := range ps {
			part := &partition{records: make(chan []*kgo.Record, 1), quit: make(chan struct{}), done: make(chan struct{})}
			g.parts[tp{topic, p}] = part
			go g.run(part)
		}
	}
}

func (g *group) revoked(_ context.Context, _ *kgo.Client, revoked map[string][]int32) {
	g.stop(revoked, false)
}

func (g *group) lost(_ context.Context, _ *kgo.Client, lost map[string][]int32) {
	g.stop(lost, true)
}

func (g *group) stop(m map[string][]int32, lost bool) {
	g.mu.Lock()
	var stopping []*partition
	for topic, ps := range m {
		for _, p := range ps {
			if part, ok := g.parts[tp{topic, p}]; ok {
				part.lost.Store(lost)
				close(part.quit)
				stopping = append(stopping, part)
				delete(g.parts, tp{topic, p})
			}
		}
	}
	g.mu.Unlock()
	for _, part := range stopping {
		<-part.done
	}
}

func (g *group) stopAll(lost bool) {
	g.mu.Lock()
	m := map[string][]int32{}
	for k := range g.parts {
		m[k.topic] = append(m[k.topic], k.partition)
	}
	g.mu.Unlock()
	g.stop(m, lost)
}

func (g *group) dispatch(ctx context.Context, k tp, recs []*kgo.Record) {
	g.mu.Lock()
	part, ok := g.parts[k]
	g.mu.Unlock()
	if !ok {
		return
	}
	select {
	case part.records <- recs:
	case <-part.done:
	case <-ctx.Done():
	}
}

// run processes one partition's records in order.
func (g *group) run(part *partition) {
	defer close(part.done)
	for {
		select {
		case <-part.quit:
			return
		case recs := <-part.records:
			if !g.batch(part, recs) {
				return
			}
		}
	}
}

// batch processes one fetched batch and reports whether to keep going.
func (g *group) batch(part *partition, recs []*kgo.Record) bool {
	if g.svc.settings.Delivery == queue.AtMostOnce {
		if err := g.cl.CommitRecords(g.work, recs[len(recs)-1]); err != nil {
			g.setErr(fmt.Errorf("kafka: committing: %w", err))
			return false
		}
		// Every record of the batch is committed, so each is processed even
		// after quit, until Shutdown's deadline cancels work.
		for _, r := range recs {
			if g.work.Err() != nil {
				return false
			}
			_ = g.process(r) // reported, not returned: the record is acknowledged
		}
		return true
	}

	var last *kgo.Record
	ok := true
	for _, r := range recs {
		select {
		case <-part.quit:
			ok = false
		default:
		}
		if !ok {
			break
		}
		if err := g.process(r); err != nil {
			g.setErr(fmt.Errorf("kafka: processing %s/%d@%d: %w", r.Topic, r.Partition, r.Offset, err))
			ok = false
			break
		}
		last = r
	}
	if last != nil && !part.lost.Load() {
		if err := g.cl.CommitRecords(g.work, last); err != nil {
			g.setErr(fmt.Errorf("kafka: committing: %w", err))
			return false
		}
	}
	return ok
}

func (g *group) process(r *kgo.Record) error {
	m := Message{Topic: r.Topic, Partition: r.Partition, Offset: r.Offset, Key: r.Key, Value: r.Value, Timestamp: r.Timestamp}
	for _, h := range r.Headers {
		m.Headers = append(m.Headers, Header{Key: h.Key, Value: h.Value})
	}
	ctx := g.svc.prop.Extract(g.work, headerCarrier{r})
	ctx, span := g.svc.tracer.Start(ctx, "process "+r.Topic,
		trace.WithSpanKind(trace.SpanKindConsumer),
		trace.WithAttributes(
			attribute.String("messaging.system", "kafka"),
			attribute.String("messaging.operation.type", "process"),
			attribute.String("messaging.destination.name", r.Topic),
			attribute.Int64("messaging.kafka.offset", r.Offset),
		))
	defer span.End()
	err := g.svc.proc.Process(ctx, m)
	if err != nil {
		span.RecordError(err)
		span.SetStatus(codes.Error, err.Error())
	}
	return err
}

type headerCarrier struct{ r *kgo.Record }

func (c headerCarrier) Get(key string) string {
	for _, h := range c.r.Headers {
		if h.Key == key {
			return string(h.Value)
		}
	}
	return ""
}

func (c headerCarrier) Set(key, val string) {
	c.r.Headers = append(c.r.Headers, kgo.RecordHeader{Key: key, Value: []byte(val)})
}

func (c headerCarrier) Keys() []string {
	ks := make([]string, len(c.r.Headers))
	for i, h := range c.r.Headers {
		ks[i] = h.Key
	}
	return ks
}

kafka/app.go

package kafka

import (
	"context"
	"time"

	"github.com/z5labs/bedrock"
	"github.com/z5labs/humus"
	"github.com/z5labs/humus/telemetry"

	"github.com/z5labs/humus/queue"
)

// App is a bedrock.App that runs one Service on humus.App.
type App struct {
	Host      humus.Host
	Telemetry telemetry.Builder
	Init      func(ctx context.Context, s *humus.Setup) (Settings, queue.Processor[Message], error)
}

func (a App) Run(ctx context.Context, p bedrock.Process) bedrock.ExitStatus {
	return humus.App{
		Host:                a.Host,
		Telemetry:           a.Telemetry,
		ShutdownGracePeriod: 20 * time.Second,
		Init: func(ctx context.Context, s *humus.Setup) ([]humus.Service, error) {
			settings, proc, err := a.Init(ctx, s)
			if err != nil {
				return nil, err
			}
			svc, err := NewService(s, settings, proc)
			if err != nil {
				return nil, err
			}
			return []humus.Service{svc}, nil
		},
	}.Run(ctx, p)
}

References

• Proposal: humus on the bedrock process model, revision 3 and the wave plan
• Proposal: the humus App lifecycle, on the two shutdown signals and an App per runtime package
• Proposal: configuration for humus apps
• Proposal: OpenTelemetry without globals
• Proposal: humustest, open question 5
• The pre-reset queue and queue/kafka
• story(queue): implement a runtime for rabbitmq #310 (RabbitMQ), story(kafka): add mTLS authentication support #352 (mTLS), story(queue/kafka): add otel metrics for message processing and commits #359 (metrics), story(queue/kafka): refactor delivery semantics so they orchestrate interface implementations #365 (orchestrators), story(queue/kafka): investigate testcontainers peformance #368 (testcontainers), story(kafka): add unit tests using kfake #431 (kfake)
• franz-go kgo: Dialer, BlockRebalanceOnPoll, OnPartitionsRevoked, DisableAutoCommit, CommitRecords, LeaveGroupContext
• franz-go kfake: ListenFn, VirtualNetwork, TLS, Cluster.GroupInfo
• franz-go kotel, not used
• OpenTelemetry semantic conventions for messaging, and Go's semconv/v1.43.0/messagingconv
• bedrock v0.24.0, the retry module

[Zaba505]

Revision: the open questions answered, on humus.Main

The charter's revision 5 moved this proposal to wave 3, behind Proposal: humus.Main, which has landed. This revision answers the five open questions, applies humus.Main, and takes in the charter's revision 7.

The open questions:

1. queue stays, as recommended. It holds Processor, ProcessorFunc, Delivery and CheckDelivery, in the core module, and Proposal: RabbitMQ (wave 4) implements the same contract.
2. An at-most-once processing error is logged and counted, and processing goes on, as proposed. The record was committed before Process ran, so stopping can't bring it back.
3. Settings keeps the fields proposed. ConsumeResetOffset, InstanceID, FetchMaxWait and ClientID are added when end users ask for them. A new field breaks no one, because DefaultSettings supplies it.
4. kotel's Meter records franz-go's broker metrics. The Service hands the client kotel.NewMeter(kotel.MeterProvider(s.Telemetry.MeterProvider)) as a hook. kotel's tracer isn't used: the record's span, the propagation through headers and the three convention instruments stay humus's own.
5. While the brokers are unreachable, franz-go keeps retrying, as proposed, and the Service keeps serving. kotel now counts each failed attempt, so a consumer that never connects shows up in its metrics. This can change as end users run it.

What humus.Main changes:

• kafka.App is gone, and the package has no Main. humus.Main has a 20s grace period and no drain, which are the defaults this proposal chose for a consumer. Like job, the package has no defaults of its own. The end user's Init builds the Service with NewService and returns it to humus.Main.
• Telemetry is no longer required. humus.Main's telemetry default replaces it, as humus.Main's revision 2 sets out.
• There is no test package. humustest.Start already runs a consumer on the defaults production uses, and putting kfake on its Host takes one option.

From the charter's revision 7, exposing health probes, producing to Kafka and RabbitMQ are wave 4. Failure injection, in wave 5, strips scenarios from the records this Service receives.

The prototype's whole-App tests are rewritten on humustest.Start, and with kotel's Meter added they pass. Each section below replaces the section of the same name in the post. Sections not named are unchanged, apart from the wording fixes at the end.

Summary

• queue keeps Processor and gains Delivery. It is the contract between the end user's code and any broker: what a message's processing returns, and when the message is acknowledged. Consumer, Acknowledger, Runtime, App and ErrEndOfQueue are removed.
• queue/kafka is a nested module with a Service. The Service joins a consumer group with franz-go, dials the brokers through the Host, and processes each partition's records one at a time, in offset order. It runs on humus.Main's defaults, so the package has no App and no Main.
• The delivery semantics are fixed and tested. Under at-least-once, a record is committed only once it and every earlier record of its partition were processed. A processing error ends the Service, so the App exits 1 and the record is redelivered. Under at-most-once, a batch is committed before its records are processed, and a processing error is logged and counted while processing goes on.
• Shutdown drains. On Serve's context the Service stops fetching, lets the record in flight on each partition finish, commits, and leaves the group, all within the grace period.
• Telemetry is injected. One consumer span per record continues the producer's trace through the record's headers, and the Service's metrics follow OpenTelemetry's messaging conventions. franz-go's broker metrics come from kotel's Meter, given the Setup's MeterProvider. kotel's tracer isn't used.
• Tests run on kfake over humustest's Host, inside synctest. kfake takes a listen function and was built for synctest. The prototype's whole-App tests, run with humustest.Start, pass under -race.

func main() {
	humus.Main(app.Init, humus.WithTelemetry(buildTelemetry))
}

app.Init sources the kafka.Settings, builds the queue.Processor[kafka.Message], and returns the Service that kafka.NewService(s, settings, p) builds.

Proposed API

queue is unchanged. In queue/kafka, Message, Header, Settings, DefaultSettings, the checks, Service, Serve and Shutdown are unchanged. App is gone, and NewService's doc gains a paragraph on what the Service reports:

// NewService checks settings and returns a Service. It opens no connection:
// Serve builds the client and joins the group.
//
// The Service reports through s.Telemetry: its spans, the three messaging
// instruments and its own logs under the scope
// github.com/z5labs/humus/queue/kafka, franz-go's logs under
// github.com/twmb/franz-go/pkg/kgo, and franz-go's broker metrics, from
// kotel's Meter, under github.com/twmb/franz-go/plugin/kotel.
func NewService(s *humus.Setup, settings Settings, p queue.Processor[Message]) (*Service, error)

So the package is Message, Header, Settings with its checks, and a Service with its constructor. It has no App, no Main and no test package.

What Serve does

Unchanged, except that step 1's client also gets kotel's Meter, with kgo.WithHooks. NewService builds the Meter, which creates its instruments and starts nothing, so NewService still acquires nothing.

Failures

Unchanged, except for the last row:

	AtLeastOnce	AtMostOnce
the brokers are unreachable	franz-go keeps retrying and logs each failure, and kotel counts each failed attempt; the Service keeps serving	the same

Telemetry

Unchanged, except that this bullet is new, after Metrics:

• franz-go's broker metrics, from kotel's Meter. NewService builds kotel.NewMeter(kotel.MeterProvider(s.Telemetry.MeterProvider)), and Serve hands it to the client.
  • By broker (node_id), it counts the connections opened, the connection errors and the connections closed, the bytes read and written, and the read and write errors. By broker and topic (topic), it counts the records and the uncompressed bytes fetched. Its produce counters stay empty, because the Service produces nothing.
  • The instruments are kotel's, under its own scope, github.com/twmb/franz-go/plugin/kotel, with names such as messaging.kafka.connects.count, messaging.kafka.connect_errors.count and messaging.kafka.fetch_records.count. They aren't conventions, because OpenTelemetry's messaging conventions define none for a client's connections or bytes.
  • kotel's options are left at its defaults. Connection errors are their own counter, not an outcome attribute on messaging.kafka.connects.count.
  • kotel uses the global MeterProvider when it is given none. The lifecycle rejects an SDK with a nil MeterProvider, so that never happens.
  • On kfake, a clean shutdown counts one read error for the broker, during shutdown and never before it. An alert on messaging.kafka.read_errors.count should allow for that.
  • As with semconv, humus pins the kotel version it uses. A kotel release that renames its instruments is a breaking change to queue/kafka.

Running on humus.Main

This section replaces The App. humus.Main runs the Services of any runtime, and its defaults are the ones this proposal chose for a consumer. So the package has no App, no Main and no defaults of its own:

	humus.Main	Why it fits a consumer
ShutdownGracePeriod	20s	With telemetry's default ShutdownTimeout of 5s, the whole shutdown takes at most 25s. That is inside Kubernetes' default terminationGracePeriodSeconds of 30s, with a margin for the process to exit. A Processor that can't finish a record within it is cut short, and the record is processed again.
Drain	none	Nothing load-balances a consumer. Its traffic stops when it leaves the group, which it does as soon as it has drained.
Telemetry	telemetry.New with DefaultSettings()	A consumer in the middle of a traced pipeline chooses its exporter with humus.WithTelemetry.
health	none	Proposal: exposing health probes, in wave 4, decides how a consumer's pod serves probes.

• Telemetry is no longer required. App required it, so that a consumer couldn't start as unknown_service by accident. humus.Main decided otherwise for every runtime. The default Builder writes to stdout, and a service that names itself or exports anything passes humus.WithTelemetry. Tests pass humustest.NewTelemetry()'s Builder, or none at all.
• The end user's Init builds the Service. App's Init returned the Settings and the Processor, and App called NewService. Now that call is in the end user's Init, which returns the Service along with any others the process runs. Two topics with different Delivery are two Services with two group IDs, returned together. More than one Service is no longer the advanced path.
• Which Main. A consumer on its own, or next to a probes or metrics server, uses humus.Main. Next to a server that a load balancer routes to, such as an API, it uses that server's package's Main, as humus.Main's revision 2 decides. The consumer still stops taking records as soon as shutdown starts, and its records in flight have the drain and the grace period to finish.
• An operator changes the defaults with HUMUS_SHUTDOWN_GRACE_PERIOD and HUMUS_SHUTDOWN_DRAIN_DELAY, as for any App. A team that wants another default in code passes humus.WithShutdownGracePeriod to humus.Main.
• No health. The process runs exactly the Services its Init returns. Proposal: exposing health probes, in wave 4, decides how a pod with no HTTP server of its own serves probes, and whether a consumer exposes them by default.

In use

func main() {
	humus.Main(Init, humus.WithTelemetry(buildTelemetry))
}

func Init(ctx context.Context, s *humus.Setup) ([]humus.Service, error) {
	d := kafka.DefaultSettings()
	env := func(name string) config.Value[string] { return config.Env(s.Process.Environment, name) }
	list := func(v string) []string { return strings.Split(v, ",") }

	settings, err := config.Struct(func(r *config.Resolver) kafka.Settings {
		k := d
		k.Brokers = r.Get(env("KAFKA_BROKERS").Map(list).Validate(kafka.CheckBrokers))
		k.GroupID = r.Get(env("KAFKA_GROUP_ID").Validate(kafka.CheckGroupID))
		k.Topics = r.Get(env("KAFKA_TOPICS").Map(list).Validate(kafka.CheckTopics))
		k.SessionTimeout = r.Get(env("KAFKA_SESSION_TIMEOUT").Parse(time.ParseDuration).Default(d.SessionTimeout).Validate(kafka.CheckSessionTimeout))
		return k
	}).Read(ctx)
	if err != nil {
		return nil, err // exit 78, naming every bad variable
	}

	pool, err := newPool(ctx, s) // dials through s.Host.Dial; closed in s.OnStop
	if err != nil {
		return nil, err
	}
	h := orders.NewHandler(pool) // idempotent: an order already stored is skipped

	svc, err := kafka.NewService(s, settings, queue.ProcessorFunc[kafka.Message](h.Handle))
	if err != nil {
		return nil, err
	}
	return []humus.Service{svc}, nil
}

What waves 4 and 5 build on this

This section is new. The charter's revision 7 has four later proposals build on this one. Nothing in this API is designed for them, and each one's stories make whatever change it needs here.

• Proposal: exposing health probes (wave 4) decides how a consumer's pod, which has no HTTP server of its own, serves probes, and whether a consumer exposes them by default. It does that in terms of humus.Main, Options that wrap Init, and the Services the end user's Init returns, as for every wave-3 runtime. Whether a consumer whose brokers are unreachable is ready (open question 5) is its decision too.

• Proposal: producing to Kafka (wave 4) is in this module. It builds on three things here:

  • the connection settings, Brokers, TLS and SASL, with their checks;
  • the unexported dialer that runs the TLS handshake over the Host's connection;
  • the header carrier, which the Service extracts the trace context through and a producer injects through.

  kotel's Meter counts a producer's records and bytes too. How a producer's settings share these fields, and whether they move into a type of their own, is that proposal's decision.

• Proposal: RabbitMQ (wave 4) implements the queue contract, as How RabbitMQ (story(queue): implement a runtime for rabbitmq #310) would fit sketches. It shares no code with queue/kafka.

• Proposal: failure injection (wave 5) strips scenarios from the records this Service receives, before the end user's Processor sees them. Its seam is the Service's processing of a record: the Service extracts the propagator's context, baggage included, from the record's headers, starts the record's span, and then calls Process. Stripping belongs after the extraction and before Process, inside the Service, so every Processor is covered without the end user wrapping anything. The wave-5 proposal designs that step and whatever configures it.

Design decisions

Unchanged, except that these bullets replace "No kotel." and "Dependencies.":

• kotel's Meter, not its tracer. franz-go's OpenTelemetry plugin has two halves.
  • Its Meter is a set of client hooks for what humus can't see from outside the client: connections, bytes, and read and write errors, per broker. It takes a MeterProvider, starts no goroutine, and works inside synctest on humustest's Host. Its names aren't conventions, but no conventions exist for these.
  • Its tracer uses semconv v1.18.0, whose names have since changed, and names its process span <topic> process. It starts a receive span for every buffered record, doubling the spans. The process span, the propagation through headers and the three convention instruments are about a hundred lines that humus controls, so they stay humus's own.
• Dependencies. github.com/twmb/franz-go is the client, as before: pure Go, maintained, with the group, commit and dialler hooks this design needs, and its pkg/sasl mechanisms live in the same module. github.com/twmb/franz-go/plugin/kotel is a module of its own. It requires franz-go and OpenTelemetry, which the module already has, and v1.7.1 builds with franz-go v1.22.2 and OpenTelemetry v1.47.0. github.com/twmb/franz-go/pkg/kfake is a test dependency. All three are in the nested module.

These bullets are new:

• No App and no Main. humus.Main's defaults are a consumer's. An App would only repeat them, and a Main would apply nothing, as for job.
• No test package. httpservertest exists because an HTTP service's defaults have to reach its tests, and HTTP/2 over a test Host is easy to get wrong. A consumer has neither problem. humustest.Start already runs it on humus.Main's defaults. kfake listens on the test Host with kfake.ListenFn(h.Listen), and a test's producer dials it with kgo.Dialer(h.Dial). The tests' helpers are a few lines each, which no API would save.
• An at-most-once processing error doesn't stop the Service. The record was committed before Process ran, so a restart doesn't redeliver it. Stopping would restart the pod for nothing. The error is recorded on the record's span and as error.type on messaging.process.duration, and logged at ERROR.
• Settings grows on request. The fields are the ones before the reset, plus SASL and MaxPollRecords. Other franz-go options become fields when end users ask for them, each with a default and a check.
• Unreachable brokers don't stop the Service. franz-go retries each connection with backoff and logs each failure, and kotel counts each failed attempt. Records are processed once the brokers are back. Failing Serve after a bound would turn a broker outage into a crash loop of every consumer, and readiness is the usual answer to a consumer that can't connect. If end users find that a consumer should fail instead, a bound can be added as a setting, which breaks no one.

Testing

The section kfake on humustest's Host and the section Integration tests are unchanged. Whole-App tests is replaced.

Whole-App tests

These tests pass under go test -race -count=3. They ran against the prototype's queue.go, settings.go and service.go, with kotel's Meter added to the client's options and app.go removed. The dependencies were humus at a155d299, bedrock v0.24.0, franz-go v1.22.2, kotel v1.7.1, kfake v0.0.0-20261013022759-7b6cac30a57a and OpenTelemetry v1.47.0. Each App runs with humustest.Start, so on humus.Main's defaults, on the Host kfake listens on:

// consumes returns an Init whose only Service consumes settings with p.
func consumes(settings kafka.Settings, p queue.ProcessorFunc[kafka.Message]) func(context.Context, *humus.Setup) ([]humus.Service, error) {
	return func(_ context.Context, s *humus.Setup) ([]humus.Service, error) {
		svc, err := kafka.NewService(s, settings, p)
		if err != nil {
			return nil, err
		}
		return []humus.Service{svc}, nil
	}
}

func TestService_at_least_once_stops_on_a_failure_and_redelivers(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		h := humustest.NewHost(t)
		c := cluster(t, h, 1) // kfake.NewCluster(kfake.ListenFn(h.Listen), kfake.SeedTopics(1, "orders"))
		defer c.Close()
		produce(t, h, c, 5) // offsets 0 to 4

		bad := errors.New("cannot process")
		run := func(fail bool) (bedrock.ExitStatus, []int64) {
			var seen []int64
			p, status := humustest.Start(t, consumes(settings(c.ListenAddrs()...), func(ctx context.Context, m kafka.Message) error {
				seen = append(seen, m.Offset)
				if fail && m.Offset == 3 {
					return bad
				}
				return nil
			}), humustest.WithApp(humus.WithHost(h.Host())))
			if !fail {
				synctest.Wait() // every record is processed and committed
				p.Signal(syscall.SIGTERM)
			}
			return <-status, seen
		}

		st, seen := run(true)
		require.Equal(t, humus.ExitFailure, st.Code())
		require.ErrorIs(t, st.Cause(), bad)
		require.Equal(t, []int64{0, 1, 2, 3}, seen)
		require.Equal(t, int64(3), committed(c), "0 to 2 are committed, 3 is not")

		st, seen = run(false)
		require.Zero(t, st.Code(), st.Cause())
		require.Equal(t, []int64{3, 4}, seen, "3 is delivered again")
		require.Equal(t, int64(5), committed(c))
	})
}

func TestService_SIGTERM_lets_the_record_in_flight_finish_and_commits_it(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		h := humustest.NewHost(t)
		c := cluster(t, h, 1)
		defer c.Close()
		produce(t, h, c, 5)

		entered := make(chan struct{}, 5)
		p, status := humustest.Start(t, consumes(settings(c.ListenAddrs()...), func(ctx context.Context, m kafka.Message) error {
			entered <- struct{}{}
			time.Sleep(3 * time.Second)
			return nil
		}), humustest.WithApp(humus.WithHost(h.Host())))
		<-entered // offset 0 is in flight
		p.Signal(syscall.SIGTERM)

		st := <-status
		require.Zero(t, st.Code(), st.Cause())
		require.Equal(t, int64(1), committed(c), "0 finished and is committed; 1 to 4 never started")
	})
}

func TestService_records_franz_go_broker_metrics_with_kotel(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		h := humustest.NewHost(t)
		c := cluster(t, h, 2)
		defer c.Close()
		produce(t, h, c, 10)

		tel := humustest.NewTelemetry()
		p, status := humustest.Start(t, consumes(settings(c.ListenAddrs()...), func(context.Context, kafka.Message) error { return nil }),
			humustest.WithApp(humus.WithHost(h.Host()), humus.WithTelemetry(tel.Builder())))
		synctest.Wait()
		p.Signal(syscall.SIGTERM)
		require.Zero(t, (<-status).Code())

		require.Len(t, tel.Spans(), 10)
		require.Equal(t, int64(10), sum(t, tel, "github.com/twmb/franz-go/plugin/kotel", "messaging.kafka.fetch_records.count"))
	})
}

func TestService_keeps_serving_while_the_brokers_are_unreachable(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		h := humustest.NewHost(t) // nothing listens on it
		tel := humustest.NewTelemetry()
		p, status := humustest.Start(t, consumes(settings("127.0.0.1:9092"), func(context.Context, kafka.Message) error { return nil }),
			humustest.WithApp(humus.WithHost(h.Host()), humus.WithTelemetry(tel.Builder())))

		time.Sleep(time.Minute)
		select {
		case st := <-status:
			t.Fatalf("the App exited while the brokers were unreachable: %v", st.Cause())
		default:
		}
		p.Signal(syscall.SIGTERM)

		require.Zero(t, (<-status).Code())
		require.Positive(t, sum(t, tel, "github.com/twmb/franz-go/plugin/kotel", "messaging.kafka.connect_errors.count"))
	})
}

The helpers are a few lines each:

• produce writes records with a franz-go client that dials h.Dial.
• settings is DefaultSettings() with the brokers given, a group ID and the topic.
• committed sums the group's committed offsets from c.GroupInfo.
• sum adds up a counter's data points from tel.Metrics.

A separate check showed that kotel's instruments reach the Setup's MeterProvider, and none reaches OpenTelemetry's global one.

The post's list of what the prototype's other tests show stands. Those tests ran with kafka.App, and the stories run them with humustest.Start. The stories add tests for what the prototype leaves out. The post's list stands, except that its last bullet becomes:

• A runnable Example, or a package main with its test, of a consumer on humus.Main, as the charter requires.

The list also gains these bullets:

• kotel's instruments are recorded under its scope with node_id and topic, and none reaches OpenTelemetry's global MeterProvider, even when one is set.
• With the brokers unreachable, franz-go's connection failures reach humustest.Telemetry's Logs() through the log adapter.

Migration from today

Unchanged, except for these rows:

Before the reset	After
queue.NewApp(runtime)	humus.Main(Init), with the end user's Init returning the Service from kafka.NewService
otel.Get*, humus.Logger, kotel, kslog	s.Telemetry; kotel's Meter alone, given s.Telemetry.MeterProvider, for franz-go's broker metrics

Wording fixes to the post

• The note at the top now reads: wave 3, building on Proposal: humus.Main as well as the four wave-1 proposals it names, and depending on no other wave-3 proposal. Proposal: exposing health probes (wave 4) decides how a consumer's pod serves probes. Proposal: producing to Kafka and Proposal: RabbitMQ (wave 4) build on this, and Proposal: failure injection (wave 5) strips scenarios at this Service.
• In Background, "What wave 1 settled":
  • "Each runtime package provides Services for composition and an App built on humus.App with its own defaults" is replaced by humus.Main: each runtime package provides Services, and a Main only where it needs other defaults than humus.Main's.
  • "No wave-2 App serves health probes" reads "No wave-3 runtime serves health probes", citing the charter's revision 7.
  • The list gains humus.Main: humus.Main, the Drain, and humustest.Start.
• In Design decisions, "One Processor and one Delivery per Service" ends "and two semantics are two Services with two group IDs, returned from one Init."
• Out of scope:
  • "Producing, and an outbound Kafka client: the charter's "Later" list" reads "Producing: Proposal: producing to Kafka (wave 4)."
  • "RabbitMQ (story(queue): implement a runtime for rabbitmq #310): "Later"" reads "RabbitMQ (story(queue): implement a runtime for rabbitmq #310): Proposal: RabbitMQ (wave 4)."
  • The bullet on health probes ends "in wave 4".
• Prototype: the post's code stands as what the first tests ran against. Its kafka/app.go is the App this revision removes.
• References gain humus.Main's revision 2 and the charter's revision 7. The kotel entry reads "franz-go kotel: NewMeter and MeterProvider; its tracer is not used".

Open questions

None.

[Zaba505]

Closing: queue and the Kafka runtime have landed

The revision's stories are done:

• story(queue): define the contract between a Processor and any broker #726 defined the contract between a Processor and any broker, in queue (story(queue): define the contract between a Processor and any broker #745);
• story(kafka): configure a Kafka consumer with Settings and a check per field #727 configured a Kafka consumer with Settings and a check per field, in the new queue/kafka module (story(kafka): configure a Kafka consumer with Settings and a check per field #746);
• story(kafka): consume Kafka topics as a humus Service, at least or at most once #728 consumed Kafka topics as a humus Service, at least or at most once (story(kafka): consume Kafka topics as a humus Service, at least or at most once #747);
• story(kafka): trace and measure every record with the Setup's telemetry #729 traced and measured every record with the Setup's telemetry, with kotel's Meter for franz-go's broker metrics (story(kafka): trace and measure every record with the Setup's telemetry #748);
• story(kafka): send franz-go's logs to the Setup's telemetry #730 sent franz-go's logs to the Setup's telemetry (story(kafka): send franz-go's logs to the Setup's telemetry #749);
• story(kafka): test the consumer against Kafka in a container, in a CI job of its own #731 tested the consumer against Kafka in a container, in a CI job of its own (story(kafka): test the consumer against Kafka in a container, in a CI job of its own #750);
• story(docs): document consuming Kafka #732 documented consuming Kafka (story(docs): document consuming Kafka #756).

The revision left no open questions. One finding came out of #731: against Kafka itself, a fenced-out member's partition comes back revoked, not lost, so its late commit can move the new owner's committed offset. At-least-once still holds, and the fix changes none of queue's or queue/kafka's API. It's tracked in #757.

What the revision hands on belongs to the wave-4 and wave-5 proposals, as its "What waves 4 and 5 build on this" lists. Exposing health probes decides how a consumer's pod serves probes. Producing to Kafka builds on the connection settings, the TLS dialer and the header carrier. RabbitMQ implements the queue contract. Failure injection strips scenarios between header extraction and Process.