KAFKA-15152: Fix incorrect format specifiers when formatting string (#14026)

Reviewers: Divij Vaidya <diviv@amazon.com>

Co-authored-by: phuchong.tran <phuchong.tran@servicenow.com>

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -1228,7 +1228,7 @@ public class KafkaAdminClient extends AdminClient {
                         call.fail(now, authException);
                     } else {
                         call.fail(now, new DisconnectException(String.format(
-                            "Cancelled %s request with correlation id %s due to node %s being disconnected",
+                            "Cancelled %s request with correlation id %d due to node %s being disconnected",
                             call.callName, correlationId, response.destination())));
                     }
                 } else {

clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java

@@ -438,7 +438,7 @@ public class MemoryRecordsBuilder implements AutoCloseable {
                 throw new IllegalArgumentException("Control records can only be appended to control batches");
 
             if (lastOffset != null && offset <= lastOffset)
-                throw new IllegalArgumentException(String.format("Illegal offset %s following previous offset %s " +
+                throw new IllegalArgumentException(String.format("Illegal offset %d following previous offset %d " +
                         "(Offsets must increase monotonically).", offset, lastOffset));
 
             if (timestamp < 0 && timestamp != RecordBatch.NO_TIMESTAMP)

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

@@ -272,7 +272,7 @@ public class HttpAccessTokenRetriever implements AccessTokenRetriever {
                 errorResponseBody);
 
             if (responseBody == null || responseBody.isEmpty())
-                throw new IOException(String.format("The token endpoint response was unexpectedly empty despite response code %s from %s and error message %s",
+                throw new IOException(String.format("The token endpoint response was unexpectedly empty despite response code %d from %s and error message %s",
                     responseCode, con.getURL(), formatErrorMessage(errorResponseBody)));
 
             return responseBody;
@@ -337,7 +337,7 @@ public class HttpAccessTokenRetriever implements AccessTokenRetriever {
             if (snippet.length() > MAX_RESPONSE_BODY_LENGTH) {
                 int actualLength = responseBody.length();
                 String s = responseBody.substring(0, MAX_RESPONSE_BODY_LENGTH);
-                snippet = String.format("%s (trimmed to first %s characters out of %s total)", s, MAX_RESPONSE_BODY_LENGTH, actualLength);
+                snippet = String.format("%s (trimmed to first %d characters out of %d total)", s, MAX_RESPONSE_BODY_LENGTH, actualLength);
             }
 
             throw new IOException(String.format("The token endpoint response did not contain an access_token value. Response: (%s)", snippet));

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

@@ -348,8 +348,8 @@ public final class RefreshingHttpsJwks implements Initable, Closeable {
             //     1. Don't try to resolve the key as the large ID will sit in our cache
             //     2. Report the issue in the logs but include only the first N characters
             int actualLength = keyId.length();
-            String s = keyId.substring(0, MISSING_KEY_ID_MAX_KEY_LENGTH);
+            String trimmedKeyId = keyId.substring(0, MISSING_KEY_ID_MAX_KEY_LENGTH);
-            String snippet = String.format("%s (trimmed to first %s characters out of %s total)", s, MISSING_KEY_ID_MAX_KEY_LENGTH, actualLength);
+            String snippet = String.format("%s (trimmed to first %d characters out of %d total)", trimmedKeyId, MISSING_KEY_ID_MAX_KEY_LENGTH, actualLength);
             log.warn("Key ID {} was too long to cache", snippet);
             return false;
         } else {

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

@@ -44,7 +44,7 @@ public class SerializedJwt {
         String[] splits = token.split("\\.");
 
         if (splits.length != 3)
-            throw new ValidateException(String.format("Malformed JWT provided (%s); expected three sections (header, payload, and signature), but %s sections provided",
+            throw new ValidateException(String.format("Malformed JWT provided (%s); expected three sections (header, payload, and signature), but %d sections provided",
                 token, splits.length));
 
         this.token = token.trim();

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

@@ -55,7 +55,7 @@ final class ControllerResultAndOffset<T> extends ControllerResult<T> {
     @Override
     public String toString() {
         return String.format(
-            "ControllerResultAndOffset(records=%s, response=%s, isAtomic=%s, offset=%s)",
+            "ControllerResultAndOffset(records=%s, response=%s, isAtomic=%s, offset=%d)",
             String.join(",", records().stream().map(ApiMessageAndVersion::toString).collect(Collectors.toList())),
             response(),
             isAtomic(),

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

@@ -168,7 +168,7 @@ public final class Batch<T> implements Iterable<T> {
         if (records.isEmpty()) {
             throw new IllegalArgumentException(
                 String.format(
-                    "Control batch must contain at least one record; baseOffset = %s; epoch = %s",
+                    "Control batch must contain at least one record; baseOffset = %d; epoch = %d",
                     baseOffset,
                     epoch
                 )
@@ -205,7 +205,7 @@ public final class Batch<T> implements Iterable<T> {
         if (records.isEmpty()) {
             throw new IllegalArgumentException(
                 String.format(
-                    "Batch must contain at least one record; baseOffset = %s; epoch = %s",
+                    "Batch must contain at least one record; baseOffset = %d; epoch = %d",
                     baseOffset,
                     epoch
                 )

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

@@ -118,12 +118,12 @@ public class FollowerState implements EpochState {
 
             if (updatedHighWatermark < 0) {
                 throw new IllegalArgumentException(
-                    String.format("Illegal negative (%s) high watermark update", updatedHighWatermark)
+                    String.format("Illegal negative (%d) high watermark update", updatedHighWatermark)
                 );
             } else if (previousHighWatermark > updatedHighWatermark) {
                 throw new IllegalArgumentException(
                     String.format(
-                        "Non-monotonic update of high watermark from %s to %s",
+                        "Non-monotonic update of high watermark from %d to %d",
                         previousHighWatermark,
                         updatedHighWatermark
                     )

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

@@ -99,7 +99,7 @@ public interface ReplicatedLog extends AutoCloseable {
              */
             OffsetAndEpoch latestSnapshotId = latestSnapshotId().orElseThrow(() -> new IllegalStateException(
                 String.format(
-                    "Log start offset (%s) is greater than zero but latest snapshot was not found",
+                    "Log start offset (%d) is greater than zero but latest snapshot was not found",
                     startOffset()
                 )
             ));

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

@@ -328,7 +328,7 @@ public class BatchBuilder<T> {
             if (expectedNextOffset - baseOffset >= Integer.MAX_VALUE) {
                 throw new IllegalArgumentException(
                     String.format(
-                        "Adding %s records to a batch with base offset of %s and next offset of %s",
+                        "Adding %d records to a batch with base offset of %d and next offset of %d",
                         records.size(),
                         baseOffset,
                         expectedNextOffset

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/CopartitionedTopicsEnforcer.java

@@ -101,8 +101,8 @@ public class CopartitionedTopicsEnforcer {
                 .orElseThrow(emptyNumberOfPartitionsExceptionSupplier(config.name()));
 
             if (numberOfPartitionsOfInternalTopic != numPartitionsToUseForRepartitionTopics) {
-                final String msg = String.format("%sNumber of partitions [%s] of repartition topic [%s] " +
+                final String msg = String.format("%sNumber of partitions [%d] of repartition topic [%s] " +
-                                                 "doesn't match number of partitions [%s] of the source topic.",
+                                                 "doesn't match number of partitions [%d] of the source topic.",
                                                  logPrefix,
                                                  numberOfPartitionsOfInternalTopic,
                                                  config.name(),