KAFKA-13947 Use %d formatting for integers rather than %s

clients/src/main/java/org/apache/kafka/common/compress/KafkaLZ4BlockInputStream.java

@@ -173,7 +173,7 @@ private void readBlock() throws IOException {
                 in.getInt(); // TODO: verify this content checksum
             return;
         } else if (blockSize > maxBlockSize) {
-            throw new IOException(String.format("Block size %s exceeded max: %s", blockSize, maxBlockSize));
+            throw new IOException(String.format("Block size %d exceeded max: %d", blockSize, maxBlockSize));
         }
 
         if (in.remaining() < blockSize) {

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/secured/Retry.java

@@ -49,13 +49,13 @@ public Retry(Time time, long retryBackoffMs, long retryBackoffMaxMs) {
         this.retryBackoffMaxMs = retryBackoffMaxMs;
 
         if (this.retryBackoffMs < 0)
-            throw new IllegalArgumentException(String.format("retryBackoffMs value (%s) must be non-negative", retryBackoffMs));
+            throw new IllegalArgumentException(String.format("retryBackoffMs value (%d) must be non-negative", retryBackoffMs));
 
         if (this.retryBackoffMaxMs < 0)
-            throw new IllegalArgumentException(String.format("retryBackoffMaxMs value (%s) must be non-negative", retryBackoffMaxMs));
+            throw new IllegalArgumentException(String.format("retryBackoffMaxMs value (%d) must be non-negative", retryBackoffMaxMs));
 
         if (this.retryBackoffMaxMs < this.retryBackoffMs)
-            throw new IllegalArgumentException(String.format("retryBackoffMaxMs value (%s) is less than retryBackoffMs value (%s)", retryBackoffMaxMs, retryBackoffMs));
+            throw new IllegalArgumentException(String.format("retryBackoffMaxMs value (%d) is less than retryBackoffMs value (%d)", retryBackoffMaxMs, retryBackoffMs));
     }
 
     public R execute(Retryable<R> retryable) throws ExecutionException {
@@ -88,7 +88,7 @@ public R execute(Retryable<R> retryable) throws ExecutionException {
                 if (waitMs <= 0)
                     break;
 
-                String message = String.format("Attempt %s to make call resulted in an error; sleeping %s ms before retrying",
+                String message = String.format("Attempt %d to make call resulted in an error; sleeping %d ms before retrying",
                     currAttempt, waitMs);
                 log.warn(message, e);
 

metadata/src/main/java/org/apache/kafka/controller/QuorumController.java

@@ -489,7 +489,7 @@ void createSnapshotGenerator(long committedOffset, int committedEpoch, long comm
             if (!snapshotRegistry.hasSnapshot(committedOffset)) {
                 throw new RuntimeException(
                     String.format(
-                        "Cannot generate a snapshot at committed offset %s because it does not exists in the snapshot registry.",
+                        "Cannot generate a snapshot at committed offset %d because it does not exists in the snapshot registry.",
                         committedOffset
                     )
                 );
@@ -857,7 +857,7 @@ public void handleSnapshot(SnapshotReader<ApiMessageAndVersion> reader) {
                     if (isActiveController()) {
                         throw new IllegalStateException(
                             String.format(
-                                "Asked to load snapshot (%s) when it is the active controller (%s)",
+                                "Asked to load snapshot (%s) when it is the active controller (%d)",
                                 reader.snapshotId(),
                                 curClaimEpoch
                             )

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

@@ -310,7 +310,7 @@ private void updateListenersProgress(long highWatermark) {
                 if (nextExpectedOffset < log.startOffset() && nextExpectedOffset < highWatermark) {
                     SnapshotReader<T> snapshot = latestSnapshot().orElseThrow(() -> new IllegalStateException(
                         String.format(
-                            "Snapshot expected since next offset of %s is %s, log start offset is %s and high-watermark is %s",
+                            "Snapshot expected since next offset of %s is %d, log start offset is %d and high-watermark is %d",
                             listenerContext.listenerName(),
                             nextExpectedOffset,
                             log.startOffset(),
@@ -1034,7 +1034,7 @@ private static OptionalInt optionalLeaderId(int leaderIdOrNil) {
     }
 
     private static String listenerName(Listener<?> listener) {
-        return String.format("%s@%s", listener.getClass().getTypeName(), System.identityHashCode(listener));
+        return String.format("%s@%d", listener.getClass().getTypeName(), System.identityHashCode(listener));
     }
 
     private boolean handleFetchResponse(
@@ -1261,7 +1261,7 @@ private FetchSnapshotResponseData handleFetchSnapshotRequest(
         if (partitionSnapshot.position() > Integer.MAX_VALUE) {
             throw new IllegalStateException(
                 String.format(
-                    "Trying to fetch a snapshot with size (%s) and a position (%s) larger than %s",
+                    "Trying to fetch a snapshot with size (%d) and a position (%d) larger than %d",
                     snapshotSize,
                     partitionSnapshot.position(),
                     Integer.MAX_VALUE
@@ -1373,7 +1373,7 @@ private boolean handleFetchSnapshotResponse(
         if (snapshot.sizeInBytes() != partitionSnapshot.position()) {
             throw new IllegalStateException(
                 String.format(
-                    "Received fetch snapshot response with an invalid position. Expected %s; Received %s",
+                    "Received fetch snapshot response with an invalid position. Expected %d; Received %d",
                     snapshot.sizeInBytes(),
                     partitionSnapshot.position()
                 )
@@ -1400,7 +1400,7 @@ private boolean handleFetchSnapshotResponse(
             } else {
                 throw new IllegalStateException(
                     String.format(
-                        "Full log truncation expected but didn't happen. Snapshot of %s, log end offset %s, last fetched %s",
+                        "Full log truncation expected but didn't happen. Snapshot of %s, log end offset %s, last fetched %d",
                         snapshot.snapshotId(),
                         log.endOffset(),
                         log.lastFetchedEpoch()

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

@@ -353,7 +353,7 @@ else if (!that.endOffset.isPresent())
         @Override
         public String toString() {
             return String.format(
-                "ReplicaState(nodeId=%s, endOffset=%s, lastFetchTimestamp=%s, hasAcknowledgedLeader=%s)",
+                "ReplicaState(nodeId=%d, endOffset=%s, lastFetchTimestamp=%s, hasAcknowledgedLeader=%s)",
                 nodeId,
                 endOffset,
                 lastFetchTimestamp,
@@ -372,7 +372,7 @@ public boolean canGrantVote(int candidateId, boolean isLogUpToDate) {
     @Override
     public String toString() {
         return String.format(
-            "Leader(localId=%s, epoch=%s, epochStartOffset=%s, highWatermark=%s, voterStates=%s)",
+            "Leader(localId=%d, epoch=%d, epochStartOffset=%d, highWatermark=%s, voterStates=%s)",
             localId,
             epoch,
             epochStartOffset,

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

@@ -91,7 +91,7 @@ public synchronized void handleCommit(BatchReader<Integer> reader) {
                     if (nextCommitted != committed + 1) {
                         throw new AssertionError(
                             String.format(
-                                "Expected next committed value to be %s, but instead found %s on node %s",
+                                "Expected next committed value to be %d, but instead found %d on node %d",
                                 committed + 1,
                                 nextCommitted,
                                 nodeId

raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java

@@ -189,7 +189,7 @@ private BatchBuilder<T> maybeAllocateBatch(
             if (bytesNeeded.isPresent() && bytesNeeded.getAsInt() > maxBatchSize) {
                 throw new RecordBatchTooLargeException(
                     String.format(
-                        "The total record(s) size of %s exceeds the maximum allowed batch size of %s",
+                        "The total record(s) size of %d exceeds the maximum allowed batch size of %d",
                         bytesNeeded.getAsInt(),
                         maxBatchSize
                     )

streams/src/main/java/org/apache/kafka/streams/processor/internals/RepartitionTopics.java

@@ -118,7 +118,7 @@ public Queue<StreamsException> missingSourceTopicExceptions() {
 
             return new StreamsException(
                 new MissingSourceTopicException(String.format(
-                    "Missing source topics %s for subtopology %s of topology %s",
+                    "Missing source topics %s for subtopology %d of topology %s",
                     missingSourceTopics, subtopologyId, topologyName)),
                 new TaskId(subtopologyId, 0, topologyName));
         }).collect(Collectors.toCollection(LinkedList::new));