harmadasg commented on code in PR #20128:
URL: https://github.com/apache/kafka/pull/20128#discussion_r2244730922
##########
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:
updated the code based on the suggestion
##########
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:
fixed it in
[809d34c](https://github.com/apache/kafka/pull/20128/commits/809d34c24244068fddc6ead90a54d785cf5b9206)
--
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
