[common] Fix equals/hashCode contract in value and kv record batches

fluss-common/src/main/java/org/apache/fluss/record/DefaultKvRecordBatch.java

@@ -23,6 +23,7 @@
 import org.apache.fluss.memory.MemorySegment;
 import org.apache.fluss.record.bytesview.BytesView;
 import org.apache.fluss.utils.CloseableIterator;
+import org.apache.fluss.utils.MurmurHashUtils;
 import org.apache.fluss.utils.crc.Crc32C;
 
 import java.nio.ByteBuffer;
@@ -219,6 +220,11 @@ public boolean equals(Object o) {
                 && segment.equalTo(that.segment, position, that.position, sizeInBytes);
     }
 
+    @Override
+    public int hashCode() {
+        return MurmurHashUtils.hashBytes(segment, position, sizeInBytes());
+    }
+
     abstract class KvRecordIterator implements CloseableIterator<KvRecord> {
         private final int numRecords;
         private int readRecords = 0;

fluss-common/src/main/java/org/apache/fluss/record/DefaultValueRecordBatch.java

@@ -21,6 +21,7 @@
 import org.apache.fluss.memory.MemorySegmentOutputView;
 import org.apache.fluss.row.BinaryRow;
 import org.apache.fluss.utils.CloseableIterator;
+import org.apache.fluss.utils.MurmurHashUtils;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -132,6 +133,11 @@ public boolean equals(Object o) {
                 && segment.equalTo(that.segment, position, that.position, sizeInBytes);
     }
 
+    @Override
+    public int hashCode() {
+        return MurmurHashUtils.hashBytes(segment, position, sizeInBytes());
+    }
+
     // ------------------------------------------------------------------------------------------
 
     abstract class ValueRecordIterator implements CloseableIterator<ValueRecord> {

fluss-common/src/test/java/org/apache/fluss/record/DefaultKvRecordBatchTest.java

@@ -27,7 +27,9 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import static org.apache.fluss.record.TestData.DATA1_ROW_TYPE;
 import static org.apache.fluss.record.TestData.DATA1_SCHEMA;
+import static org.apache.fluss.testutils.DataTestUtils.compactedRow;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** Test for {@link DefaultKvRecordBatch}. */
@@ -76,4 +78,39 @@ KvFormat.COMPACTED, new TestingSchemaGetter(1, DATA1_SCHEMA)))) {
 
         builder.close();
     }
+
+    @Test
+    void testEqualsAndHashCode() throws Exception {
+        DefaultKvRecordBatch batch1 = buildBatch();
+        DefaultKvRecordBatch batch2 = buildBatch();
+
+        assertThat(batch1).isEqualTo(batch2);
+        assertThat(batch1).hasSameHashCodeAs(batch2);
+    }
+
+    @Test
+    void testHashCodeDiffersForDifferentContents() throws Exception {
+        DefaultKvRecordBatch batch1 = buildBatch();
+        DefaultKvRecordBatch batch2 = buildBatch(new byte[] {(byte) 9, (byte) 9});
+
+        assertThat(batch1).isNotEqualTo(batch2);
+        assertThat(batch1.hashCode()).isNotEqualTo(batch2.hashCode());
+    }
+
+    private DefaultKvRecordBatch buildBatch() throws Exception {
+        return buildBatch(new byte[] {(byte) 1, (byte) 1});
+    }
+
+    private DefaultKvRecordBatch buildBatch(byte[] key) throws Exception {
+        KvRecordBatchBuilder builder =
+                KvRecordBatchBuilder.builder(
+                        schemaId,
+                        Integer.MAX_VALUE,
+                        new UnmanagedPagedOutputView(100),
+                        KvFormat.COMPACTED);
+        builder.append(key, compactedRow(DATA1_ROW_TYPE, new Object[] {1, "a1"}));
+        DefaultKvRecordBatch batch = DefaultKvRecordBatch.pointToBytesView(builder.build());
+        builder.close();
+        return batch;
+    }
 }

fluss-common/src/test/java/org/apache/fluss/record/DefaultValueRecordBatchTest.java

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.record;
+
+import org.junit.jupiter.api.Test;
+
+import static org.apache.fluss.record.TestData.DATA1_ROW_TYPE;
+import static org.apache.fluss.record.TestData.DEFAULT_SCHEMA_ID;
+import static org.apache.fluss.testutils.DataTestUtils.compactedRow;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link DefaultValueRecordBatch}. */
+class DefaultValueRecordBatchTest {
+
+    @Test
+    void testEqualsAndHashCode() throws Exception {
+        DefaultValueRecordBatch batch1 = buildBatch();
+        DefaultValueRecordBatch batch2 = buildBatch();
+
+        assertThat(batch1).isEqualTo(batch2);
+        assertThat(batch1).hasSameHashCodeAs(batch2);
+    }
+
+    @Test
+    void testHashCodeDiffersForDifferentContents() throws Exception {
+        DefaultValueRecordBatch.Builder builder = DefaultValueRecordBatch.builder();
+        builder.append(DEFAULT_SCHEMA_ID, compactedRow(DATA1_ROW_TYPE, new Object[] {1, "a1"}));
+        DefaultValueRecordBatch batch1 = builder.build();
+
+        DefaultValueRecordBatch.Builder otherBuilder = DefaultValueRecordBatch.builder();
+        otherBuilder.append(
+                DEFAULT_SCHEMA_ID, compactedRow(DATA1_ROW_TYPE, new Object[] {2, "a2"}));
+        DefaultValueRecordBatch batch2 = otherBuilder.build();
+
+        assertThat(batch1).isNotEqualTo(batch2);
+        assertThat(batch1.hashCode()).isNotEqualTo(batch2.hashCode());
+    }
+
+    @Test
+    void testEmptyBatchesAreEqualAndShareHashCode() throws Exception {
+        DefaultValueRecordBatch batch1 = DefaultValueRecordBatch.builder().build();
+        DefaultValueRecordBatch batch2 = DefaultValueRecordBatch.builder().build();
+
+        assertThat(batch1).isEqualTo(batch2);
+        assertThat(batch1).hasSameHashCodeAs(batch2);
+    }
+
+    private static DefaultValueRecordBatch buildBatch() throws Exception {
+        DefaultValueRecordBatch.Builder builder = DefaultValueRecordBatch.builder();
+        builder.append(DEFAULT_SCHEMA_ID, compactedRow(DATA1_ROW_TYPE, new Object[] {1, "a1"}));
+        builder.append(DEFAULT_SCHEMA_ID, compactedRow(DATA1_ROW_TYPE, new Object[] {2, "a2"}));
+        return builder.build();
+    }
+}