From c034cf2953e691ce4ecd94bf00ac5810167354bc Mon Sep 17 00:00:00 2001
From: Alok Thatikunta <athatikunta@confluent.io>
Date: Sat, 13 Apr 2024 18:25:48 +0530
Subject: [PATCH] MINOR: Fix incorrect Java equals comparison of Uuid by
 reference (#15707)

Reviewers: Justine Olshan <jolshan@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>
---
 .../org/apache/kafka/clients/admin/KafkaAdminClient.java    | 2 +-
 .../org/apache/kafka/common/requests/MetadataResponse.java  | 2 +-
 .../common/telemetry/internals/ClientTelemetryUtils.java    | 2 +-
 .../storage/internals/checkpoint/PartitionMetadataFile.java | 6 +++---
 .../src/main/java/org/apache/kafka/tools/TopicCommand.java  | 4 ++--
 5 files changed, 8 insertions(+), 8 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index d98ad8ac04e..846c2264eb1 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -1702,7 +1702,7 @@ public class KafkaAdminClient extends AdminClient {
     }
 
     private static boolean topicIdIsUnrepresentable(Uuid topicId) {
-        return topicId == null || topicId == Uuid.ZERO_UUID;
+        return topicId == null || topicId.equals(Uuid.ZERO_UUID);
     }
 
     // for testing
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java
index bc6c387c5e0..6e58e209b35 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java
@@ -112,7 +112,7 @@ public class MetadataResponse extends AbstractResponse {
     public Map<Uuid, Errors> errorsByTopicId() {
         Map<Uuid, Errors> errors = new HashMap<>();
         for (MetadataResponseTopic metadata : data.topics()) {
-            if (metadata.topicId() == Uuid.ZERO_UUID) {
+            if (metadata.topicId().equals(Uuid.ZERO_UUID)) {
                 throw new IllegalStateException("Use errors() when managing topic using topic name");
             }
             if (metadata.errorCode() != Errors.NONE.code())
diff --git a/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtils.java b/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtils.java
index 6f3cbd18d94..9911a00a2f9 100644
--- a/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtils.java
+++ b/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtils.java
@@ -133,7 +133,7 @@ public class ClientTelemetryUtils {
     }
 
     public static Uuid validateClientInstanceId(Uuid clientInstanceId) {
-        if (clientInstanceId == null || clientInstanceId == Uuid.ZERO_UUID) {
+        if (clientInstanceId == null || clientInstanceId.equals(Uuid.ZERO_UUID)) {
             throw new IllegalArgumentException("clientInstanceId is not valid");
         }
 
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/checkpoint/PartitionMetadataFile.java b/storage/src/main/java/org/apache/kafka/storage/internals/checkpoint/PartitionMetadataFile.java
index eaee2decad0..da294d457c3 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/checkpoint/PartitionMetadataFile.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/checkpoint/PartitionMetadataFile.java
@@ -63,9 +63,9 @@ public class PartitionMetadataFile {
     public void record(Uuid topicId) {
         // Topic IDs should not differ, but we defensively check here to fail earlier in the case that the IDs somehow differ.
         dirtyTopicIdOpt.ifPresent(dirtyTopicId -> {
-            if (dirtyTopicId != topicId) {
-                throw new InconsistentTopicIdException("Tried to record topic ID $topicId to file " +
-                    "but had already recorded $dirtyTopicId");
+            if (!dirtyTopicId.equals(topicId)) {
+                throw new InconsistentTopicIdException("Tried to record topic ID " + topicId + " to file " +
+                        "but had already recorded " + dirtyTopicId);
             }
         });
         dirtyTopicIdOpt = Optional.of(topicId);
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 66650cb9db5..ce3fb3d8d17 100644
--- a/tools/src/main/java/org/apache/kafka/tools/TopicCommand.java
+++ b/tools/src/main/java/org/apache/kafka/tools/TopicCommand.java
@@ -296,7 +296,7 @@ public abstract class TopicCommand {
                 .map(ce -> ce.name() + "=" + ce.value())
                 .collect(Collectors.joining(","));
             System.out.print("Topic: " +  topic);
-            if (topicId != Uuid.ZERO_UUID)
+            if (!topicId.equals(Uuid.ZERO_UUID))
                 System.out.print("\tTopicId: " + topicId);
             System.out.print("\tPartitionCount: " + numPartitions);
             System.out.print("\tReplicationFactor: " + replicationFactor);
@@ -539,7 +539,7 @@ public abstract class TopicCommand {
         public void describeTopic(TopicCommandOptions opts) throws ExecutionException, InterruptedException {
             // If topicId is provided and not zero, will use topicId regardless of topic name
             Optional<Uuid> inputTopicId = opts.topicId()
-                .map(Uuid::fromString).filter(uuid -> uuid != Uuid.ZERO_UUID);
+                .map(Uuid::fromString).filter(uuid -> !uuid.equals(Uuid.ZERO_UUID));
             Boolean useTopicId = inputTopicId.isPresent();
 
             List<Uuid> topicIds;