MINOR: Cleanup DelayedOperationKey (#20947)

Refactor the subclasses of DelayedOperationKey using `record`.

Reviewers: Chia-Ping Tsai <[email protected]>

Author: Yunyung

server-common/src/main/java/org/apache/kafka/server/purgatory/TopicPartitionOperationKey.java

@@ -19,20 +19,10 @@
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.TopicPartition;
 
-import java.util.Objects;
-
 /**
  * Used by delayed-produce and delayed-fetch operations
  */
-public class TopicPartitionOperationKey implements DelayedOperationKey {
-
-    public final String topic;
-    public final int partition;
-
-    public TopicPartitionOperationKey(String topic, int partition) {
-        this.topic = topic;
-        this.partition = partition;
-    }
+public record TopicPartitionOperationKey(String topic, int partition) implements DelayedOperationKey {
 
     public TopicPartitionOperationKey(TopicPartition tp) {
         this(tp.topic(), tp.partition());
@@ -46,17 +36,4 @@ public TopicPartitionOperationKey(TopicIdPartition tp) {
     public String keyLabel() {
         return topic + "-" + partition;
     }
-
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) return true;
-        if (o == null || getClass() != o.getClass()) return false;
-        TopicPartitionOperationKey that = (TopicPartitionOperationKey) o;
-        return partition == that.partition && Objects.equals(topic, that.topic);
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(topic, partition);
-    }
 }

server-common/src/test/java/org/apache/kafka/server/purgatory/DelayedOperationTest.java

@@ -24,7 +24,6 @@
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Objects;
 import java.util.Optional;
 import java.util.Random;
 import java.util.concurrent.ExecutionException;
@@ -63,26 +62,7 @@ public void tearDown() throws Exception {
             executorService.shutdown();
     }
 
-    private static class MockKey implements DelayedOperationKey {
-        @Override
-        public boolean equals(Object o) {
-            if (this == o) return true;
-            if (o == null || getClass() != o.getClass()) return false;
-            MockKey mockKey = (MockKey) o;
-            return Objects.equals(key, mockKey.key);
-        }
-
-        @Override
-        public int hashCode() {
-            return key != null ? key.hashCode() : 0;
-        }
-
-        final String key;
-
-        MockKey(String key) {
-            this.key = key;
-        }
-
+    private record MockKey(String key) implements DelayedOperationKey {
         @Override
         public String keyLabel() {
             return key;

server/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchGroupKey.java

@@ -19,47 +19,15 @@
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.Uuid;
 
-import java.util.Objects;
-
 /**
  * A key for delayed share fetch purgatory that refers to the share partition.
  */
-public class DelayedShareFetchGroupKey implements DelayedShareFetchKey {
-    private final String groupId;
-    private final Uuid topicId;
-    private final int partition;
+public record DelayedShareFetchGroupKey(String groupId, Uuid topicId, int partition) implements DelayedShareFetchKey {
 
     public DelayedShareFetchGroupKey(String groupId, TopicIdPartition topicIdPartition) {
         this(groupId, topicIdPartition.topicId(), topicIdPartition.partition());
     }
 
-    public DelayedShareFetchGroupKey(String groupId, Uuid topicId, int partition) {
-        this.groupId = groupId;
-        this.topicId = topicId;
-        this.partition = partition;
-    }
-
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) return true;
-        if (o == null || getClass() != o.getClass()) return false;
-        DelayedShareFetchGroupKey that = (DelayedShareFetchGroupKey) o;
-        return topicId.equals(that.topicId) && partition == that.partition && groupId.equals(that.groupId);
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(topicId, partition, groupId);
-    }
-
-    @Override
-    public String toString() {
-        return "DelayedShareFetchGroupKey(groupId=" + groupId +
-            ", topicId=" + topicId +
-            ", partition=" + partition +
-            ")";
-    }
-
     @Override
     public String keyLabel() {
         return String.format("groupId=%s, topicId=%s, partition=%s", groupId, topicId, partition);

server/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchPartitionKey.java

@@ -19,43 +19,15 @@
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.Uuid;
 
-import java.util.Objects;
-
 /**
  * A key for delayed share fetch purgatory that refers to the topic partition.
  */
-public class DelayedShareFetchPartitionKey implements DelayedShareFetchKey {
-    private final Uuid topicId;
-    private final int partition;
+public record DelayedShareFetchPartitionKey(Uuid topicId, int partition) implements DelayedShareFetchKey {
 
     public DelayedShareFetchPartitionKey(TopicIdPartition topicIdPartition) {
         this(topicIdPartition.topicId(), topicIdPartition.partition());
     }
 
-    public DelayedShareFetchPartitionKey(Uuid topicId, int partition) {
-        this.topicId = topicId;
-        this.partition = partition;
-    }
-
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) return true;
-        if (o == null || getClass() != o.getClass()) return false;
-        DelayedShareFetchPartitionKey that = (DelayedShareFetchPartitionKey) o;
-        return topicId.equals(that.topicId) && partition == that.partition;
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(topicId, partition);
-    }
-
-    @Override
-    public String toString() {
-        return "DelayedShareFetchPartitionKey(topicId=" + topicId +
-            ", partition=" + partition + ")";
-    }
-
     @Override
     public String keyLabel() {
         return String.format("topicId=%s, partition=%s", topicId, partition);

storage/src/test/java/org/apache/kafka/server/purgatory/DelayedRemoteListOffsetsTest.java

@@ -100,7 +100,7 @@ public void testResponseOnRequestExpiration() throws InterruptedException {
         assertEquals(numResponse.get(), listOffsetsRequestKeys.size());
         assertEquals(listOffsetsRequestKeys.size(), DelayedRemoteListOffsets.AGGREGATE_EXPIRATION_METER.count());
         listOffsetsRequestKeys.forEach(key -> {
-            TopicPartition tp = new TopicPartition(key.topic, key.partition);
+            TopicPartition tp = new TopicPartition(key.topic(), key.partition());
             assertEquals(1, DelayedRemoteListOffsets.PARTITION_EXPIRATION_METERS.get(tp).count());
         });
     }