fix: latch Kafka fatal-error flag so error bursts cannot mask it (#4227)

[iancooper]

Symptom

The Kafka consumer's _hasFatalError flag (and the producer's _hasFatalProducerError) is unconditionally overwritten on every librdkafka error-callback invocation. librdkafka emits errors in bursts, so a fatal error can be silently cleared by a subsequent non-fatal one before the consume/publish path observes it. When that happens the intended ChannelFailureException is never thrown and the consumer/producer keeps running against a connection librdkafka has declared unrecoverable.

Confirmed root cause

_hasFatalError = error.IsFatal; assigns rather than latches. Sequence:

1. Fatal error fires → flag true.
2. A non-fatal error fires immediately after (e.g. Local_TimedOut on an idle socket) → flag false.
3. Receive() / Send() runs, sees false, does not throw → the fatal condition is masked.

The flag is meant to be a one-way "the client is dead" latch, so it must never be reset to false by a later non-fatal error.

Evidence

Proven by code-trace and reproduced at the unit level (see tests): invoking the extracted error handler with a fatal error followed by a non-fatal one and asserting Receive()/Send() still throw ChannelFailureException. The defect is a single-threaded logic error — it reproduces without any threading.

Fix

• Latch the flag: only ever set it true on a fatal error; never clear it.
• Decouple the log branch from the latch — drive it off error.IsFatal, not _hasFatalError. Without this, once latched, every subsequent non-fatal error would be logged as fatal and the non-fatal log branch would become dead code (a defect the naive one-line fix would introduce).

Scope

• Applied to both the consumer (_hasFatalError) and the producer (_hasFatalProducerError) — the same non-latched pattern, surfaced during confirmation.
• The error-callback bodies were first extracted into public HandleError(Error) methods as a separate, behaviour-preserving refactor: commit (InternalsVisibleTo is banned in this repo, so the seam is a public method). The behavioural change is isolated in the fix: commit.
• Deliberately not changed: the fields remain non-volatile. The confirmed diagnosis established this is a single-threaded logic bug, not a memory-visibility one; volatile would be optional hardening beyond the proven cause.

Tests

• Consumer + producer regression tests: a fatal-then-non-fatal sequence still causes Receive()/Send() to throw ChannelFailureException.
• Consumer + producer log-guard tests (via Serilog.Sinks.TestCorrelator): a non-fatal error after a fatal one is still logged at Warning, locking in the log/latch decoupling.

Verification

• Regression + broker-free unit tests: green.
• Full suite against a live Kafka broker (Podman): 85/94. The 9 failures are a pre-existing timezone assertion in KafkaMessageAssertion.cs:50 (host BST vs UTC, off by exactly one hour) — unrelated to this change and a candidate for a separate issue.

Fixes #4227

[claude]

Review — PR #4229: latch Kafka fatal-error flag

Thorough, well-documented fix. I verified the core claims against the source and the diagnosis holds up. Summary: approve with a couple of considerations, none blocking.

What is correct

• Root cause is real and the fix is right. _hasFatalError = error.IsFatal; (assign) becomes if (error.IsFatal) _hasFatalError = true; (latch), correctly turning the flag into the one-way "client is dead" latch that Receive() (KafkaMessageConsumer.cs:486) and Send()/SendWithDelayAsync() (KafkaMessageProducer.cs:264, :340) already assume.
• Decoupling the log branch from the latch is the non-obvious detail that makes this a good fix, not a naive one. Driving the log off error.IsFatal rather than the latched field keeps NonFatalError reachable after a fatal error. I confirmed the levels: FatalError/FatalProducerError = LogLevel.Error, NonFatalError/NonFatalProducerError = LogLevel.Warning (KafkaMessageConsumer.cs:1232-1236), so the Warning-level assertions in the log-guard tests are pinned to the right thing.
• The log-guard tests are a nice touch — green today but red against the naive one-line fix, locking in the decoupling requirement. That kind of test protects intent, not just current behavior.
• public HandleError seam is justified. I confirmed InternalsVisibleTo is genuinely not a pattern here (the only hit in src/ is a comment in SpannerBoxMigrationRunner.cs:129 explaining its avoidance). Extraction is behaviour-preserving and the XML doc flags it as not-for-direct-use.
• Scope discipline is good — parity fix across consumer + producer, both sync and async producer guards covered, and the bugfixes/000N/ artifacts follow the established repo convention (0001-0003 already tracked).

Considerations (non-blocking)

1. The "single-threaded, so volatile is unnecessary" argument is weaker for the producer than for the consumer. For the consumer it is solid: confluent-kafka-dotnet dispatches the error callback synchronously inside Consume(), i.e. on the same thread that later reads the latch. But the producer serves error callbacks on librdkafka background poll thread (the producer is not pumped by the app thread the way the consumer is), so the write (callback thread) and the read in Send() (app thread) are genuinely cross-thread with no barrier. The bug is still a logic bug that reproduces single-threaded, and the field was already non-volatile, so this fix does not make anything worse — but the "correctness does not depend on volatile" claim does not fully carry over to the producer. Reasonable to defer; worth an accurate note in the follow-up.

2. Test collection/category placement. The four tests are broker-free (they exercise HandleError + the guard directly) but carry [Collection("Kafka")] / [Trait("Category","Kafka")], which serializes them with the live-broker integration tests and gates them behind the Kafka category in CI. Since they need no broker, a lighter category would let them run in the fast unit lane. Minor.

3. Micro-nit (ignore if intentional): the two consecutive if (error.IsFatal) blocks in each HandleError could be one block, but keeping them separate documents "latch" vs "log" as distinct concerns — fine as-is.

Verdict

Correct fix, good tests, honest write-up (including flagging the pre-existing KafkaMessageAssertion.cs:50 timezone failures as unrelated — a host BST/UTC issue, not caused by this change). Ship it; consider the producer thread-visibility note for a follow-up.

Reviewed by Claude — automated PR review.