OmniaGM commented on code in PR #23639:
URL: https://github.com/apache/kafka/pull/23639#discussion_r4455389813
##########
core/src/main/scala/kafka/zk/KafkaZkClient.scala:
##########
@@ -1764,11 +1764,46 @@ class KafkaZkClient private[zk] (
initialState.withMigrationZkVersion(0)
}
+ /**
+ * When there is a ConnectionLossException while updating the /migration
znode, the update may have actually succeeded
+ * on the server before the connection dropped, so the retried request can
fail with BadVersionException even though
+ * our own write went through. This checker reads back the current data and,
if its migration-relevant fields match
+ * what we intended to write, treats it as our own write having succeeded,
returning the real current version.
+ */
+ private def checkMigrationStateZkData(
+ zkClient: KafkaZkClient,
+ path: String,
+ expectedData: Array[Byte]
+ ): (Boolean, Int) = {
+ try {
+ val (writtenDataOpt, writtenStat) = zkClient.getDataAndStat(path)
+ val expected = MigrationZNode.decode(expectedData,
writtenStat.getVersion, -1)
+ val succeeded = writtenDataOpt.exists { writtenData =>
+ val written = MigrationZNode.decode(writtenData,
writtenStat.getVersion, -1)
+ expected.kraftControllerId() == written.kraftControllerId() &&
+ expected.kraftControllerEpoch() == written.kraftControllerEpoch() &&
+ expected.kraftMetadataOffset() == written.kraftMetadataOffset() &&
+ expected.kraftMetadataEpoch() == written.kraftMetadataEpoch()
+ }
+ if (succeeded) (true, writtenStat.getVersion) else (false,
ZkVersion.UnknownVersion)
+ } catch {
+ case _: Exception => (false, ZkVersion.UnknownVersion)
Review Comment:
I think catching all exceptions here is too broad. If `getDataAndStat` fails
for any reason (for example a session expiry, or `NoAuthException`), the
checker returns false, and `updateMigrationState` then throws
`MigrationConflictException`. So a failed read would make the controller
renounce leadership as if another controller had written to /migration. Before
this PR, the same error would have been retried as a
`MigrationClientException`, or handled as a `MigrationClientAuthException`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
