Michael Westerby created KAFKA-21062:
----------------------------------------
Summary: Potential data loss during KRaft migration due to a lost
ZooKeeper acknowledgement on /migration write
Key: KAFKA-21062
URL: https://issues.apache.org/jira/browse/KAFKA-21062
Project: Kafka
Issue Type: Bug
Components: kraft, migration, zkclient
Affects Versions: 3.9.2
Reporter: Michael Westerby
A lost ZooKeeper acknowledgment on a successful {{/migration}} znode write can
cause the KRaft migration driver to mistake its own prior write for a
conflicting one, permanently stalling it once it reaches the {{DUAL_WRITE}}
stage. This prevents further writes to ZooKeeper, allowing the state in between
KRaft and ZooKeeper to diverge during the migration.
This happens because the client the migration uses to talk to ZooKeeper will
retry an identical request to {{/migration}} if it sees a connection loss. If
the original request had actually already succeeded on the server, but only its
acknowledgment was lost, the retry sends the same expected version of the
{{/migration}} znode again. Since that version has already moved on, ZooKeeper
rejects the retry as a version conflict, even though nothing is actually wrong.
The migration driver currently has no way to tell this apart from a genuine
conflict caused by a second controller writing at the same time, so it treats
every such rejection as a hard failure. Because the write is treated as a
failure, the driver's own record of what version the {{/migration}} znode is
currently at never gets corrected, and stays at the old, stale value. Every
subsequent write then reuses that same stale value and hits the same rejection,
indefinitely. The driver becomes permanently stuck, unable to make further
migration progress, until an unrelated event, such as a new controller
election, forces an unconditional resynchronization of {{{}/migration{}}}.
While stuck, ZooKeeper stops receiving updates from the controller, and any
ZK-mode brokers that rely on ZooKeeper for their view of cluster state are
never notified of the corresponding changes via RPCs either. This causes the
states in ZooKeeper and KRaft to diverge, which can destabilize the cluster
further.
This is of particular concern when some, but not all, brokers have already been
restarted into their KRaft mode. The remaining ZK-mode brokers are left running
on an increasingly outdated view of the cluster state. In the worst case, this
can cause real data loss if the KRaft side elects a new partition leader while
the driver is stuck. As that change is never propagated to ZooKeeper, and the
ZK-mode brokers are never notified of it via RPC either, the old leader can
continue to operate under the stale, lower epoch. The old leader can keep
accepting and acknowledging produce requests it is no longer authorized to
serve. When it eventually discovers the true, higher-epoch leader, it must
truncate its own log to reconcile, silently discarding any writes it had
already acknowledged to producers in the meantime.
h2. Walkthrough
[{{KafkaZkClient}}
|https://github.com/apache/kafka/blob/3.9.2/core/src/main/scala/kafka/zk/KafkaZkClient.scala]exposes
two methods used to write to the {{/migration}} znode in ZooKeeper during a
ZK-to-KRaft migration:
# the standalone
[{{updateMigrationState}}|https://github.com/apache/kafka/blob/5e9866f43ab8e7e41ef39e5584ac50019381328d/core/src/main/scala/kafka/zk/KafkaZkClient.scala#L1767]
# the bundled
[{{{}retryMigrationRequestsUntilConnected{}}}|https://github.com/apache/kafka/blob/5e9866f43ab8e7e41ef39e5584ac50019381328d/core/src/main/scala/kafka/zk/KafkaZkClient.scala#L2012].
Both take a
[{{ZkMigrationLeadershipState}}|https://github.com/apache/kafka/blob/3.9.2/metadata/src/main/java/org/apache/kafka/metadata/migration/ZkMigrationLeadershipState.java]
and send its {{migrationZkVersion}} field to ZooKeeper as the expected version
for a conditional write to {{{}/migration{}}}.
Both methods ultimately use {{{}KafkaZkClient{}}}'s own retry loop
[({{{}retryRequestsUntilConnected{}}})|https://github.com/apache/kafka/blob/5e9866f43ab8e7e41ef39e5584ac50019381328d/core/src/main/scala/kafka/zk/KafkaZkClient.scala#L2123],
which repeatedly calls into the lower-level {{ZooKeeperClient}} and resends
whenever it sees {{{}CONNECTIONLOSS{}}}. If the connection is lost
({{{}ConnectionLossException{}}}) after a write has already committed
server-side, but before its acknowledgment reaches the client, this loop
transparently resends the identical request, still carrying the same
{{{}migrationZkVersion{}}}, with no way of knowing the original attempt already
succeeded. Since the version has genuinely moved on, this resend is rejected
with {{{}BadVersionException{}}}, even though nothing is actually wrong.
The two methods surface this differently:
# {{updateMigrationState}} issues a single {{{}SetDataRequest{}}}. On the
resend's {{{}BadVersionException{}}}, this propagates directly out via
{{{}resp.maybeThrow(){}}}.
# {{retryMigrationRequestsUntilConnected}} wraps each request in a multi-op
transaction that includes a {{{}CheckOp{}}}/{{{}SetDataOp{}}} on
{{{}/migration{}}}. On the resend's {{{}BadVersionException{}}},
{{handleUnwrappedMigrationResult}} throws an unconditional
{{{}RuntimeException{}}}, assuming a second KRaft controller must be writing to
ZooKeeper.
Both methods are called from the
[{{{}KRaftMigrationDriver{}}}|https://github.com/apache/kafka/blob/3.9.2/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java],
which caches the {{ZkMigrationLeadershipState}} (and its
{{{}migrationZkVersion{}}}) in {{{}this.migrationLeadershipState{}}}. Once the
driver reaches {{{}DUAL_WRITE{}}},
[{{MetadataChangeEvent.run()}}|https://github.com/apache/kafka/blob/5e9866f43ab8e7e41ef39e5584ac50019381328d/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java#L510]
calls into them continuously, from three places:
#
[{{{}handleDelta{}}}|https://github.com/apache/kafka/blob/5e9866f43ab8e7e41ef39e5584ac50019381328d/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java#L574],
for each incremental metadata change, writes topic/config/ACL/delegation-token
updates via {{{}retryMigrationRequestsUntilConnected{}}}.
#
[{{{}handleSnapshot{}}}|https://github.com/apache/kafka/blob/5e9866f43ab8e7e41ef39e5584ac50019381328d/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java#L570],
the same mechanism, used when a full metadata snapshot is loaded instead of an
incremental delta.
# The dedicated checkpoint write at the end of the same event,
[{{{}zkMigrationClient.setMigrationRecoveryState(zkStateAfterDualWrite){}}}|https://github.com/apache/kafka/blob/5e9866f43ab8e7e41ef39e5584ac50019381328d/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java#L593],
via {{{}updateMigrationState{}}}.
Each of these calls is wrapped in {{{}applyMigrationOperation{}}}, which only
[reassigns {{this.migrationLeadershipState}} when the call returns
{*}successfully{*}|https://github.com/apache/kafka/blob/5e9866f43ab8e7e41ef39e5584ac50019381328d/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java#L257].
As these methods will now always throw an exception, this leaves the driver's
cached {{migrationZkVersion}} stale, and is never corrected. Every subsequent
write, from any of the three call sites, reuses that same stale version and
fails identically, so the driver becomes stuck indefinitely, unable to make
further migration progress.
Both exceptions will eventually surface at
[{{MigrationEvent.handleException}}|https://github.com/apache/kafka/blob/5e9866f43ab8e7e41ef39e5584ac50019381328d/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java#L416]
:
{code:java}
public void handleException(Throwable e) {
if (e instanceof MigrationClientAuthException) {
KRaftMigrationDriver.this.faultHandler.handleFault("Encountered
ZooKeeper authentication in " + this, e);
} else if (e instanceof MigrationClientException) {
log.info(String.format("Encountered ZooKeeper error during
event %s. Will retry.", this), e.getCause());
} else if (e instanceof RejectedExecutionException) {
log.debug("Not processing {} because the event queue is
closed.", this);
} else {
KRaftMigrationDriver.this.faultHandler.handleFault("Unhandled
error in " + this, e);
}
}
{code}
# The {{MigrationClientException}} which wraps {{updateMigrationState}} ‘s
{{BadVersionException}} (via
[{{{}ZkMigrationClient.wrapZkExeception{}}}|https://github.com/apache/kafka/blob/5e9866f43ab8e7e41ef39e5584ac50019381328d/core/src/main/scala/kafka/zk/ZkMigrationClient.scala#L68])
is simply logged at INFO level as “will retry”.
# The unwrapped {{RuntimeException}} from
{{retryMigrationRequestsUntilConnected}} instead falls through to
{{KRaftMigrationDriver.this.faultHandler.handleFault(...)}} . The controller
which builds that fault handler provides it with [{{fatal =
false}}|https://github.com/apache/kafka/blob/5e9866f43ab8e7e41ef39e5584ac50019381328d/core/src/main/scala/kafka/server/ControllerServer.scala#L304]
in
[{{ControllerServer.scala}}|https://github.com/apache/kafka/blob/3.9.2/core/src/main/scala/kafka/server/ControllerServer.scala]
which when provided to the {{StandardFaultHandlerFactory}} will create a
{{LoggingFaultHandler}} . This will simply log exceptions as errors, without
terminating the process.
{code:java}
val migrationDriver = KRaftMigrationDriver.newBuilder()
...
.setFaultHandler(sharedServer.faultHandlerFactory.build(
"zk migration",
fatal = false,
() => {}
))
{code}
Given both paths result in errors which are just logged, nothing actually
causes the process to exit fatally, meaning the current driver will persist in
this stalled state indefinitely. It only recovers once an unrelated event (e.g.
a new controller election, which performs an unconditional overwrite of
{{{}/migration{}}}) happens to resynchronize it.
This {{ConnectionLossException}} / {{BadVersionException}} pattern has been
seen before, with an existing precedent on how to address it.
[{{KafkaZkClient.conditionalUpdatePath}}|https://github.com/apache/kafka/blob/5e9866f43ab8e7e41ef39e5584ac50019381328d/core/src/main/scala/kafka/zk/KafkaZkClient.scala#L965]
supports an optional checker function that, on a {{BadVersionException}} ,
re-reads the znode and compares its content against what the caller intended to
write, so a genuine conflict can be told apart from the caller’s own retried
write having already landed.
[{{ReplicationUtils.updateLeaderAndIsr}}|https://github.com/apache/kafka/blob/5e9866f43ab8e7e41ef39e5584ac50019381328d/core/src/main/scala/kafka/utils/ReplicationUtils.scala#L33-L34]
already rely on this via
[{{checkLeaderAndIsrZkData}}|https://github.com/apache/kafka/blob/5e9866f43ab8e7e41ef39e5584ac50019381328d/core/src/main/scala/kafka/utils/ReplicationUtils.scala#L38]
for leader/ISR updates. However, neither the {{updateMigrationState}} nor
{{retryMigrationRequestsUntilConnected}} currently use it for
{{{}/migration{}}}.
h2. ZkWriteBehindLag Metric
When the controller is in this stuck state, the {{ZkWriteBehindLag}} metric can
fail to report the actual lag present. The metric works by comparing the
controller's latest committed offset against the offset it believes it has
already mirrored to ZooKeeper, but that mirrored offset is recorded before the
checkpoint write to {{/migration}} is attempted, not after it succeeds. On an
event that doesn't otherwise involve any topic, config, ACL, quota, producer
ID, or delegation token changes, such as one driven by a no-op record, this
recording happens regardless, immediately before the checkpoint write that then
fails. The metric can therefore report at or near zero lag even while the
driver is completely stuck, giving operators no indication that dual-write has
failed.
The {{ZkWriteBehindLag}} gauge in
[{{QuorumControllerMetrics}}|https://github.com/apache/kafka/blob/3.9.2/metadata/src/main/java/org/apache/kafka/controller/metrics/QuorumControllerMetrics.java]
is computed as
[{{{}lastCommittedRecordOffset() -
dualWriteOffset(){}}}|https://github.com/apache/kafka/blob/5e9866f43ab8e7e41ef39e5584ac50019381328d/metadata/src/main/java/org/apache/kafka/controller/metrics/QuorumControllerMetrics.java#L167].
{{dualWriteOffset}} is updated via
[{{{}controllerMetrics.updateDualWriteOffset(image.highestOffsetAndEpoch().offset()){}}}|https://github.com/apache/kafka/blob/5e9866f43ab8e7e41ef39e5584ac50019381328d/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java#L590],
called in
{{MetadataChangeEvent.run()}} immediately before the checkpoint write described
above.
Whether that update is actually reached, for any given event, depends on what
the event contains:
* If the event includes any topic, config, client quota, SCRAM credential,
producer ID, ACL, or delegation token change,
{{KRaftMigrationZkWriter.handleDelta}} calls into the bundled write path
({{{}retryMigrationRequestsUntilConnected{}}}) for each one. Once the driver is
already stuck, these calls fail the same way as described above, and the
exception propagates out of {{handleDelta}} before {{updateDualWriteOffset}} is
ever reached. The metric correctly stays where it was, and the reported lag
keeps growing as expected.
* If the event contains none of those changes (for example, one driven only by
a broker registration change or a {{{}NoOpRecord{}}}), {{handleDelta}} has
nothing to write, returns normally, and execution reaches
{{updateDualWriteOffset}} regardless. {{dualWriteOffset}} advances to the
latest offset, and only then does the checkpoint write immediately after it
fail. On these events, the lag metric is misleadingly reset toward zero, even
though the driver is still stuck.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
