Produce with acks=0 no longer gets an unsolicited response; failure closes the connection

[solace-aross]

Bug

A Kafka producer configured with acks=0 never reads a response. Nisshi's broker wrote one anyway for every Produce request. On success this served no purpose; on failure, the broker silently dropped the batch instead of signaling anything. Worse, the unsolicited success response sits on the connection and corrupts correlation-id matching on the client's next request, since a real acks=0 client never consumes it.

Fix

ProduceService (for acks=0 requests only):

• On success: marks a new SuppressResponseExtension, a self-resetting AtomicBool-backed marker, so TcpBytesService::req skips writing the already-assembled response. A plain bool can't be inserted once and reused because rama's Extensions is shared (not forked) across every request on a connection and has no way to remove an entry once inserted.
• On any partition error anywhere in the assembled response, matching real Kafka's own handleProduceRequest (which closes the connection whenever unauthorizedTopicResponses/nonExistingTopicResponses/invalidRequestResponses/per-partition errors merge into a non-empty error set): returns a new Error::AcksZeroProduceFailed, reusing the existing error-closes-connection path instead of embedding the error in a response nobody reads.

Both decisions (broad scope matching every error path, not just storage; a dedicated error variant) were checked against real Kafka source and independently mutation-tested: disabling the response suppression makes all 4 success tests fail with a stray Produce response's correlation id showing up where it shouldn't, and narrowing the error scope back to storage-only makes all 4 failure tests fail with a timeout since the connection incorrectly stays open. Both were restored to green afterward.

broker.rs logs the new failure variant at info, matching real Kafka's own "Closing connection due to error during produce request..." level, instead of falling through to the generic error! path. A pre-existing deviation was also checked in this review: TcpBytesService::process's error! on a connection failure was deliberately left alone rather than downgraded to debug!, because TcpListenerService (the proxy's own accept loop) only logs connection failures at debug! already -- downgrading the broker side too would remove the proxy operator's only error-level signal for this class of failure.

Proxy batch-forwarding fix (included here)

Audited every existing acks=0 call site in the repo. nisshi-generator and nisshi-perf both built a default (acks=0) ProduceRequest and awaited a response that would now never arrive from a nisshi broker, so both now request Ack::Leader explicitly.

One more call site needed the same fix: nisshi-proxy/src/produce.rs's produce_request(), used by the batch-forwarding path (tansu.batch=true), also built a default ProduceRequest with no acks(...). send_pending_batch awaits the origin's response to fan out base offsets to the waiting downstream clients, so forwarding with acks=0 against an origin that correctly suppresses its response (this PR's fix) would hang that await forever. Fixed by requesting Ack::Leader explicitly, same pattern as generator/perf -- the batch path needs the origin's real response regardless of what kind of origin is on the other end.

Deferred: proxy pass-through case

The proxy's plain pass-through forwarding (not the batching path above) goes through nisshi-client's BytesConnectionService (reached via ConnectionManager), which does write_all then an unconditional read_exact for every forwarded request. Forwarding a pass-through acks=0 Produce to an origin that correctly never responds (a nisshi origin after this fix) will hang on that read_exact.

This is a known, deliberately-deferred consequence of this fix for the pass-through path specifically, not something newly discovered and unrelated. Fixing it properly means adding acks-awareness to BytesConnectionService itself, a separate, larger piece of work, tracked as a separate follow-up (description corrected to name the right component and framing).

Tests

• nisshi-broker/tests/it/produce_acks_zero.rs: real-socket tests across all 4 storage engines (in-memory, libSQL, Postgres, SlateDB).
  • success_is_not_written: sends an acks=0 Produce then a Metadata request, asserts the first (and only) frame read back carries the Metadata correlation id, not the Produce one -- positive proof nothing was interposed -- then confirms via ListOffsets(Latest) the record was still stored.
  • failure_closes_connection: a record-less batch, rejected by ProduceService::partition's pre-storage rejection() validation, asserts the connection closes.
  • storage_failure_closes_connection (new in this round): an idempotent producer id never registered via InitProducerId, exercising a genuine storage.produce()-layer failure (UnknownProducerId) rather than pre-storage validation, across all 4 engines.
• Bumped 5 produce.rs tests that relied on an acks=0 response embedding an error code (their actual subject is idempotent-producer/batch-rejection error codes, not acks semantics) to a non-zero acks so they keep testing that.
• nisshi-proxy's batch tests updated to expect acks=1 (Ack::Leader) on the forwarded request.

Verified against a real client: librdkafka's compat suite's 0008_reqacks (acks=-1/0/1) passes, along with the rest of the allowlist bar one pre-existing failure, 0060_op_prio (reproduced identically against an unmodified main build) -- a legacy-consumer test that does call produce, but fails during its later consume phase, not in the produce path itself.

Local run this round:

cargo nextest run -p nisshi-broker --all-features -E 'test(/^produce_acks_zero::/)'  # 12 passed
cargo nextest run -p nisshi-broker --all-features -E 'test(/^produce::/)'            # 28 passed
cargo nextest run -p nisshi-proxy --all-features                                      # 10 passed
just fmt / just clippy / just build-all                                               # clean

CHANGELOG

The broker no longer writes a Produce response for an acks=0 request. A storage or validation failure on an acks=0 Produce now closes the connection (matching Apache Kafka) instead of silently dropping the batch.

🤖 Generated with Claude Code

https://claude.ai/code/session_01D2qdgPhMyLGZN5dLR8CVHs

[sgamelin]

The broker change looks right to me. I checked it against Kafka 3.9.1: KafkaApis.handleProduceRequest sends nothing on success and closes the connection when any partition in the merged map has an error, and ProduceRequest.getErrorResponse returns null for acks=0, so request-level failures close the connection too. Nisshi's failures before the handler runs (frame decode, unauthenticated, unknown API key) already return Err and close, so those match as well.

I'm requesting changes for the proxy, because merging this as it stands breaks acks=0 producers that go through it:

• Pass-through (topics without tansu.batch): BytesConnectionService writes the request and then waits for a response, with no timeout. A nisshi origin no longer sends one, so the proxied request stalls until the origin's connection idle timeout (600 s by default) closes the upstream connection. The same applies to anyone calling nisshi_client::Client::call with a default ProduceRequest, which is acks=0.
• Batch path: BatchProduceService returns the origin's response to the downstream client whatever acks that client asked for, so an acks=0 client behind the proxy still gets the unsolicited response this PR removes from the broker.
• Upgrade order: a proxy built before this PR forwards batches with acks=0 and waits for the reply, so it stalls every tansu.batch=true producer once its origin is upgraded. Proxies have to be upgraded before brokers.

Kafka's own client decides this per request: the producer's Sender passes expectResponse = acks != 0. A predicate in nisshi-sans-io (say Frame::expects_response) that BytesConnectionService checks before reading would fix the pass-through and the published client together. For the batch path, the downstream suppression needs per-connection state that the proxy doesn't have yet (see the inline comment on stream.rs). Checking acks in BytesFrameService, which every Produce on both the broker and the proxy passes through, would keep the rule in one place instead of in each handler.

If you'd rather land the broker change first, could you open an issue for the proxy, link it here, and add the pass-through limitation and the upgrade order to the CHANGELOG entry?

[solace-aross]

On the proxy points from the summary, in 5beae8b:

• The batch path forwards the strictest acks among batched requests, at least 1.
• CHANGELOG ### Changed notes the upgrade order: upgrade proxies before brokers, since an older proxy forwards batches with acks=0 and would stall against an upgraded broker.
• The pass-through stall is not fixed here; it needs a no-response path through nisshi_client and the proxy (a Frame::expects_response predicate, as Kafka's Sender does with expectResponse = acks != 0). The CHANGELOG lists it as a known limitation, and it needs a follow-up issue before this lands.

[solace-aross]

Rebased on main. broker.rs has both the pre-authentication arm from main and the acks=0 arm. No code change otherwise.

[sgamelin]

Thanks, this covers everything from my last review. Forking the extensions once per connection in TcpBytesService::serve fixes the cross-connection leak. acks_zero_then_acks_one_on_one_connection covers the reused marker. The batch path keeps the strictest acks, and #890 tracks the proxy pass-through and the response the proxy still returns downstream. Approving.

Two things before it merges, neither blocking:

• #853's new tests will fail once this lands, and git won't show a conflict. The produce helper in produce_invalid_record.rs sends a default ProduceRequest, which is acks=0, and asserts INVALID_RECORD in the response. With this PR, ProduceService returns Err(AcksZeroProduceFailed) for that request, so every test in that file fails at the ?. Whichever PR lands second needs .acks(-1) in that helper, as you did for the five produce.rs tests. #847's new tests already use acks=-1, so they're fine.
• The description is out of date. It still says the marker is shared across a connection's requests, not forked per connection, and that the batch path forwards Ack::Leader, not the strictest acks. The deferred section doesn't link #890. Could you update those?