add equals hashCode to BoundedToUnboundedSourceAdapter (#34057)

Author: scwhittle

sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/UnboundedReadFromBoundedSource.java

@@ -28,6 +28,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.NoSuchElementException;
+import java.util.Objects;
 import java.util.stream.Collectors;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.CoderException;
@@ -121,6 +122,20 @@ public BoundedToUnboundedSourceAdapter(BoundedSource<T> boundedSource) {
       this.boundedSource = boundedSource;
     }
 
+    @Override
+    public boolean equals(Object other) {
+      if (!(other instanceof BoundedToUnboundedSourceAdapter)) {
+        return false;
+      }
+      return Objects.equals(
+          boundedSource, ((BoundedToUnboundedSourceAdapter<?>) other).boundedSource);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(boundedSource);
+    }
+
     @Override
     public void validate() {
       boundedSource.validate();

sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/UnboundedReadFromBoundedSourceTest.java

@@ -18,7 +18,6 @@
 package org.apache.beam.sdk.util.construction;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
@@ -262,7 +261,7 @@ public void testInvokesSplitWithDefaultNumSplitsTooLarge() throws Exception {
     PipelineOptions options = PipelineOptionsFactory.create();
     List<?> splits = unboundedCountingSource.split(100, options);
     assertEquals(1, splits.size());
-    assertNotEquals(splits.get(0), unboundedCountingSource);
+    assertEquals(splits.get(0), unboundedCountingSource);
   }
 
   @Test
@@ -272,7 +271,7 @@ public void testInvokingSplitProducesAtLeastOneSplit() throws Exception {
     PipelineOptions options = PipelineOptionsFactory.create();
     List<?> splits = unboundedCountingSource.split(100, options);
     assertEquals(1, splits.size());
-    assertNotEquals(splits.get(0), unboundedCountingSource);
+    assertEquals(splits.get(0), unboundedCountingSource);
   }
 
   @Test