You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "muralibasani (via GitHub)" <gi...@apache.org> on 2024/06/13 13:23:35 UTC
[PR] KAFKA-16518 : Adding standalone argument for storage [kafka]
muralibasani opened a new pull request, #16325:
URL: https://github.com/apache/kafka/pull/16325
Resolves : https://issues.apache.org/jira/browse/KAFKA-16518
Adds a new argument "standalone" to kafka-storage.sh
If standalone mode, creates a checkpoint file in metadata dir ${kafkaConfig.metadataLogDir}/__cluster_metadata-0/
### 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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16518; Adding standalone argument for storage [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #16325:
URL: https://github.com/apache/kafka/pull/16325#discussion_r1665647113
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -503,24 +537,29 @@ object StorageTool extends Logging {
}
def formatCommand(
- stream: PrintStream,
- directories: Seq[String],
- metaProperties: MetaProperties,
- metadataVersion: MetadataVersion,
- ignoreFormatted: Boolean
- ): Int = {
+ stream: PrintStream,
+ directories: Seq[String],
+ metaProperties: MetaProperties,
+ metadataVersion: MetadataVersion,
+ ignoreFormatted: Boolean,
+ advertisedListenerEndpoints: scala.collection.Seq[kafka.cluster.EndPoint],
+ controllersQuorumVoters: String
+ ): Int = {
val bootstrapMetadata = buildBootstrapMetadata(metadataVersion, None, "format command")
- formatCommand(stream, directories, metaProperties, bootstrapMetadata, metadataVersion, ignoreFormatted)
+ formatCommand(stream, directories, metaProperties, bootstrapMetadata, metadataVersion, ignoreFormatted,
+ advertisedListenerEndpoints, controllersQuorumVoters)
}
def formatCommand(
- stream: PrintStream,
- directories: Seq[String],
- metaProperties: MetaProperties,
- bootstrapMetadata: BootstrapMetadata,
- metadataVersion: MetadataVersion,
- ignoreFormatted: Boolean
- ): Int = {
+ stream: PrintStream,
+ directories: Seq[String],
+ metaProperties: MetaProperties,
+ bootstrapMetadata: BootstrapMetadata,
+ metadataVersion: MetadataVersion,
+ ignoreFormatted: Boolean,
+ advertisedListenerEndpoints: scala.collection.Seq[kafka.cluster.EndPoint],
+ controllersQuorumVoters: String
Review Comment:
Tried considering but directoryId is derived only in format method. So updated format method with only listeners param.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16518; Adding standalone argument for storage [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #16325:
URL: https://github.com/apache/kafka/pull/16325#discussion_r1672845320
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -698,6 +663,42 @@ object StorageTool extends Logging {
voterSet
}
+ def createStandaloneVoterMap(config: KafkaConfig): util.Map[ListenerName, InetSocketAddress] = {
+ val advertisedListenerEndpoints = config.effectiveAdvertisedControllerListeners
+ val listeners: util.Map[ListenerName, InetSocketAddress] = new util.HashMap()
+ advertisedListenerEndpoints.foreach(endpoint => {
+ val host: String = endpoint.host
+ listeners.put(endpoint.listenerName, new InetSocketAddress(host, endpoint.port))
+ })
+ listeners
+ }
+
+ private def parseControllerQuorumVotersMap(controllerQuorumVoterMap: util.Map[Integer, InetSocketAddress],
+ metaProperties: MetaProperties,
+ config: KafkaConfig): util.Map[ListenerName, InetSocketAddress] = {
Review Comment:
From the kip description
"When the format command is executed with this option it will read the node.id configured in the properties file specified by the --config option and compare it against the <replica-id> specified in --controller-quorum-voters. If there is a match, it will write the <replica-directory-id> specified to the directory.id property in the meta.properties for the metadata.log.dir directory."
I tried adding the if condition
if (metaProperties.nodeId().getAsInt == replicaId) )
in the method
May be am wrong. Can you pls suggest code maybe?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16518; Adding standalone argument for storage [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #16325:
URL: https://github.com/apache/kafka/pull/16325#discussion_r1672833818
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -112,20 +112,21 @@ object StorageTool extends Logging {
setNodeId(config.nodeId).
build()
val standaloneMode = namespace.getBoolean("standalone")
- var advertisedListenerEndpoints: collection.Seq[kafka.cluster.EndPoint] = List()
val controllersQuorumVoters = namespace.getString("controller_quorum_voters")
if(standaloneMode && controllersQuorumVoters != null) {
throw new TerseFailure("Both --standalone and --controller-quorum-voters were set. Only one of the two flags can be set.")
}
+ var listeners: util.Map[ListenerName, InetSocketAddress] = new util.HashMap()
if (standaloneMode) {
- advertisedListenerEndpoints = config.effectiveAdvertisedBrokerListeners
+ listeners = createStandaloneVoterMap(config)
} else if(controllersQuorumVoters != null) {
if (!validateControllerQuorumVoters(controllersQuorumVoters)) {
throw new TerseFailure("Expected schema for --controller-quorum-voters is <replica-id>[-<replica-directory-id>]@<host>:<port>")
}
- advertisedListenerEndpoints = config.effectiveAdvertisedControllerListeners
+ val controllerQuorumVoterMap: util.Map[Integer, InetSocketAddress] = parseVoterConnections(Collections.singletonList(controllersQuorumVoters))
+ listeners = parseControllerQuorumVotersMap(controllerQuorumVoterMap, metaProperties, config)
Review Comment:
This means we would have to update https://github.com/apache/kafka/blob/25d775b742406477a0ff678b9990ed149d2157cc/raft/src/main/java/org/apache/kafka/raft/QuorumConfig.java#L178 to return Uuid too ?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16518; Adding standalone argument for storage [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani closed pull request #16325: KAFKA-16518; Adding standalone argument for storage
URL: https://github.com/apache/kafka/pull/16325
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16518; Adding standalone argument for storage [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on PR #16325:
URL: https://github.com/apache/kafka/pull/16325#issuecomment-2200108407
@jsancio , updated with controller option
```
./bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties -q 1@localhost:9093
metaPropertiesEnsemble=MetaPropertiesEnsemble(metadataLogDir=Optional.empty, dirs={/tmp/kraft-combined-logs: EMPTY})
Formatting /tmp/kraft-combined-logs with metadata.version 3.8-IV0.
Snapshot written to /tmp/kraft-combined-logs/__cluster_metadata
muralidharbasani@Muralidhars-MacBook-Pro kafka % ./bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties -s
metaPropertiesEnsemble=MetaPropertiesEnsemble(metadataLogDir=Optional.empty, dirs={/tmp/kraft-combined-logs: EMPTY})
Formatting /tmp/kraft-combined-logs with metadata.version 3.8-IV0.
Snapshot written to /tmp/kraft-combined-logs/__cluster_metadata
muralidharbasani@Muralidhars-MacBook-Pro kafka % ./bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties -q 1@localhost:9093 -s
Both --standalone and --controller-quorum-voters were set. Only one of the two flags can be set.
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16518; Adding standalone argument for storage [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #16325:
URL: https://github.com/apache/kafka/pull/16325#discussion_r1668893365
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -300,19 +301,19 @@ object StorageTool extends Logging {
}
private def getUserScramCredentialRecord(
- mechanism: String,
- config: String
- ) : UserScramCredentialRecord = {
+ mechanism: String,
+ config: String
+ ) : UserScramCredentialRecord = {
/*
* Remove '[' amd ']'
* Split K->V pairs on ',' and no K or V should contain ','
* Split K and V on '=' but V could contain '=' if inside ""
* Create Map of K to V and replace all " in V
*/
val argMap = config.substring(1, config.length - 1)
- .split(",")
- .map(_.split("=(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)"))
- .map(args => args(0) -> args(1).replaceAll("\"", "")).toMap
+ .split(",")
+ .map(_.split("=(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)"))
+ .map(args => args(0) -> args(1).replaceAll("\"", "")).toMap
Review Comment:
Again, this is not the correct indentation.
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -112,20 +112,21 @@ object StorageTool extends Logging {
setNodeId(config.nodeId).
build()
val standaloneMode = namespace.getBoolean("standalone")
- var advertisedListenerEndpoints: collection.Seq[kafka.cluster.EndPoint] = List()
val controllersQuorumVoters = namespace.getString("controller_quorum_voters")
if(standaloneMode && controllersQuorumVoters != null) {
throw new TerseFailure("Both --standalone and --controller-quorum-voters were set. Only one of the two flags can be set.")
}
+ var listeners: util.Map[ListenerName, InetSocketAddress] = new util.HashMap()
if (standaloneMode) {
- advertisedListenerEndpoints = config.effectiveAdvertisedBrokerListeners
+ listeners = createStandaloneVoterMap(config)
} else if(controllersQuorumVoters != null) {
if (!validateControllerQuorumVoters(controllersQuorumVoters)) {
throw new TerseFailure("Expected schema for --controller-quorum-voters is <replica-id>[-<replica-directory-id>]@<host>:<port>")
}
- advertisedListenerEndpoints = config.effectiveAdvertisedControllerListeners
+ val controllerQuorumVoterMap: util.Map[Integer, InetSocketAddress] = parseVoterConnections(Collections.singletonList(controllersQuorumVoters))
+ listeners = parseControllerQuorumVotersMap(controllerQuorumVoterMap, metaProperties, config)
Review Comment:
This doesn't look correct. `VoterSet` is basically a `Map[Integer, (Uuid, Map[ListenerName, InetSocketAddress])]` where `Integer` is the replica id and `Uuid` is the replica directory id. The type for listener doesn't contain all of the information needed to generate a `VoterSet` for all the possible configuration cases.
Note that the value for `--controller-quorum-voters` has the follow schema: `<replica-id>-<replica-directory-id>@<host>:<port>`
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -300,19 +301,19 @@ object StorageTool extends Logging {
}
private def getUserScramCredentialRecord(
- mechanism: String,
- config: String
- ) : UserScramCredentialRecord = {
+ mechanism: String,
+ config: String
+ ) : UserScramCredentialRecord = {
Review Comment:
Extra space between `)` and `:`. It should be `): User... = {`
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -335,10 +336,10 @@ object StorageTool extends Logging {
if (argMap.contains("salt")) {
val iterations = argMap("iterations").toInt
if (iterations < scramMechanism.minIterations()) {
- throw new TerseFailure(s"The 'iterations' value must be >= ${scramMechanism.minIterations()} for add-scram")
+ throw new TerseFailure(s"The 'iterations' value must be >= ${scramMechanism.minIterations()} for add-scram")
}
if (iterations > scramMechanism.maxIterations()) {
- throw new TerseFailure(s"The 'iterations' value must be <= ${scramMechanism.maxIterations()} for add-scram")
+ throw new TerseFailure(s"The 'iterations' value must be <= ${scramMechanism.maxIterations()} for add-scram")
Review Comment:
Looks like you have extra two spaces!
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -506,8 +507,8 @@ object StorageTool extends Logging {
val metadataRecords = new util.ArrayList[ApiMessageAndVersion]
metadataRecords.add(new ApiMessageAndVersion(new FeatureLevelRecord().
- setName(MetadataVersion.FEATURE_NAME).
- setFeatureLevel(metadataVersion.featureLevel()), 0.toShort))
+ setName(MetadataVersion.FEATURE_NAME).
+ setFeatureLevel(metadataVersion.featureLevel()), 0.toShort))
Review Comment:
Incorrect indentation.
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -347,22 +348,22 @@ object StorageTool extends Logging {
}
def getSaltedPassword(
- argMap: Map[String,String],
- scramMechanism : ScramMechanism,
- salt : Array[Byte],
- iterations: Int
- ) : Array[Byte] = {
+ argMap: Map[String,String],
+ scramMechanism : ScramMechanism,
+ salt : Array[Byte],
+ iterations: Int
+ ) : Array[Byte] = {
if (argMap.contains("password")) {
if (argMap.contains("saltedpassword")) {
- throw new TerseFailure(s"You must only supply one of 'password' or 'saltedpassword' to add-scram")
+ throw new TerseFailure(s"You must only supply one of 'password' or 'saltedpassword' to add-scram")
}
new ScramFormatter(scramMechanism).saltedPassword(argMap("password"), salt, iterations)
} else {
if (!argMap.contains("saltedpassword")) {
- throw new TerseFailure(s"You must supply one of 'password' or 'saltedpassword' to add-scram")
+ throw new TerseFailure(s"You must supply one of 'password' or 'saltedpassword' to add-scram")
}
if (!argMap.contains("salt")) {
- throw new TerseFailure(s"You must supply 'salt' with 'saltedpassword' to add-scram")
+ throw new TerseFailure(s"You must supply 'salt' with 'saltedpassword' to add-scram")
Review Comment:
Extra spaces.
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -698,6 +663,42 @@ object StorageTool extends Logging {
voterSet
}
+ def createStandaloneVoterMap(config: KafkaConfig): util.Map[ListenerName, InetSocketAddress] = {
+ val advertisedListenerEndpoints = config.effectiveAdvertisedControllerListeners
+ val listeners: util.Map[ListenerName, InetSocketAddress] = new util.HashMap()
+ advertisedListenerEndpoints.foreach(endpoint => {
+ val host: String = endpoint.host
+ listeners.put(endpoint.listenerName, new InetSocketAddress(host, endpoint.port))
+ })
+ listeners
+ }
+
+ private def parseControllerQuorumVotersMap(controllerQuorumVoterMap: util.Map[Integer, InetSocketAddress],
+ metaProperties: MetaProperties,
+ config: KafkaConfig): util.Map[ListenerName, InetSocketAddress] = {
+ val listeners: util.Map[ListenerName, InetSocketAddress] = new util.HashMap()
+ controllerQuorumVoterMap.keySet().forEach(replicaId => {
+ if (metaProperties.nodeId().getAsInt == replicaId) {
+ val listenerNameOption = config.effectiveAdvertisedControllerListeners.
Review Comment:
We should assume that the listener name is the default listener name. The default listener name is the first listener in https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/KafkaConfig.scala#L730-L737.
There is also this code for an example of getting the first listener name: https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/raft/RaftManager.scala#L267
There is one validation that we should do for the local replica. The local replica's default listener (name, host and port) matches the entry specified in `--controller-quorum-voters`.
##########
core/src/test/scala/unit/kafka/tools/StorageToolTest.scala:
##########
@@ -713,7 +733,7 @@ Found problem:
StorageTool.main(args)
} catch {
case e: StorageToolTestException => assertEquals("", exitString)
- assertEquals(0, exitStatus)
+ assertEquals(0, exitStatus)
Review Comment:
Incorrect indentation!
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -199,7 +200,7 @@ object StorageTool extends Logging {
val level: java.lang.Short = specifiedFeatures.getOrElse(feature.featureName, feature.defaultValue(metadataVersionForDefault))
// Only set feature records for levels greater than 0. 0 is assumed if there is no record. Throw an error if level < 0.
if (level != 0) {
- allNonZeroFeaturesAndLevels.append(feature.fromFeatureLevel(level, unstableFeatureVersionsEnabled))
+ allNonZeroFeaturesAndLevels.append(feature.fromFeatureLevel(level, unstableFeatureVersionsEnabled))
Review Comment:
Looks like we are missing a space.
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -347,22 +348,22 @@ object StorageTool extends Logging {
}
def getSaltedPassword(
- argMap: Map[String,String],
- scramMechanism : ScramMechanism,
- salt : Array[Byte],
- iterations: Int
- ) : Array[Byte] = {
+ argMap: Map[String,String],
+ scramMechanism : ScramMechanism,
+ salt : Array[Byte],
+ iterations: Int
+ ) : Array[Byte] = {
Review Comment:
There shouldn't be a space between `)` and `:`.
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -377,12 +378,12 @@ object StorageTool extends Logging {
val formatter = new ScramFormatter(scramMechanism)
new UserScramCredentialRecord()
- .setName(name)
- .setMechanism(scramMechanism.`type`)
- .setSalt(salt)
- .setStoredKey(formatter.storedKey(formatter.clientKey(saltedPassword)))
- .setServerKey(formatter.serverKey(saltedPassword))
- .setIterations(iterations)
+ .setName(name)
+ .setMechanism(scramMechanism.`type`)
+ .setSalt(salt)
+ .setStoredKey(formatter.storedKey(formatter.clientKey(saltedPassword)))
+ .setServerKey(formatter.serverKey(saltedPassword))
+ .setIterations(iterations)
Review Comment:
For Scala in Kafka we indent two spaces.
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -698,6 +663,42 @@ object StorageTool extends Logging {
voterSet
}
+ def createStandaloneVoterMap(config: KafkaConfig): util.Map[ListenerName, InetSocketAddress] = {
+ val advertisedListenerEndpoints = config.effectiveAdvertisedControllerListeners
+ val listeners: util.Map[ListenerName, InetSocketAddress] = new util.HashMap()
+ advertisedListenerEndpoints.foreach(endpoint => {
+ val host: String = endpoint.host
+ listeners.put(endpoint.listenerName, new InetSocketAddress(host, endpoint.port))
+ })
+ listeners
+ }
+
+ private def parseControllerQuorumVotersMap(controllerQuorumVoterMap: util.Map[Integer, InetSocketAddress],
+ metaProperties: MetaProperties,
+ config: KafkaConfig): util.Map[ListenerName, InetSocketAddress] = {
Review Comment:
Incorrect indentation:
```java
private def parseControllerQuorumVotersMap(
controllerQuorumVoterMap: util.Map[Integer, InetSocketAddress],
metaProperties: MetaProperties,
config: KafkaConfig
): util.Map[ListenerName, InetSocketAddress] = {
```
Aren't you losing information if you map from `util.Map[Integer, InetSocketAddress]` to `util.Map[ListenerName, InetSocketAddress]`? Why are you removing replicas that are not the local replica? The `VoterSet` must contain all of the voters in `--controller-quorum-voters` not just the local replica.
Why would Kafka require all of the voters in `--controller-quorum-voters` to only use the local voter?
##########
core/src/test/scala/unit/kafka/tools/StorageToolTest.scala:
##########
@@ -598,9 +618,9 @@ Found problem:
// Validate we can add multiple SCRAM creds.
scramRecords = parseAddScram("-S",
- "SCRAM-SHA-256=[name=alice,salt=\"MWx2NHBkbnc0ZndxN25vdGN4bTB5eTFrN3E=\",saltedpassword=\"mT0yyUUxnlJaC99HXgRTSYlbuqa4FSGtJCJfTMvjYCE=\",iterations=8192]",
- "-S",
- "SCRAM-SHA-256=[name=george,salt=\"MWx2NHBkbnc0ZndxN25vdGN4bTB5eTFrN3E=\",saltedpassword=\"mT0yyUUxnlJaC99HXgRTSYlbuqa4FSGtJCJfTMvjYCE=\",iterations=8192]")
+ "SCRAM-SHA-256=[name=alice,salt=\"MWx2NHBkbnc0ZndxN25vdGN4bTB5eTFrN3E=\",saltedpassword=\"mT0yyUUxnlJaC99HXgRTSYlbuqa4FSGtJCJfTMvjYCE=\",iterations=8192]",
+ "-S",
+ "SCRAM-SHA-256=[name=george,salt=\"MWx2NHBkbnc0ZndxN25vdGN4bTB5eTFrN3E=\",saltedpassword=\"mT0yyUUxnlJaC99HXgRTSYlbuqa4FSGtJCJfTMvjYCE=\",iterations=8192]")
Review Comment:
Incorrect indentation.
##########
core/src/test/scala/unit/kafka/tools/StorageToolTest.scala:
##########
@@ -342,8 +340,8 @@ Found problem:
val config = new KafkaConfig(newSelfManagedProperties())
assertEquals("Cluster ID string invalid does not appear to be a valid UUID: " +
"Input string `invalid` decoded as 5 bytes, which is not equal to the expected " +
- "16 bytes of a base64-encoded UUID", assertThrows(classOf[TerseFailure],
- () => StorageTool.buildMetadataProperties("invalid", config)).getMessage)
+ "16 bytes of a base64-encoded UUID", assertThrows(classOf[TerseFailure],
+ () => StorageTool.buildMetadataProperties("invalid", config)).getMessage)
Review Comment:
Indentation doesn't look correct.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16518; Adding standalone argument for storage [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on PR #16325:
URL: https://github.com/apache/kafka/pull/16325#issuecomment-2266231744
We ended up doing this in: #16669
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16518; Adding standalone argument for storage [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #16325:
URL: https://github.com/apache/kafka/pull/16325#discussion_r1661035915
##########
core/src/test/scala/unit/kafka/tools/StorageToolTest.scala:
##########
@@ -285,6 +362,49 @@ Found problem:
"Expected the default metadata.version to be 3.3-IV2")
}
+ @Test
+ def testStandaloneModeWithArguments(): Unit = {
+ val namespace = StorageTool.parseArguments(Array("format", "-c", "config.props", "-t", "XcZZOzUqS4yPOjhMQB6JAT",
+ "-s"))
+ val config = Mockito.spy(new KafkaConfig(TestUtils.createBrokerConfig(1, null)))
+ val exitCode = StorageTool.runFormatCommand(namespace, config)
+ val tempDirs = config.logDirs
+ tempDirs.foreach(tempDir => {
+ val checkpointDir = tempDir + "/" + CLUSTER_METADATA_TOPIC_NAME
+ val checkpointFilePath = Snapshots.snapshotPath(Paths.get(checkpointDir), BOOTSTRAP_SNAPSHOT_ID)
+ assertTrue(checkpointFilePath.toFile.exists)
+ assertTrue(Utils.readFileAsString(checkpointFilePath.toFile.getPath).contains("localhost"))
+ Utils.delete(new File(tempDir))
+ })
+ assertEquals(0, exitCode)
+ }
+
+// @Test TODO
+// def testControllerQuorumVotersWithArguments(): Unit = {
+// val namespace = StorageTool.parseArguments(Array("format", "-c", "config.props", "-t", "XcZZOzUqS4yQQjhMQB6JAT",
+// "--controller-quorum-voters", "1@localhost:9092"))
+// val config = Mockito.spy(new KafkaConfig(TestUtils.createBrokerConfig(1, null)))
Review Comment:
Not sure how to set controller listener
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16518; Adding standalone argument for storage [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #16325:
URL: https://github.com/apache/kafka/pull/16325#discussion_r1665644755
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -102,6 +111,23 @@ object StorageTool extends Logging {
setClusterId(clusterId).
setNodeId(config.nodeId).
build()
+ val standaloneMode = namespace.getBoolean("standalone")
+ var advertisedListenerEndpoints: collection.Seq[kafka.cluster.EndPoint] = List()
+
+ val controllersQuorumVoters = namespace.getString("controller_quorum_voters")
+ if(standaloneMode && controllersQuorumVoters != null) {
+ throw new TerseFailure("Both --standalone and --controller-quorum-voters were set. Only one of the two flags can be set.")
+ }
+
+ if (standaloneMode) {
+ advertisedListenerEndpoints = config.effectiveAdvertisedBrokerListeners
+ } else if(controllersQuorumVoters != null) {
+ if (!validateControllerQuorumVoters(controllersQuorumVoters)) {
+ throw new TerseFailure("Expected schema for --controller-quorum-voters is <replica-id>[-<replica-directory-id>]@<host>:<port>")
+ }
+ advertisedListenerEndpoints = config.effectiveAdvertisedControllerListeners
Review Comment:
Fixed.
##########
core/src/test/scala/unit/kafka/tools/StorageToolTest.scala:
##########
@@ -285,6 +362,49 @@ Found problem:
"Expected the default metadata.version to be 3.3-IV2")
}
+ @Test
+ def testStandaloneModeWithArguments(): Unit = {
+ val namespace = StorageTool.parseArguments(Array("format", "-c", "config.props", "-t", "XcZZOzUqS4yPOjhMQB6JAT",
+ "-s"))
+ val config = Mockito.spy(new KafkaConfig(TestUtils.createBrokerConfig(1, null)))
+ val exitCode = StorageTool.runFormatCommand(namespace, config)
+ val tempDirs = config.logDirs
+ tempDirs.foreach(tempDir => {
+ val checkpointDir = tempDir + "/" + CLUSTER_METADATA_TOPIC_NAME
+ val checkpointFilePath = Snapshots.snapshotPath(Paths.get(checkpointDir), BOOTSTRAP_SNAPSHOT_ID)
+ assertTrue(checkpointFilePath.toFile.exists)
+ assertTrue(Utils.readFileAsString(checkpointFilePath.toFile.getPath).contains("localhost"))
+ Utils.delete(new File(tempDir))
+ })
+ assertEquals(0, exitCode)
+ }
+
+// @Test TODO
+// def testControllerQuorumVotersWithArguments(): Unit = {
+// val namespace = StorageTool.parseArguments(Array("format", "-c", "config.props", "-t", "XcZZOzUqS4yQQjhMQB6JAT",
+// "--controller-quorum-voters", "1@localhost:9092"))
+// val config = Mockito.spy(new KafkaConfig(TestUtils.createBrokerConfig(1, null)))
Review Comment:
Not sure how to set controller listener
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16518 : Adding standalone argument for storage [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on PR #16325:
URL: https://github.com/apache/kafka/pull/16325#issuecomment-2170896788
@jsancio Output looks like below.
./bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties -s
metaPropertiesEnsemble=MetaPropertiesEnsemble(metadataLogDir=Optional.empty, dirs={/tmp/kraft-combined-logs: EMPTY})
Snapshot written to /tmp/kraft-combined-logs/__cluster_metadata-0
Formatting /tmp/kraft-combined-logs with metadata.version 3.7-IV4.
% cat /tmp/kraft-combined-logs/__cluster_metadata-0/00000000000000000000-0000000000.checkpoint
�[� �%�%��������������*
l�#�� �ﱃS�=��7
PLAINTEXT
localhost#�?)��� �% �% �������������%
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16518 : Adding standalone argument for storage [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on PR #16325:
URL: https://github.com/apache/kafka/pull/16325#issuecomment-2173806713
> cat /tmp/kraft-combined-logs/__cluster_metadata-0/00000000000000000000-0000000000.checkpoint
@muralibasani You can use `bin/kafka-dump-log --cluster-metadata-decoder --files /tmp/kraft-combined-logs/__cluster_metadata-0/00000000000000000000-0000000000.checkpoint`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16518; Adding standalone argument for storage [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #16325:
URL: https://github.com/apache/kafka/pull/16325#discussion_r1651072253
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -102,6 +109,11 @@ object StorageTool extends Logging {
setClusterId(clusterId).
setNodeId(config.nodeId).
build()
+ val standaloneMode = namespace.getBoolean("standalone")
+ val advertisedListenerEndpoints: scala.collection.Seq[kafka.cluster.EndPoint] = config.effectiveAdvertisedListeners
+
+ // effectiveAdvertisedControllerListeners to be added
Review Comment:
I think we should wait. The bootstrap checkpoint file is incorrect without https://github.com/apache/kafka/pull/16235https://github.com/apache/kafka/pull/16235
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16518 : Adding standalone argument for storage [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #16325:
URL: https://github.com/apache/kafka/pull/16325#discussion_r1642996134
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -46,6 +51,8 @@ import scala.collection.mutable.ArrayBuffer
object StorageTool extends Logging {
+ val clusterMetadataDir: String = "/__cluster_metadata-0"
Review Comment:
This is defined in https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/internals/Topic.java#L31-L35
And we resolve the actual directory like this: https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/raft/RaftManager.scala#L57-L62
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -226,7 +238,12 @@ object StorageTool extends Logging {
help(s"A KRaft release version to use for the initial metadata.version. The minimum is ${MetadataVersion.IBP_3_0_IV0}, the default is ${MetadataVersion.LATEST_PRODUCTION}")
formatParser.addArgument("--feature", "-f").
help("A feature upgrade we should perform, in feature=level format. For example: `metadata.version=5`.").
- action(append());
+ action(append())
+ formatParser.addArgument("--standalone", "-s").
+ help("This command will 1) create a meta.properties file in metadata.log.dir with a randomly generated" +
+ " directory.id, 2) create a snapshot at 00000000000000000000-0000000000.checkpoint with the necessary control" +
+ " records (KRaftVersionRecord and VotersRecord) to make this Kafka node the only voter for the quorum").
Review Comment:
I think this is too low level. Most users wont understand what this mean. How about?
> This flag will bootstrap the controller in standalone as the only KRaft controller if the Kafka cluster. Use the --controller-quorum-voters flag instead to bootstrap a controller cluster with more than one controller.
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -548,6 +570,7 @@ object StorageTool extends Logging {
if (metaPropertiesEnsemble.emptyLogDirs().isEmpty) {
stream.println("All of the log directories are already formatted.")
} else {
+ val directoryId = copier.generateValidDirectoryId()
Review Comment:
This doesn't look correct. We need to use the directoryId of the metadata log directory. After the `copier` has generated all of the directory ids in `copier.writeLogDirChanges()`, you want to read the directory id in the `meta.properties` of the `metadataLogDir`.
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -589,4 +616,56 @@ object StorageTool extends Logging {
(nameAndLevel._1, nameAndLevel._2)
}.toMap
}
-}
+
+ def writeCheckpointFile(stream: PrintStream, logDir: String, directoryId: Uuid,
+ advertisedListenerEndpoints: scala.collection.Seq[kafka.cluster.EndPoint]): Unit = {
+ val snapshotCheckpointDir = logDir + clusterMetadataDir
+
+ // Ensure the directory exists
+ val snapshotDir = Paths.get(snapshotCheckpointDir)
+ if (!Files.exists(snapshotDir)) {
+ Files.createDirectories(snapshotDir)
+ }
+
+ // Create the full path for the checkpoint file
+ val checkpointFilePath = snapshotDir.resolve(snapshotDir)
+
+ // Create the raw snapshot writer
+ val rawSnapshotWriter = FileRawSnapshotWriter.create(checkpointFilePath, new OffsetAndEpoch(0, 0))
+
+ if(advertisedListenerEndpoints.nonEmpty){
+ val voterSet: VoterSet = getVoterSet(directoryId, advertisedListenerEndpoints)
+
+ val builder = new RecordsSnapshotWriter.Builder()
+ .setKraftVersion(1)
+ .setVoterSet(Optional.of(voterSet))
+ .setRawSnapshotWriter(rawSnapshotWriter).build(new StringSerde)
+
+ // Close the builder to finalize the snapshot
+ builder.freeze()
+ builder.close()
+ stream.println(s"Snapshot written to $checkpointFilePath")
+ }
+ }
+
+ private def getVoterSet(directoryId: Uuid, advertisedListenerEndpoints: collection.Seq[EndPoint]) = {
+ // Create a VotersRecord endpoint collection
+ val endpointCollection = new VotersRecord.EndpointCollection()
+ advertisedListenerEndpoints.foreach(endpoint => {
+ endpointCollection.add(new VotersRecord.Endpoint().setName(endpoint.listenerName.value())
+ .setHost(endpoint.host).setPort(endpoint.port))
+ })
+
+ // Create voters
+ val voters: util.List[VotersRecord.Voter] = new util.ArrayList()
+ voters.add(new VotersRecord.Voter()
+ .setVoterId(1)
+ .setVoterDirectoryId(directoryId)
+ .setEndpoints(endpointCollection))
+
+ // Create Voter set
+ val voterNewRecord = new VotersRecord().setVersion(1).setVoters(voters)
+ val voterSet = VoterSet.fromVotersRecord(voterNewRecord)
+ voterSet
+ }
+}
Review Comment:
Missing newline character.
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -589,4 +616,56 @@ object StorageTool extends Logging {
(nameAndLevel._1, nameAndLevel._2)
}.toMap
}
-}
+
+ def writeCheckpointFile(stream: PrintStream, logDir: String, directoryId: Uuid,
+ advertisedListenerEndpoints: scala.collection.Seq[kafka.cluster.EndPoint]): Unit = {
+ val snapshotCheckpointDir = logDir + clusterMetadataDir
+
+ // Ensure the directory exists
+ val snapshotDir = Paths.get(snapshotCheckpointDir)
+ if (!Files.exists(snapshotDir)) {
+ Files.createDirectories(snapshotDir)
+ }
+
+ // Create the full path for the checkpoint file
+ val checkpointFilePath = snapshotDir.resolve(snapshotDir)
+
+ // Create the raw snapshot writer
+ val rawSnapshotWriter = FileRawSnapshotWriter.create(checkpointFilePath, new OffsetAndEpoch(0, 0))
+
+ if(advertisedListenerEndpoints.nonEmpty){
+ val voterSet: VoterSet = getVoterSet(directoryId, advertisedListenerEndpoints)
+
+ val builder = new RecordsSnapshotWriter.Builder()
+ .setKraftVersion(1)
+ .setVoterSet(Optional.of(voterSet))
+ .setRawSnapshotWriter(rawSnapshotWriter).build(new StringSerde)
+
+ // Close the builder to finalize the snapshot
+ builder.freeze()
+ builder.close()
Review Comment:
Let's use `try ... finally ...` so that the builder is always closed.
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -589,4 +616,56 @@ object StorageTool extends Logging {
(nameAndLevel._1, nameAndLevel._2)
}.toMap
}
-}
+
+ def writeCheckpointFile(stream: PrintStream, logDir: String, directoryId: Uuid,
+ advertisedListenerEndpoints: scala.collection.Seq[kafka.cluster.EndPoint]): Unit = {
+ val snapshotCheckpointDir = logDir + clusterMetadataDir
+
+ // Ensure the directory exists
+ val snapshotDir = Paths.get(snapshotCheckpointDir)
+ if (!Files.exists(snapshotDir)) {
+ Files.createDirectories(snapshotDir)
+ }
+
+ // Create the full path for the checkpoint file
+ val checkpointFilePath = snapshotDir.resolve(snapshotDir)
+
+ // Create the raw snapshot writer
+ val rawSnapshotWriter = FileRawSnapshotWriter.create(checkpointFilePath, new OffsetAndEpoch(0, 0))
+
+ if(advertisedListenerEndpoints.nonEmpty){
+ val voterSet: VoterSet = getVoterSet(directoryId, advertisedListenerEndpoints)
+
+ val builder = new RecordsSnapshotWriter.Builder()
+ .setKraftVersion(1)
+ .setVoterSet(Optional.of(voterSet))
+ .setRawSnapshotWriter(rawSnapshotWriter).build(new StringSerde)
+
+ // Close the builder to finalize the snapshot
+ builder.freeze()
+ builder.close()
+ stream.println(s"Snapshot written to $checkpointFilePath")
+ }
+ }
+
+ private def getVoterSet(directoryId: Uuid, advertisedListenerEndpoints: collection.Seq[EndPoint]) = {
+ // Create a VotersRecord endpoint collection
+ val endpointCollection = new VotersRecord.EndpointCollection()
+ advertisedListenerEndpoints.foreach(endpoint => {
+ endpointCollection.add(new VotersRecord.Endpoint().setName(endpoint.listenerName.value())
+ .setHost(endpoint.host).setPort(endpoint.port))
+ })
+
+ // Create voters
+ val voters: util.List[VotersRecord.Voter] = new util.ArrayList()
+ voters.add(new VotersRecord.Voter()
+ .setVoterId(1)
+ .setVoterDirectoryId(directoryId)
+ .setEndpoints(endpointCollection))
+
+ // Create Voter set
+ val voterNewRecord = new VotersRecord().setVersion(1).setVoters(voters)
+ val voterSet = VoterSet.fromVotersRecord(voterNewRecord)
Review Comment:
See. Let's avoid creating a `VotersRecord` just to get a `VoterSet`. Feel free to add a static method to `VoterSet` like `VoterSet fromMap(Map<Integer, VoterNode> voters)`.
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -102,6 +109,11 @@ object StorageTool extends Logging {
setClusterId(clusterId).
setNodeId(config.nodeId).
build()
+ val standaloneMode = namespace.getBoolean("standalone")
+ val advertisedListenerEndpoints: scala.collection.Seq[kafka.cluster.EndPoint] = config.effectiveAdvertisedListeners
+
+ // effectiveAdvertisedControllerListeners to be added
Review Comment:
Okay. Are you going to wait for this PR to get merged and update this PR then?
https://github.com/apache/kafka/pull/16235
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -589,4 +616,56 @@ object StorageTool extends Logging {
(nameAndLevel._1, nameAndLevel._2)
}.toMap
}
-}
+
+ def writeCheckpointFile(stream: PrintStream, logDir: String, directoryId: Uuid,
+ advertisedListenerEndpoints: scala.collection.Seq[kafka.cluster.EndPoint]): Unit = {
+ val snapshotCheckpointDir = logDir + clusterMetadataDir
+
+ // Ensure the directory exists
+ val snapshotDir = Paths.get(snapshotCheckpointDir)
+ if (!Files.exists(snapshotDir)) {
+ Files.createDirectories(snapshotDir)
+ }
+
+ // Create the full path for the checkpoint file
+ val checkpointFilePath = snapshotDir.resolve(snapshotDir)
+
+ // Create the raw snapshot writer
+ val rawSnapshotWriter = FileRawSnapshotWriter.create(checkpointFilePath, new OffsetAndEpoch(0, 0))
+
+ if(advertisedListenerEndpoints.nonEmpty){
+ val voterSet: VoterSet = getVoterSet(directoryId, advertisedListenerEndpoints)
+
+ val builder = new RecordsSnapshotWriter.Builder()
+ .setKraftVersion(1)
+ .setVoterSet(Optional.of(voterSet))
+ .setRawSnapshotWriter(rawSnapshotWriter).build(new StringSerde)
+
+ // Close the builder to finalize the snapshot
+ builder.freeze()
+ builder.close()
+ stream.println(s"Snapshot written to $checkpointFilePath")
+ }
+ }
+
+ private def getVoterSet(directoryId: Uuid, advertisedListenerEndpoints: collection.Seq[EndPoint]) = {
+ // Create a VotersRecord endpoint collection
+ val endpointCollection = new VotersRecord.EndpointCollection()
+ advertisedListenerEndpoints.foreach(endpoint => {
+ endpointCollection.add(new VotersRecord.Endpoint().setName(endpoint.listenerName.value())
+ .setHost(endpoint.host).setPort(endpoint.port))
+ })
+
+ // Create voters
+ val voters: util.List[VotersRecord.Voter] = new util.ArrayList()
+ voters.add(new VotersRecord.Voter()
+ .setVoterId(1)
Review Comment:
You can't assume that the voter id is 1.
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -589,4 +616,56 @@ object StorageTool extends Logging {
(nameAndLevel._1, nameAndLevel._2)
}.toMap
}
-}
+
+ def writeCheckpointFile(stream: PrintStream, logDir: String, directoryId: Uuid,
+ advertisedListenerEndpoints: scala.collection.Seq[kafka.cluster.EndPoint]): Unit = {
+ val snapshotCheckpointDir = logDir + clusterMetadataDir
+
+ // Ensure the directory exists
+ val snapshotDir = Paths.get(snapshotCheckpointDir)
+ if (!Files.exists(snapshotDir)) {
+ Files.createDirectories(snapshotDir)
+ }
+
+ // Create the full path for the checkpoint file
+ val checkpointFilePath = snapshotDir.resolve(snapshotDir)
Review Comment:
What does this mean?
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -589,4 +616,56 @@ object StorageTool extends Logging {
(nameAndLevel._1, nameAndLevel._2)
}.toMap
}
-}
+
+ def writeCheckpointFile(stream: PrintStream, logDir: String, directoryId: Uuid,
+ advertisedListenerEndpoints: scala.collection.Seq[kafka.cluster.EndPoint]): Unit = {
+ val snapshotCheckpointDir = logDir + clusterMetadataDir
+
+ // Ensure the directory exists
+ val snapshotDir = Paths.get(snapshotCheckpointDir)
+ if (!Files.exists(snapshotDir)) {
+ Files.createDirectories(snapshotDir)
+ }
+
+ // Create the full path for the checkpoint file
+ val checkpointFilePath = snapshotDir.resolve(snapshotDir)
+
+ // Create the raw snapshot writer
+ val rawSnapshotWriter = FileRawSnapshotWriter.create(checkpointFilePath, new OffsetAndEpoch(0, 0))
+
+ if(advertisedListenerEndpoints.nonEmpty){
+ val voterSet: VoterSet = getVoterSet(directoryId, advertisedListenerEndpoints)
+
+ val builder = new RecordsSnapshotWriter.Builder()
+ .setKraftVersion(1)
+ .setVoterSet(Optional.of(voterSet))
+ .setRawSnapshotWriter(rawSnapshotWriter).build(new StringSerde)
Review Comment:
Add newline between the characters `)` and `.`. E.g.
```java
.setRawSnapshotWriter(rawSnapshotWriter)
.build(new StringSerde)
```
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -507,10 +524,13 @@ object StorageTool extends Logging {
directories: Seq[String],
metaProperties: MetaProperties,
metadataVersion: MetadataVersion,
- ignoreFormatted: Boolean
+ ignoreFormatted: Boolean,
+ standaloneMode: Boolean,
+ advertisedListenerEndpoints: scala.collection.Seq[kafka.cluster.EndPoint]
Review Comment:
This is not general enough. How are you planning to implement --controller-quorum-voters with this signature? Why do you need `standaloneMode` if you are passing the local endpoints?
##########
core/src/main/scala/kafka/tools/StorageTool.scala:
##########
@@ -589,4 +616,56 @@ object StorageTool extends Logging {
(nameAndLevel._1, nameAndLevel._2)
}.toMap
}
-}
+
+ def writeCheckpointFile(stream: PrintStream, logDir: String, directoryId: Uuid,
+ advertisedListenerEndpoints: scala.collection.Seq[kafka.cluster.EndPoint]): Unit = {
+ val snapshotCheckpointDir = logDir + clusterMetadataDir
+
+ // Ensure the directory exists
+ val snapshotDir = Paths.get(snapshotCheckpointDir)
+ if (!Files.exists(snapshotDir)) {
+ Files.createDirectories(snapshotDir)
+ }
+
+ // Create the full path for the checkpoint file
+ val checkpointFilePath = snapshotDir.resolve(snapshotDir)
+
+ // Create the raw snapshot writer
+ val rawSnapshotWriter = FileRawSnapshotWriter.create(checkpointFilePath, new OffsetAndEpoch(0, 0))
Review Comment:
Instead of hard-coding `OffsetAndEpoch(0, 0)` let's add a constant to `o.a.k.s.Snapshots` named `BOOTSTRAP_SNAPSHOT_ID`.
##########
core/src/test/scala/unit/kafka/tools/StorageToolTest.scala:
##########
@@ -182,23 +188,60 @@ Found problem:
val stream = new ByteArrayOutputStream()
val bootstrapMetadata = StorageTool.buildBootstrapMetadata(MetadataVersion.latestTesting(), None, "test format command")
assertEquals(0, StorageTool.
- formatCommand(new PrintStream(stream), Seq(tempDir.toString), metaProperties, bootstrapMetadata, MetadataVersion.latestTesting(), ignoreFormatted = false))
+ formatCommand(new PrintStream(stream), Seq(tempDir.toString), metaProperties, bootstrapMetadata,
+ MetadataVersion.latestTesting(), ignoreFormatted = false, standaloneMode = false,
+ advertisedListenerEndpoints = scala.collection.Seq.empty[kafka.cluster.EndPoint]))
assertTrue(stringAfterFirstLine(stream.toString()).startsWith("Formatting %s".format(tempDir)))
try assertEquals(1, StorageTool.
- formatCommand(new PrintStream(new ByteArrayOutputStream()), Seq(tempDir.toString), metaProperties, bootstrapMetadata, MetadataVersion.latestTesting(), ignoreFormatted = false)) catch {
+ formatCommand(new PrintStream(new ByteArrayOutputStream()), Seq(tempDir.toString), metaProperties,
+ bootstrapMetadata, MetadataVersion.latestTesting(), ignoreFormatted = false, standaloneMode = false,
+ advertisedListenerEndpoints = scala.collection.Seq.empty[kafka.cluster.EndPoint])) catch {
case e: TerseFailure => assertEquals(s"Log directory ${tempDir} is already " +
"formatted. Use --ignore-formatted to ignore this directory and format the " +
"others.", e.getMessage)
}
val stream2 = new ByteArrayOutputStream()
assertEquals(0, StorageTool.
- formatCommand(new PrintStream(stream2), Seq(tempDir.toString), metaProperties, bootstrapMetadata, MetadataVersion.latestTesting(), ignoreFormatted = true))
+ formatCommand(new PrintStream(stream2), Seq(tempDir.toString), metaProperties, bootstrapMetadata,
+ MetadataVersion.latestTesting(), ignoreFormatted = true, standaloneMode = false,
+ advertisedListenerEndpoints = scala.collection.Seq.empty[kafka.cluster.EndPoint]))
assertEquals("All of the log directories are already formatted.%n".format(), stringAfterFirstLine(stream2.toString()))
} finally Utils.delete(tempDir)
}
+ @Test
+ def testFormatEmptyDirectoryWithStandaloneMode(): Unit = {
+ val tempDir = TestUtils.tempDir()
+ try {
+ val metaProperties = new MetaProperties.Builder().
+ setVersion(MetaPropertiesVersion.V1).
+ setClusterId("XcZZOzUqS5yHOjhMQB2JLR").
+ setNodeId(2).
+ build()
+ val stream = new ByteArrayOutputStream()
+ val bootstrapMetadata = StorageTool.buildBootstrapMetadata(MetadataVersion.latestTesting(), None,
+ "test format command")
+ val advertisedListenerEndpoints: scala.collection.Seq[kafka.cluster.EndPoint] = Seq.empty[EndPoint]
+ val host = "PLAINTEXT"
+ val newEndPoint = new EndPoint(host = host, port = 9092, listenerName = new ListenerName("PLAINTEXT"),
+ SecurityProtocol.PLAINTEXT)
+ val updatedAdvertisedListenerEndpoints: scala.collection.Seq[EndPoint] = advertisedListenerEndpoints :+ newEndPoint
+
+ assertEquals(0, StorageTool.
+ formatCommand(new PrintStream(stream), Seq(tempDir.toString), metaProperties, bootstrapMetadata,
+ MetadataVersion.latestTesting(), ignoreFormatted = false, standaloneMode = true,
+ updatedAdvertisedListenerEndpoints))
+ val checkpointDir = tempDir + clusterMetadataDir
+ assertTrue(stringAfterFirstLine(stream.toString()).startsWith("Snapshot written to %s".format(checkpointDir) +
+ "\n" + "Formatting %s".format(tempDir)))
+ val checkpointFilePath = Paths.get(checkpointDir + "/"+ checkpointFileName)
Review Comment:
Let's use `Snapshots.snapshotPath`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16518; Adding standalone argument for storage [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on PR #16325:
URL: https://github.com/apache/kafka/pull/16325#issuecomment-2265007883
Closing this, as the StorageTool class will be refactored in a different PR.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org