From 4d0c726ccaf23984c7fdb9a8fc48c3c57fa5d16f Mon Sep 17 00:00:00 2001
From: Chia-Ping Tsai <chia7712@gmail.com>
Date: Wed, 17 May 2023 18:45:21 +0800
Subject: [PATCH] [PRODUCER] rewrite proudcer.Metadata by java 17 record
 (#1751)

---
 .../org/astraea/common/producer/Metadata.java | 94 +++++--------------
 1 file changed, 26 insertions(+), 68 deletions(-)

diff --git a/common/src/main/java/org/astraea/common/producer/Metadata.java b/common/src/main/java/org/astraea/common/producer/Metadata.java
index ad013a3a85..2f2fb5afc7 100644
--- a/common/src/main/java/org/astraea/common/producer/Metadata.java
+++ b/common/src/main/java/org/astraea/common/producer/Metadata.java
@@ -19,73 +19,31 @@
 import java.util.Optional;
 import org.apache.kafka.clients.producer.RecordMetadata;
 
-public interface Metadata {
-
-  static Metadata of(RecordMetadata metadata) {
-    return new Metadata() {
-      @Override
-      public long offset() {
-        return metadata.offset();
-      }
-
-      @Override
-      public int serializedKeySize() {
-        return metadata.serializedKeySize();
-      }
-
-      @Override
-      public int serializedValueSize() {
-        return metadata.serializedValueSize();
-      }
-
-      @Override
-      public Optional<Long> timestamp() {
-        return metadata.hasTimestamp() ? Optional.of(metadata.timestamp()) : Optional.empty();
-      }
-
-      @Override
-      public String topic() {
-        return metadata.topic();
-      }
-
-      @Override
-      public int partition() {
-        return metadata.partition();
-      }
-    };
+/**
+ * @param topic The topic the record was appended to
+ * @param partition The partition the record was sent to
+ * @param offset The offset of the record in the topic/partition.
+ * @param serializedKeySize The size of the serialized, uncompressed key in bytes. If key is null,
+ *     the returned size is -1.
+ * @param serializedValueSize The size of the serialized, uncompressed value in bytes. If value is
+ *     null, the returned size is -1.
+ * @param timestamp the timestamp of the record
+ */
+public record Metadata(
+    String topic,
+    int partition,
+    long offset,
+    int serializedKeySize,
+    int serializedValueSize,
+    Optional<Long> timestamp) {
+
+  public static Metadata of(RecordMetadata metadata) {
+    return new Metadata(
+        metadata.topic(),
+        metadata.partition(),
+        metadata.offset(),
+        metadata.serializedKeySize(),
+        metadata.serializedValueSize(),
+        metadata.hasTimestamp() ? Optional.of(metadata.timestamp()) : Optional.empty());
   }
-
-  /**
-   * The offset of the record in the topic/partition.
-   *
-   * @return the offset of the record, or -1
-   */
-  long offset();
-
-  /**
-   * @return The size of the serialized, uncompressed key in bytes. If key is null, the returned
-   *     size is -1.
-   */
-  int serializedKeySize();
-
-  /**
-   * @return The size of the serialized, uncompressed value in bytes. If value is null, the returned
-   *     size is -1.
-   */
-  int serializedValueSize();
-
-  /**
-   * @return the timestamp of the record
-   */
-  Optional<Long> timestamp();
-
-  /**
-   * @return The topic the record was appended to
-   */
-  String topic();
-
-  /**
-   * @return The partition the record was sent to
-   */
-  int partition();
 }