KAFKA-20646: Use Locale.ROOT in String.format() calls in tools/ and s…

server-common/src/main/java/org/apache/kafka/common/ClientIdAndBroker.java

@@ -16,6 +16,8 @@
  */
 package org.apache.kafka.common;
 
+import java.util.Locale;
+
 /**
  * Convenience case class since (clientId, brokerInfo) pairs are used to create
  * SyncProducer Request Stats and SimpleConsumer Request and Response Stats.
@@ -33,6 +35,6 @@ public ClientIdAndBroker(String clientId, String brokerHost, int brokerPort) {
 
     @Override
     public String toString() {
-        return String.format("%s-%s-%d", clientId, brokerHost, brokerPort);
+        return String.format(Locale.ROOT, "%s-%s-%d", clientId, brokerHost, brokerPort);
     }
 }

server-common/src/main/java/org/apache/kafka/logger/StateChangeLogger.java

@@ -19,6 +19,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Locale;
+
 /**
  * Simple class that sets logIdent appropriately depending on whether the state change logger is being used in the
  * context of the broker (e.g. ReplicaManager and Partition).
@@ -29,7 +31,7 @@ public class StateChangeLogger {
     private final String logIdent;
 
     public StateChangeLogger(int brokerId) {
-        this.logIdent = String.format("[Broker id=%d] ", brokerId);
+        this.logIdent = String.format(Locale.ROOT, "[Broker id=%d] ", brokerId);
     }
 
     public void trace(String message) {

server-common/src/main/java/org/apache/kafka/server/common/CheckpointFile.java

@@ -32,6 +32,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.Locale;
 import java.util.Optional;
 
 /**
@@ -186,7 +187,7 @@ private int toInt(String line) throws IOException {
         }
 
         private IOException buildMalformedLineException(String line) {
-            return new IOException(String.format("Malformed line in checkpoint file [%s]: %s", location, line));
+            return new IOException(String.format(Locale.ROOT, "Malformed line in checkpoint file [%s]: %s", location, line));
         }
     }
 

server-common/src/main/java/org/apache/kafka/server/common/Feature.java

@@ -21,6 +21,7 @@
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 
 import static org.apache.kafka.server.common.UnitTestFeatureVersion.FV0.UT_FV0_0;
@@ -265,12 +266,12 @@ public static void validateDefaultValueAndLatestProductionValue(
         FeatureVersion latestProduction = feature.latestProduction;
 
         if (!feature.hasFeatureVersion(latestProduction)) {
-            throw new IllegalArgumentException(String.format("Feature %s has latest production version %s " +
+            throw new IllegalArgumentException(String.format(Locale.ROOT, "Feature %s has latest production version %s " +
                     "which is not one of its feature versions.", feature.name(), latestProduction));
         }
 
         if (latestProduction.featureLevel() < defaultVersion.featureLevel()) {
-            throw new IllegalArgumentException(String.format("Feature %s has latest production value %s " +
+            throw new IllegalArgumentException(String.format(Locale.ROOT, "Feature %s has latest production value %s " +
                     "smaller than its default version %s with latest production MV.",
                 feature.name(), latestProduction, defaultVersion));
         }
@@ -280,14 +281,14 @@ public static void validateDefaultValueAndLatestProductionValue(
             if (!dependencyFeatureName.equals(MetadataVersion.FEATURE_NAME)) {
                 Feature dependencyFeature = featureFromName(dependencyFeatureName);
                 if (!dependencyFeature.isProductionReady(dependency.getValue())) {
-                    throw new IllegalArgumentException(String.format("Feature %s has latest production FeatureVersion %s " +
+                    throw new IllegalArgumentException(String.format(Locale.ROOT, "Feature %s has latest production FeatureVersion %s " +
                             "with dependency %s that is not production ready. (%s latest production: %s)",
                         feature.name(), latestProduction, dependencyFeature.fromFeatureLevel(dependency.getValue(), true),
                         dependencyFeature, dependencyFeature.latestProduction));
                 }
             } else {
                 if (dependency.getValue() > MetadataVersion.LATEST_PRODUCTION.featureLevel()) {
-                    throw new IllegalArgumentException(String.format("Feature %s has latest production FeatureVersion %s " +
+                    throw new IllegalArgumentException(String.format(Locale.ROOT, "Feature %s has latest production FeatureVersion %s " +
                             "with MV dependency %s that is not production ready. (MV latest production: %s)",
                         feature.name(), latestProduction, MetadataVersion.fromFeatureLevel(dependency.getValue()),
                         MetadataVersion.LATEST_PRODUCTION));
@@ -302,15 +303,15 @@ public static void validateDefaultValueAndLatestProductionValue(
                 if (!dependencyFeatureName.equals(MetadataVersion.FEATURE_NAME)) {
                     Feature dependencyFeature = featureFromName(dependencyFeatureName);
                     if (dependency.getValue() > dependencyFeature.defaultLevel(metadataVersion)) {
-                        throw new IllegalArgumentException(String.format("Feature %s has default FeatureVersion %s " +
+                        throw new IllegalArgumentException(String.format(Locale.ROOT, "Feature %s has default FeatureVersion %s " +
                                 "when MV=%s with dependency %s that is behind its default version %s.",
                             feature.name(), defaultVersion, metadataVersion,
                             dependencyFeature.fromFeatureLevel(dependency.getValue(), true),
                             dependencyFeature.defaultVersion(metadataVersion)));
                     }
                 } else {
                     if (dependency.getValue() > defaultVersion.bootstrapMetadataVersion().featureLevel()) {
-                        throw new IllegalArgumentException(String.format("Feature %s has default FeatureVersion %s " +
+                        throw new IllegalArgumentException(String.format(Locale.ROOT, "Feature %s has default FeatureVersion %s " +
                                 "when MV=%s with MV dependency %s that is behind its bootstrap MV %s.",
                             feature.name(), defaultVersion, metadataVersion,
                             MetadataVersion.fromFeatureLevel(dependency.getValue()),

server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java

@@ -20,6 +20,7 @@
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 import java.util.regex.Pattern;
@@ -177,7 +178,7 @@ public enum MetadataVersion {
         if (subVersion.isEmpty()) {
             this.ibpVersion = release;
         } else {
-            this.ibpVersion = String.format("%s-%s", release, subVersion);
+            this.ibpVersion = String.format(Locale.ROOT, "%s-%s", release, subVersion);
         }
         this.didMetadataChange = didMetadataChange;
     }

server-common/src/main/java/org/apache/kafka/server/config/QuotaConfig.java

@@ -21,6 +21,7 @@
 import org.apache.kafka.common.security.scram.internals.ScramMechanism;
 
 import java.util.List;
+import java.util.Locale;
 import java.util.Set;
 
 import static org.apache.kafka.common.config.ConfigDef.Importance.LOW;
@@ -74,12 +75,12 @@ public class QuotaConfig {
 
     public static final String LEADER_REPLICATION_THROTTLED_RATE_CONFIG = "leader.replication.throttled.rate";
     public static final String LEADER_REPLICATION_THROTTLED_RATE_DOC = "A long representing the upper bound (bytes/sec) on replication traffic for leaders enumerated in the " +
-            String.format("property %s (for each topic). This property can be only set dynamically. It is suggested that the ", LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG) +
+            String.format(Locale.ROOT, "property %s (for each topic). This property can be only set dynamically. It is suggested that the ", LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG) +
             "limit be kept above 1MB/s for accurate behaviour.";
 
     public static final String FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG = "follower.replication.throttled.rate";
     public static final String FOLLOWER_REPLICATION_THROTTLED_RATE_DOC = "A long representing the upper bound (bytes/sec) on replication traffic for followers enumerated in the " +
-            String.format("property %s (for each topic). This property can be only set dynamically. It is suggested that the ", FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG) +
+            String.format(Locale.ROOT, "property %s (for each topic). This property can be only set dynamically. It is suggested that the ", FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG) +
             "limit be kept above 1MB/s for accurate behaviour.";
     public static final String REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG = "replica.alter.log.dirs.io.max.bytes.per.second";
     public static final String REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_DOC = "A long representing the upper bound (bytes/sec) on disk IO used for moving replica between log directories on the same broker. " +

server-common/src/main/java/org/apache/kafka/server/share/SharePartitionKey.java

@@ -21,6 +21,7 @@
 import org.apache.kafka.common.Uuid;
 
 import java.util.Arrays;
+import java.util.Locale;
 import java.util.Objects;
 
 /**
@@ -133,7 +134,7 @@ public String asCoordinatorKey() {
     }
 
     public static String asCoordinatorKey(String groupId, Uuid topicId, int partition) {
-        return String.format("%s:%s:%d", groupId, topicId, partition);
+        return String.format(Locale.ROOT, "%s:%s:%d", groupId, topicId, partition);
     }
 
     @Override

server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java

@@ -34,6 +34,7 @@
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
@@ -661,7 +662,7 @@ private static void validateGroupTopicPartitionData(String prefix, GroupTopicPar
             for (PartitionIdData partitionData : topicData.partitions()) {
                 if (partitionData.partition() < 0) {
                     throw new IllegalArgumentException(
-                        String.format("%s has invalid partitionId - %s %s %d", prefix, groupId, topicData.topicId(), partitionData.partition()));
+                        String.format(Locale.ROOT, "%s has invalid partitionId - %s %s %d", prefix, groupId, topicData.topicId(), partitionData.partition()));
                 }
             }
         }

server-common/src/main/java/org/apache/kafka/server/util/CommandLineUtils.java

@@ -25,6 +25,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Locale;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
@@ -81,7 +82,7 @@ public static void maybePrintHelpOrVersion(CommandDefaultOptions commandOpts, St
     public static void checkRequiredArgs(OptionParser parser, OptionSet options, OptionSpec<?>... requiredList) {
         for (OptionSpec<?> arg : requiredList) {
             if (!options.has(arg)) {
-                printUsageAndExit(parser, String.format("Missing required argument \"%s\"", arg));
+                printUsageAndExit(parser, String.format(Locale.ROOT, "Missing required argument \"%s\"", arg));
             }
         }
     }
@@ -96,7 +97,7 @@ public static void checkInvalidArgs(OptionParser parser,
         if (options.has(usedOption)) {
             for (OptionSpec<?> arg : invalidOptions) {
                 if (options.has(arg)) {
-                    printUsageAndExit(parser, String.format("Option \"%s\" can't be used with option \"%s\"", usedOption, arg));
+                    printUsageAndExit(parser, String.format(Locale.ROOT, "Option \"%s\" can't be used with option \"%s\"", usedOption, arg));
                 }
             }
         }
@@ -125,7 +126,7 @@ public static void checkInvalidArgsSet(OptionParser parser,
         if (usedOptions.stream().filter(options::has).count() == usedOptions.size()) {
             for (OptionSpec<?> arg : invalidOptions) {
                 if (options.has(arg)) {
-                    printUsageAndExit(parser, String.format("Option combination \"%s\" can't be used with option \"%s\"%s",
+                    printUsageAndExit(parser, String.format(Locale.ROOT, "Option combination \"%s\" can't be used with option \"%s\"%s",
                             usedOptions, arg, trailingAdditionalMessage.orElse("")));
                 }
             }
@@ -201,7 +202,7 @@ public static Properties parseKeyValueArgs(List<String> args, boolean acceptMiss
                 if (acceptMissingValue) {
                     props.put(split[0], "");
                 } else {
-                    throw new IllegalArgumentException(String.format("Missing value for key %s}", split[0]));
+                    throw new IllegalArgumentException(String.format(Locale.ROOT, "Missing value for key %s}", split[0]));
                 }
             } else {
                 props.put(split[0], split[1]);

server-common/src/main/java/org/apache/kafka/server/util/json/DecodeJson.java

@@ -24,6 +24,7 @@
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 
@@ -36,7 +37,7 @@ public interface DecodeJson<T> {
     T decode(JsonNode node) throws JsonMappingException;
 
     static JsonMappingException throwJsonMappingException(String expectedType, JsonNode node) {
-        return new JsonMappingException(null, String.format("Expected `%s` value, received %s", expectedType, node));
+        return new JsonMappingException(null, String.format(Locale.ROOT, "Expected `%s` value, received %s", expectedType, node));
     }
 
     final class DecodeBoolean implements DecodeJson<Boolean> {

server-common/src/main/java/org/apache/kafka/server/util/json/JsonValue.java

@@ -21,6 +21,7 @@
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 
+import java.util.Locale;
 import java.util.Optional;
 
 /**
@@ -54,7 +55,7 @@ default <T> T to(DecodeJson<T> decodeJson) throws JsonMappingException {
      */
     default JsonObject asJsonObject() throws JsonMappingException {
         return asJsonObjectOptional()
-                .orElseThrow(() -> new JsonMappingException(null, String.format("Expected JSON object, received %s", node())));
+                .orElseThrow(() -> new JsonMappingException(null, String.format(Locale.ROOT, "Expected JSON object, received %s", node())));
     }
 
     /**
@@ -76,7 +77,7 @@ default Optional<JsonObject> asJsonObjectOptional() {
      */
     default JsonArray asJsonArray() throws JsonMappingException {
         return asJsonArrayOptional()
-                .orElseThrow(() -> new JsonMappingException(null, String.format("Expected JSON array, received %s", node())));
+                .orElseThrow(() -> new JsonMappingException(null, String.format(Locale.ROOT, "Expected JSON array, received %s", node())));
 
     }
 

server-common/src/main/java/org/apache/kafka/timeline/BaseHashTable.java

@@ -19,6 +19,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Locale;
 
 /**
  * A hash table which uses separate chaining.
@@ -238,7 +239,7 @@ String baseToDebugString() {
         bld.append("BaseHashTable{");
         for (int i = 0; i < elements.length; i++) {
             Object slotObject = elements[i];
-            bld.append(String.format("%n%d: ", i));
+            bld.append(String.format(Locale.ROOT, "%n%d: ", i));
             if (slotObject == null) {
                 bld.append("null");
             } else if (slotObject instanceof Object[] array) {
@@ -252,7 +253,7 @@ String baseToDebugString() {
                 bld.append(slotObject);
             }
         }
-        bld.append(String.format("%n}"));
+        bld.append(String.format(Locale.ROOT, "%n}"));
         return bld.toString();
     }
 }

tools/src/main/java/org/apache/kafka/tools/AclCommand.java

@@ -44,6 +44,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
@@ -401,7 +402,7 @@ private static void validateOperation(AclCommandOptions opts, Map<ResourcePatter
                 }
             }
             if (!unsupportedOps.isEmpty()) {
-                String msg = String.format("ResourceType %s only supports operations %s", resource.resourceType(), validOps);
+                String msg = String.format(Locale.ROOT, "ResourceType %s only supports operations %s", resource.resourceType(), validOps);
                 CommandLineUtils.printUsageAndExit(opts.parser, msg);
             }
         }

tools/src/main/java/org/apache/kafka/tools/ClientCompatibilityTest.java

@@ -63,6 +63,7 @@
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Properties;
@@ -212,7 +213,7 @@ public static void main(String[] args) throws Exception {
     private static String toHexString(byte[] buf) {
         StringBuilder bld = new StringBuilder();
         for (byte b : buf) {
-            bld.append(String.format("%02x", b));
+            bld.append(String.format(Locale.ROOT, "%02x", b));
         }
         return bld.toString();
     }
@@ -367,7 +368,7 @@ private boolean topicExists(Collection<TopicListing> listings, String topicName)
         for (TopicListing listing : listings) {
             if (listing.name().equals(topicName)) {
                 if (listing.isInternal())
-                    throw new KafkaException(String.format("Did not expect %s to be an internal topic.", topicName));
+                    throw new KafkaException(String.format(Locale.ROOT, "Did not expect %s to be an internal topic.", topicName));
                 foundTopic = true;
             }
         }

tools/src/main/java/org/apache/kafka/tools/DeleteRecordsCommand.java

@@ -39,6 +39,7 @@
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
@@ -130,7 +131,7 @@ static void execute(Admin adminClient, String offsetJsonString, PrintStream out)
             StringJoiner duplicates = new StringJoiner(",");
             duplicatePartitions.forEach(tp -> duplicates.add(tp.toString()));
             throw new AdminCommandFailedException(
-                String.format("Offset json file contains duplicate topic partitions: %s", duplicates)
+                String.format(Locale.ROOT, "Offset json file contains duplicate topic partitions: %s", duplicates)
             );
         }