Bug

The Kafka async producer's worker accumulates messages into a local batch buffer and flushes on size/linger. On a graceful Stop(), that buffer was silently dropped:

• Stop() sets shuttingDown=true before closing the publish channel.
• flushBuffered — including the final drain triggered on channel-close — bailed early on shuttingDown.

So up to a full linger window / batch of messages (txmeta, blocks-final, rejected-tx, …) was lost on every graceful shutdown, across every service using the producer. It's a logic drop, not a panic, so recover() doesn't help. (Found during a concurrency/lifecycle audit; the send-on-closed race that #1009 patched is a different, already-handled issue — this is the buffered batch behind it.)

Two compounding causes, both removed:

1. flushBuffered returned early on shuttingDown, defeating the final drain — and also discarding any batch the size/linger paths cleared right after that no-op flush.
2. The worker loop breaked the instant shuttingDown was observed, which could also strand messages still queued in the channel.

Fix

flushBuffered now gates only on closed (client gone), not shuttingDown. This is provably safe: Stop closes the client only after publishWg.Wait() — i.e. after the worker has returned — so producing while shutting-down is always valid, and the subsequent client.Flush() delivers it. The worker now keeps draining until Stop closes the channel; the close drains channel-resident messages into the buffer and the final drain produces them.

Steady-state behaviour is unchanged (shuttingDown is false then, so the removed gate never fired).

Testability

The buffered loop is extracted into runProducerWorker (behaviour-preserving move) and a produceHook test seam captures what the loop emits without a live broker (the in-memory producer path is separate and doesn't reproduce the bug, and project policy is no kafka mocking).

TestAsyncProducer_DrainsBufferedBatchOnStop: linger/batch set huge so messages only ever leave via the final drain; asserts all are produced on Stop.

Verification

• New test fails on the old code (0 of 25 produced), passes on the fix.
• Full util/kafka suite passes under -race (81s).
• go vet, gofmt, gci clean.

Note: the large line count in kafka_producer_async.go is mostly the runProducerWorker extraction (the loop moved out of Start), not new logic.