MINOR: Remove unnecessary toString(); fix comment references (#13212)

Reviewers: Mickael Maison <mickael.maison@gmail.com>, Divij Vaidya <diviv@amazon.com>, Lucas Brutschy <lbrutschy@confluent.io>

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

@@ -19,6 +19,8 @@ package org.apache.kafka.clients.admin;
 
 import org.apache.kafka.common.annotation.InterfaceStability;
 
+import java.util.Collection;
+
 /**
  * Options for {@link Admin#alterClientQuotas(Collection, AlterClientQuotasOptions)}.
  *

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

@@ -21,6 +21,7 @@ import org.apache.kafka.common.KafkaFuture;
 import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.common.quota.ClientQuotaEntity;
 
+import java.util.Collection;
 import java.util.Map;
 
 /**

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

@@ -20,6 +20,7 @@ package org.apache.kafka.clients.admin;
 import org.apache.kafka.common.KafkaFuture;
 import org.apache.kafka.common.annotation.InterfaceStability;
 
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java

@@ -1153,7 +1153,7 @@ public abstract class AbstractStickyAssignor extends AbstractPartitionAssignor {
                 List<String> path = new ArrayList<>(Collections.singleton(pair.srcMemberId));
                 if (isLinked(pair.dstMemberId, pair.srcMemberId, reducedPairs, path) && !in(path, cycles)) {
                     cycles.add(new ArrayList<>(path));
-                    log.error("A cycle of length {} was found: {}", path.size() - 1, path.toString());
+                    log.error("A cycle of length {} was found: {}", path.size() - 1, path);
                 }
             }
 

clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java

@@ -670,7 +670,7 @@ public class Metrics implements Closeable {
         
         if (!runtimeTagKeys.equals(templateTagKeys)) {
             throw new IllegalArgumentException("For '" + template.name() + "', runtime-defined metric tags do not match the tags in the template. "
-                    + "Runtime = " + runtimeTagKeys.toString() + " Template = " + templateTagKeys.toString());
+                    + "Runtime = " + runtimeTagKeys + " Template = " + templateTagKeys.toString());
         }
                 
         return this.metricName(template.name(), template.group(), template.description(), tags);

clients/src/main/java/org/apache/kafka/common/network/TransportLayer.java

@@ -67,7 +67,7 @@ public interface TransportLayer extends ScatteringByteChannel, TransferableChann
     /**
      * This a no-op for the non-secure PLAINTEXT implementation. For SSL, this performs
      * SSL handshake. The SSL handshake includes client authentication if configured using
-     * {@link org.apache.kafka.common.config.SslConfigs#SSL_CLIENT_AUTH_CONFIG}.
+     * {@link org.apache.kafka.common.config.internals.BrokerSecurityConfigs#SSL_CLIENT_AUTH_CONFIG}.
      * @throws AuthenticationException if handshake fails due to an {@link javax.net.ssl.SSLException}.
      * @throws IOException if read or write fails with an I/O error.
     */

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/unsecured/OAuthBearerValidationUtils.java

@@ -176,7 +176,7 @@ public class OAuthBearerValidationUtils {
             if (!tokenScope.contains(requiredScopeElement))
                 return OAuthBearerValidationResult.newFailure(String.format(
                         "The provided scope (%s) was mising a required scope (%s).  All required scope elements: %s",
-                        String.valueOf(tokenScope), requiredScopeElement, requiredScope.toString()),
+                        String.valueOf(tokenScope), requiredScopeElement, requiredScope),
                         requiredScope.toString(), null);
         }
         return OAuthBearerValidationResult.newSuccess();

clients/src/test/java/org/apache/kafka/common/protocol/types/ProtocolSerializationTest.java

@@ -154,7 +154,7 @@ public class ProtocolSerializationTest {
                 if (!f.def.type.isNullable())
                     fail("Should not allow serialization of null value.");
             } catch (SchemaException e) {
-                assertFalse(f.def.type.isNullable(), f.toString() + " should not be nullable");
+                assertFalse(f.def.type.isNullable(), f + " should not be nullable");
             } finally {
                 this.struct.set(f, o);
             }

generator/src/main/java/org/apache/kafka/message/EntityType.java

@@ -54,7 +54,7 @@ public enum EntityType {
         } else {
             if (!type.toString().equals(baseType.toString())) {
                 throw new RuntimeException("Field " + fieldName + " has entity type " +
-                    name() + ", but field type " + type.toString() + ", which does " +
+                    name() + ", but field type " + type + ", which does " +
                     "not match.");
             }
         }

generator/src/main/java/org/apache/kafka/message/FieldSpec.java

@@ -454,7 +454,7 @@ public final class FieldSpec {
                     name + ".  The only valid default for a struct field " +
                     "is the empty struct or null.");
             }
-            return "new " + type.toString() + "()";
+            return "new " + type + "()";
         } else if (type.isArray()) {
             if (fieldDefault.equals("null")) {
                 validateNullDefault();

generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java

@@ -585,7 +585,7 @@ public final class MessageDataGenerator implements MessageClassGenerator {
         } else if (type instanceof FieldType.Float64FieldType) {
             return "_readable.readDouble()";
         } else if (type.isStruct()) {
-            return String.format("new %s(_readable, _version)", type.toString());
+            return String.format("new %s(_readable, _version)", type);
         } else {
             throw new RuntimeException("Unsupported field type " + type);
         }

generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java

@@ -317,7 +317,7 @@ final class SchemaGenerator {
                         fieldTypeToSchemaType(arrayType.elementType(), false, version, fieldFlexibleVersions, false));
             }
         } else if (type.isStruct()) {
-            return String.format("%s.SCHEMA_%d", type.toString(),
+            return String.format("%s.SCHEMA_%d", type,
                 floorVersion(type.toString(), version));
         } else {
             throw new RuntimeException("Unsupported type " + type);

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

@@ -594,10 +594,10 @@ public class ReplicationControlManager {
             resultsPrefix = ", ";
         }
         if (request.validateOnly()) {
-            log.info("Validate-only CreateTopics result(s): {}", resultsBuilder.toString());
+            log.info("Validate-only CreateTopics result(s): {}", resultsBuilder);
             return ControllerResult.atomicOf(Collections.emptyList(), data);
         } else {
-            log.info("CreateTopics result(s): {}", resultsBuilder.toString());
+            log.info("CreateTopics result(s): {}", resultsBuilder);
             return ControllerResult.atomicOf(records, data);
         }
     }
@@ -1679,7 +1679,7 @@ public class ReplicationControlManager {
                         append(record.partitionId());
                     prefix = ", ";
                 }
-                log.debug("{}: changing partition(s): {}", context, bld.toString());
+                log.debug("{}: changing partition(s): {}", context, bld);
             } else if (log.isInfoEnabled()) {
                 log.info("{}: changing {} partition(s)", context, records.size() - oldSize);
             }

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

@@ -121,7 +121,7 @@ public class QuorumState {
             // For exceptions during state file loading (missing or not readable),
             // we could assume the file is corrupted already and should be cleaned up.
             log.warn("Clearing local quorum state store after error loading state {}",
-                store.toString(), e);
+                store, e);
             store.clear();
             election = ElectionState.withUnknownLeader(0, voters);
         }

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

@@ -69,7 +69,7 @@ public class CheckpointFile<T> {
             // Ignore if file already exists.
         }
         absolutePath = file.toPath().toAbsolutePath();
-        tempPath = Paths.get(absolutePath.toString() + ".tmp");
+        tempPath = Paths.get(absolutePath + ".tmp");
     }
 
     public void write(Collection<T> entries) throws IOException {

shell/src/main/java/org/apache/kafka/shell/LsCommandHandler.java

@@ -179,7 +179,7 @@ public final class LsCommandHandler implements Commands.Handler {
                     }
                 }
             }
-            writer.println(output.toString());
+            writer.println(output);
         }
     }
 

shell/src/main/java/org/apache/kafka/shell/MetadataShell.java

@@ -149,12 +149,12 @@ public final class MetadataShell {
             builder.setSnapshotPath(res.getString("snapshot"));
             Path tempDir = Files.createTempDirectory("MetadataShell");
             Exit.addShutdownHook("agent-shutdown-hook", () -> {
-                log.debug("Removing temporary directory " + tempDir.toAbsolutePath().toString());
+                log.debug("Removing temporary directory " + tempDir.toAbsolutePath());
                 try {
                     Utils.delete(tempDir.toFile());
                 } catch (Exception e) {
                     log.error("Got exception while removing temporary directory " +
-                        tempDir.toAbsolutePath().toString());
+                        tempDir.toAbsolutePath());
                 }
             });
             MetadataShell shell = builder.build();

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

@@ -146,7 +146,7 @@ public class GlobalStateManagerImpl implements GlobalStateManager {
                         " state store, topic-partition: {}, checkpoint file: {}. If this topic-partition is no longer valid," +
                         " an application reset and state store directory cleanup will be required.",
                     tp.topic(),
-                    checkpointFile.toString()
+                    checkpointFile
                 );
                 throw new StreamsException("Encountered a topic-partition not associated with any global state store");
             }

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

@@ -1430,7 +1430,7 @@ public class StreamsPartitionAssignor implements ConsumerPartitionAssignor, Conf
                     "%sNumber of assigned partitions %d is not equal to "
                         + "the number of active taskIds %d, assignmentInfo=%s",
                     logPrefix, partitions.size(),
-                    info.activeTasks().size(), info.toString()
+                    info.activeTasks().size(), info
                 )
             );
         }

streams/src/main/java/org/apache/kafka/streams/state/internals/CompositeReadOnlySessionStore.java

@@ -185,7 +185,7 @@ public class CompositeReadOnlySessionStore<K, V> implements ReadOnlySessionStore
                 throw new InvalidStateStoreException("State store  [" + storeName + "] is not available anymore" +
                                                              " and may have been migrated to another instance; " +
                                                              "please re-discover its location from the state metadata. " +
-                                                             "Original error message: " + ise.toString());
+                                                             "Original error message: " + ise);
             }
         }
         return KeyValueIterators.emptyIterator();

streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStoreTest.java

@@ -225,7 +225,7 @@ public class MeteredKeyValueStoreTest {
             STORE_LEVEL_GROUP,
             THREAD_ID_TAG_KEY,
             threadId,
-            taskId.toString(),
+            taskId,
             STORE_TYPE,
             STORE_NAME
         )));

streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreTest.java

@@ -229,7 +229,7 @@ public class MeteredSessionStoreTest {
             STORE_LEVEL_GROUP,
             THREAD_ID_TAG_KEY,
             threadId,
-            taskId.toString(),
+            taskId,
             STORE_TYPE,
             STORE_NAME
         )));

streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreTest.java

@@ -238,7 +238,7 @@ public class MeteredTimestampedKeyValueStoreTest {
             STORE_LEVEL_GROUP,
             THREAD_ID_TAG_KEY,
             threadId,
-            taskId.toString(),
+            taskId,
             STORE_TYPE,
             STORE_NAME
         )));

trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConsumeBenchWorker.java

@@ -163,7 +163,7 @@ public class ConsumeBenchWorker implements TaskWorker {
 
         private String consumerGroup() {
             return toUseRandomConsumeGroup()
-                ? "consume-bench-" + UUID.randomUUID().toString()
+                ? "consume-bench-" + UUID.randomUUID()
                 : spec.consumerGroup();
         }
 

trogdor/src/test/java/org/apache/kafka/trogdor/coordinator/CoordinatorTest.java

@@ -332,7 +332,7 @@ public class CoordinatorTest {
         public ExpectedLines waitFor(final String nodeName,
                 final CapturingCommandRunner runner) throws InterruptedException {
             TestUtils.waitForCondition(() -> linesMatch(nodeName, runner.lines(nodeName)),
-                "failed to find the expected lines " + this.toString());
+                "failed to find the expected lines " + this);
             return this;
         }