[
https://issues.apache.org/jira/browse/KAFKA-19785?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18122115#comment-18122115
]
Rishi Kant Rana commented on KAFKA-19785:
-----------------------------------------
h3. Root cause (verified against current trunk)
The crash signature in this ticket (e.g. {{Received request or response with
leader OptionalInt[3] and epoch 55 which is inconsistent with current leader
OptionalInt.empty}}) comes from {{KafkaRaftClient.maybeTransition}}:
{code:java}
private void maybeTransition(
OptionalInt leaderId,
int epoch,
Endpoints leaderEndpoints,
long currentTimeMs
) {
if (!hasConsistentLeader(epoch, leaderId)) {
throw new IllegalStateException("Received request or response with
leader " + leaderId +
" and epoch " + epoch + " which is inconsistent with current leader
" +
quorum.leaderId() + " and epoch " + quorum.epoch());
}
...
}
{code}
and {{hasConsistentLeader}}:
{code:java}
private boolean hasConsistentLeader(int epoch, OptionalInt leaderId) {
if (leaderId.isPresent() && leaderId.getAsInt() ==
quorum.localIdOrSentinel()) {
// The response indicates that we should be the leader, so we verify
that is the case
return quorum.isLeader();
} else {
...
}
}
{code}
So whenever an inbound request or response asserts "the leader is you" for a
node that does *not* currently believe it is the leader, this throws
unconditionally.
h3. Why this is fatal, not just logged
{{maybeTransition}} is invoked from the raft I/O thread. Any exception it
throws propagates out of {{KafkaRaftClient.poll()}} to
{{KafkaRaftClientDriver.doWork()}}:
{code:java}
} catch (Throwable t) {
throw fatalFaultHandler.handleFault("Unexpected error in raft IO thread",
t);
}
{code}
In production broker/controller wiring ({{SharedServer.scala}}),
{{raftManagerFaultHandler}} is built from {{ProcessTerminatingFaultHandler}}.
So this {{IllegalStateException}} does not just fail one request — it
terminates the broker/controller process. That matches every reporter's symptom
in this ticket: the node crashes, restarts, and (per the metrics quoted in the
description) can come back up fenced with no leader known, requiring manual
intervention.
h3. This is reachable without any data corruption or misconfiguration
There are two call sites with no epoch-equality guard at all before
{{maybeTransition}}: {{handleBeginQuorumEpochRequest}} and
{{handleEndQuorumEpochRequest}}. Both validate the incoming request only via:
{code:java}
private Optional<Errors> validateVoterOnlyRequest(int remoteNodeId, int
requestEpoch) {
if (requestEpoch < quorum.epoch()) {
return Optional.of(Errors.FENCED_LEADER_EPOCH);
} else if (remoteNodeId < 0) {
return Optional.of(Errors.INVALID_REQUEST);
} else {
return Optional.empty();
}
}
{code}
Note this only rejects *older* epochs. A request where {{requestEpoch ==
quorum.epoch()}} and {{requestLeaderId == localId}} sails through this check
and falls straight into {{maybeTransition}}. If the local node does not
currently consider itself the leader of that epoch (e.g. it has already moved
to {{ResignedState}} — where {{isLeader()}} is {{false}} even though
{{quorum.epoch()}} has *not* been bumped yet), this is fatal.
This is a real, unprivileged distributed-systems hazard, not an edge case
requiring a buggy/malicious peer: a duplicate or delayed
{{BeginQuorumEpoch}}/{{EndQuorumEpoch}} request (completely normal under
retries, slow networks, or partitions) that still names this node as leader of
the current epoch is exactly the kind of message every Raft implementation has
to tolerate as "stale" rather than treat as an invariant violation.
Notably, the Kafka maintainers are already aware this class of hazard exists —
see the comment in {{pollResigned}}:
{code:java}
// The replica stays in resigned state for an election timeout period to allow
end quorum requests
// to be processed, and to give other replicas a chance to become leader. When
transitioning out
// of resigned state, the epoch must be increased to avoid FETCH responses with
the leader
// being this replica, and to avoid this replica attempting to transition into
follower state with
// itself as the leader.
transitionToUnattached(quorum.epoch() + 1, OptionalInt.empty());
{code}
That epoch bump specifically dodges the hazard for the Resigned → Unattached
transition (any stale "leader=self" message at the old epoch is afterwards
rejected gracefully as {{FENCED_LEADER_EPOCH}}, since the epoch moved on). But
it only covers that one transition, and only once the election timeout has
elapsed. For the window *while still in* {{ResignedState}} (same epoch,
{{isLeader() == false}} already), and for any other path that reaches
{{maybeTransition}} with a same-epoch self-referential leaderId, there is no
equivalent protection — the fatal throw is still the only outcome.
I also want to flag:
{{raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java}} has no
test coverage at all exercising this throw, which suggests it was never
deliberately exercised as intended behavior — it reads as an unguarded internal
assumption rather than an intentional fail-fast decision.
h3. Proposed fix directions (looking for maintainer input before writing a
patch)
# *Make the inconsistency recoverable instead of fatal.* Every other
stale/unexpected condition nearby in this same code ({{UNKNOWN_LEADER_EPOCH}},
{{FENCED_LEADER_EPOCH}}, {{NOT_LEADER_OR_FOLLOWER}}, {{INVALID_VOTER_KEY}}) is
handled by logging and returning/ignoring, not by throwing. Replacing the
{{throw}} in {{maybeTransition}} with a logged warning + no-op (dropping the
inconsistent request/response) would bring this in line with the rest of the
method and remove the process-termination blast radius entirely.
# *Tighten request validation before {{maybeTransition}} is reached.*
{{validateVoterOnlyRequest}} could additionally reject (with e.g.
{{FENCED_LEADER_EPOCH}} or a new/existing error code) the specific case of
{{requestEpoch == quorum.epoch() && requestLeaderId == localId &&
!quorum.isLeader()}}, so
{{handleBeginQuorumEpochRequest}}/{{handleEndQuorumEpochRequest}} never call
into {{maybeTransition}} with a self-inconsistent leader claim in the first
place.
# Do both: (2) as defense-in-depth for the two request handlers specifically,
plus (1) so the shared {{maybeTransition}}/{{hasConsistentLeader}} path is safe
for any other caller (including the response path via
{{maybeHandleCommonResponse}}) that might hit the same condition in the future.
I'd like maintainer input on which direction (or combination) is preferred, and
whether there's a reason this was intentionally left fatal that I'm missing,
before I put together a patch + tests (including a regression test reproducing
the stale same-epoch self-leader request against a {{Resigned}}/{{Unattached}}
node).
One more note on scope: a couple of the later comments on this ticket describe
what looks like a *separate*, harder-to-pin-down contributing issue — the
on-disk {{quorum-state}} epoch appearing to regress across a restart (e.g.
"Epoch from quorum store file is 0, which is smaller than last written epoch
66"). That would make the crash reachable even more easily (a wider epoch gap
re-triggers the same {{hasConsistentLeader}} failure), but I'm treating it as a
distinct question from the crash-on-inconsistency design gap above, since I
don't have a verified root cause for the epoch regression itself yet.
> Two Kafka brokers were not active in 3 node cluster setup
> ---------------------------------------------------------
>
> Key: KAFKA-19785
> URL: https://issues.apache.org/jira/browse/KAFKA-19785
> Project: Kafka
> Issue Type: Bug
> Components: core, kraft
> Affects Versions: 4.0.0
> Reporter: Sravani
> Priority: Major
> Labels: kraft
>
> Hi Team,
> We were facing kafka issue where two of the kafka brokers were fenced and
> Kafka was not able to process messages. We are using Kafka 4.0.0 version.
> Below are the errors.
>
> Sep 22 09:41:42 host kafka[42245]: [2025-09-22 07:41:42,419] ERROR
> Encountered fatal fault: Unexpected error in raft IO thread
> (org.apache.kafka.server.fault.ProcessTerminatingFaultHandler)
> Sep 22 09:41:42 host kafka[42245]: java.lang.IllegalStateException: Received
> request or response with leader OptionalInt[3] and epoch 55 which is
> inconsistent with current leader OptionalInt.empty and epoch 55
> Sep 22 09:41:42 host kafka[42245]: #011at
> org.apache.kafka.raft.KafkaRaftClient.maybeTransition(KafkaRaftClient.java:2528)
> ~[kafka-raft-4.0.0.jar:?]
> Sep 22 09:41:42 host kafka[42245]: #011at
> org.apache.kafka.raft.KafkaRaftClient.maybeHandleCommonResponse(KafkaRaftClient.java:2484)
> ~[kafka-raft-4.0.0.jar:?]
> Sep 22 09:41:42 host kafka[42245]: #011at
> org.apache.kafka.raft.KafkaRaftClient.handleFetchResponse(KafkaRaftClient.java:1707)
> ~[kafka-raft-4.0.0.jar:?]
> Sep 22 09:41:42 host kafka[42245]: #011at
> org.apache.kafka.raft.KafkaRaftClient.handleResponse(KafkaRaftClient.java:2568)
> ~[kafka-raft-4.0.0.jar:?]
> Sep 22 09:41:42 host kafka[42245]: #011at
> org.apache.kafka.raft.KafkaRaftClient.handleInboundMessage(KafkaRaftClient.java:2724)
> ~[kafka-raft-4.0.0.jar:?]
> Sep 22 09:41:42 host kafka[42245]: #011at
> org.apache.kafka.raft.KafkaRaftClient.poll(KafkaRaftClient.java:3460)
> ~[kafka-raft-4.0.0.jar:?]
> Sep 22 09:41:42 host kafka[42245]: #011at
> org.apache.kafka.raft.KafkaRaftClientDriver.doWork(KafkaRaftClientDriver.java:64)
> [kafka-raft-4.0.0.jar:?]
> Sep 22 09:41:42 host kafka[42245]: #011at
> org.apache.kafka.server.util.ShutdownableThread.run(ShutdownableThread.java:136)
> [kafka-server-common-4.0.0.jar:?]
> Below metrics shows Fenceborker count as 2.0
> kafka_controller_KafkaController_Value\{name="ActiveBrokerCount",} 1.0
> kafka_controller_KafkaController_Value\{name="GlobalTopicCount",} 23.0
> kafka_controller_KafkaController_Value\{name="FencedBrokerCount",} 2.0
> Please help us to resolve this issue.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
