diff --git a/server-common/src/main/java/org/apache/kafka/common/ClientIdAndBroker.java b/server-common/src/main/java/org/apache/kafka/common/ClientIdAndBroker.java index 5ef9bb227b37f..5f78835f3e179 100644 --- a/server-common/src/main/java/org/apache/kafka/common/ClientIdAndBroker.java +++ b/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); } } diff --git a/server-common/src/main/java/org/apache/kafka/logger/StateChangeLogger.java b/server-common/src/main/java/org/apache/kafka/logger/StateChangeLogger.java index a852fc30eb889..33377902fe16d 100644 --- a/server-common/src/main/java/org/apache/kafka/logger/StateChangeLogger.java +++ b/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) { diff --git a/server-common/src/main/java/org/apache/kafka/server/common/CheckpointFile.java b/server-common/src/main/java/org/apache/kafka/server/common/CheckpointFile.java index 1047877c9deb0..86a8c425b6f8e 100644 --- a/server-common/src/main/java/org/apache/kafka/server/common/CheckpointFile.java +++ b/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)); } } diff --git a/server-common/src/main/java/org/apache/kafka/server/common/Feature.java b/server-common/src/main/java/org/apache/kafka/server/common/Feature.java index 9744ff7f9df70..8f1e7ffc9ff27 100644 --- a/server-common/src/main/java/org/apache/kafka/server/common/Feature.java +++ b/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,7 +303,7 @@ 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), @@ -310,7 +311,7 @@ public static void validateDefaultValueAndLatestProductionValue( } } 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()), diff --git a/server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java b/server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java index 4177d1bc859ab..c52b0cf09d66f 100644 --- a/server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java +++ b/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; } diff --git a/server-common/src/main/java/org/apache/kafka/server/config/QuotaConfig.java b/server-common/src/main/java/org/apache/kafka/server/config/QuotaConfig.java index ede737cb390ec..73ae1e4caddaa 100644 --- a/server-common/src/main/java/org/apache/kafka/server/config/QuotaConfig.java +++ b/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. " + diff --git a/server-common/src/main/java/org/apache/kafka/server/share/SharePartitionKey.java b/server-common/src/main/java/org/apache/kafka/server/share/SharePartitionKey.java index 8565064a5973b..8b0eb5a6e6f6b 100644 --- a/server-common/src/main/java/org/apache/kafka/server/share/SharePartitionKey.java +++ b/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 diff --git a/server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java index 7e1a6d7d14807..e813932f534ef 100644 --- a/server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java +++ b/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())); } } } diff --git a/server-common/src/main/java/org/apache/kafka/server/util/CommandLineUtils.java b/server-common/src/main/java/org/apache/kafka/server/util/CommandLineUtils.java index 7e231989d63ed..edb698e91374e 100644 --- a/server-common/src/main/java/org/apache/kafka/server/util/CommandLineUtils.java +++ b/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 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]); diff --git a/server-common/src/main/java/org/apache/kafka/server/util/json/DecodeJson.java b/server-common/src/main/java/org/apache/kafka/server/util/json/DecodeJson.java index de9995532e0e2..d25f4401a5a0a 100644 --- a/server-common/src/main/java/org/apache/kafka/server/util/json/DecodeJson.java +++ b/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 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 { diff --git a/server-common/src/main/java/org/apache/kafka/server/util/json/JsonValue.java b/server-common/src/main/java/org/apache/kafka/server/util/json/JsonValue.java index 55ee460959eb7..ffec1e6a8d268 100644 --- a/server-common/src/main/java/org/apache/kafka/server/util/json/JsonValue.java +++ b/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 to(DecodeJson 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 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()))); } diff --git a/server-common/src/main/java/org/apache/kafka/timeline/BaseHashTable.java b/server-common/src/main/java/org/apache/kafka/timeline/BaseHashTable.java index b8d2c9fc5fb58..2461976c55169 100644 --- a/server-common/src/main/java/org/apache/kafka/timeline/BaseHashTable.java +++ b/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(); } } diff --git a/tools/src/main/java/org/apache/kafka/tools/AclCommand.java b/tools/src/main/java/org/apache/kafka/tools/AclCommand.java index 8799a4f5d959b..2de2261e8aea2 100644 --- a/tools/src/main/java/org/apache/kafka/tools/AclCommand.java +++ b/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 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; } } diff --git a/tools/src/main/java/org/apache/kafka/tools/DeleteRecordsCommand.java b/tools/src/main/java/org/apache/kafka/tools/DeleteRecordsCommand.java index 510818f39fea5..39c76820cfc1e 100644 --- a/tools/src/main/java/org/apache/kafka/tools/DeleteRecordsCommand.java +++ b/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) ); } diff --git a/tools/src/main/java/org/apache/kafka/tools/JmxTool.java b/tools/src/main/java/org/apache/kafka/tools/JmxTool.java index 57e825ed50b5f..28f6d2c63a129 100644 --- a/tools/src/main/java/org/apache/kafka/tools/JmxTool.java +++ b/tools/src/main/java/org/apache/kafka/tools/JmxTool.java @@ -151,7 +151,7 @@ private static MBeanServerConnection connectToBeanServer(JmxToolOptions options) } while (System.currentTimeMillis() - connectTestStarted < connectTimeoutMs && !connected); if (!connected) { - throw new TerseException(String.format("Could not connect to JMX url %s after %d ms.", + throw new TerseException(String.format(Locale.ROOT, "Could not connect to JMX url %s after %d ms.", options.jmxServiceURL(), connectTimeoutMs)); } return serverConn; @@ -177,7 +177,7 @@ private static Set findObjects(JmxToolOptions options, if (!hasPatternQueries && options.hasWait() && !foundAllObjects.test(querySet, result)) { querySet.removeAll(result); String missing = mkString(querySet.stream().map(Object::toString), ","); - throw new TerseException(String.format("Could not find all requested object names after %d ms. Missing %s", waitTimeoutMs, missing)); + throw new TerseException(String.format(Locale.ROOT, "Could not find all requested object names after %d ms. Missing %s", waitTimeoutMs, missing)); } return result; } @@ -235,7 +235,7 @@ private static Map findNumExpectedAttributes(MBeanServerCon } if (result.isEmpty()) { - throw new TerseException(String.format("No matched attributes for the queried objects %s.", queries)); + throw new TerseException(String.format(Locale.ROOT, "No matched attributes for the queried objects %s.", queries)); } return result; } @@ -251,11 +251,11 @@ private static Map queryAttributes(MBeanServerConnection conn, for (Attribute attribute : attributes.asList()) { if (attributesInclude.isPresent()) { if (List.of(attributesInclude.get()).contains(attribute.getName())) { - result.put(String.format("%s:%s", objectName.toString(), attribute.getName()), + result.put(String.format(Locale.ROOT, "%s:%s", objectName.toString(), attribute.getName()), attribute.getValue()); } } else { - result.put(String.format("%s:%s", objectName.toString(), attribute.getName()), + result.put(String.format(Locale.ROOT, "%s:%s", objectName.toString(), attribute.getName()), attribute.getValue()); } } @@ -265,7 +265,7 @@ private static Map queryAttributes(MBeanServerConnection conn, private static void maybePrintCsvHeader(String reportFormat, List keys, Map numExpectedAttributes) { if (reportFormat.equals("original") && keys.size() == sumValues(numExpectedAttributes) + 1) { - System.out.println(mkString(keys.stream().map(key -> String.format("\"%s\"", key)), ",")); + System.out.println(mkString(keys.stream().map(key -> String.format(Locale.ROOT, "\"%s\"", key)), ",")); } } diff --git a/tools/src/main/java/org/apache/kafka/tools/LeaderElectionCommand.java b/tools/src/main/java/org/apache/kafka/tools/LeaderElectionCommand.java index b76b1cc0e542d..6a0f0a956c427 100644 --- a/tools/src/main/java/org/apache/kafka/tools/LeaderElectionCommand.java +++ b/tools/src/main/java/org/apache/kafka/tools/LeaderElectionCommand.java @@ -45,6 +45,7 @@ import java.util.HashSet; import java.util.Iterator; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Optional; import java.util.Properties; @@ -165,7 +166,7 @@ private static void electLeaders(Admin client, ElectionType electionType, Option String partitionsAsString = succeeded.stream() .map(TopicPartition::toString) .collect(Collectors.joining(", ")); - System.out.println(String.format("Successfully completed leader election (%s) for partitions %s", + System.out.println(String.format(Locale.ROOT, "Successfully completed leader election (%s) for partitions %s", electionType, partitionsAsString)); } @@ -173,15 +174,15 @@ private static void electLeaders(Admin client, ElectionType electionType, Option String partitionsAsString = noop.stream() .map(TopicPartition::toString) .collect(Collectors.joining(", ")); - System.out.println(String.format("Valid replica already elected for partitions %s", partitionsAsString)); + System.out.println(String.format(Locale.ROOT, "Valid replica already elected for partitions %s", partitionsAsString)); } if (!failed.isEmpty()) { AdminCommandFailedException rootException = - new AdminCommandFailedException(String.format("%s replica(s) could not be elected", failed.size())); + new AdminCommandFailedException(String.format(Locale.ROOT, "%s replica(s) could not be elected", failed.size())); failed.forEach((key, value) -> { System.out.println( - String.format( + String.format(Locale.ROOT, "Error completing leader election (%s) for partition: %s: %s", electionType, key, @@ -238,7 +239,7 @@ private static Set toTopicPartition(JsonValue partitionsList) th .collect(Collectors.toSet()); if (duplicatePartitions.size() > 0) { - throw new AdminOperationException(String.format( + throw new AdminOperationException(String.format(Locale.ROOT, "Replica election data contains duplicate partitions: %s", String.join(",", duplicatePartitions.toString())) ); } @@ -384,12 +385,12 @@ public void validate() { } // --partition if and only if --topic is used if (options.has(topic) && !options.has(partition)) { - throw new AdminCommandFailedException(String.format("Missing required option(s): %s", + throw new AdminCommandFailedException(String.format(Locale.ROOT, "Missing required option(s): %s", partition.options().get(0))); } if (!options.has(topic) && options.has(partition)) { - throw new AdminCommandFailedException(String.format("Option %s is only allowed if %s is used", + throw new AdminCommandFailedException(String.format(Locale.ROOT, "Option %s is only allowed if %s is used", partition.options().get(0), topic.options().get(0) )); diff --git a/tools/src/main/java/org/apache/kafka/tools/LogDirsCommand.java b/tools/src/main/java/org/apache/kafka/tools/LogDirsCommand.java index d6cb07222c44c..303829ba85d57 100644 --- a/tools/src/main/java/org/apache/kafka/tools/LogDirsCommand.java +++ b/tools/src/main/java/org/apache/kafka/tools/LogDirsCommand.java @@ -36,6 +36,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; @@ -83,7 +84,7 @@ static void execute(LogDirsCommandOptions options, Admin adminClient) throws Exc if (!nonExistingBrokers.isEmpty()) { throw new TerseException( - String.format( + String.format(Locale.ROOT, "ERROR: The given brokers do not exist from --broker-list: %s. Current existent brokers: %s", commaDelimitedStringFromIntegerSet(nonExistingBrokers), commaDelimitedStringFromIntegerSet(clusterBrokers))); diff --git a/tools/src/main/java/org/apache/kafka/tools/OAuthCompatibilityTool.java b/tools/src/main/java/org/apache/kafka/tools/OAuthCompatibilityTool.java index 3babe5444bb12..52cae754462ae 100644 --- a/tools/src/main/java/org/apache/kafka/tools/OAuthCompatibilityTool.java +++ b/tools/src/main/java/org/apache/kafka/tools/OAuthCompatibilityTool.java @@ -38,6 +38,7 @@ import java.util.HashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import javax.security.auth.login.AppConfigurationEntry; @@ -206,7 +207,7 @@ private static JwtValidator createValidator(Map configs, List metrics) { } String doubleOutputFormat = "%-" + maxLengthOfDisplayName + "s : %.3f"; String defaultOutputFormat = "%-" + maxLengthOfDisplayName + "s : %s"; - System.out.println(String.format("\n%-" + maxLengthOfDisplayName + "s %s", "Metric Name", "Value")); + System.out.println(String.format(Locale.ROOT, "\n%-" + maxLengthOfDisplayName + "s %s", "Metric Name", "Value")); for (Map.Entry entry : sortedMetrics.entrySet()) { String outputFormat; @@ -54,7 +55,7 @@ public static void printMetrics(Map metrics) { outputFormat = doubleOutputFormat; else outputFormat = defaultOutputFormat; - System.out.println(String.format(outputFormat, entry.getKey(), entry.getValue())); + System.out.println(String.format(Locale.ROOT, outputFormat, entry.getKey(), entry.getValue())); } } } diff --git a/tools/src/main/java/org/apache/kafka/tools/TopicCommand.java b/tools/src/main/java/org/apache/kafka/tools/TopicCommand.java index 81ceecf2401ef..b832c4abe84df 100644 --- a/tools/src/main/java/org/apache/kafka/tools/TopicCommand.java +++ b/tools/src/main/java/org/apache/kafka/tools/TopicCommand.java @@ -65,6 +65,7 @@ import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Objects; import java.util.Optional; @@ -204,7 +205,7 @@ private static void ensureTopicExists(List foundTopics, Optional // If no topic name was mentioned, do not need to throw exception. if (requestedTopic.isPresent() && !requestedTopic.get().isEmpty() && requireTopicExists && foundTopics.isEmpty()) { // If given topic doesn't exist then throw exception - throw new IllegalArgumentException(String.format("Topic '%s' does not exist as expected", requestedTopic.get())); + throw new IllegalArgumentException(String.format(Locale.ROOT, "Topic '%s' does not exist as expected", requestedTopic.get())); } } @@ -234,7 +235,7 @@ private static void ensureTopicIdExists(List foundTopicIds, Uuid requested // If no topic id was mentioned, do not need to throw exception. if (requestedTopicId != null && requireTopicIdExists && foundTopicIds.isEmpty()) { // If given topicId doesn't exist then throw exception - throw new IllegalArgumentException(String.format("TopicId '%s' does not exist as expected", requestedTopicId)); + throw new IllegalArgumentException(String.format(Locale.ROOT, "TopicId '%s' does not exist as expected", requestedTopicId)); } } diff --git a/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java b/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java index 35a9a32fc47c6..f27e030c89b07 100644 --- a/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java +++ b/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java @@ -50,6 +50,7 @@ import java.util.Date; import java.util.HashMap; import java.util.LinkedHashMap; +import java.util.Locale; import java.util.Map; import java.util.Properties; import java.util.Random; @@ -398,7 +399,7 @@ public void onPartitionsAssigned(Collection partitions) { totalMessageProcessed.getAndAdd(messagesSentWithinCurrentTxn); } } catch (ProducerFencedException e) { - throw new KafkaException(String.format("The transactional.id %s has been claimed by another process", transactionalId), e); + throw new KafkaException(String.format(Locale.ROOT, "The transactional.id %s has been claimed by another process", transactionalId), e); } catch (KafkaException e) { log.debug("Aborting transaction after catching exception", e); abortTransactionAndResetPosition(producer, consumer); diff --git a/tools/src/main/java/org/apache/kafka/tools/VerifiableConsumer.java b/tools/src/main/java/org/apache/kafka/tools/VerifiableConsumer.java index 98b1ac00e2a6a..8134770d9c90e 100644 --- a/tools/src/main/java/org/apache/kafka/tools/VerifiableConsumer.java +++ b/tools/src/main/java/org/apache/kafka/tools/VerifiableConsumer.java @@ -60,6 +60,7 @@ import java.util.Collection; import java.util.HashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Properties; import java.util.concurrent.CountDownLatch; @@ -544,7 +545,7 @@ private static ArgumentParser argParser() { .setDefault(ConsumerConfig.DEFAULT_GROUP_PROTOCOL) .dest("groupProtocol") .metavar("GROUP-PROTOCOL") - .help(String.format("Group protocol (must be one of %s)", Arrays.stream(GroupProtocol.values()) + .help(String.format(Locale.ROOT, "Group protocol (must be one of %s)", Arrays.stream(GroupProtocol.values()) .map(Object::toString).collect(Collectors.joining(", ")))); parser.addArgument("--group-remote-assignor") @@ -554,7 +555,7 @@ private static ArgumentParser argParser() { .setDefault(ConsumerConfig.DEFAULT_GROUP_REMOTE_ASSIGNOR) .dest("groupRemoteAssignor") .metavar("GROUP-REMOTE-ASSIGNOR") - .help(String.format("Group remote assignor; only used if the group protocol is %s", GroupProtocol.CONSUMER.name())); + .help(String.format(Locale.ROOT, "Group remote assignor; only used if the group protocol is %s", GroupProtocol.CONSUMER.name())); parser.addArgument("--group-id") .action(store()) @@ -618,7 +619,7 @@ private static ArgumentParser argParser() { .type(String.class) .dest("assignmentStrategy") .metavar("ASSIGNMENT-STRATEGY") - .help(String.format("Set assignment strategy (e.g. %s); only used if the group protocol is %s", RoundRobinAssignor.class.getName(), GroupProtocol.CLASSIC.name())); + .help(String.format(Locale.ROOT, "Set assignment strategy (e.g. %s); only used if the group protocol is %s", RoundRobinAssignor.class.getName(), GroupProtocol.CLASSIC.name())); parser.addArgument("--consumer.config") .action(store()) diff --git a/tools/src/main/java/org/apache/kafka/tools/VerifiableProducer.java b/tools/src/main/java/org/apache/kafka/tools/VerifiableProducer.java index db2d8817f50ce..5af19599e9032 100644 --- a/tools/src/main/java/org/apache/kafka/tools/VerifiableProducer.java +++ b/tools/src/main/java/org/apache/kafka/tools/VerifiableProducer.java @@ -40,6 +40,7 @@ import java.io.InputStream; import java.nio.file.Files; import java.nio.file.Paths; +import java.util.Locale; import java.util.Properties; import static net.sourceforge.argparse4j.impl.Arguments.store; @@ -303,9 +304,9 @@ record = new ProducerRecord<>(topic, key, value); /** Returns a string to publish: ether 'valuePrefix'.'val' or 'val' */ public String getValue(long val) { if (this.valuePrefix != null) { - return String.format("%d.%d", this.valuePrefix, val); + return String.format(Locale.ROOT, "%d.%d", this.valuePrefix, val); } - return String.format("%d", val); + return String.format(Locale.ROOT, "%d", val); } public String getKey() { diff --git a/tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java b/tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java index f8705c9e7a4d2..fe7c859d59883 100644 --- a/tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java +++ b/tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java @@ -61,6 +61,7 @@ import java.util.HashSet; import java.util.Iterator; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Map.Entry; import java.util.Optional; @@ -97,7 +98,7 @@ public static void main(String[] args) { if (actions.stream().filter(opts.options::has).count() != 1) { CommandLineUtils.printUsageAndExit( opts.parser, - String.format( + String.format(Locale.ROOT, "Command must include exactly one action: %s", actions.stream().map(opt -> "--" + opt.options().get(0) @@ -991,7 +992,7 @@ private Map prepareOffsetsToReset(String grou return groupOffsetsResetter.resetToCurrent(partitionsToReset, currentCommittedOffsets); } - CommandLineUtils.printUsageAndExit(opts.parser, String.format("Option '%s' requires one of the following scenarios: %s", opts.resetOffsetsOpt, opts.allResetOffsetScenarioOpts)); + CommandLineUtils.printUsageAndExit(opts.parser, String.format(Locale.ROOT, "Option '%s' requires one of the following scenarios: %s", opts.resetOffsetsOpt, opts.allResetOffsetScenarioOpts)); return null; } diff --git a/tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java b/tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java index aa02dd22d3bcb..befcc0e01edcf 100644 --- a/tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java +++ b/tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java @@ -56,6 +56,7 @@ import java.util.HashSet; import java.util.LinkedHashSet; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Map.Entry; import java.util.Optional; @@ -387,7 +388,7 @@ void resetOffsets() { try { ShareGroupDescription shareGroupDescription = describeShareGroups(List.of(groupId)).get(groupId); if (!(GroupState.EMPTY.equals(shareGroupDescription.groupState()) || GroupState.DEAD.equals(shareGroupDescription.groupState()))) { - CommandLineUtils.printErrorAndExit(String.format("Share group '%s' is not empty.", groupId)); + CommandLineUtils.printErrorAndExit(String.format(Locale.ROOT, "Share group '%s' is not empty.", groupId)); } resetOffsetsForInactiveGroup(groupId); } catch (InterruptedException ie) { @@ -458,7 +459,7 @@ private Map prepareOffsetsToReset(Collection< return groupOffsetsResetter.resetToDateTime(partitionsToReset); } CommandLineUtils - .printUsageAndExit(opts.parser, String.format("Option '%s' requires one of the following scenarios: %s", opts.resetOffsetsOpt, opts.allResetOffsetScenarioOpts)); + .printUsageAndExit(opts.parser, String.format(Locale.ROOT, "Option '%s' requires one of the following scenarios: %s", opts.resetOffsetsOpt, opts.allResetOffsetScenarioOpts)); return null; } diff --git a/tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandOptions.java b/tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandOptions.java index b227ac713746d..08bb97448d80b 100644 --- a/tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandOptions.java +++ b/tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandOptions.java @@ -23,6 +23,7 @@ import org.slf4j.LoggerFactory; import java.util.List; +import java.util.Locale; import java.util.Set; import java.util.stream.Collectors; @@ -176,10 +177,10 @@ public void checkArgs() { if (options.has(deleteOpt)) { if (!options.has(groupOpt) && !options.has(allGroupsOpt)) CommandLineUtils.printUsageAndExit(parser, - String.format("Option %s takes the options %s or %s", deleteOpt, groupOpt, allGroupsOpt)); + String.format(Locale.ROOT, "Option %s takes the options %s or %s", deleteOpt, groupOpt, allGroupsOpt)); if (options.has(allGroupsOpt) && options.has(groupOpt)) CommandLineUtils.printUsageAndExit(parser, - String.format("Option %s takes either %s or %s, not both.", deleteOpt, groupOpt, allGroupsOpt)); + String.format(Locale.ROOT, "Option %s takes either %s or %s, not both.", deleteOpt, groupOpt, allGroupsOpt)); if (options.has(topicOpt)) CommandLineUtils.printUsageAndExit(parser, "Option " + deleteOpt + " does not take the option: " + topicOpt); diff --git a/tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java b/tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java index 89595b63750ef..948c28cc5069b 100644 --- a/tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java +++ b/tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java @@ -66,6 +66,7 @@ import java.util.Iterator; import java.util.LinkedHashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Map.Entry; import java.util.Objects; @@ -281,7 +282,7 @@ static String partitionReassignmentStatesToString(Map, List> parseGenerateAssignmentArgs(String rea List brokerListToReassign = Stream.of(brokerList.split(",")).map(Integer::parseInt).collect(Collectors.toList()); Set duplicateReassignments = ToolsUtils.duplicates(brokerListToReassign); if (!duplicateReassignments.isEmpty()) - throw new AdminCommandFailedException(String.format("Broker list contains duplicate entries: %s", duplicateReassignments)); + throw new AdminCommandFailedException(String.format(Locale.ROOT, "Broker list contains duplicate entries: %s", duplicateReassignments)); List topicsToReassign = parseTopicsData(reassignmentJson); Set duplicateTopicsToReassign = ToolsUtils.duplicates(topicsToReassign); if (!duplicateTopicsToReassign.isEmpty()) - throw new AdminCommandFailedException(String.format("List of topics to reassign contains duplicate entries: %s", + throw new AdminCommandFailedException(String.format(Locale.ROOT, "List of topics to reassign contains duplicate entries: %s", duplicateTopicsToReassign)); return Map.entry(brokerListToReassign, topicsToReassign); } @@ -810,7 +811,7 @@ public static void executeAssignment(Admin adminClient, Map errors = alterPartitionReassignments(adminClient, proposedParts, disallowReplicationFactorChange); if (!errors.isEmpty()) { throw new TerseException( - String.format("Error reassigning partition(s):%n%s", + String.format(Locale.ROOT, "Error reassigning partition(s):%n%s", errors.keySet().stream() .sorted(ReassignPartitionsCommand::compareTopicPartitions) .map(part -> part + ": " + errors.get(part).getMessage()) @@ -856,7 +857,7 @@ private static void executeMoves(Admin adminClient, if (pendingReplicas.isEmpty()) { done = true; } else if (time.milliseconds() >= startTimeMs + timeoutMs) { - throw new TerseException(String.format( + throw new TerseException(String.format(Locale.ROOT, "Timed out before log directory move%s could be started for: %s", pendingReplicas.size() == 1 ? "" : "s", pendingReplicas.keySet().stream() @@ -896,13 +897,13 @@ static String curReassignmentsToString(Admin adminClient) throws ExecutionExcept List addingReplicas = reassignment.addingReplicas(); List removingReplicas = reassignment.removingReplicas(); - return String.format("%s: replicas: %s.%s%s", + return String.format(Locale.ROOT, "%s: replicas: %s.%s%s", part, replicas.stream().map(Object::toString).collect(Collectors.joining(",")), - addingReplicas.isEmpty() ? "" : String.format(" adding: %s.", addingReplicas.stream() + addingReplicas.isEmpty() ? "" : String.format(Locale.ROOT, " adding: %s.", addingReplicas.stream() .map(Object::toString) .collect(Collectors.joining(","))), - removingReplicas.isEmpty() ? "" : String.format(" removing: %s.", removingReplicas.stream() + removingReplicas.isEmpty() ? "" : String.format(Locale.ROOT, " removing: %s.", removingReplicas.stream() .map(Object::toString) .collect(Collectors.joining(","))) ); @@ -910,7 +911,7 @@ static String curReassignmentsToString(Admin adminClient) throws ExecutionExcept return text.isEmpty() ? "No partition reassignments found." - : String.format("Current partition reassignments:%n%s", text); + : String.format(Locale.ROOT, "Current partition reassignments:%n%s", text); } /** @@ -956,7 +957,7 @@ static String currentPartitionReplicaAssignmentToString( Map> currentParts = toReplicaIds(partitionsToBeReassigned); - return String.format("Current partition replica assignment%n%n%s%n%nSave this to use as the %s", + return String.format(Locale.ROOT, "Current partition replica assignment%n%n%s%n%nSave this to use as the %s", formatAsReassignmentJson(currentParts, currentReplicaLogDirs), "--reassignment-json-file option during rollback"); } @@ -1086,7 +1087,7 @@ static Map calculateLeaderThrottles(Map { Set components = new TreeSet<>(); partMoveMap.forEach((partId, move) -> - move.sources().forEach(source -> components.add(String.format("%d:%d", partId, source)))); + move.sources().forEach(source -> components.add(String.format(Locale.ROOT, "%d:%d", partId, source)))); results.put(topicName, String.join(",", components)); }); return results; @@ -1105,7 +1106,7 @@ static Map calculateFollowerThrottles(Map move.destinations().forEach(destination -> { if (!move.sources().contains(destination)) { - components.add(String.format("%d:%d", partId, destination)); + components.add(String.format(Locale.ROOT, "%d:%d", partId, destination)); } }) ); @@ -1251,7 +1252,7 @@ static Entry>, Map duplicateReassignedPartitions = ToolsUtils.duplicates(partitionsToBeReassigned.stream().map( Entry::getKey).collect(Collectors.toList())); if (!duplicateReassignedPartitions.isEmpty()) { - throw new AdminCommandFailedException(String.format( + throw new AdminCommandFailedException(String.format(Locale.ROOT, "Partition reassignment contains duplicate topic partitions: %s", duplicateReassignedPartitions.stream().map(Object::toString).collect(Collectors.joining(","))) ); @@ -1261,11 +1262,11 @@ static Entry>, Map !t.getValue().isEmpty()).toList(); if (!duplicateEntries.isEmpty()) { String duplicatesMsg = duplicateEntries.stream().map(t -> - String.format("%s contains multiple entries for %s", + String.format(Locale.ROOT, "%s contains multiple entries for %s", t.getKey(), t.getValue().stream().map(Object::toString).collect(Collectors.joining(","))) ).collect(Collectors.joining(". ")); - throw new AdminCommandFailedException(String.format("Partition replica lists may not contain duplicate entries: %s", duplicatesMsg)); + throw new AdminCommandFailedException(String.format(Locale.ROOT, "Partition replica lists may not contain duplicate entries: %s", duplicatesMsg)); } return Map.entry(partitionsToBeReassigned.stream().collect(Collectors.toMap(Entry::getKey, Entry::getValue)), replicaAssignment); } @@ -1302,7 +1303,7 @@ static Entry, Set> cancelAssignment(A if (!curReassigningParts.isEmpty()) { Map errors = cancelPartitionReassignments(adminClient, curReassigningParts); if (!errors.isEmpty()) { - throw new TerseException(String.format( + throw new TerseException(String.format(Locale.ROOT, "Error cancelling partition reassignment%s for:%n%s", errors.size() == 1 ? "" : "s", errors.keySet().stream() @@ -1453,7 +1454,7 @@ static ReassignPartitionsCommandOptions validateAndParseArgs(String[] args) { .filter(a -> opts.options.has(a)).toList(); if (allActions.size() != 1) { - CommandLineUtils.printUsageAndExit(opts.parser, String.format("Command must include exactly one action: %s", + CommandLineUtils.printUsageAndExit(opts.parser, String.format(Locale.ROOT, "Command must include exactly one action: %s", validActions.stream().map(a -> "--" + a.options().get(0)).collect(Collectors.joining(", ")))); } @@ -1515,7 +1516,7 @@ else if (!opts.options.has(opts.bootstrapServerOpt) && !opts.options.has(opts.bo !requiredArgs.getOrDefault(action, List.of()).contains(opt) && !permittedArgs.getOrDefault(action, List.of()).contains(opt)) { CommandLineUtils.printUsageAndExit(opts.parser, - String.format("Option \"%s\" can't be used with action \"%s\"", opt, action)); + String.format(Locale.ROOT, "Option \"%s\" can't be used with action \"%s\"", opt, action)); } }); return opts; diff --git a/tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java b/tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java index 8e40d81228d90..89ec691ccaafe 100644 --- a/tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java +++ b/tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java @@ -73,6 +73,7 @@ import java.util.Iterator; import java.util.LinkedHashSet; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Optional; import java.util.Properties; @@ -986,7 +987,7 @@ private Map prepareOffsetsToReset(String grou } CommandLineUtils - .printUsageAndExit(opts.parser, String.format("Option '%s' requires one of the following scenarios: %s", opts.resetOffsetsOpt, opts.allResetOffsetScenarioOpts)); + .printUsageAndExit(opts.parser, String.format(Locale.ROOT, "Option '%s' requires one of the following scenarios: %s", opts.resetOffsetsOpt, opts.allResetOffsetScenarioOpts)); return null; }