You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Michael Westerby (Jira)" <ji...@apache.org> on 2026/09/11 09:49:00 UTC
[jira] [Updated] (KAFKA-21062) Potential data loss during KRaft migration due to a lost ZooKeeper acknowledgement on /migration write
[ https://issues.apache.org/jira/browse/KAFKA-21062?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Michael Westerby updated KAFKA-21062:
-------------------------------------
Description:
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.
h2. Tests
I've written some tests which show this behaviour, both for the impact of the lost ack, and how the ZkWriteBehindLag metric can report no lag in this scenario:
[https://github.com/apache/kafka/commit/7ca7e81bc608b3db7ece3486506f425216fa5220]
was:
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.
> 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
> Assignee: Michael Westerby
> Priority: Major
>
> 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.
> h2. Tests
> I've written some tests which show this behaviour, both for the impact of the lost ack, and how the ZkWriteBehindLag metric can report no lag in this scenario:
> [https://github.com/apache/kafka/commit/7ca7e81bc608b3db7ece3486506f425216fa5220]
--
This message was sent by Atlassian Jira
(v8.20.10#820010)