Proposal: producing to Kafka

[Zaba505]

Note

Part of the humus re-evaluation; see Proposal: humus on the bedrock process model, wave 4. It builds on Proposal: the humus App lifecycle, Proposal: configuration for humus apps, Proposal: OpenTelemetry without globals, Proposal: humustest, Proposal: humus.Main and Proposal: queue processing and the Kafka runtime, whose module, connection settings, TLS dialer and header carrier it uses. It depends on no other wave-4 proposal. It is discussed beside Proposal: outbound HTTP clients and Proposal: outbound gRPC clients, and keeps to the three points the charter's revision 7 sets for the clients. It blocks Proposal: failure injection (wave 5), which adds an injection point at the seam named below.

Summary

queue/kafka gets a producer, next to its consumer:

• kafka.Producer produces records to one Kafka cluster. The end user's Init builds it with NewProducer, from ProducerSettings, and any Service uses it: a Processor that writes to another topic or to a dead-letter topic, an HTTP handler that publishes an event, a job. It is not a Service. It registers its flush and close with s.OnStop, so it outlives every Service that uses it.
• Produce(ctx, records...) is synchronous. It returns once the cluster has acknowledged every record, with acks from all in-sync replicas, or once one has failed. A Processor that produces before it returns nil has the consumed record committed only after the produced one is stored.
• It reaches the cluster the way the consumer does. Brokers, TLS and SASL mean what they mean in Settings, with the same checks, and the client dials through s.Host with the consumer's TLS dialer.
• franz-go retries, not bedrock's retry. franz-go's idempotent producer retries a batch on the sequence numbers that let the broker drop duplicates, and keeps each partition in order. A retry around it would send the records again under new sequence numbers. DeliveryTimeout, 10s by default, bounds each call, retries included.
• A bedrock circuit breaker guards the cluster. One per Producer. While the cluster keeps failing, Produce fails at once with an error matching circuit.ErrRejected, rather than holding every caller for the delivery timeout.
• Telemetry follows OpenTelemetry's messaging conventions. Each record gets a PRODUCER span, send <topic>, whose context goes into the record's headers through the Setup's propagator, so the consumer's process span is its child. messaging.client.sent.messages and messaging.client.operation.duration are recorded, kotel's produce counters fill in, and each breaker transition is a log record.

func Init(ctx context.Context, s *humus.Setup) ([]humus.Service, error) {
	cs, ps, err := readSettings(ctx, s) // kafka.Settings and kafka.ProducerSettings, sourced with config.Struct
	if err != nil {
		return nil, err
	}
	dlq, err := kafka.NewProducer(s, ps)
	if err != nil {
		return nil, err
	}
	svc, err := kafka.NewService(s, cs, queue.ProcessorFunc[kafka.Message](func(ctx context.Context, m kafka.Message) error {
		order, err := decode(m.Value)
		if err != nil {
			// A dead letter, stored before the record is committed.
			return dlq.Produce(ctx, kafka.Record{Topic: "orders.dlq", Key: m.Key, Value: m.Value, Headers: m.Headers})
		}
		return store.Save(ctx, order)
	}))
	if err != nil {
		return nil, err
	}
	return []humus.Service{svc}, nil
}

Background

• humus has no producer. The consumer landed with #683, which deferred producing to this proposal. Its package doc says kotel's produce counters "stay empty, because the Service produces nothing" (doc.go#L189).
• Before the reset, there was no producer either. The Kafka examples' tools built a franz-go client with kgo.NewClient and called ProduceSync themselves, on the real network and with no telemetry (example/queue/kafka-at-least-once/tool/main.go).
• Dead letters are the Processor's. queue's contract and the Kafka page leave them to the end user: a Processor that would rather skip a record than stop on it "stores the record wherever it keeps dead letters, such as a table or a topic of its own, and returns nil". Under at-least-once that store has to finish before Process returns. The walkthroughs (wave 6) rely on it: their tests treat the group's committed offset as a barrier after which the dead-letter topic holds every dead letter.
• What Proposal: queue processing and the Kafka runtime #683 hands on. Its revision lists what a producer builds on: the connection settings Brokers, TLS and SASL with their checks (settings.go#L31-L63); the unexported tlsDialer, which runs the TLS handshake over the Host's connection because franz-go refuses kgo.Dialer beside kgo.DialTLSConfig (dial.go#L25); and the unexported headerCarrier, which the Service extracts through and a producer injects through (carrier.go#L16). "How a producer's settings share these fields, and whether they move into a type of their own, is that proposal's decision."
• Outbound instrumentation. OpenTelemetry without globals left the instrumentation of outbound clients out of scope. The charter's revision 7 gives it to the three client proposals.
• bedrock v0.24.0's retry and circuit ship the models with no I/O, and leave the protocol wiring, the instruments and the settings to humus (circuit, retry). retry's doc says it doesn't replay requests, which "belong to the framework that wraps a transport".
• franz-go's produce path. These facts carry the design. Each was checked with franz-go v1.22.2 and kfake v0.0.0-20261013022759-7b6cac30a57a, inside synctest.Test on a humustest Host, in a prototype that isn't kept:
  • The producer is idempotent by default, with acks=all. It retries a partition's batch without limit, on the same sequence numbers, so the broker drops a batch it already stored, and the partition's order holds.
  • By default it refuses to fail a record that has been sent and hasn't been answered, because it can't tell "never received" from "stored, reply lost". A ProduceSync whose context ran out after 2s, against a broker holding its produce request, was still blocked 5 minutes later.
  • With kgo.AllowIdempotentProduceCancellation, the same call returned at 2s with context.DeadlineExceeded. The broker stored the record once it was released, so the caller was told of a failure for a record that landed.
  • kgo.RecordDeliveryTimeout is checked only at safe points. With a 10s timeout and a broker holding its request, the record failed after 20s: the 10s produce timeout plus the 10s request overhead. A record context with a 10s deadline, against a broker answering every produce with the retriable NOT_ENOUGH_REPLICAS, failed each call at exactly 10s.
  • A broker that refuses the SASL credentials doesn't fail a produce: franz-go retries the connection, and the call ends only when its context does, with no sign of the refusal.
  • Flush(ctx) returns ctx's error when ctx is done first, and Close then fails what is left with kgo.ErrClientClosed.

Proposed API

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

// Record is a Kafka record to produce.
type Record struct {
	// Topic is the topic the record is produced to. It is required.
	Topic string

	// Key is the record's key, nil when it has none. Records with the same
	// key go to the same partition of Topic, hashed with murmur2 as Kafka's
	// Java producer hashes them.
	Key []byte

	// Value is the record's value, nil when it has none, which a compacted
	// topic reads as a tombstone.
	Value []byte

	// Headers are the record's headers, in order. A key may repeat. Any
	// header the Setup's propagator writes, such as traceparent or baggage,
	// is replaced by the record's own trace context.
	Headers []Header
}

// ProducerSettings configures a Producer. Every field is a plain value, and a
// field's zero value means zero, never "use the default": start from
// DefaultProducerSettings and change the fields you source. Brokers has no
// default, so it is always among them.
type ProducerSettings struct {
	// Brokers are the addresses of the brokers the Producer bootstraps
	// from, each as host:port. They are required.
	//
	// Default: none. Valid: at least one; each host:port, with a host and a
	// port from 1 to 65535. Check: CheckBrokers.
	Brokers []string

	// TLS is the config the Producer dials the brokers with. nil means
	// plaintext. A client certificate in Certificates makes it mTLS.
	//
	// Default: nil, which means plaintext. Valid: any. Check: none.
	TLS *tls.Config

	// SASL is the mechanism the Producer authenticates to the brokers with,
	// one of franz-go's from github.com/twmb/franz-go/pkg/sasl, carrying
	// the end user's credentials. nil means no authentication.
	//
	// Default: nil, which means no authentication. Valid: any. Check: none.
	SASL sasl.Mechanism

	// DeliveryTimeout bounds how long the cluster has to acknowledge the
	// records of one call to Produce, franz-go's retries included. A record
	// that isn't acknowledged by then fails, even one already sent, and
	// may still have been stored.
	//
	// Default: 10s. Valid: positive. Check: CheckDeliveryTimeout.
	DeliveryTimeout time.Duration

	// Breaker configures the circuit breaker in front of the cluster.
	//
	// Default: circuit.Config{Window: 20, Probes: 1, OpenFor: 10 *
	// time.Second, ProbeTimeout: 10 * time.Second}. Valid: Window and
	// Probes from 1 to circuit.MaxCount, OpenFor and ProbeTimeout positive.
	// Checks: CheckBreakerWindow, CheckBreakerProbes, CheckBreakerOpenFor,
	// CheckBreakerProbeTimeout.
	Breaker circuit.Config

	// BreakerPolicy decides when the breaker opens and closes. It is set in
	// code, not sourced.
	//
	// Default: open after 5 failures in a row, or once half of the last 20
	// or more calls failed; while half-open, reopen on a failed probe and
	// close after 3 successful ones. Valid: Trip and Probe set, which
	// NewProducer checks. Check: none.
	BreakerPolicy circuit.Policy
}

// DefaultProducerSettings returns the Producer's defaults. Brokers has none,
// so it is nil, which CheckBrokers rejects: the end user always sources it.
func DefaultProducerSettings() ProducerSettings

// The new checks. CheckBrokers is the consumer's. Each reports what is
// wrong, without naming the field.
func CheckDeliveryTimeout(d time.Duration) error
func CheckBreakerWindow(n int) error
func CheckBreakerProbes(n int) error
func CheckBreakerOpenFor(d time.Duration) error
func CheckBreakerProbeTimeout(d time.Duration) error

// Producer produces records to one Kafka cluster. It is safe for concurrent
// use: one Producer serves every Service of the process.
type Producer struct{ /* unexported */ }

// NewProducer runs every check on settings, reporting every bad field in
// one error that names each one, and builds the franz-go client and the
// circuit breaker. It opens no connection: the first Produce does. It
// registers a function with s.OnStop that flushes the client and closes it.
//
// The Producer reports through s.Telemetry: its spans, the two 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 NewProducer(s *humus.Setup, settings ProducerSettings) (*Producer, error)

// Produce produces records, and returns nil once the cluster has
// acknowledged every one. Records to one partition are stored in the order
// they are given.
//
// It returns an error, naming each record that failed by its topic, when a
// record failed or ProducerSettings.DeliveryTimeout passed; an error that
// matches circuit.ErrRejected, at once and with nothing sent, while the
// breaker is open; and an error that wraps ctx's error when ctx is done
// first. A record that failed may still have been stored, so a caller that
// produces it again may store it twice.
//
// Produce rejects a record with no Topic before it sends anything. With no
// records, it returns nil.
func (p *Producer) Produce(ctx context.Context, records ...Record) error

The Producer is in package kafka, in the queue/kafka module. The one new dependency is bedrock's circuit module, github.com/z5labs/bedrock/circuit v0.24.0, which has no dependency outside the standard library.

Producing

The client. NewProducer builds one franz-go client with kgo.SeedBrokers(settings.Brokers...), kgo.Dialer of the consumer's tlsDialer over s.Host.Dial and settings.TLS, kgo.SASL when settings.SASL is set, the consumer's franz-go logger, kotel's Meter as hooks, a connect hook described under Refused credentials, and kgo.AllowIdempotentProduceCancellation(). Every other producer option keeps franz-go's default, among them:

franz-go option	Default	What it means here
RequiredAcks	AllISRAcks()	a record is acknowledged once every in-sync replica has it
idempotent writes	on	the broker drops a batch franz-go sends twice
RecordRetries	no limit	a batch is retried until it is stored, fails for good, or its records' context ends
RetryBackoffFn	jittered, 250ms to 5s	the wait between retries of a batch
RecordPartitioner	UniformBytesPartitioner(64KiB, adaptive, keys, murmur2)	keyed records hash like Java's; unkeyed ones fill a partition to 64 KiB, then move, away from backed-up partitions
ProducerLinger	10ms	ProduceSync, which Produce uses, stops lingering on the partitions it produced to
ProducerBatchCompression	snappy, then none
ProducerBatchMaxBytes	1,000,012	Kafka's default max.message.bytes; a larger record fails at once with MESSAGE_TOO_LARGE
MaxBufferedRecords	50,000	Produce waits for room, or for ctx

A call. Produce:

1. rejects a record with no Topic, and returns at once, with nothing sent, when the client was closed or the cluster refused the credentials;
2. builds a context for the call from ctx: one that keeps ctx's values, isn't cancelled with it, and is done after DeliveryTimeout;
3. starts each record's span, and injects the span's context and ctx's baggage into the record's headers through the Setup's propagator, after removing any header the propagator writes;
4. asks the breaker to admit the call, and fails every record's span and returns the rejection when it doesn't;
5. hands every record to franz-go with the call's context, as ProduceSync does, and waits for every record's result or for ctx, whichever comes first.

Records outlive a cancelled call. When ctx is done first, Produce returns an error that wraps ctx's error, at once. Its records stay in the client, with the call's context, until the cluster acknowledges them or DeliveryTimeout fails them, and their spans and measurements are recorded then. So a handler whose request is cancelled returns at once, and the grace period reaches a Process blocked in Produce, as the lifecycle requires. And one caller's cancellation never fails another's records: franz-go fails every buffered record of a partition when it fails one, and decides from the context of a batch's first record, so records handed over with their callers' contexts would fail with whichever caller gave up first.

DeliveryTimeout bounds what is in flight. kgo.AllowIdempotentProduceCancellation lets the call's context fail a record that was sent and not answered. Without it, as the prototype showed, a hung broker holds a call, and the App's shutdown, until it answers. With it, a record can fail and still be stored. A caller that produces it again may store it twice, which is at-least-once, and is what a Processor under queue.AtLeastOnce already handles. franz-go bumps its producer epoch before producing again, so its own sequence numbers stay valid.

Order. franz-go keeps each partition's records in the order it was handed them, across its retries. Within one call that is the order of records. Between calls on different goroutines it is the order they reach the client. When a record of a partition fails, franz-go fails the records buffered after it on that partition too, so a later record is never stored after an earlier one failed. A caller that produces a failed call again sends its records after any that other calls have produced since.

Errors. When every record was acknowledged, Produce returns nil. Otherwise it returns one error joining each failed record's error, with its topic, and its partition when one was chosen, so errors.Is finds kgo.ErrRecordTimeout, context.DeadlineExceeded, circuit.ErrRejected or a kerr code in it. A failed call doesn't say which of its records were stored. Kafka returns no partial result a caller could act on safely, and the records that failed may have been stored anyway.

Settings

Field	Default	Valid	Check
Brokers	none	at least one host:port	CheckBrokers
TLS	nil, plaintext	any	none
SASL	nil, no authentication	any	none
DeliveryTimeout	10s	positive	CheckDeliveryTimeout
Breaker.Window	20	1 to circuit.MaxCount	CheckBreakerWindow
Breaker.Probes	1	1 to circuit.MaxCount	CheckBreakerProbes
Breaker.OpenFor	10s	positive	CheckBreakerOpenFor
Breaker.ProbeTimeout	10s	positive	CheckBreakerProbeTimeout
BreakerPolicy	see the field's doc	Trip and Probe set	NewProducer

• Brokers, TLS and SASL are the consumer's fields, repeated. They mean the same, take the same check, and dial through the same unexported dialer. A process that consumes and produces on one cluster sources them once and copies the three into ProducerSettings. Settings is unchanged, so no Init breaks. Open question 2 asks whether they should move into a type of their own instead.
• Breaker is bedrock's own circuit.Config, with a check for each of its fields, so an operator's bad OpenFor fails startup naming the variable it came from, as #631 requires. NewProducer also passes it to circuit.New, which rejects the same values.
• BreakerPolicy is a circuit.Policy, set in code. A policy is two functions, so it can't come from a variable, as SASL can't. bedrock ships no policy, because "how many failures are too many is the caller's decision", so humus ships its default in DefaultProducerSettings and the end user replaces it in their Init. A policy whose Trip always returns false turns the breaker off.
• No retry settings. There is no MaxAttempts, backoff or retry.BudgetConfig, because franz-go retries. See Design decisions.

Sourced in the end user's Init, with the consumer's variables reused for the cluster:

dp := kafka.DefaultProducerSettings()
ps, err := config.Struct(func(r *config.Resolver) kafka.ProducerSettings {
	p := dp
	// No .Default: Brokers is required, so an unset variable fails as not found.
	p.Brokers = r.Get(env("KAFKA_BROKERS").Map(list).Validate(kafka.CheckBrokers))
	p.DeliveryTimeout = r.Get(env("KAFKA_DELIVERY_TIMEOUT").Parse(time.ParseDuration).Default(dp.DeliveryTimeout).Validate(kafka.CheckDeliveryTimeout))
	p.Breaker.OpenFor = r.Get(env("KAFKA_BREAKER_OPEN_FOR").Parse(time.ParseDuration).Default(dp.Breaker.OpenFor).Validate(kafka.CheckBreakerOpenFor))
	return p
}).Read(ctx)

env and list are the end user's helpers, as on the Kafka page.

Retries and the breaker

franz-go's retries, and no retry.Retrier. The charter's rule for the clients is that each picks the library's retries or bedrock's, never both. For Kafka it is franz-go's:

• Duplicates. franz-go retries a batch on its sequence numbers, and the broker drops a batch it has already stored. A retry around Produce would hand the records to the client again, under new sequence numbers, and a record whose first send landed would be stored twice.
• Order. franz-go retries a partition's batches in order. A retry around Produce sends its records after whatever other calls produced in the meantime.
• Retries don't multiply with callers. franz-go retries a partition's batch, which holds every caller's records for that partition, once per backoff. A thousand callers waiting on a dead partition cost one retry per backoff, not a thousand, so the storm a retry.Budget exists to stop doesn't happen. There is no budget.

DeliveryTimeout is the one retry setting: how long a call lets franz-go keep trying.

What the breaker guards. The dependency is the cluster: one Producer is one client to one cluster, with one breaker, built in NewProducer. A process that produces to two clusters builds two Producers. Each call to Produce is one call through the breaker, admitted before any of its records reach franz-go and settled once all of them have a result, which may be after Produce has returned. The ticket is handed off to the records' results, so the guard is the flag circuit's doc shows for a result settled after the function returns.

While the breaker is open, Produce fails at once with a *circuit.RejectedError, which says how much longer it stays open. Without the breaker, every call during an outage would wait out DeliveryTimeout: an HTTP handler that publishes an event would hold its request for 10s before answering 503. With it, the first few calls wait, and the rest fail at once until a probe gets through. Produce doesn't retry a rejected call. Whether to wait and try again is the caller's, as queue's contract leaves retries to the Processor.

What counts. The breaker's classifier sees each call's result with the call's own context, which a caller giving up doesn't cancel:

• Success: every record was acknowledged.
• Ignored: a record failed for a reason that is the record's or its topic's, not the cluster's: a Kafka error code franz-go doesn't retry, such as MESSAGE_TOO_LARGE, INVALID_RECORD or TOPIC_AUTHORIZATION_FAILED, or UNKNOWN_TOPIC_OR_PARTITION once franz-go has given up on the topic. One topic's mistake doesn't open the breaker for the others.
• Failure: anything else, which is the cluster not storing records in time: DeliveryTimeout passing, kgo.ErrRecordTimeout, a retriable error code franz-go stopped on, or a connection that failed.

The defaults follow the charter's starting values for the clients. ProbeTimeout matches DeliveryTimeout: a probe's result comes within the delivery timeout, so a probe holds its slot no longer than the call does.

A partition's leader going down. One breaker covers every partition. When one broker of three is down, the calls to the partitions it leads fail and the rest succeed, which the default policy may or may not trip on. franz-go moves unkeyed records away from backed-up partitions, and a keyed record has only one partition. A breaker per broker or per partition would follow Kafka's failures more closely, but franz-go picks the partition inside the client, after the breaker has admitted the call.

Shutdown

The Producer is not a Service, so the lifecycle doesn't stop it. The Services that use it stop first. The App calls their Shutdown, waits for every Serve to return, and only then runs the OnStop functions, so no Service is producing when the Producer closes.

NewProducer registers one function with s.OnStop. It calls Flush with the grace period's context, which waits for the records still in the client: those of a call that returned when its ctx was done, at most DeliveryTimeout old. Then it calls Close. When the grace period runs out first, Close fails what is left with kgo.ErrClientClosed, those records are lost, and the function returns an error that wraps the context's error, which the lifecycle reports as a *humus.GracePeriodError. A Produce after that, from a goroutine the end user left running, fails with an error wrapping kgo.ErrClientClosed.

DeliveryTimeout's default fits the budgets the runtime packages set. It is half of humus.Main's 20s grace period, which a consumer runs on, so a Process blocked in Produce at SIGTERM finishes inside it. It equals httpserver.Main's 10s grace period, and a handler's Produce ends sooner anyway when the grace period cancels the request.

If the end user's Init fails after NewProducer, the lifecycle runs the OnStop functions registered so far, and the client is closed. Building the client starts franz-go's goroutines, and closing it stops them, so nothing outlives the App.

Telemetry

The Producer takes everything from s.Telemetry, under the module's scope, github.com/z5labs/humus/queue/kafka, and never reads or sets OpenTelemetry's globals.

A span per record. Named send <topic>, of kind PRODUCER, a child of the span in Produce's ctx. Its attributes are messaging.system=kafka, messaging.operation.type=send, messaging.operation.name=send and messaging.destination.name, and, once the record is acknowledged, messaging.destination.partition.id and messaging.kafka.offset. The key is not recorded, as the consumer doesn't record it. It ends when the record is acknowledged or fails, which may be after Produce returns. A failure, a rejection by the breaker included, is recorded on it as an exception event, with error.type and an error status.

The span's context goes into the record's headers. The Setup's propagator injects it, with ctx's baggage, through headerCarrier. With the default propagator that is traceparent, tracestate and baggage. Before injecting, the Producer removes every header whose key is one of the propagator's Fields(). A dead letter that copies the consumed record's headers would otherwise carry two traceparent headers, and the carrier's Get returns the first, the old one. So the consumer's process span on any topic is a child of the send span that produced its record. Inside a Process, the send span is a child of the process span, and one trace runs from the first producer through every topic a record passes through.

Two instruments of the messaging conventions, as semconv v1.43.0 names them, the version the consumer pins:

• messaging.client.sent.messages counts each record once its result is known, with messaging.operation.name=send, and error.type when it failed;
• messaging.client.operation.duration is how long each record took, from Produce taking it to its result, in seconds, with messaging.operation.name=send, messaging.operation.type=send, and error.type when it failed.

Each also has messaging.system, messaging.destination.name and, once known, messaging.destination.partition.id. A record the breaker rejected, or that Produce rejected for having no topic, was never sent, so neither instrument counts it. The conventions say a sent-messages count "MUST NOT count messages that were created but haven't yet been sent". Its span records the rejection.

kotel's produce counters fill in. The Producer builds kotel's Meter with the Setup's MeterProvider, as the consumer does. messaging.kafka.produce_records.count and messaging.kafka.produce_bytes.count, by broker and topic, now count, next to the connection counters by broker. The consumer's doc and the Kafka page change where they say those counters stay empty.

Logs. franz-go's logs go through the consumer's adapter, under github.com/twmb/franz-go/pkg/kgo, so a retried produce request and a broker it can't reach are logged as the consumer's are. The Producer's own records are:

• one per breaker transition, in the context of the call that made it: circuit breaker open at WARN, and circuit breaker half-open and circuit breaker closed at INFO, the bodies the other two client proposals use. Its attributes are messaging.system=kafka, circuit.previous_state, and the counts of the state it left: circuit.successes, circuit.failures, circuit.consecutive_successes, circuit.consecutive_failures and circuit.timed_out;
• one ERROR record when a broker refuses the credentials, described next.

There are no instruments of humus's own for the breaker. Its transitions are in the logs, and a rejected call is an error on its span.

Refused credentials

franz-go retries a connection whose credentials a broker refuses, and the produce waits until its context ends with no sign of why. That is the consumer's problem too, and its Service stops on the first refusal (service.go#L437). A Producer can't stop the App, so it fails loudly instead. The same connect hook watches for SASL_AUTHENTICATION_FAILED, UNSUPPORTED_SASL_MECHANISM and ILLEGAL_SASL_STATE. On the first refusal, the Producer emits one ERROR record naming the broker, fails the calls in flight with the refusal, and from then on every Produce returns it at once, before the breaker, so the breaker doesn't count it. Credentials don't fix themselves while the process runs: a Processor that produces fails its record, and the consumer stops and is restarted with whatever the secret holds then.

The seam for failure injection

Proposal: failure injection (wave 5) adds an injection point to Produce. Its seam is step 3 of a call: once each record's span has started and its headers carry the trace context and the baggage, and before the call goes through the breaker. An injected abort there fails the records' spans like a real failure. Whether it counts against the breaker, and the point's name, are wave 5's to decide, and its stories add them here.

Scenarios cross Kafka with no further change. The Producer injects ctx's baggage into the record's headers, and the consumer extracts it before Process. Stripping scenarios from records produced to a cluster another organisation owns happens at the same step, and is wave 5's too.

Changes to the rest of the module

This proposal's stories make them:

• doc.go's first sentence says the package consumes and produces, and its paragraph on kotel's produce counters changes. A section, Producing, describes the Producer.
• headerCarrier is unchanged. The Producer removes the propagator's fields from a record's headers itself, before Inject.
• The connect hook that detects a refusal is shared by the Service and the Producer.
• go.mod adds github.com/z5labs/bedrock/circuit v0.24.0.
• The root package doc and the README, which say queue/kafka consumes Kafka topics, say it produces too.
• The Kafka page gets a section, Producing, and its Dead letters section shows a Processor producing a dead letter with the Producer.

Design decisions

• In package kafka. The module already has franz-go, the dialer and the carrier, and a sub-package would need those exported. A Producer beside the Service reads as kafka.NewProducer and kafka.NewService.
• A Producer, not a Service. Producing has no loop of its own to serve. It is called from the Services that do, so it lives as long as they do, and closes in OnStop, after them. The charter sets the same rule for every client.
• Synchronous only. A record is either stored or failed when Produce returns, which is what at-least-once needs: a Processor that produced before returning nil never commits a record whose output was lost. franz-go's promise, the asynchronous alternative, runs on the client's own goroutine, one at a time, and blocks the client when it blocks. A caller that wants not to wait runs Produce in a goroutine of its own. Open question 1.
• A Record of its own. Message has a partition, an offset and a timestamp, which a producer doesn't set. Header is shared, so a dead letter copies m.Headers as it is. The record's timestamp is the time it is produced, which franz-go sets.
• acks=all and idempotence are not settings. They are franz-go's and Kafka's defaults since Kafka 3.0, and turning either off trades durability for latency in a way that is invisible until a broker fails.
• No setting for the rest of franz-go's options. Compression, linger, batch size, the partitioner and the buffer size keep franz-go's defaults. A field is added when a user asks, as the consumer's revision did for ConsumeResetOffset and the others.
• Records outlive a cancelled call. See Producing. Open question 3 asks whether records should follow the caller's context instead.
• DeliveryTimeout is a context on the call, not kgo.RecordDeliveryTimeout. franz-go checks its own timeout only at safe points, which in the prototype made a 10s timeout take 20s. The call's context fails the records when it says.
• The breaker counts the cluster, not the records or the callers. A record's own error is Ignored, and a caller giving up never reaches the classifier.
• Refused credentials fail every call. The consumer stops on them. A Producer can't stop the App, so it says so on every call.
• No panic is recovered. No code of the end user's runs in the produce path: there is no promise, partitioner or hook of theirs. A panic in Produce's caller is the caller's.
• One new dependency, circuit. It is bedrock's, with no dependency of its own, and the charter assigns it to the clients. retry isn't used, so it isn't added.

Testing

These tests are sketches and have not been run, apart from the franz-go behaviour listed in Background, which the prototype checked. Every test runs inside a testing/synctest bubble, on humustest's Host and Start, with a kfake cluster built in the bubble with kfake.ListenFn(h.Listen), and asserts with testify/require. kfake runs in the bubble, as the consumer's tests already show. The helpers cluster, committed and settings are the consumer's test helpers. readAll is a franz-go client that dials h.Dial and polls a topic to its end. spanNamed finds a recorded span by name.

A dead letter is stored before the consumed record is committed, and the trace runs through it:

func TestInit_produces_a_dead_letter_before_committing(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		h := humustest.NewHost(t)
		c := cluster(t, h, 1, kfake.SeedTopics(1, "orders.dlq"))
		produce(t, h, c, 1) // a record whose value isn't an order

		tel := humustest.NewTelemetry()
		p, status := humustest.Start(t, Init, // the Init in the Summary
			humustest.WithApp(humus.WithHost(h.Host()), humus.WithTelemetry(tel.Builder())),
			humustest.WithEnv(map[string]string{
				"KAFKA_BROKERS":  strings.Join(c.ListenAddrs(), ","),
				"KAFKA_GROUP_ID": "orders-consumer",
				"KAFKA_TOPICS":   "orders",
			}),
		)
		synctest.Wait() // processed, dead-lettered and committed
		p.Signal(syscall.SIGTERM)
		require.Zero(t, (<-status).Code())

		require.Equal(t, int64(1), committed(c)[0])
		dead := readAll(t, h, c, "orders.dlq")
		require.Len(t, dead, 1)

		send, process := spanNamed(t, tel, "send orders.dlq"), spanNamed(t, tel, "process orders")
		require.Equal(t, process.SpanContext().SpanID(), send.Parent().SpanID())
		require.Equal(t, trace.SpanKindProducer, send.SpanKind())

		// The dead letter carries the send span's context, not the copied one.
		var traceparents []string
		for _, hdr := range dead[0].Headers {
			if hdr.Key == "traceparent" {
				traceparents = append(traceparents, string(hdr.Value))
			}
		}
		require.Len(t, traceparents, 1)
		require.Contains(t, traceparents[0], send.SpanContext().SpanID().String())
	})
}

The breaker opens after five failed calls, and the sixth fails at once:

func TestProduce_fails_at_once_once_the_breaker_opens(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		h := humustest.NewHost(t)
		c := cluster(t, h, 1)
		answerProduce(c, kerr.NotEnoughReplicas.Code) // every Produce, a retriable error

		tel := humustest.NewTelemetry()
		var errs []error
		var took []time.Duration
		init := func(_ context.Context, s *humus.Setup) ([]humus.Service, error) {
			ps := kafka.DefaultProducerSettings()
			ps.Brokers = c.ListenAddrs()
			pr, err := kafka.NewProducer(s, ps)
			if err != nil {
				return nil, err
			}
			return []humus.Service{job.NewService(s, func(ctx context.Context) error {
				for range 6 {
					start := time.Now()
					errs = append(errs, pr.Produce(ctx, kafka.Record{Topic: "orders", Value: []byte("x")}))
					took = append(took, time.Since(start))
				}
				return nil
			})}, nil
		}
		_, status := humustest.Start(t, init,
			humustest.WithApp(humus.WithHost(h.Host()), humus.WithTelemetry(tel.Builder())))
		require.Zero(t, (<-status).Code())

		for i := range 5 {
			require.ErrorIs(t, errs[i], context.DeadlineExceeded)
			require.Equal(t, 10*time.Second, took[i]) // DeliveryTimeout, in fake time
		}
		require.ErrorIs(t, errs[5], circuit.ErrRejected)
		require.Zero(t, took[5])
		require.Len(t, logsWithBody(tel, "circuit breaker open"), 1)
	})
}

Produce returns when its context is done, and the record is flushed in OnStop:

func TestProduce_returns_when_ctx_is_done_and_OnStop_flushes(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		h := humustest.NewHost(t)
		c := cluster(t, h, 1)
		release := make(chan struct{})
		holdProduce(c, release) // SleepControl until release is closed

		var err error
		var took time.Duration
		init := func(_ context.Context, s *humus.Setup) ([]humus.Service, error) {
			ps := kafka.DefaultProducerSettings()
			ps.Brokers = c.ListenAddrs()
			pr, nerr := kafka.NewProducer(s, ps)
			if nerr != nil {
				return nil, nerr
			}
			return []humus.Service{job.NewService(s, func(ctx context.Context) error {
				ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
				defer cancel()
				start := time.Now()
				err = pr.Produce(ctx, kafka.Record{Topic: "orders", Value: []byte("held")})
				took = time.Since(start)
				close(release) // the cluster answers once the call has given up
				return nil
			})}, nil
		}
		_, status := humustest.Start(t, init, humustest.WithApp(humus.WithHost(h.Host())))
		require.Zero(t, (<-status).Code()) // OnStop's flush waited for the record

		require.ErrorIs(t, err, context.DeadlineExceeded)
		require.Equal(t, 2*time.Second, took)
		require.Len(t, readAll(t, h, c, "orders"), 1) // it was stored all the same
	})
}

The tests also show the following:

• DeliveryTimeout bounds a call against a hung broker. With the broker holding every produce request and no deadline on ctx, Produce fails after exactly DeliveryTimeout, not the 20s franz-go's own timeout took.
• A caller giving up doesn't fail another's records. Two calls to the same partition, held by the broker, one with a 1s deadline: the first returns at 1s, and the second succeeds once the broker answers.
• Refused credentials. Against kfake with SASL and a wrong password, generated with crypto/rand.Text like the right one, the first Produce fails with an error matching kerr.SaslAuthenticationFailed, naming the broker, rather than its deadline. One ERROR record is emitted, and a second call fails at once.
• A record's own failure doesn't open the breaker. Ten calls with a value larger than ProducerBatchMaxBytes fail with MESSAGE_TOO_LARGE, and an eleventh, valid call succeeds.
• The breaker closes again. After OpenFor, three successful probes close it, with an INFO record for half-open and one for closed.
• Order. Ten records with one key, in one call, are stored in order on one partition.
• The instruments. messaging.client.sent.messages and messaging.client.operation.duration are recorded per record with their attributes, and with error.type for a failed one. A rejected record is in neither. kotel's messaging.kafka.produce_records.count counts the records by broker and topic.
• The grace period. With the broker holding a record and HUMUS_SHUTDOWN_GRACE_PERIOD=2s, the OnStop flush is cut short, the record fails with kgo.ErrClientClosed, and the App exits 1 with a *humus.GracePeriodError.
• Init failing after NewProducer. The client is closed by its OnStop function, and synctest finds none of franz-go's goroutines left when the bubble ends.
• The settings convention. Each new check accepts an in-range value and rejects an out-of-range one without naming the field. NewProducer names every bad field together. DefaultProducerSettings() with Brokers set passes. The package still reads no environment and no file, and the hygiene tests cover the new files.
• No globals. With OpenTelemetry's global providers and propagator set to recorders of their own, a produce records nothing in them.

answerProduce, holdProduce, readAll, spanNamed and logsWithBody are a few lines each in the tests, like the consumer's answerOffsetCommit.

Against Kafka. The container tests of #731, in their own CI job, add one test: a Producer produces to apache/kafka-native with idempotence and acks=all, and the consumer's Service consumes the records back, in order, each with its process span the child of its send span. kfake accepts idempotent produce requests, but the producer ID, the sequence numbers and the epoch bump after a cancelled record are worth one run against the real broker.

The package's example becomes a consumer that produces dead letters, and the module stays under the 80% coverage gate.

Migration from today

Before the reset	With this proposal
no producer in humus; the examples' tools called kgo.NewClient and ProduceSync	kafka.NewProducer(s, settings) and Produce(ctx, records...)
the real network	s.Host.Dial, with the consumer's TLS dialer
no telemetry on producing	a send <topic> span per record, its context in the headers, the messaging conventions' instruments, kotel's produce counters
a producer's retries were franz-go's defaults, unbounded	franz-go's retries, bounded by DeliveryTimeout, behind a circuit breaker
	kafka.Settings and kafka.Service are unchanged

Open questions

1. An asynchronous Produce. Keep one synchronous call, or also offer one that takes a callback for each record's result? An asynchronous call produces more per goroutine, but its callback runs on the client's goroutine, where blocking stalls every producer of the process, and a Processor that used it could commit a record whose output is later lost. Recommendation: synchronous only. A caller that wants not to wait runs Produce in a goroutine of its own.
2. The connection fields. Repeat Brokers, TLS and SASL in ProducerSettings, as proposed, or move them into a type, say Connection, that Settings and ProducerSettings both embed? Embedding keeps settings.Brokers working, but breaks every kafka.Settings{Brokers: …} literal and every message that names Settings.Brokers, to save the end user three assignments. Recommendation: repeat them.
3. Records outlive a cancelled call. When a caller's ctx is done first, Produce returns at once and its records stay in the client until they are acknowledged or DeliveryTimeout passes, as proposed. The alternative is to hand franz-go the caller's context, so the records fail when the caller gives up. But franz-go fails a batch, and every record buffered behind it on that partition, by the context of the batch's first record, so one caller's cancelled request would fail other callers' records. Recommendation: records outlive the call. Either way, a record that failed may have been stored.

Out of scope

• Transactions and exactly-once. Consuming, producing and committing in one Kafka transaction would need the consumer to commit through the producer's transaction, which changes queue's contract. A Processor gets at-least-once end to end by producing before it returns nil, and handles duplicates downstream as it already does.
• Sharing one client between the consumer and the Producer. franz-go can consume and produce on one client, but the consumer's client lives for one Serve and the Producer's for the whole App. Two clients cost a second set of connections.
• Serialization and schema registries. Key and Value are bytes, as in Message.
• Creating topics and other admin requests. A topic that doesn't exist fails its records with UNKNOWN_TOPIC_OR_PARTITION, unless the cluster creates topics itself.
• A dead-letter topic convention. Where dead letters go, and which headers they carry, stays the end user's, as Proposal: queue processing and the Kafka runtime #683 decided.
• Health. No client feeds health, as the charter's revision 7 decides: an open breaker doesn't make the pod unready. A readiness monitor for the cluster is the end user's, as health monitors and probes left a dependency's health.
• Failure injection. Its injection point, its stripping of scenarios, and whether an injected abort counts against the breaker: Proposal: failure injection (wave 5), at the seam named above.
• The walkthroughs. A Kafka processor walkthrough that produces dead letters: Proposal: walkthroughs and a humus MIGRATING.md (wave 6).
• RabbitMQ publishing. Proposal: RabbitMQ (wave 4) decides whether it publishes.
• The consumer's late commit after a fence. It is the consumer's: #757, fixed by story(kafka): a member fenced out of its group commits nothing more for the partitions it held #758.

References

• Proposal: humus on the bedrock process model, the charter, and its revision 7: the client proposals, the three points they share, and failure injection's seams
• Proposal: queue processing and the Kafka runtime, and its revision: "What waves 4 and 5 build on this"
• The configuration revision: Settings, DefaultSettings, Check<Field>
• Proposal: OpenTelemetry without globals: telemetry.SDK, and outbound instrumentation left out of scope
• Proposal: humus.Main: the grace period, and OnStop after every Service
• The landed consumer: settings.go, dial.go, carrier.go, service.go, logger.go
• bedrock's circuit and retry at v0.24.0, and their discussions, A circuit breaker should be provided by bedrock bedrock#595 and A retry package owned by bedrock bedrock#606; fit and Feature request: request-scoped failure injection (Netflix FIT) on OpenTelemetry baggage bedrock#578 for wave 5
• franz-go v1.22.2's kgo producer options, among them AllowIdempotentProduceCancellation, RecordDeliveryTimeout and ProduceSync, and kotel v1.7.1's Meter
• OpenTelemetry's semantic conventions for messaging spans, for Kafka and for messaging metrics
• Before the reset: example/queue/kafka-at-least-once/tool/main.go

[Zaba505]

Revision 1: the open questions answered

The open questions:

1. Produce stays synchronous, with no asynchronous call, as recommended. A smaller API leaves room to change the client under it later. At franz-go's level the two cost the same. ProduceSync is Produce with a promise that waits on a sync.WaitGroup, so both go through the same buffers, batches, requests in flight, acks, retries and buffer limits. Background below says what differs. A caller that wants more throughput puts more records in one call, or calls from more goroutines. An asynchronous Produce goes out of scope.
2. Brokers, TLS and SASL are repeated in ProducerSettings, as recommended. A Producer may produce to a cluster other than the one the consumer reads, so each set is sourced on its own, and Settings doesn't change.
3. Records outlive a cancelled call, as recommended. Produce now also honours ctx up to the moment it hands its records to franz-go: a call whose ctx is already done sends nothing. After the hand-off, franz-go can't cancel one caller's records without failing other callers' records and cutting the broker's produce connection, as Background now says. So a caller that gives up gets its answer at once, and its records go on to their result.

Each section below replaces or adds to the part of the same name in the post. Retries and the breaker, Shutdown, Refused credentials, The seam for failure injection, Changes to the rest of the module, Migration from today and References are unchanged.

Summary

The second bullet becomes:

• Produce(ctx, records...) is synchronous. It returns once the cluster has acknowledged every record, with acks from all in-sync replicas, or once one has failed. A Processor that produces before it returns nil has the consumed record committed only after the produced one is stored. A call whose ctx is already done sends nothing. A call whose ctx ends while it waits returns at once, and its records go on to their result.

Background

These are added under "franz-go's produce path". They were read in franz-go v1.22.2's source, not prototyped:

• ProduceSync hands each record to Produce with a promise that appends the result and calls Done on a sync.WaitGroup. It then stops the linger on every partition it produced to, so those are sent at once. The code that stops the linger is unexported (producer.go#L380). A record handed to Produce alone lingers up to ProducerLinger, unless one of three things comes first: a batch fills, another partition on the same broker starts a request, or the client reaches MaxBufferedRecords.
• Both go through the same buffers, the same batches and the same produce requests. With idempotence there are at most five requests in flight per broker. Under load, records pile up while five are in flight, and the next request takes them all, so a synchronous caller's records are batched too.
• Every promise runs on one goroutine of the client, one at a time.
• franz-go decides whether to fail a batch from its first record's context alone, and failing a batch fails every record buffered on its partition (sink.go#L2054, sink.go#L1884).
• With kgo.AllowIdempotentProduceCancellation, a produce request in flight is cancelled by the first cancellable context among its batches' first records (sink.go#L593). On the produce connection, a cancelled read closes the connection and fails every request in flight on it (broker.go#L2006). Those requests are sent again after a new dial, TLS handshake and SASL authentication.
• Failing a batch that was sent bumps the client's producer epoch (sink.go#L1921). The next produce, on any broker, waits for an InitProducerID, and every partition restarts its sequence numbers.

Proposed API

Produce's doc becomes:

// Produce produces records, and returns nil once the cluster has
// acknowledged every one. Records to one partition are stored in the order
// they are given.
//
// It returns an error, naming each record that failed by its topic, when a
// record failed or ProducerSettings.DeliveryTimeout passed; an error that
// matches circuit.ErrRejected, at once and with nothing sent, while the
// breaker is open; and an error that wraps ctx's error, at once, when ctx is
// done first. A call whose ctx is already done sends nothing. Records handed
// to the client before ctx is done go on to their result, within
// DeliveryTimeout, and never fail another call's records. A record that
// failed may still have been stored, so a caller that produces it again may
// store it twice.
//
// Produce rejects a record with no Topic before it sends anything. With no
// records, it returns nil.
func (p *Producer) Produce(ctx context.Context, records ...Record) error

Producing

In A call, step 1 becomes:

rejects a record with no Topic, and returns at once, with nothing sent, when ctx is already done, the client was closed or the cluster refused the credentials;

and step 5 becomes:

calls franz-go's ProduceSync with the records and the call's context, on a goroutine of its own, and waits for its results or for ctx, whichever comes first.

Produce calls ProduceSync, not franz-go's Produce, for the unlinger, which is unexported. Each record's span and measurements end in an OnProduceRecordUnbuffered hook as that record's result arrives, not when the last one does.

Records outlive a cancelled call becomes:

Records outlive a cancelled call. Produce honours ctx up to the hand-off to franz-go. A call whose ctx is already done sends nothing, starts no span and doesn't reach the breaker. After the hand-off, a ctx that ends returns the call at once, with an error that wraps ctx's error. Its records stay in the client, with the call's context, until the cluster acknowledges them or DeliveryTimeout fails them, and their spans and measurements are recorded then. So a handler whose request is cancelled returns at once, and the grace period reaches a Process blocked in Produce, as the lifecycle requires.

The records don't follow ctx after the hand-off, because franz-go can't cancel one caller's records on their own:

• A sent record can't be taken back. The broker stores a request it has read, whether the client still waits or not. Cancelling the request changes what the caller is told, not what is stored.
• Only a batch's first record counts. A cancelled record in the middle of a batch is sent anyway.
• One cancelled record fails its partition. franz-go fails every record buffered behind it on that partition, whoever produced them.
• A cancelled request cuts the connection. It closes the broker's produce connection, and fails every request in flight on it, up to five, for every partition that broker leads. They are sent again after a new dial, TLS handshake and SASL authentication.
• It bumps the epoch. The next produce, on any broker, waits for an InitProducerID.

One Producer serves the whole process. If a caller's ctx reached franz-go, any caller could fail any other caller's records. For example, an HTTP client that disconnects could fail a dead letter that a Processor buffered behind that handler's event on the same partition. The Processor would get a context.Canceled that isn't its own. Under queue.AtLeastOnce that error ends the consumer, and the App exits 1. A handler that reads context.Canceled as "the client went away", and so doesn't log it, would hide a real failure. And the breaker's classifier would have to tell a caller's own cancellation from another caller's.

This is added to DeliveryTimeout bounds what is in flight:

When DeliveryTimeout passes, the call's context does what a cancelled ctx would. It fails its partition's buffered records, closes the produce connection when its request is in flight, and bumps the epoch. That happens only once the cluster hasn't stored a record for that long, which the breaker counts as a failure anyway, so it follows the cluster's failures, not a caller's choices.

Settings

The bullet on the connection fields becomes:

• Brokers, TLS and SASL are the consumer's fields, repeated. They mean the same, take the same check, and dial through the same unexported dialer. A Producer may produce to a cluster other than the one the consumer reads, so each set is sourced on its own. A process that consumes and produces on one cluster copies the three into ProducerSettings, or reads the same variables, as the example below does. Settings is unchanged, so no Init breaks.

Telemetry

This is added after the paragraph on the two instruments:

A call whose ctx was already done sends nothing and starts no span, so neither instrument counts it.

Design decisions

The bullets "Synchronous only" and "Records outlive a cancelled call" become:

• Synchronous only. A record is either stored or failed when Produce returns, which is what at-least-once needs: a Processor that produced before returning nil never commits a record whose output was lost. A smaller API also leaves room to change the client under it. A synchronous call costs no throughput at franz-go's level: apart from the linger, it takes the same path as an asynchronous one. A caller that wants more throughput puts more records in one call, or calls from more goroutines.
• Records outlive a cancelled call, and a done ctx sends nothing. ctx is honoured wherever it can be without touching other callers' records: before the hand-off, and in deciding when Produce returns. See Producing.

These are added:

• The connection fields are repeated. A Producer may produce to another cluster than the consumer's. A type that both embed would break every kafka.Settings{Brokers: …} literal and every message that names Settings.Brokers, to save three assignments.
• Rejected: an asynchronous Produce. Its callback runs on franz-go's one promise goroutine, where blocking stalls every producer of the process. A Processor that used it could commit a record whose output is later lost. It would also tie the API to franz-go's promises.
• Rejected: records that follow the caller's ctx. franz-go decides from a batch's first record, fails a whole partition's buffered records at once, and closes a broker's produce connection when it cancels a request in flight. One caller giving up would fail other callers' records, and a record already sent would be stored anyway.

Testing

These bullets are added:

• A call whose ctx is already done sends nothing. With ctx cancelled before the call, Produce returns an error matching context.Canceled at once. The topic stays empty, no send span is recorded, and six such calls in a row leave the breaker closed.
• A caller giving up keeps the connection. With the broker holding a produce request, a call whose ctx ends at 1s returns then. Once the broker answers, kotel's messaging.kafka.connects.count for that broker is still one.

Open questions

All three are answered above. None are left.

Out of scope

This is added:

• An asynchronous Produce. It is reconsidered only when end users report needing it, by a proposal of its own.

[Zaba505]

Closing: the Kafka Producer has landed

The revision's stories are done:

• story(kafka): produce records synchronously to a cluster with kafka.Producer #775 produced records synchronously to a cluster with kafka.Producer, including the check that sends nothing for a ctx that is already done (story(kafka): produce records synchronously to a cluster with kafka.Producer #781);
• story(kafka): trace and measure every produced record with the Setup's telemetry #776 traced and measured every produced record with the Setup's telemetry, with a send <topic> span whose context goes into the headers (story(kafka): trace and measure every produced record with the Setup's telemetry #784);
• story(kafka): a Producer whose credentials a broker refuses fails every produce at once #777 made a Producer whose credentials a broker refuses fail every produce at once, sharing the connect hook with the Service (story(kafka): a Producer whose credentials a broker refuses fails every produce at once #787);
• story(kafka): a circuit breaker fails produces at once while the cluster keeps failing #778 put a circuit breaker in front of produces while the cluster keeps failing, with Breaker and BreakerPolicy (story(kafka): a circuit breaker fails produces at once while the cluster keeps failing #788);
• story(kafka): produce to Kafka itself in the container tests #779 produced to Kafka itself in the container tests, with an idempotent Producer and acks=all (story(kafka): produce to Kafka itself in the container tests #790);
• story(docs): document producing to Kafka #780 documented producing to Kafka. The Kafka page gained a Producing section, and the package example now produces dead letters (story(docs): document producing to Kafka #792).

The revision left no open questions. Produce is synchronous only, ProducerSettings repeats Brokers, TLS and SASL, and records outlive a cancelled call.

The rest is out of scope or belongs to other proposals, as "Out of scope" lists:

• Failure injection's seam is Proposal: failure injection (wave 5).
• A Kafka walkthrough that produces dead letters is Proposal: walkthroughs and a humus MIGRATING.md (wave 6).
• RabbitMQ publishing is Proposal: RabbitMQ #765's.
• An asynchronous Produce waits for end users to report needing it.