KAFKA-16518; Adding standalone argument for storage #16325

muralibasani · 2024-06-13T13:23:33Z

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)

muralibasani · 2024-06-15T21:47:56Z

@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#�?)�� % �% ��%

jsancio

Thanks for the changes @muralibasani . We should also implement --controller-quorum-voters in this PR as it will inform the signature needed to implement both flags.

core/src/main/scala/kafka/tools/StorageTool.scala

core/src/test/scala/unit/kafka/tools/StorageToolTest.scala

jsancio · 2024-06-17T16:09:32Z

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.

core/src/main/scala/kafka/tools/StorageTool.scala

core/src/test/scala/unit/kafka/tools/StorageToolTest.scala

muralibasani · 2024-07-01T13:12:38Z

@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.

jsancio

Thanks for the changes.

jsancio · 2024-07-02T14:35:04Z

core/src/main/scala/kafka/tools/StorageTool.scala

+      if (!validateControllerQuorumVoters(controllersQuorumVoters)) {
+        throw new TerseFailure("Expected schema for --controller-quorum-voters is <replica-id>[-<replica-directory-id>]@<host>:<port>")
+      }
+      advertisedListenerEndpoints = config.effectiveAdvertisedControllerListeners


This is not correct, right? We need to parse the string that you validated in validateControllerQuorumVoters(controllersQuorumVoters).

core/src/main/scala/kafka/tools/StorageTool.scala

jsancio · 2024-07-02T14:48:46Z

core/src/main/scala/kafka/tools/StorageTool.scala

+                     advertisedListenerEndpoints: scala.collection.Seq[kafka.cluster.EndPoint],
+                     controllersQuorumVoters: String


Did you consider making the input to this method just voters: VoterSet instead of advertisedListenerEndpoints and controllerQuorrumVoters.

When the --standalone flag is used the set of voters can be constructed from VoterSet.fromMap(createStandaloneVoterMap(nodeId, nodeDirectoryId, config.effectiveAdvertisedControllerListeners)).

When the --controller-quorum-voters flag is used the set of voters can be constructed from VoterSet.fromMap(parseVotersToMap(controllersQuorumVoters))

Tried considering but directoryId is derived only in format method. So updated format method with only listeners param.

jsancio · 2024-07-02T14:51:49Z

core/src/main/scala/kafka/tools/StorageTool.scala

+    }
+
+    if (standaloneMode) {
+      advertisedListenerEndpoints = config.effectiveAdvertisedBrokerListeners


The bootstrapping checkpoint should only be generated on controllers (config.processRoles.contains(ProcessRole.ControllerRole)). If the node is a broker only then the bootstrapping checkpoint should not get generated.

jsancio · 2024-07-08T15:49:49Z

core/src/main/scala/kafka/tools/StorageTool.scala

+    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)


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>

This means we would have to update

kafka/raft/src/main/java/org/apache/kafka/raft/QuorumConfig.java

Line 178 in 25d775b

public static Map<Integer, InetSocketAddress> parseVoterConnections(List<String> voterEntries) {

to return Uuid too ?

jsancio · 2024-07-08T15:50:01Z

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))


Looks like we are missing a space.

jsancio · 2024-07-08T15:51:01Z

core/src/main/scala/kafka/tools/StorageTool.scala

-                                          ) : UserScramCredentialRecord = {
+    mechanism: String,
+    config: String
+  ) : UserScramCredentialRecord = {


Extra space between ) and :. It should be ): User... = {

jsancio · 2024-07-08T15:51:25Z

core/src/main/scala/kafka/tools/StorageTool.scala

    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


Again, this is not the correct indentation.

jsancio · 2024-07-08T15:51:49Z

core/src/main/scala/kafka/tools/StorageTool.scala

-          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")


Looks like you have extra two spaces!

jsancio · 2024-07-08T15:58:06Z

core/src/main/scala/kafka/tools/StorageTool.scala

+  private def parseControllerQuorumVotersMap(controllerQuorumVoterMap: util.Map[Integer, InetSocketAddress],
+                           metaProperties: MetaProperties,
+                           config: KafkaConfig): util.Map[ListenerName, InetSocketAddress] = {


Incorrect indentation:

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?

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 specified in --controller-quorum-voters. If there is a match, it will write the 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?

jsancio · 2024-07-08T17:38:24Z

core/src/main/scala/kafka/tools/StorageTool.scala

+    val listeners: util.Map[ListenerName, InetSocketAddress] = new util.HashMap()
+    controllerQuorumVoterMap.keySet().forEach(replicaId => {
+      if (metaProperties.nodeId().getAsInt == replicaId) {
+        val listenerNameOption = config.effectiveAdvertisedControllerListeners.


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.

jsancio · 2024-07-08T17:40:30Z

core/src/test/scala/unit/kafka/tools/StorageToolTest.scala

-      "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)


Indentation doesn't look correct.

jsancio · 2024-07-08T17:41:49Z

core/src/test/scala/unit/kafka/tools/StorageToolTest.scala

-      "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]")


Incorrect indentation.

jsancio · 2024-07-08T17:42:01Z

core/src/test/scala/unit/kafka/tools/StorageToolTest.scala

-        assertEquals(0, exitStatus)
+      assertEquals(0, exitStatus)


Incorrect indentation!

muralibasani · 2024-08-02T09:54:58Z

Closing this, as the StorageTool class will be refactored in a different PR.

cmccabe · 2024-08-02T22:49:24Z

We ended up doing this in: #16669

muralibasani marked this pull request as ready for review June 13, 2024 16:52

cmccabe added the KIP-853 KRaft Controller Membership Changes label Jun 13, 2024

jsancio added the kraft label Jun 14, 2024

jsancio reviewed Jun 17, 2024

View reviewed changes

jsancio changed the title ~~KAFKA-16518 : Adding standalone argument for storage~~ KAFKA-16518; Adding standalone argument for storage Jun 17, 2024

jsancio self-assigned this Jun 17, 2024

muralibasani commented Jun 24, 2024

View reviewed changes

core/src/main/scala/kafka/tools/StorageTool.scala Outdated Show resolved Hide resolved

jsancio reviewed Jun 25, 2024

View reviewed changes

ahuang98 reviewed Jun 27, 2024

View reviewed changes

core/src/test/scala/unit/kafka/tools/StorageToolTest.scala Outdated Show resolved Hide resolved

muralibasani added 7 commits June 28, 2024 16:02

Fix review

448ddfe

Fix conflicts

14d845e

Update voterset, test

3c27ba3

Update voterset, test

fe66c00

Fix review

35510aa

Fix review

31e9762

moving pattern check to QuorumConfig

799bc77

muralibasani force-pushed the kafka-16518 branch from 51e8bc4 to 799bc77 Compare June 28, 2024 14:33

muralibasani added 3 commits July 1, 2024 11:19

Adding controller arg options

bc32d53

Adding controller arg test

98b3976

Adding controller arg test

99a8dcf

muralibasani requested review from jsancio and ahuang98 July 1, 2024 16:25

jsancio reviewed Jul 2, 2024

View reviewed changes

muralibasani added 4 commits July 3, 2024 13:06

Fix indents

4d234b3

Fix indents

34f9f49

Fix indents

21c022c

From review

b8a5ba3

Remove commented line

fee30ec

muralibasani requested a review from jsancio July 4, 2024 12:33

jsancio reviewed Jul 8, 2024

View reviewed changes

muralibasani closed this Aug 2, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-16518; Adding standalone argument for storage #16325

KAFKA-16518; Adding standalone argument for storage #16325

muralibasani commented Jun 13, 2024

muralibasani commented Jun 15, 2024

jsancio left a comment

jsancio commented Jun 17, 2024

muralibasani commented Jul 1, 2024

jsancio left a comment

jsancio Jul 2, 2024

muralibasani Jul 4, 2024

jsancio Jul 2, 2024

muralibasani Jul 4, 2024

jsancio Jul 2, 2024

jsancio Jul 8, 2024

muralibasani Jul 10, 2024

jsancio Jul 8, 2024

jsancio Jul 8, 2024

jsancio Jul 8, 2024

jsancio Jul 8, 2024

jsancio Jul 8, 2024

muralibasani Jul 10, 2024

jsancio Jul 8, 2024

jsancio Jul 8, 2024

jsancio Jul 8, 2024

jsancio Jul 8, 2024

muralibasani commented Aug 2, 2024

cmccabe commented Aug 2, 2024

		advertisedListenerEndpoints: scala.collection.Seq[kafka.cluster.EndPoint],
		controllersQuorumVoters: String

KAFKA-16518; Adding standalone argument for storage #16325

KAFKA-16518; Adding standalone argument for storage #16325

Conversation

muralibasani commented Jun 13, 2024

Committer Checklist (excluded from commit message)

muralibasani commented Jun 15, 2024

jsancio left a comment

Choose a reason for hiding this comment

jsancio commented Jun 17, 2024

muralibasani commented Jul 1, 2024

jsancio left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

muralibasani commented Aug 2, 2024

cmccabe commented Aug 2, 2024