You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/01/25 19:33:48 UTC
[GitHub] [kafka] hachikuji opened a new pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
hachikuji opened a new pull request #9967:
URL: https://github.com/apache/kafka/pull/9967
This patch contains the new handling of `meta.properties` required by the KIP-500 server as specified in KIP-631. When using the self-managed quorum, the `meta.properties` file is required in each log directory with the new `version` property set to 1. It must include the `cluster.id` property and it must have a `broker.id` and `controller.id` matching those in the configuration.
The behavior of `meta.properties` for the Zookeeper-based `KafkaServer` does not change. We treat the old `meta.properties` as if it were `version=0` and we generate the clusterId and/or the brokerId through Zookeeper as needed.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#discussion_r567254453
##########
File path: core/src/main/scala/kafka/server/BrokerMetadataCheckpoint.scala
##########
@@ -21,38 +21,203 @@ import java.io._
import java.nio.file.{Files, NoSuchFileException}
import java.util.Properties
+import kafka.common.{InconsistentBrokerMetadataException, KafkaException}
+import kafka.server.RawMetaProperties._
import kafka.utils._
+import org.apache.kafka.common.Uuid
import org.apache.kafka.common.utils.Utils
-case class BrokerMetadata(brokerId: Int,
- clusterId: Option[String]) {
+import scala.collection.mutable
+import scala.jdk.CollectionConverters._
+
+object RawMetaProperties {
+ val ClusterIdKey = "cluster.id"
+ val BrokerIdKey = "broker.id"
+ val NodeIdKey = "node.id"
Review comment:
Both "broker.id" and "node.id" are defined by ```KafkaConfig``` also. Should we unify them?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#discussion_r565032694
##########
File path: core/src/main/scala/kafka/server/KafkaConfig.scala
##########
@@ -1495,6 +1517,13 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
distinctRoles
}
+ def metadataLogDir: String = {
+ Option(getString(KafkaConfig.MetadataLogDirProp)) match {
+ case Some(dir) => dir
+ case None => logDirs.head
Review comment:
What I am thinking for the moment is the following:
1. Assume a deterministic mapping to the metadata dir.
2. Raise an exception on startup if we find the `@metadata` log dir in another directory.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji edited a comment on pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji edited a comment on pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#issuecomment-768034713
@ijuma To answer your question, newly generated ClusterIds do indeed use the Base64 encoding. I think it probably needs to be smart enough to accept the old encoding as well though. Let me add a test case for this.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#issuecomment-768742967
@chia7712 @ijuma Thanks for the comments thus far. This is ready for another look when you have time.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#issuecomment-768034713
@ijuma To answer your question, newly generated ClusterIds do indeed use the Base64 encoding. I think it probably needs to be smart enough to accept both versions though. Let me add a test case for this.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#discussion_r567309420
##########
File path: clients/src/main/java/org/apache/kafka/common/Uuid.java
##########
@@ -106,6 +106,9 @@ public String toString() {
*/
public static Uuid fromString(String str) {
ByteBuffer uuidBytes = ByteBuffer.wrap(Base64.getUrlDecoder().decode(str));
+ if (uuidBytes.remaining() != 16) {
+ throw new IllegalArgumentException("Failed to parse `" + str + "` as a base64-encoded UUID");
Review comment:
Thanks. Tried to improve this in the latest commit. Please take a look.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#discussion_r564573301
##########
File path: core/src/main/scala/kafka/common/InconsistentControllerIdException.scala
##########
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.common
+
+class InconsistentControllerIdException(message: String, cause: Throwable) extends RuntimeException(message, cause) {
+ def this(message: String) = this(message, null)
+ def this(cause: Throwable) = this(null, cause)
+ def this() = this(null, null)
Review comment:
Do we need this constructor?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#discussion_r565032694
##########
File path: core/src/main/scala/kafka/server/KafkaConfig.scala
##########
@@ -1495,6 +1517,13 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
distinctRoles
}
+ def metadataLogDir: String = {
+ Option(getString(KafkaConfig.MetadataLogDirProp)) match {
+ case Some(dir) => dir
+ case None => logDirs.head
Review comment:
What I am thinking for the moment is the following:
1. Assume a deterministic mapping to the metadata dir.
2. Raise an exception on startup if we find the `@metadata` log dir in another directory.
As a future improvement, we might relax this in case the metadata log dir is offline.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#discussion_r565032694
##########
File path: core/src/main/scala/kafka/server/KafkaConfig.scala
##########
@@ -1495,6 +1517,13 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
distinctRoles
}
+ def metadataLogDir: String = {
+ Option(getString(KafkaConfig.MetadataLogDirProp)) match {
+ case Some(dir) => dir
+ case None => logDirs.head
Review comment:
What I am thinking for the moment is the following:
1. Assume a deterministic mapping to the metadata dir from the configuration.
2. Raise an exception on startup if we find the `@metadata` log dir in another directory.
As a future improvement, we might relax this in case the metadata log dir has not been explicitly defined and is offline.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji edited a comment on pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji edited a comment on pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#issuecomment-768033295
> Pardon me, what did you mean "in each log directory"? I read this PR and it seems to me the new meta.properties must exists in "only one" log directory (the head of directories by default)
I guess this is proper chastisement for the shortage of tests verifying this behavior. I have added some and caught a bug in the process. Perhaps the bug was the source of the confusion.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#issuecomment-768034167
Another thing we need to do here is replace `controller.id` and `broker.id` with a single `node.id`. I will get to this tomorrow.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#issuecomment-770279434
@ijuma Yep, got a little lazy verifying the case class conversion. Hopefully fixed now. We were in fact relying on the `equals`, but I modified the test case.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji merged pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji merged pull request #9967:
URL: https://github.com/apache/kafka/pull/9967
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#discussion_r566168366
##########
File path: core/src/main/scala/kafka/server/BrokerMetadataCheckpoint.scala
##########
@@ -21,38 +21,203 @@ import java.io._
import java.nio.file.{Files, NoSuchFileException}
import java.util.Properties
+import kafka.common.{InconsistentBrokerMetadataException, KafkaException}
+import kafka.server.RawMetaProperties._
import kafka.utils._
+import org.apache.kafka.common.Uuid
import org.apache.kafka.common.utils.Utils
-case class BrokerMetadata(brokerId: Int,
- clusterId: Option[String]) {
+import scala.collection.mutable
+import scala.jdk.CollectionConverters._
+
+object RawMetaProperties {
+ val ClusterIdKey = "cluster.id"
+ val BrokerIdKey = "broker.id"
+ val NodeIdKey = "node.id"
+ val VersionKey = "version"
+}
+
+case class RawMetaProperties(props: Properties = new Properties()) {
Review comment:
Does it make sense for this to be a case class? We override `toString`, so what do we actually get from making it a case class here? Do we rely on the generated equals/hashCode?
##########
File path: clients/src/main/java/org/apache/kafka/common/Uuid.java
##########
@@ -106,6 +106,9 @@ public String toString() {
*/
public static Uuid fromString(String str) {
ByteBuffer uuidBytes = ByteBuffer.wrap(Base64.getUrlDecoder().decode(str));
+ if (uuidBytes.remaining() != 16) {
+ throw new IllegalArgumentException("Failed to parse `" + str + "` as a base64-encoded UUID");
Review comment:
Also, would it be better to check the string size before decoding? If there's a bug and a really large string is passed, it's a bit safer to error before parsing (and to only include the first N characters in the exception).
##########
File path: clients/src/main/java/org/apache/kafka/common/Uuid.java
##########
@@ -106,6 +106,9 @@ public String toString() {
*/
public static Uuid fromString(String str) {
ByteBuffer uuidBytes = ByteBuffer.wrap(Base64.getUrlDecoder().decode(str));
+ if (uuidBytes.remaining() != 16) {
+ throw new IllegalArgumentException("Failed to parse `" + str + "` as a base64-encoded UUID");
Review comment:
It is worth being a bit more explicit about the issue, maybe something like "Expected 16 bytes, received Y bytes."? It may be a bit difficult to figure out why the parsing failed otherwise (since the string will have only valid characters).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#discussion_r567306870
##########
File path: core/src/main/scala/kafka/server/BrokerMetadataCheckpoint.scala
##########
@@ -21,38 +21,203 @@ import java.io._
import java.nio.file.{Files, NoSuchFileException}
import java.util.Properties
+import kafka.common.{InconsistentBrokerMetadataException, KafkaException}
+import kafka.server.RawMetaProperties._
import kafka.utils._
+import org.apache.kafka.common.Uuid
import org.apache.kafka.common.utils.Utils
-case class BrokerMetadata(brokerId: Int,
- clusterId: Option[String]) {
+import scala.collection.mutable
+import scala.jdk.CollectionConverters._
+
+object RawMetaProperties {
+ val ClusterIdKey = "cluster.id"
+ val BrokerIdKey = "broker.id"
+ val NodeIdKey = "node.id"
+ val VersionKey = "version"
+}
+
+case class RawMetaProperties(props: Properties = new Properties()) {
Review comment:
I can't think of a reason to keep it as a case class. Will change.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#discussion_r565786393
##########
File path: core/src/main/scala/kafka/server/Server.scala
##########
@@ -46,6 +46,22 @@ object Server {
new Metrics(metricConfig, reporters, time, true, metricsContext)
}
+ def initializeMetrics(
+ config: KafkaConfig,
+ time: Time,
+ metaProps: MetaProperties
Review comment:
It seems to me the properties in ```MetaProperties``` is duplicate to ```KafkaConfig``` in this case. Is there any reason that we need to pass ```MetaProperties```?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#issuecomment-770307353
One test failure fixed by https://github.com/apache/kafka/pull/10006. Merging to trunk.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#discussion_r565019178
##########
File path: core/src/main/scala/kafka/server/KafkaConfig.scala
##########
@@ -1495,6 +1517,13 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
distinctRoles
}
+ def metadataLogDir: String = {
+ Option(getString(KafkaConfig.MetadataLogDirProp)) match {
+ case Some(dir) => dir
+ case None => logDirs.head
Review comment:
Good question. I think the implementation will have to be smart enough to find the metadata log dir. Let me see if it is straightforward to add that here or if we should leave it for a follow-up. I guess the alternative would be to require `metadata.log.dir`, but I think we were trying to avoid that.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#discussion_r567301638
##########
File path: core/src/main/scala/kafka/server/BrokerMetadataCheckpoint.scala
##########
@@ -21,38 +21,203 @@ import java.io._
import java.nio.file.{Files, NoSuchFileException}
import java.util.Properties
+import kafka.common.{InconsistentBrokerMetadataException, KafkaException}
+import kafka.server.RawMetaProperties._
import kafka.utils._
+import org.apache.kafka.common.Uuid
import org.apache.kafka.common.utils.Utils
-case class BrokerMetadata(brokerId: Int,
- clusterId: Option[String]) {
+import scala.collection.mutable
+import scala.jdk.CollectionConverters._
+
+object RawMetaProperties {
+ val ClusterIdKey = "cluster.id"
+ val BrokerIdKey = "broker.id"
+ val NodeIdKey = "node.id"
Review comment:
Yeah, that is the plan. Once KIP-500 is enabled, we will deprecate `broker.id` so that `node.id` is the only ID going forward. That said, I'm trying to give us a little wiggle room by defining this config as internal for now. So we still have an opportunity to change this. If you have any thoughts about it, definitely share them.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#issuecomment-768033295
> Pardon me, what did you mean "in each log directory"? I read this PR and it seems to me the new meta.properties must exists in "only one" log directory (the head of directories by default)
I guess this is chastisement for the shortage of tests verifying this behavior. I have added some and caught a bug in the process. Perhaps the bug was the source of the confusion.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#discussion_r565019178
##########
File path: core/src/main/scala/kafka/server/KafkaConfig.scala
##########
@@ -1495,6 +1517,13 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
distinctRoles
}
+ def metadataLogDir: String = {
+ Option(getString(KafkaConfig.MetadataLogDirProp)) match {
+ case Some(dir) => dir
+ case None => logDirs.head
Review comment:
Good question. I think the implementation will have to be smart enough to find the metadata log dir in case we have already created the partition directory. Let me see if it is straightforward to add that here or if we should leave it for a follow-up. I guess the alternative would be to require `metadata.log.dir`, but I think we were trying to avoid that.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#discussion_r564703919
##########
File path: core/src/main/scala/kafka/server/BrokerMetadataCheckpoint.scala
##########
@@ -21,38 +21,220 @@ import java.io._
import java.nio.file.{Files, NoSuchFileException}
import java.util.Properties
+import kafka.common.InconsistentBrokerMetadataException
+import kafka.server.KafkaRaftServer.{BrokerRole, ControllerRole, ProcessRole}
+import kafka.server.RawMetaProperties._
import kafka.utils._
+import org.apache.kafka.common.Uuid
import org.apache.kafka.common.utils.Utils
-case class BrokerMetadata(brokerId: Int,
- clusterId: Option[String]) {
+import scala.collection.mutable
+import scala.jdk.CollectionConverters._
+
+object RawMetaProperties {
+ val ClusterIdKey = "cluster.id"
+ val BrokerIdKey = "broker.id"
+ val ControllerIdKey = "controller.id"
+ val VersionKey = "version"
+}
+
+case class RawMetaProperties(props: Properties = new Properties()) {
+
+ def clusterId: Option[String] = {
+ Option(props.getProperty(ClusterIdKey))
+ }
+
+ def clusterId_=(id: String): Unit = {
+ props.setProperty(ClusterIdKey, id)
+ }
+
+ def brokerId: Option[Int] = {
+ intValue(BrokerIdKey)
+ }
+
+ def brokerId_=(id: Int): Unit = {
+ props.setProperty(BrokerIdKey, id.toString)
+ }
+
+ def controllerId: Option[Int] = {
+ intValue(ControllerIdKey)
+ }
+
+ def controllerId_=(id: Int): Unit = {
+ props.setProperty(ControllerIdKey, id.toString)
+ }
+
+ def version: Int = {
+ intValue(VersionKey).getOrElse(0)
+ }
+
+ def version_=(ver: Int): Unit = {
+ props.setProperty(VersionKey, ver.toString)
+ }
+
+ def requireVersion(expectedVersion: Int): Unit = {
+ if (version != expectedVersion) {
+ throw new RuntimeException(s"Expected version $expectedVersion, but got "+
+ s"version $version")
+ }
+ }
+
+ private def intValue(key: String): Option[Int] = {
+ try {
+ Option(props.getProperty(key)).map(Integer.parseInt)
+ } catch {
+ case e: Throwable => throw new RuntimeException(s"Failed to parse $key property " +
+ s"as an int: ${e.getMessage}")
+ }
+ }
+
+ override def toString: String = {
+ "RawMetaProperties(" + props.keySet().asScala.toList.asInstanceOf[List[String]].sorted.map {
+ key => key + "=" + props.get(key)
+ }.mkString(", ") + ")"
+ }
+}
+
+object MetaProperties {
+ def parse(
+ properties: RawMetaProperties,
+ processRoles: Set[ProcessRole]
+ ): MetaProperties = {
+ properties.requireVersion(expectedVersion = 1)
+ val clusterId = requireClusterId(properties)
+
+ if (processRoles.contains(BrokerRole)) {
+ require(BrokerIdKey, properties.brokerId)
+ }
+
+ if (processRoles.contains(ControllerRole)) {
+ require(ControllerIdKey, properties.controllerId)
+ }
+
+ new MetaProperties(clusterId, properties.brokerId, properties.controllerId)
+ }
+
+ def require[T](key: String, value: Option[T]): T = {
+ value.getOrElse(throw new RuntimeException(s"Failed to find required property $key."))
+ }
+
+ def requireClusterId(properties: RawMetaProperties): Uuid = {
+ val value = require(ClusterIdKey, properties.clusterId)
+ try {
+ Uuid.fromString(value)
+ } catch {
+ case e: Throwable => throw new RuntimeException(s"Failed to parse $ClusterIdKey property " +
+ s"as a UUID: ${e.getMessage}")
+ }
+ }
+}
+
+case class ZkMetaProperties(
+ clusterId: String,
+ brokerId: Int
+) {
+ def toProperties: Properties = {
+ val properties = new RawMetaProperties()
+ properties.version = 0
+ properties.clusterId = clusterId
+ properties.brokerId = brokerId
+ properties.props
+ }
+
+ override def toString: String = {
+ s"LegacyMetaProperties(brokerId=$brokerId, clusterId=$clusterId)"
+ }
+}
+
+case class MetaProperties(
+ clusterId: Uuid,
+ brokerId: Option[Int] = None,
+ controllerId: Option[Int] = None
+) {
+ def toProperties: Properties = {
+ val properties = new RawMetaProperties()
+ properties.version = 1
+ properties.clusterId = clusterId.toString
+ brokerId.foreach(properties.brokerId = _)
+ controllerId.foreach(properties.controllerId = _)
+ properties.props
+ }
override def toString: String = {
- s"BrokerMetadata(brokerId=$brokerId, clusterId=${clusterId.map(_.toString).getOrElse("None")})"
+ s"MetaProperties(clusterId=$clusterId" +
+ s", brokerId=${brokerId.getOrElse("none")}" +
+ s", controllerId=${controllerId.getOrElse("none")}" +
+ ")"
+ }
+}
+
+object BrokerMetadataCheckpoint extends Logging {
+ def getBrokerMetadataAndOfflineDirs(
+ logDirs: collection.Seq[String],
+ ignoreMissing: Boolean
+ ): (RawMetaProperties, collection.Seq[String]) = {
+ require(logDirs.nonEmpty, "Must have at least one log dir to read meta.properties")
+
+ val brokerMetadataMap = mutable.HashMap[String, Properties]()
+ val offlineDirs = mutable.ArrayBuffer.empty[String]
+
+ for (logDir <- logDirs) {
+ val brokerCheckpointFile = new File(logDir + File.separator + "meta.properties")
Review comment:
How about using ```new File(logDir, "meta.properties")```?
##########
File path: core/src/main/scala/kafka/server/BrokerMetadataCheckpoint.scala
##########
@@ -21,38 +21,220 @@ import java.io._
import java.nio.file.{Files, NoSuchFileException}
import java.util.Properties
+import kafka.common.InconsistentBrokerMetadataException
+import kafka.server.KafkaRaftServer.{BrokerRole, ControllerRole, ProcessRole}
+import kafka.server.RawMetaProperties._
import kafka.utils._
+import org.apache.kafka.common.Uuid
import org.apache.kafka.common.utils.Utils
-case class BrokerMetadata(brokerId: Int,
- clusterId: Option[String]) {
+import scala.collection.mutable
+import scala.jdk.CollectionConverters._
+
+object RawMetaProperties {
+ val ClusterIdKey = "cluster.id"
+ val BrokerIdKey = "broker.id"
+ val ControllerIdKey = "controller.id"
+ val VersionKey = "version"
+}
+
+case class RawMetaProperties(props: Properties = new Properties()) {
+
+ def clusterId: Option[String] = {
+ Option(props.getProperty(ClusterIdKey))
+ }
+
+ def clusterId_=(id: String): Unit = {
+ props.setProperty(ClusterIdKey, id)
+ }
+
+ def brokerId: Option[Int] = {
+ intValue(BrokerIdKey)
+ }
+
+ def brokerId_=(id: Int): Unit = {
+ props.setProperty(BrokerIdKey, id.toString)
+ }
+
+ def controllerId: Option[Int] = {
+ intValue(ControllerIdKey)
+ }
+
+ def controllerId_=(id: Int): Unit = {
+ props.setProperty(ControllerIdKey, id.toString)
+ }
+
+ def version: Int = {
+ intValue(VersionKey).getOrElse(0)
+ }
+
+ def version_=(ver: Int): Unit = {
+ props.setProperty(VersionKey, ver.toString)
+ }
+
+ def requireVersion(expectedVersion: Int): Unit = {
+ if (version != expectedVersion) {
+ throw new RuntimeException(s"Expected version $expectedVersion, but got "+
+ s"version $version")
+ }
+ }
+
+ private def intValue(key: String): Option[Int] = {
+ try {
+ Option(props.getProperty(key)).map(Integer.parseInt)
+ } catch {
+ case e: Throwable => throw new RuntimeException(s"Failed to parse $key property " +
+ s"as an int: ${e.getMessage}")
+ }
+ }
+
+ override def toString: String = {
+ "RawMetaProperties(" + props.keySet().asScala.toList.asInstanceOf[List[String]].sorted.map {
+ key => key + "=" + props.get(key)
+ }.mkString(", ") + ")"
+ }
+}
+
+object MetaProperties {
+ def parse(
+ properties: RawMetaProperties,
+ processRoles: Set[ProcessRole]
+ ): MetaProperties = {
+ properties.requireVersion(expectedVersion = 1)
+ val clusterId = requireClusterId(properties)
+
+ if (processRoles.contains(BrokerRole)) {
+ require(BrokerIdKey, properties.brokerId)
+ }
+
+ if (processRoles.contains(ControllerRole)) {
+ require(ControllerIdKey, properties.controllerId)
+ }
+
+ new MetaProperties(clusterId, properties.brokerId, properties.controllerId)
+ }
+
+ def require[T](key: String, value: Option[T]): T = {
+ value.getOrElse(throw new RuntimeException(s"Failed to find required property $key."))
+ }
+
+ def requireClusterId(properties: RawMetaProperties): Uuid = {
+ val value = require(ClusterIdKey, properties.clusterId)
+ try {
+ Uuid.fromString(value)
+ } catch {
+ case e: Throwable => throw new RuntimeException(s"Failed to parse $ClusterIdKey property " +
+ s"as a UUID: ${e.getMessage}")
+ }
+ }
+}
+
+case class ZkMetaProperties(
+ clusterId: String,
+ brokerId: Int
+) {
+ def toProperties: Properties = {
+ val properties = new RawMetaProperties()
+ properties.version = 0
+ properties.clusterId = clusterId
+ properties.brokerId = brokerId
+ properties.props
+ }
+
+ override def toString: String = {
+ s"LegacyMetaProperties(brokerId=$brokerId, clusterId=$clusterId)"
+ }
+}
+
+case class MetaProperties(
+ clusterId: Uuid,
+ brokerId: Option[Int] = None,
+ controllerId: Option[Int] = None
+) {
+ def toProperties: Properties = {
+ val properties = new RawMetaProperties()
+ properties.version = 1
+ properties.clusterId = clusterId.toString
+ brokerId.foreach(properties.brokerId = _)
+ controllerId.foreach(properties.controllerId = _)
+ properties.props
+ }
override def toString: String = {
- s"BrokerMetadata(brokerId=$brokerId, clusterId=${clusterId.map(_.toString).getOrElse("None")})"
+ s"MetaProperties(clusterId=$clusterId" +
+ s", brokerId=${brokerId.getOrElse("none")}" +
+ s", controllerId=${controllerId.getOrElse("none")}" +
+ ")"
+ }
+}
+
+object BrokerMetadataCheckpoint extends Logging {
+ def getBrokerMetadataAndOfflineDirs(
+ logDirs: collection.Seq[String],
+ ignoreMissing: Boolean
+ ): (RawMetaProperties, collection.Seq[String]) = {
+ require(logDirs.nonEmpty, "Must have at least one log dir to read meta.properties")
+
+ val brokerMetadataMap = mutable.HashMap[String, Properties]()
+ val offlineDirs = mutable.ArrayBuffer.empty[String]
+
+ for (logDir <- logDirs) {
+ val brokerCheckpointFile = new File(logDir + File.separator + "meta.properties")
+ val brokerCheckpoint = new BrokerMetadataCheckpoint(brokerCheckpointFile)
+
+ try {
+ brokerCheckpoint.read() match {
+ case Some(properties) => brokerMetadataMap += logDir -> properties
+ case None => if (ignoreMissing) {
+ logDir -> new Properties()
Review comment:
unnecessary body?
##########
File path: core/src/main/scala/kafka/server/KafkaConfig.scala
##########
@@ -1495,6 +1517,13 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
distinctRoles
}
+ def metadataLogDir: String = {
+ Option(getString(KafkaConfig.MetadataLogDirProp)) match {
+ case Some(dir) => dir
+ case None => logDirs.head
Review comment:
Does it cause trouble if we change the order of directories from "log.dirs" when restarting server?
##########
File path: core/src/main/scala/kafka/server/KafkaRaftServer.scala
##########
@@ -100,4 +104,36 @@ object KafkaRaftServer {
sealed trait ProcessRole
case object BrokerRole extends ProcessRole
case object ControllerRole extends ProcessRole
+
+ def loadMetaProperties(config: KafkaConfig): (MetaProperties, Seq[String]) = {
+ val logDirs = config.logDirs ++ Seq(config.metadataLogDir)
Review comment:
How about ```config.logDirs :+ config.metadataLogDir```?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#issuecomment-770276918
I think the case class change caused a compiler error, we probably need to make the field accessible.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#discussion_r565032694
##########
File path: core/src/main/scala/kafka/server/KafkaConfig.scala
##########
@@ -1495,6 +1517,13 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
distinctRoles
}
+ def metadataLogDir: String = {
+ Option(getString(KafkaConfig.MetadataLogDirProp)) match {
+ case Some(dir) => dir
+ case None => logDirs.head
Review comment:
What I am thinking for the moment is the following:
1. Assume a deterministic mapping to the metadata dir.
2. Raise an exception on startup if we find the `@metadata` log dir in another directory.
As a future improvement, we might relax this in case the metadata log dir has not been explicitly defined and is offline.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9967: KAFKA-12236; New meta.properties logic for KIP-500
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9967:
URL: https://github.com/apache/kafka/pull/9967#discussion_r566587032
##########
File path: core/src/main/scala/kafka/server/Server.scala
##########
@@ -46,6 +46,22 @@ object Server {
new Metrics(metricConfig, reporters, time, true, metricsContext)
}
+ def initializeMetrics(
+ config: KafkaConfig,
+ time: Time,
+ metaProps: MetaProperties
Review comment:
@chia7712 We need the clusterId from `MetaProperties`. I modified this to pass it through directly avoid the duplicate methods.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org