mumrah commented on code in PR #20128:
URL: https://github.com/apache/kafka/pull/20128#discussion_r2240026615
##########
core/src/main/scala/kafka/zk/KafkaZkClient.scala:
##########
@@ -1734,20 +1734,27 @@ class KafkaZkClient private[zk] (
val getDataResponse = retryRequestUntilConnected(getDataRequest)
getDataResponse.resultCode match {
case Code.OK =>
- MigrationZNode.decode(getDataResponse.data,
getDataResponse.stat.getVersion, getDataResponse.stat.getMtime)
+ Option(getDataResponse.data) match {
+ case Some(data) =>
+ MigrationZNode.decode(data, getDataResponse.stat.getVersion,
getDataResponse.stat.getMtime)
+ case None =>
+ createInitialMigrationState(initialState, removeFirst = true)
Review Comment:
Can we log a message here that we are fixing the `/migration` znode?
##########
core/src/main/scala/kafka/zk/KafkaZkClient.scala:
##########
@@ -1734,20 +1734,27 @@ class KafkaZkClient private[zk] (
val getDataResponse = retryRequestUntilConnected(getDataRequest)
getDataResponse.resultCode match {
case Code.OK =>
- MigrationZNode.decode(getDataResponse.data,
getDataResponse.stat.getVersion, getDataResponse.stat.getMtime)
+ Option(getDataResponse.data) match {
+ case Some(data) =>
+ MigrationZNode.decode(data, getDataResponse.stat.getVersion,
getDataResponse.stat.getMtime)
+ case None =>
+ createInitialMigrationState(initialState, removeFirst = true)
+ }
case Code.NONODE =>
createInitialMigrationState(initialState)
case _ => throw getDataResponse.resultException.get
}
}
- private def createInitialMigrationState(initialState:
ZkMigrationLeadershipState): ZkMigrationLeadershipState = {
- val createRequest = CreateRequest(
+ private def createInitialMigrationState(initialState:
ZkMigrationLeadershipState, removeFirst: Boolean = false):
ZkMigrationLeadershipState = {
+ val createOp = CreateOp(
MigrationZNode.path,
MigrationZNode.encode(initialState),
defaultAcls(MigrationZNode.path),
CreateMode.PERSISTENT)
- val response = retryRequestUntilConnected(createRequest)
+ val deleteOp = DeleteOp(MigrationZNode.path, ZkVersion.MatchAnyVersion)
+ val multi = MultiRequest((if (removeFirst) Some(deleteOp) else None).toSeq
++ Seq(createOp))
Review Comment:
Similar to the co-pilot suggestion, we can simplify this to something like
```
var multi = if (removeFirst) {
MultiRequest(deleteOp, createOp)
} else {
MultiRequest(createOp)
}
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
