From ac475d29d26e6238792855ec7cb93039f3c7defb Mon Sep 17 00:00:00 2001 From: Bhanu Chander Vallabaneni Date: Wed, 12 Aug 2026 20:25:48 -0400 Subject: [PATCH] [server] Fix equals/hashCode contract in ISR states and remote log manifest handle CommittedIsrState, PendingExpandIsrState, PendingShrinkIsrState and RemoteLogManifestHandle override equals() with value semantics but do not override hashCode(). Each extends Object and none delegates to super.equals(), so hashCode() is the identity hash and equal instances hash differently, violating the general contract of Object.hashCode(). The two pending ISR states compare lastCommittedState in equals(), so their hash codes have to fold in CommittedIsrState.hashCode(); fixing the pending states alone would leave the composed hash identity-derived. LeaderAndIsr, the other compared field, already implements both methods. Each hashCode() covers exactly the fields its own equals() compares, via Objects.hash to match the existing idiom in LeaderAndIsr. In particular CommittedIsrState hashes only isr, because that is all its equals() compares today; widening equality is left alone deliberately. --- .../apache/fluss/server/replica/IsrState.java | 16 +++ .../zk/data/RemoteLogManifestHandle.java | 7 ++ .../fluss/server/replica/IsrStateTest.java | 119 ++++++++++++++++++ .../zk/data/RemoteLogManifestHandleTest.java | 70 +++++++++++ 4 files changed, 212 insertions(+) create mode 100644 fluss-server/src/test/java/org/apache/fluss/server/replica/IsrStateTest.java create mode 100644 fluss-server/src/test/java/org/apache/fluss/server/zk/data/RemoteLogManifestHandleTest.java diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/IsrState.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/IsrState.java index b8b3b0a8274..8d29a6f5deb 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/replica/IsrState.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/IsrState.java @@ -22,6 +22,7 @@ import java.util.ArrayList; import java.util.List; +import java.util.Objects; /* This file is based on source code of Apache Kafka Project (https://kafka.apache.org/), licensed by the Apache * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for @@ -93,6 +94,11 @@ public boolean equals(Object o) { return isr.equals(that.isr); } + @Override + public int hashCode() { + return Objects.hash(isr); + } + @Override public String toString() { return "CommittedIsrState{" + "isr=" + isr + '}'; @@ -181,6 +187,11 @@ public boolean equals(Object o) { && lastCommittedState.equals(that.lastCommittedState); } + @Override + public int hashCode() { + return Objects.hash(newInSyncReplicaId, sentLeaderAndIsr, lastCommittedState); + } + @Override public String toString() { return "PendingExpandIsrState{" @@ -254,6 +265,11 @@ public boolean equals(Object o) { && lastCommittedState.equals(that.lastCommittedState); } + @Override + public int hashCode() { + return Objects.hash(outOfSyncReplicaIds, sentLeaderAndIsr, lastCommittedState); + } + @Override public String toString() { return "PendingShrinkIsrState{" diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/RemoteLogManifestHandle.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/RemoteLogManifestHandle.java index a3a6e2cee4d..865bd048071 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/RemoteLogManifestHandle.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/RemoteLogManifestHandle.java @@ -19,6 +19,8 @@ import org.apache.fluss.fs.FsPath; +import java.util.Objects; + /** * The remote log manifest handle of a table bucket stored in {@link ZkData.BucketRemoteLogsZNode}. * @@ -58,6 +60,11 @@ public boolean equals(Object o) { && remoteLogEndOffset == that.remoteLogEndOffset; } + @Override + public int hashCode() { + return Objects.hash(remoteLogManifestPath, remoteLogEndOffset); + } + @Override public String toString() { return "RemoteLogManifestHandle{" diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/IsrStateTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/IsrStateTest.java new file mode 100644 index 00000000000..8ead13ea148 --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/IsrStateTest.java @@ -0,0 +1,119 @@ +/* + * 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.server.replica; + +import org.apache.fluss.server.replica.IsrState.CommittedIsrState; +import org.apache.fluss.server.replica.IsrState.PendingExpandIsrState; +import org.apache.fluss.server.replica.IsrState.PendingShrinkIsrState; +import org.apache.fluss.server.zk.data.LeaderAndIsr; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Test the {@code equals}/{@code hashCode} contract of the {@link IsrState} implementations. */ +class IsrStateTest { + + private static final List ISR = Arrays.asList(1, 2, 3); + private static final List STANDBY = Collections.singletonList(4); + + private static LeaderAndIsr leaderAndIsr() { + return new LeaderAndIsr(1, 0, Arrays.asList(1, 2, 3), Collections.singletonList(4), 0, 0); + } + + @Test + void testCommittedIsrStateEqualsAndHashCode() { + CommittedIsrState state1 = new CommittedIsrState(ISR, STANDBY); + CommittedIsrState state2 = new CommittedIsrState(Arrays.asList(1, 2, 3), STANDBY); + + assertThat(state1).isEqualTo(state2); + assertThat(state1).hasSameHashCodeAs(state2); + } + + @Test + void testCommittedIsrStateHashCodeDiffersForDifferentIsr() { + CommittedIsrState state1 = new CommittedIsrState(ISR, STANDBY); + CommittedIsrState state2 = new CommittedIsrState(Arrays.asList(1, 2), STANDBY); + + assertThat(state1).isNotEqualTo(state2); + assertThat(state1.hashCode()).isNotEqualTo(state2.hashCode()); + } + + @Test + void testPendingExpandIsrStateEqualsAndHashCode() { + CommittedIsrState committed = new CommittedIsrState(ISR, STANDBY); + PendingExpandIsrState state1 = + new PendingExpandIsrState(5, leaderAndIsr(), new CommittedIsrState(ISR, STANDBY)); + PendingExpandIsrState state2 = new PendingExpandIsrState(5, leaderAndIsr(), committed); + + assertThat(state1).isEqualTo(state2); + assertThat(state1).hasSameHashCodeAs(state2); + } + + @Test + void testPendingExpandIsrStateHashCodeDiffersForDifferentReplica() { + CommittedIsrState committed = new CommittedIsrState(ISR, STANDBY); + PendingExpandIsrState state1 = new PendingExpandIsrState(5, leaderAndIsr(), committed); + PendingExpandIsrState state2 = new PendingExpandIsrState(6, leaderAndIsr(), committed); + + assertThat(state1).isNotEqualTo(state2); + assertThat(state1.hashCode()).isNotEqualTo(state2.hashCode()); + } + + @Test + void testPendingShrinkIsrStateEqualsAndHashCode() { + CommittedIsrState committed = new CommittedIsrState(ISR, STANDBY); + PendingShrinkIsrState state1 = + new PendingShrinkIsrState( + Collections.singletonList(3), + leaderAndIsr(), + new CommittedIsrState(ISR, STANDBY)); + PendingShrinkIsrState state2 = + new PendingShrinkIsrState(Collections.singletonList(3), leaderAndIsr(), committed); + + assertThat(state1).isEqualTo(state2); + assertThat(state1).hasSameHashCodeAs(state2); + } + + @Test + void testPendingShrinkIsrStateHashCodeDiffersForDifferentOutOfSyncReplicas() { + CommittedIsrState committed = new CommittedIsrState(ISR, STANDBY); + PendingShrinkIsrState state1 = + new PendingShrinkIsrState(Collections.singletonList(3), leaderAndIsr(), committed); + PendingShrinkIsrState state2 = + new PendingShrinkIsrState(Arrays.asList(2, 3), leaderAndIsr(), committed); + + assertThat(state1).isNotEqualTo(state2); + assertThat(state1.hashCode()).isNotEqualTo(state2.hashCode()); + } + + @Test + void testEqualStatesDeduplicateInHashSet() { + Set states = new HashSet<>(); + states.add(new CommittedIsrState(ISR, STANDBY)); + states.add(new CommittedIsrState(Arrays.asList(1, 2, 3), STANDBY)); + + assertThat(states).hasSize(1); + } +} diff --git a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/RemoteLogManifestHandleTest.java b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/RemoteLogManifestHandleTest.java new file mode 100644 index 00000000000..d2c29719285 --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/RemoteLogManifestHandleTest.java @@ -0,0 +1,70 @@ +/* + * 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.server.zk.data; + +import org.apache.fluss.fs.FsPath; + +import org.junit.jupiter.api.Test; + +import java.util.HashSet; +import java.util.Set; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Test the {@code equals}/{@code hashCode} contract of {@link RemoteLogManifestHandle}. */ +class RemoteLogManifestHandleTest { + + private static final String PATH = "/tmp/remote/log/manifest"; + + @Test + void testEqualsAndHashCode() { + RemoteLogManifestHandle handle1 = new RemoteLogManifestHandle(new FsPath(PATH), 100L); + RemoteLogManifestHandle handle2 = new RemoteLogManifestHandle(new FsPath(PATH), 100L); + + assertThat(handle1).isEqualTo(handle2); + assertThat(handle1).hasSameHashCodeAs(handle2); + } + + @Test + void testHashCodeDiffersForDifferentEndOffset() { + RemoteLogManifestHandle handle1 = new RemoteLogManifestHandle(new FsPath(PATH), 100L); + RemoteLogManifestHandle handle2 = new RemoteLogManifestHandle(new FsPath(PATH), 101L); + + assertThat(handle1).isNotEqualTo(handle2); + assertThat(handle1.hashCode()).isNotEqualTo(handle2.hashCode()); + } + + @Test + void testHashCodeDiffersForDifferentPath() { + RemoteLogManifestHandle handle1 = new RemoteLogManifestHandle(new FsPath(PATH), 100L); + RemoteLogManifestHandle handle2 = + new RemoteLogManifestHandle(new FsPath(PATH + "/other"), 100L); + + assertThat(handle1).isNotEqualTo(handle2); + assertThat(handle1.hashCode()).isNotEqualTo(handle2.hashCode()); + } + + @Test + void testEqualHandlesDeduplicateInHashSet() { + Set handles = new HashSet<>(); + handles.add(new RemoteLogManifestHandle(new FsPath(PATH), 100L)); + handles.add(new RemoteLogManifestHandle(new FsPath(PATH), 100L)); + + assertThat(handles).hasSize(1); + } +}