MINOR: Small cleanups in clients

mimaison · mimaison · commit 768b31b9478b · 2025-09-12T19:00:42.000+02:00
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteShareGroupsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteShareGroupsOptions.java
@@ -22,7 +22,7 @@
 import java.util.Collection;
 
 /**
- * Options for the {@link Admin#deleteShareGroups(Collection <String>, DeleteShareGroupsOptions)} call.
+ * Options for the {@link Admin#deleteShareGroups(Collection, DeleteShareGroupsOptions)} call.
  * <p>
  * The API of this class is evolving, see {@link Admin} for details.
  */
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteShareGroupsResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteShareGroupsResult.java
@@ -25,7 +25,7 @@
 import java.util.Map;
 
 /**
- * The result of the {@link Admin#deleteShareGroups(Collection <String>, DeleteShareGroupsOptions)} call.
+ * The result of the {@link Admin#deleteShareGroups(Collection, DeleteShareGroupsOptions)} call.
  * <p></p>
  * The API of this class is evolving, see {@link Admin} for details.
  */
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteStreamsGroupsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteStreamsGroupsOptions.java
@@ -21,7 +21,7 @@
 import java.util.Collection;
 
 /**
- * Options for the {@link Admin#deleteStreamsGroups(Collection<String>, DeleteStreamsGroupsOptions)} call.
+ * Options for the {@link Admin#deleteStreamsGroups(Collection, DeleteStreamsGroupsOptions)} call.
  * <p>
  * The API of this class is evolving, see {@link Admin} for details.
  */
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ListStreamsGroupOffsetsSpec.java b/clients/src/main/java/org/apache/kafka/clients/admin/ListStreamsGroupOffsetsSpec.java
@@ -21,6 +21,7 @@
 import org.apache.kafka.common.annotation.InterfaceStability;
 
 import java.util.Collection;
+import java.util.Map;
 
 /**
  * Specification of streams group offsets to list using {@link Admin#listStreamsGroupOffsets(Map, ListStreamsGroupOffsetsOptions)}.
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteShareGroupOffsetsHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteShareGroupOffsetsHandler.java
@@ -122,8 +122,8 @@ public ApiResult<CoordinatorKey, Map<String, ApiException>> handleResponse(
                 if (topic.errorCode() != Errors.NONE.code()) {
                     final Errors topicError = Errors.forCode(topic.errorCode());
                     final String topicErrorMessage = topic.errorMessage();
-                    log.debug("DeleteShareGroupOffsets request for group id {} and topic {} failed and returned error {}." + topicErrorMessage,
-                        groupId.idValue, topic.topicName(), topicError);
+                    log.debug("DeleteShareGroupOffsets request for group id {} and topic {} failed and returned error {}. {}",
+                        groupId.idValue, topic.topicName(), topicError, topicErrorMessage);
                 }
                 topicResults.put(
                     topic.topicName(),
@@ -147,14 +147,14 @@ private void handleGroupError(
             case REBALANCE_IN_PROGRESS:
                 // If the coordinator is in the middle of loading, then we just need to retry
                 log.debug("DeleteShareGroupOffsets request for group id {} failed because the coordinator" +
-                    " is still in the process of loading state. Will retry. " + errorMessage, groupId.idValue);
+                    " is still in the process of loading state. Will retry. {}", groupId.idValue, errorMessage);
                 break;
             case COORDINATOR_NOT_AVAILABLE:
             case NOT_COORDINATOR:
                 // If the coordinator is unavailable or there was a coordinator change, then we unmap
                 // the key so that we retry the `FindCoordinator` request
-                log.debug("DeleteShareGroupOffsets request for group id {} returned error {}. Will rediscover the coordinator and retry. " + errorMessage,
-                    groupId.idValue, error);
+                log.debug("DeleteShareGroupOffsets request for group id {} returned error {}. Will rediscover the coordinator and retry. {}",
+                    groupId.idValue, error, errorMessage);
                 groupsToUnmap.add(groupId);
                 break;
             case INVALID_GROUP_ID:
@@ -164,11 +164,11 @@ private void handleGroupError(
             case UNKNOWN_SERVER_ERROR:
             case KAFKA_STORAGE_ERROR:
             case GROUP_AUTHORIZATION_FAILED:
-                log.debug("DeleteShareGroupOffsets request for group id {} failed due to error {}. " + errorMessage, groupId.idValue, error);
+                log.debug("DeleteShareGroupOffsets request for group id {} failed due to error {}. {}", groupId.idValue, error, errorMessage);
                 failed.put(groupId, error.exception(errorMessage));
                 break;
             default:
-                log.error("DeleteShareGroupOffsets request for group id {} failed due to unexpected error {}. " + errorMessage, groupId.idValue, error);
+                log.error("DeleteShareGroupOffsets request for group id {} failed due to unexpected error {}. {}", groupId.idValue, error, errorMessage);
                 failed.put(groupId, error.exception(errorMessage));
         }
     }
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java
@@ -202,7 +202,7 @@ private void handlePartitionError(
     public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
         int brokerId, UnsupportedVersionException exception, Set<TopicPartition> keys
     ) {
-        log.warn("Broker " + brokerId + " does not support MAX_TIMESTAMP offset specs");
+        log.warn("Broker {} does not support MAX_TIMESTAMP offset specs", brokerId);
         Map<TopicPartition, Throwable> maxTimestampPartitions = new HashMap<>();
         for (TopicPartition topicPartition : keys) {
             Long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java
@@ -988,7 +988,7 @@ boolean maybeAbortReconciliation() {
             String reason = rejoinedWhileReconciliationInProgress ?
                 "the member has re-joined the group" :
                 "the member already transitioned out of the reconciling state into " + state;
-            log.info("Interrupting reconciliation that is not relevant anymore because " + reason);
+            log.info("Interrupting reconciliation that is not relevant anymore because {}", reason);
             markReconciliationCompleted();
         }
         return shouldAbort;
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java
@@ -965,8 +965,8 @@ private class GeneralAssignmentBuilder extends AbstractAssignmentBuilder {
             super(partitionsPerTopic, rackInfo, currentAssignment);
             this.subscriptions = subscriptions;
 
-            topic2AllPotentialConsumers = new HashMap<>(partitionsPerTopic.keySet().size());
-            consumer2AllPotentialTopics = new HashMap<>(subscriptions.keySet().size());
+            topic2AllPotentialConsumers = new HashMap<>(partitionsPerTopic.size());
+            consumer2AllPotentialTopics = new HashMap<>(subscriptions.size());
 
             // initialize topic2AllPotentialConsumers and consumer2AllPotentialTopics
             partitionsPerTopic.keySet().forEach(
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java
@@ -541,7 +541,7 @@ private void fetchOffsetsWithRetries(final OffsetFetchRequestState fetchRequest,
             boolean inflightRemoved = pendingRequests.inflightOffsetFetches.remove(fetchRequest);
             if (!inflightRemoved) {
                 log.warn("A duplicated, inflight, request was identified, but unable to find it in the " +
-                    "outbound buffer:" + fetchRequest);
+                    "outbound buffer: {}", fetchRequest);
             }
             if (error == null) {
                 maybeUpdateLastSeenEpochIfNewer(res);
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java
@@ -287,7 +287,7 @@ private static List<StreamsGroupHeartbeatRequestData.TopicInfo> getChangelogTopi
 
     private final HeartbeatMetricsManager metricsManager;
 
-    private StreamsRebalanceData streamsRebalanceData;
+    private final StreamsRebalanceData streamsRebalanceData;
 
     /**
      * Timer for tracking the time since the last consumer poll.  If the timer expires, the consumer will stop
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java
@@ -1010,8 +1010,8 @@ private void maybeReconcile() {
             return;
         }
         if (reconciliationInProgress) {
-            log.trace("Ignoring reconciliation attempt. Another reconciliation is already in progress. Assignment " +
-                targetAssignment + " will be handled in the next reconciliation loop.");
+            log.trace("Ignoring reconciliation attempt. Another reconciliation is already in progress. Assignment {}" +
+                " will be handled in the next reconciliation loop.", targetAssignment);
             return;
         }
 
@@ -1131,12 +1131,12 @@ private CompletableFuture<Void> assignTasks(final SortedSet<StreamsRebalanceData
         );
 
         final SortedSet<TopicPartition> partitionsToAssign = topicPartitionsForActiveTasks(activeTasksToAssign);
-        final SortedSet<TopicPartition> partitionsToAssigneNotPreviouslyOwned =
+        final SortedSet<TopicPartition> partitionsToAssignNotPreviouslyOwned =
             partitionsToAssignNotPreviouslyOwned(partitionsToAssign, topicPartitionsForActiveTasks(ownedActiveTasks));
 
         subscriptionState.assignFromSubscribedAwaitingCallback(
             partitionsToAssign,
-            partitionsToAssigneNotPreviouslyOwned
+            partitionsToAssignNotPreviouslyOwned
         );
         notifyAssignmentChange(partitionsToAssign);
 
@@ -1152,10 +1152,10 @@ private CompletableFuture<Void> assignTasks(final SortedSet<StreamsRebalanceData
             if (callbackError == null) {
                 subscriptionState.enablePartitionsAwaitingCallback(partitionsToAssign);
             } else {
-                if (!partitionsToAssigneNotPreviouslyOwned.isEmpty()) {
+                if (!partitionsToAssignNotPreviouslyOwned.isEmpty()) {
                     log.warn("Leaving newly assigned partitions {} marked as non-fetchable and not " +
                             "requiring initializing positions after onTasksAssigned callback failed.",
-                        partitionsToAssigneNotPreviouslyOwned, callbackError);
+                        partitionsToAssignNotPreviouslyOwned, callbackError);
                 }
             }
         });
@@ -1205,9 +1205,9 @@ private SortedSet<TopicPartition> topicPartitionsForActiveTasks(final SortedSet<
             Stream.concat(
                 streamsRebalanceData.subtopologies().get(task.subtopologyId()).sourceTopics().stream(),
                 streamsRebalanceData.subtopologies().get(task.subtopologyId()).repartitionSourceTopics().keySet().stream()
-            ).forEach(topic -> {
-                topicPartitions.add(new TopicPartition(topic, task.partitionId()));
-            })
+            ).forEach(topic ->
+                topicPartitions.add(new TopicPartition(topic, task.partitionId()))
+            )
         );
         return topicPartitions;
     }
@@ -1223,7 +1223,7 @@ private boolean maybeAbortReconciliation() {
             String reason = rejoinedWhileReconciliationInProgress ?
                 "the member has re-joined the group" :
                 "the member already transitioned out of the reconciling state into " + state;
-            log.info("Interrupting reconciliation that is not relevant anymore because " + reason);
+            log.info("Interrupting reconciliation that is not relevant anymore because {}", reason);
             markReconciliationCompleted();
         }
         return shouldAbort;
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java b/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
@@ -553,7 +553,7 @@ public synchronized KafkaMetric removeMetric(MetricName metricName) {
                 try {
                     reporter.metricRemoval(metric);
                 } catch (Exception e) {
-                    log.error("Error when removing metric from " + reporter.getClass().getName(), e);
+                    log.error("Error when removing metric from {}", reporter.getClass().getName(), e);
                 }
             }
             log.trace("Removed metric named {}", metricName);
@@ -596,7 +596,7 @@ synchronized KafkaMetric registerMetric(KafkaMetric metric) {
             try {
                 reporter.metricChange(metric);
             } catch (Exception e) {
-                log.error("Error when registering metric on " + reporter.getClass().getName(), e);
+                log.error("Error when registering metric on {}", reporter.getClass().getName(), e);
             }
         }
         log.trace("Registered metric named {}", metricName);
@@ -688,7 +688,7 @@ public void close() {
                 log.info("Closing reporter {}", reporter.getClass().getName());
                 reporter.close();
             } catch (Exception e) {
-                log.error("Error when closing " + reporter.getClass().getName(), e);
+                log.error("Error when closing {}", reporter.getClass().getName(), e);
             }
         }
         log.info("Metrics reporters closed");
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/ByteBufferUnmapper.java b/clients/src/main/java/org/apache/kafka/common/utils/ByteBufferUnmapper.java
@@ -40,15 +40,15 @@ public final class ByteBufferUnmapper {
     private static final RuntimeException UNMAP_NOT_SUPPORTED_EXCEPTION;
 
     static {
-        Object unmap = null;
+        MethodHandle unmap = null;
         RuntimeException exception = null;
         try {
             unmap = lookupUnmapMethodHandle();
         } catch (RuntimeException e) {
             exception = e;
         }
         if (unmap != null) {
-            UNMAP = (MethodHandle) unmap;
+            UNMAP = unmap;
             UNMAP_NOT_SUPPORTED_EXCEPTION = null;
         } else {
             UNMAP = null;
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/ConfigUtils.java b/clients/src/main/java/org/apache/kafka/common/utils/ConfigUtils.java
@@ -75,7 +75,8 @@ public static boolean getBoolean(final Map<String, Object> configs, final String
         } else if (value instanceof String) {
             return Boolean.parseBoolean((String) value);
         } else {
-            log.error("Invalid value (" + value + ") on configuration '" + key + "'. The default value '" + defaultValue + "' will be used instead. Please specify a true/false value.");
+            log.error("Invalid value ({}) on configuration '{}'. The default value '{}' will be used instead. Please specify a true/false value.",
+                    value, key, defaultValue);
             return defaultValue;
         }
     }
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/LoggingSignalHandler.java b/clients/src/main/java/org/apache/kafka/common/utils/LoggingSignalHandler.java
@@ -69,7 +69,7 @@ public void register() throws ReflectiveOperationException {
         for (String signal : SIGNALS) {
             register(signal, jvmSignalHandlers);
         }
-        log.info("Registered signal handlers for " + String.join(", ", SIGNALS));
+        log.info("Registered signal handlers for {}", String.join(", ", SIGNALS));
     }
 
     private Object createSignalHandler(final Map<String, Object> jvmSignalHandlers) {
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/SecurityUtils.java b/clients/src/main/java/org/apache/kafka/common/utils/SecurityUtils.java
@@ -33,7 +33,7 @@
 
 public class SecurityUtils {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(SecurityConfig.class);
+    private static final Logger LOGGER = LoggerFactory.getLogger(SecurityUtils.class);
 
     private static final Map<String, ResourceType> NAME_TO_RESOURCE_TYPES;
     private static final Map<String, AclOperation> NAME_TO_OPERATIONS;
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
@@ -857,7 +857,7 @@ public static Properties mkObjectProperties(final Map<String, Object> properties
     public static void delete(final File rootFile) throws IOException {
         if (rootFile == null)
             return;
-        Files.walkFileTree(rootFile.toPath(), new SimpleFileVisitor<Path>() {
+        Files.walkFileTree(rootFile.toPath(), new SimpleFileVisitor<>() {
             @Override
             public FileVisitResult visitFileFailed(Path path, IOException exc) throws IOException {
                 if (exc instanceof NoSuchFileException) {
@@ -1403,7 +1403,7 @@ public static Set<Byte> from32BitField(final int intValue) {
      * @return new Collector<Map.Entry<K, V>, M, M>
      */
     public static <K, V, M extends Map<K, V>> Collector<Map.Entry<K, V>, M, M> entriesToMap(final Supplier<M> mapSupplier) {
-        return new Collector<Map.Entry<K, V>, M, M>() {
+        return new Collector<>() {
             @Override
             public Supplier<M> supplier() {
                 return mapSupplier;
diff --git a/clients/src/test/java/org/apache/kafka/common/KafkaFutureTest.java b/clients/src/test/java/org/apache/kafka/common/KafkaFutureTest.java
@@ -318,7 +318,7 @@ public void testThenApplyOnSucceededFutureAndFunctionThrowsCompletionException()
         awaitAndAssertResult(future, 21, null);
         Throwable cause = awaitAndAssertFailure(dependantFuture, CompletionException.class, "java.lang.RuntimeException: We require more vespene gas");
         assertInstanceOf(RuntimeException.class, cause.getCause());
-        assertEquals(cause.getCause().getMessage(), "We require more vespene gas");
+        assertEquals("We require more vespene gas", cause.getCause().getMessage());
     }
 
     @Test
diff --git a/clients/src/test/java/org/apache/kafka/common/UuidTest.java b/clients/src/test/java/org/apache/kafka/common/UuidTest.java
@@ -35,8 +35,8 @@ public class UuidTest {
     public void testSignificantBits() {
         Uuid id = new Uuid(34L, 98L);
 
-        assertEquals(id.getMostSignificantBits(), 34L);
-        assertEquals(id.getLeastSignificantBits(), 98L);
+        assertEquals(34L, id.getMostSignificantBits());
+        assertEquals(98L, id.getLeastSignificantBits());
     }
 
     @Test
@@ -74,7 +74,7 @@ public void testStringConversion() {
 
         String zeroIdString = Uuid.ZERO_UUID.toString();
 
-        assertEquals(Uuid.fromString(zeroIdString), Uuid.ZERO_UUID);
+        assertEquals(Uuid.ZERO_UUID, Uuid.fromString(zeroIdString));
     }
 
     @RepeatedTest(value = 100, name = RepeatedTest.LONG_DISPLAY_NAME)
diff --git a/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java b/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java
@@ -379,8 +379,8 @@ public void testOriginalWithOverrides() {
         Properties props = new Properties();
         props.put("config.providers", "file");
         TestIndirectConfigResolution config = new TestIndirectConfigResolution(props);
-        assertEquals(config.originals().get("config.providers"), "file");
-        assertEquals(config.originals(Collections.singletonMap("config.providers", "file2")).get("config.providers"), "file2");
+        assertEquals("file", config.originals().get("config.providers"));
+        assertEquals("file2", config.originals(Collections.singletonMap("config.providers", "file2")).get("config.providers"));
     }
 
     @Test
diff --git a/clients/src/test/java/org/apache/kafka/common/network/ChannelBuildersTest.java b/clients/src/test/java/org/apache/kafka/common/network/ChannelBuildersTest.java
diff --git a/clients/src/test/java/org/apache/kafka/common/protocol/ErrorsTest.java b/clients/src/test/java/org/apache/kafka/common/protocol/ErrorsTest.java
diff --git a/clients/src/test/java/org/apache/kafka/common/record/FileRecordsTest.java b/clients/src/test/java/org/apache/kafka/common/record/FileRecordsTest.java
diff --git a/clients/src/test/java/org/apache/kafka/common/replica/ReplicaSelectorTest.java b/clients/src/test/java/org/apache/kafka/common/replica/ReplicaSelectorTest.java

Original file line number	Diff line number	Diff line change
`@@ -988,7 +988,7 @@ boolean maybeAbortReconciliation() {`
`988`	`988`	`String reason = rejoinedWhileReconciliationInProgress ?`
`989`	`989`	`"the member has re-joined the group" :`
`990`	`990`	`"the member already transitioned out of the reconciling state into " + state;`
`991`		`- log.info("Interrupting reconciliation that is not relevant anymore because " + reason);`
	`991`	`+ log.info("Interrupting reconciliation that is not relevant anymore because {}", reason);`
`992`	`992`	`markReconciliationCompleted();`
`993`	`993`	`}`
`994`	`994`	`return shouldAbort;`
Original file line number	Diff line number	Diff line change
`@@ -541,7 +541,7 @@ private void fetchOffsetsWithRetries(final OffsetFetchRequestState fetchRequest,`
`541`	`541`	`boolean inflightRemoved = pendingRequests.inflightOffsetFetches.remove(fetchRequest);`
`542`	`542`	`if (!inflightRemoved) {`
`543`	`543`	`log.warn("A duplicated, inflight, request was identified, but unable to find it in the " +`
`544`		`- "outbound buffer:" + fetchRequest);`
	`544`	`+ "outbound buffer: {}", fetchRequest);`
`545`	`545`	`}`
`546`	`546`	`if (error == null) {`
`547`	`547`	`maybeUpdateLastSeenEpochIfNewer(res);`