KAFKA-14275: KRaft Controllers should crash after failing to apply any metadata record

[niket-goel]

Make all faults in metadata processing on standby controllers be fatal. This is the same behavior-wise as the active controller.

[cmccabe]

Thanks, @niket-goel.

• I'd prefer to keep the name fatalFaultHandler since it accurately describes what the handler does. It makes it clear that you're terminating the process (when not in test context) which is good to know

• as per our earlier discussion, we shouldn't need to remove metadataFaultHandler from ControllerServer, etc. since we'll still be using it there (later). Just from QuorumController, at the moment. This should also make the PR smaller

• as you mentioned offline, we do need a test for this. it shouldn't be too hard to get a bad record committed and then wait for everyone to invoke the FaultHandler. I believe there is a test that does this already, but maybe just for the active. I can't recall if we use the same fault handler for everyone when in testkit -- if so, we can just wait for that this thing to be called 3 times (or whatever)

[niket-goel]

Thanks for the review @cmccabe

I'd prefer to keep the name fatalFaultHandler

Agreed! I don't think I changed that. Please let me know if I did.

Just from QuorumController, at the moment

Sounds good. I will add it back in.

if so, we can just wait for that this thing to be called 3 times (or whatever)

Yeah, I guess that is one way of doing it. Will play around to get a test in the next iteration.

[niket-goel]

The two tests that are failing in the build pass locally reliably:

kafka.api.TransactionsTest.testCommitTransactionTimeout(String).quorum=zk
org.apache.kafka.controller.QuorumControllerTest.testFenceMultipleBrokers()

[hachikuji]

Left a minor comment. LGTM overall.

[hachikuji]

nit: perhaps better to call this INVALID_RECORD with a brief comment mentioning the topicId is not set. Also, since it's not used in any other tests, maybe we can keep it local to testFatalMetadataReplayErrorOnStandbys? It's always possible to factor it out if it does get used in other tests.

[hachikuji]

LGTM

[niket-goel]

The failed tests (5) on the most recent version of the PR seem unrelated and flaky.

[niket-goel]

Two different tests failing this time (timeout and metric not updated)

org.apache.kafka.common.network.SslTransportLayerTest
org.apache.kafka.streams.processor.internals.DefaultStateUpdaterTest