diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy index a5b3e34571ac..26a7cbc32755 100644 --- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy +++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy @@ -1546,7 +1546,6 @@ class BeamModulePlugin implements Plugin { "PreferJavaTimeOverload", "NonCanonicalType", "Slf4jSignOnlyFormat", - "UndefinedEquals", "UnescapedEntity", "UnrecognisedJavadocTag", // errorprone 3.2.0+ checks diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/InMemoryMultimapSideInputViewTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/InMemoryMultimapSideInputViewTest.java index 869febf5601a..72e497855e6d 100644 --- a/runners/core-java/src/test/java/org/apache/beam/runners/core/InMemoryMultimapSideInputViewTest.java +++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/InMemoryMultimapSideInputViewTest.java @@ -27,6 +27,7 @@ import org.apache.beam.sdk.transforms.Materializations.MultimapView; import org.apache.beam.sdk.values.KV; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.hamcrest.Description; import org.hamcrest.Matchers; import org.hamcrest.TypeSafeMatcher; @@ -43,9 +44,9 @@ public void testStructuralKeyEquality() { InMemoryMultimapSideInputView.fromIterable( ByteArrayCoder.of(), ImmutableList.of(KV.of(new byte[] {0x00}, 0), KV.of(new byte[] {0x01}, 1))); - assertEquals(view.get(new byte[] {0x00}), ImmutableList.of(0)); - assertEquals(view.get(new byte[] {0x01}), ImmutableList.of(1)); - assertEquals(view.get(new byte[] {0x02}), ImmutableList.of()); + assertEquals(Lists.newArrayList(view.get(new byte[] {0x00})), ImmutableList.of(0)); + assertEquals(Lists.newArrayList(view.get(new byte[] {0x01})), ImmutableList.of(1)); + assertEquals(Lists.newArrayList(view.get(new byte[] {0x02})), ImmutableList.of()); assertThat( view.get(), Matchers.containsInAnyOrder( @@ -77,9 +78,9 @@ public void testValueGrouping() { InMemoryMultimapSideInputView.fromIterable( StringUtf8Coder.of(), ImmutableList.of(KV.of("A", "a1"), KV.of("A", "a2"), KV.of("B", "b1"))); - assertEquals(view.get("A"), ImmutableList.of("a1", "a2")); - assertEquals(view.get("B"), ImmutableList.of("b1")); - assertEquals(view.get("C"), ImmutableList.of()); + assertEquals(Lists.newArrayList(view.get("A")), ImmutableList.of("a1", "a2")); + assertEquals(Lists.newArrayList(view.get("B")), ImmutableList.of("b1")); + assertEquals(Lists.newArrayList(view.get("C")), ImmutableList.of()); assertThat(view.get(), containsInAnyOrder("A", "B")); } } diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java index e2db5550ff4a..0163b46a4d22 100644 --- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java +++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java @@ -1699,7 +1699,10 @@ public boolean equals(@Nullable Object other) { } TimerUpdate that = (TimerUpdate) other; return Objects.equals(this.key, that.key) - && Objects.equals(this.completedTimers, that.completedTimers) + && (this.completedTimers == that.completedTimers + || (this.completedTimers != null + && that.completedTimers != null + && Iterables.elementsEqual(this.completedTimers, that.completedTimers))) && Objects.equals(this.setTimers, that.setTimers) && Objects.equals(this.deletedTimers, that.deletedTimers); } diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRegistrarTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRegistrarTest.java index 86384c7fe082..90895ffa596f 100644 --- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRegistrarTest.java +++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRegistrarTest.java @@ -38,14 +38,14 @@ public class DirectRegistrarTest { public void testCorrectOptionsAreReturned() { assertEquals( ImmutableList.of(DirectOptions.class, DirectTestOptions.class), - new Options().getPipelineOptions()); + Lists.newArrayList(new Options().getPipelineOptions())); } @Test public void testCorrectRunnersAreReturned() { assertEquals( ImmutableList.of(org.apache.beam.runners.direct.DirectRunner.class), - new Runner().getPipelineRunners()); + Lists.newArrayList(new Runner().getPipelineRunners())); } @Test diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineRegistrarTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineRegistrarTest.java index c91ad665cd06..44daca79297b 100644 --- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineRegistrarTest.java +++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineRegistrarTest.java @@ -37,14 +37,14 @@ public class DataflowPipelineRegistrarTest { public void testCorrectOptionsAreReturned() { assertEquals( ImmutableList.of(DataflowPipelineOptions.class), - new DataflowPipelineRegistrar.Options().getPipelineOptions()); + Lists.newArrayList(new DataflowPipelineRegistrar.Options().getPipelineOptions())); } @Test public void testCorrectRunnersAreReturned() { assertEquals( ImmutableList.of(DataflowRunner.class, TestDataflowRunner.class), - new DataflowPipelineRegistrar.Runner().getPipelineRunners()); + Lists.newArrayList(new DataflowPipelineRegistrar.Runner().getPipelineRunners())); } @Test diff --git a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/ShuffleSinkTest.java b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/ShuffleSinkTest.java index 025eb7c2fbb1..661ff8781430 100644 --- a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/ShuffleSinkTest.java +++ b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/ShuffleSinkTest.java @@ -117,7 +117,8 @@ private void runTestWriteUngroupingShuffleSink(List expected) throws Ex ByteString valueBytes = record.getValue(); WindowedValue value = CoderUtils.decodeFromByteString(windowedValueCoder, valueBytes); - Assert.assertEquals(Lists.newArrayList(GlobalWindow.INSTANCE), value.getWindows()); + Assert.assertEquals( + Lists.newArrayList(GlobalWindow.INSTANCE), Lists.newArrayList(value.getWindows())); actual.add(value.getValue()); } diff --git a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java index 5b63e408d790..3a25a671ca92 100644 --- a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java +++ b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java @@ -1940,7 +1940,7 @@ public void testMergeWindows() throws Exception { PaneInfo.createPane(true, true, Timing.ON_TIME), PaneInfoCoder.INSTANCE.decode(inStream)); assertEquals( Collections.singletonList(WINDOW_AT_ZERO), - DEFAULT_WINDOW_COLLECTION_CODER.decode(inStream, Coder.Context.OUTER)); + Lists.newArrayList(DEFAULT_WINDOW_COLLECTION_CODER.decode(inStream, Coder.Context.OUTER))); // Data was deleted assertThat( @@ -2230,7 +2230,7 @@ public void testMergeWindowsCaching() throws Exception { PaneInfo.createPane(true, true, Timing.ON_TIME), PaneInfoCoder.INSTANCE.decode(inStream)); assertEquals( Collections.singletonList(WINDOW_AT_ZERO), - DEFAULT_WINDOW_COLLECTION_CODER.decode(inStream, Coder.Context.OUTER)); + Lists.newArrayList(DEFAULT_WINDOW_COLLECTION_CODER.decode(inStream, Coder.Context.OUTER))); // Data was deleted assertThat( diff --git a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/UngroupedShuffleReaderTest.java b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/UngroupedShuffleReaderTest.java index 5e8fe648e7d7..492fe110503c 100644 --- a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/UngroupedShuffleReaderTest.java +++ b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/UngroupedShuffleReaderTest.java @@ -106,7 +106,7 @@ void runTestReadFromShuffle(List expected) throws Exception { for (boolean more = iter.start(); more; more = iter.advance()) { WindowedValue elem = iter.getCurrent(); Assert.assertEquals(timestamp, elem.getTimestamp()); - Assert.assertEquals(Lists.newArrayList(window), elem.getWindows()); + Assert.assertEquals(Lists.newArrayList(window), Lists.newArrayList(elem.getWindows())); actual.add(elem.getValue()); } Assert.assertFalse(iter.advance()); diff --git a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillStateInternalsTest.java b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillStateInternalsTest.java index 7a06d3a29493..de036214b958 100644 --- a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillStateInternalsTest.java +++ b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillStateInternalsTest.java @@ -3540,7 +3540,10 @@ public boolean equals(Object o) { MultimapEntryUpdate that = (MultimapEntryUpdate) o; return deleteAll == that.deleteAll && Objects.equals(key, that.key) - && Objects.equals(values, that.values); + && (values == that.values + || (values != null + && that.values != null + && Iterables.elementsEqual(values, that.values))); } @Override diff --git a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/artifact/ArtifactStagingServiceTest.java b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/artifact/ArtifactStagingServiceTest.java index bd75b9c24db6..5610e4f5bb52 100644 --- a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/artifact/ArtifactStagingServiceTest.java +++ b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/artifact/ArtifactStagingServiceTest.java @@ -20,7 +20,6 @@ import static org.junit.Assert.assertEquals; import java.nio.file.Path; -import java.util.Collection; import java.util.Iterator; import java.util.List; import java.util.Map; @@ -168,7 +167,7 @@ public void testStageArtifacts() throws InterruptedException, ExecutionException } private void checkArtifacts( - Collection expectedContents, List staged) { + List expectedContents, List staged) { assertEquals( expectedContents, Lists.transform(staged, RunnerApi.ArtifactInformation::getRoleUrn)); assertEquals(expectedContents, Lists.transform(staged, this::getArtifact)); diff --git a/runners/prism/java/src/test/java/org/apache/beam/runners/prism/PrismRegistrarTest.java b/runners/prism/java/src/test/java/org/apache/beam/runners/prism/PrismRegistrarTest.java index 4ed69b14edf8..f9af9c07e0b3 100644 --- a/runners/prism/java/src/test/java/org/apache/beam/runners/prism/PrismRegistrarTest.java +++ b/runners/prism/java/src/test/java/org/apache/beam/runners/prism/PrismRegistrarTest.java @@ -38,14 +38,14 @@ public class PrismRegistrarTest { public void testCorrectOptionsAreReturned() { assertEquals( ImmutableList.of(PrismPipelineOptions.class, TestPrismPipelineOptions.class), - new Options().getPipelineOptions()); + Lists.newArrayList(new Options().getPipelineOptions())); } @Test public void testCorrectRunnersAreReturned() { assertEquals( ImmutableList.of(PrismRunner.class, TestPrismRunner.class), - new Runner().getPipelineRunners()); + Lists.newArrayList(new Runner().getPipelineRunners())); } @Test diff --git a/runners/samza/src/test/java/org/apache/beam/runners/samza/runtime/SamzaTimerInternalsFactoryTest.java b/runners/samza/src/test/java/org/apache/beam/runners/samza/runtime/SamzaTimerInternalsFactoryTest.java index c7998a281f7c..ecedd7ae79f1 100644 --- a/runners/samza/src/test/java/org/apache/beam/runners/samza/runtime/SamzaTimerInternalsFactoryTest.java +++ b/runners/samza/src/test/java/org/apache/beam/runners/samza/runtime/SamzaTimerInternalsFactoryTest.java @@ -249,7 +249,8 @@ public void testRestoreEventBufferSize() throws Exception { ByteArrayOutputStream baos = new ByteArrayOutputStream(); StringUtf8Coder.of().encode(key, baos); byte[] keyBytes = baos.toByteArray(); - assertEquals(readyTimers, Arrays.asList(new KeyedTimerData<>(keyBytes, key, timer2))); + assertEquals( + new ArrayList<>(readyTimers), Arrays.asList(new KeyedTimerData<>(keyBytes, key, timer2))); store.close(); } @@ -301,7 +302,7 @@ public void testRestore() throws Exception { StringUtf8Coder.of().encode(key, baos); byte[] keyBytes = baos.toByteArray(); assertEquals( - readyTimers, + new ArrayList<>(readyTimers), Arrays.asList( new KeyedTimerData<>(keyBytes, key, timer1), new KeyedTimerData<>(keyBytes, key, timer2))); diff --git a/runners/samza/src/test/java/org/apache/beam/runners/samza/util/FutureUtilsTest.java b/runners/samza/src/test/java/org/apache/beam/runners/samza/util/FutureUtilsTest.java index 2adc9f9e65e9..c261fc623af3 100644 --- a/runners/samza/src/test/java/org/apache/beam/runners/samza/util/FutureUtilsTest.java +++ b/runners/samza/src/test/java/org/apache/beam/runners/samza/util/FutureUtilsTest.java @@ -23,6 +23,7 @@ import java.util.concurrent.CompletionStage; import java.util.concurrent.CountDownLatch; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.junit.Assert; import org.junit.Test; @@ -42,7 +43,9 @@ public void testFlattenFuturesForCollection() { resultFuture.thenAccept( actualResults -> { Assert.assertEquals( - "Expected flattened results to contain {hello, world}", RESULTS, actualResults); + "Expected flattened results to contain {hello, world}", + RESULTS, + Lists.newArrayList(actualResults)); }); validationFuture.toCompletableFuture().join(); @@ -93,7 +96,9 @@ public void testWaitForAllFutures() { resultFuture.thenAccept( actualResults -> { Assert.assertEquals( - "Expected flattened results to contain {hello, world}", RESULTS, actualResults); + "Expected flattened results to contain {hello, world}", + RESULTS, + Lists.newArrayList(actualResults)); }); validationFuture.toCompletableFuture().join(); diff --git a/runners/spark/3/src/test/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingRunnerRegistrarTest.java b/runners/spark/3/src/test/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingRunnerRegistrarTest.java index bc7b561eea2d..23a5427fc437 100644 --- a/runners/spark/3/src/test/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingRunnerRegistrarTest.java +++ b/runners/spark/3/src/test/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingRunnerRegistrarTest.java @@ -36,14 +36,16 @@ public class SparkStructuredStreamingRunnerRegistrarTest { public void testOptions() { assertEquals( ImmutableList.of(SparkStructuredStreamingPipelineOptions.class), - new SparkStructuredStreamingRunnerRegistrar.Options().getPipelineOptions()); + Lists.newArrayList( + new SparkStructuredStreamingRunnerRegistrar.Options().getPipelineOptions())); } @Test public void testRunners() { assertEquals( ImmutableList.of(SparkStructuredStreamingRunner.class), - new SparkStructuredStreamingRunnerRegistrar.Runner().getPipelineRunners()); + Lists.newArrayList( + new SparkStructuredStreamingRunnerRegistrar.Runner().getPipelineRunners())); } @Test diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/SparkRunnerRegistrarTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/SparkRunnerRegistrarTest.java index c0b62be045fe..fb6df4e1b127 100644 --- a/runners/spark/src/test/java/org/apache/beam/runners/spark/SparkRunnerRegistrarTest.java +++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/SparkRunnerRegistrarTest.java @@ -36,14 +36,14 @@ public class SparkRunnerRegistrarTest { public void testOptions() { assertEquals( ImmutableList.of(SparkPipelineOptions.class, SparkPortableStreamingPipelineOptions.class), - new SparkRunnerRegistrar.Options().getPipelineOptions()); + Lists.newArrayList(new SparkRunnerRegistrar.Options().getPipelineOptions())); } @Test public void testRunners() { assertEquals( ImmutableList.of(SparkRunner.class, TestSparkRunner.class), - new SparkRunnerRegistrar.Runner().getPipelineRunners()); + Lists.newArrayList(new SparkRunnerRegistrar.Runner().getPipelineRunners())); } @Test diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/ValueAndCoderLazySerializableTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/ValueAndCoderLazySerializableTest.java index 0f2ef2ce5f45..8588a780357b 100644 --- a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/ValueAndCoderLazySerializableTest.java +++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/ValueAndCoderLazySerializableTest.java @@ -34,6 +34,7 @@ import org.apache.beam.sdk.transforms.windowing.PaneInfo; import org.apache.beam.sdk.values.WindowedValue; import org.apache.beam.sdk.values.WindowedValues; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.joda.time.Instant; import org.junit.Test; @@ -67,7 +68,9 @@ public void serializableAccumulatorSerializationTest() @SuppressWarnings("unchecked") ValueAndCoderLazySerializable>> materialized = (ValueAndCoderLazySerializable>>) ois.readObject(); - assertEquals(accumulatedValue, materialized.getOrDecode(iterAccumCoder)); + assertEquals( + Lists.newArrayList(accumulatedValue), + Lists.newArrayList(materialized.getOrDecode(iterAccumCoder))); } @Test @@ -101,7 +104,9 @@ public void serializableAccumulatorKryoTest() { kryo.readObject(input, ValueAndCoderLazySerializable.class); input.close(); - assertEquals(accumulatedValue, materialized.getOrDecode(iterAccumCoder)); + assertEquals( + Lists.newArrayList(accumulatedValue), + Lists.newArrayList(materialized.getOrDecode(iterAccumCoder))); } private WindowedValue winVal(T val) { diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/StaticWindows.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/StaticWindows.java index fe95616bec9d..74c4a2ada6b7 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/StaticWindows.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/StaticWindows.java @@ -21,7 +21,6 @@ import java.util.Collection; import java.util.Collections; -import java.util.Objects; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.IncompatibleWindowException; @@ -94,7 +93,7 @@ public boolean isCompatible(WindowFn other) { return false; } StaticWindows that = (StaticWindows) other; - return Objects.equals(this.windows.get(), that.windows.get()); + return Iterables.elementsEqual(this.windows.get(), that.windows.get()); } @Override diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/Timer.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/Timer.java index bd6e0d22a719..9ea6dfd1c8b4 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/Timer.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/Timer.java @@ -37,6 +37,7 @@ import org.apache.beam.sdk.transforms.windowing.PaneInfo.PaneInfoCoder; import org.apache.beam.sdk.util.common.ElementByteSizeObserver; import org.apache.beam.sdk.values.CausedByDrain; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables; import org.checkerframework.checker.nullness.qual.Nullable; import org.joda.time.Instant; @@ -136,7 +137,7 @@ public final boolean equals(@Nullable Object other) { Timer that = (Timer) other; return Objects.equals(this.getUserKey(), that.getUserKey()) && Objects.equals(this.getDynamicTimerTag(), that.getDynamicTimerTag()) - && Objects.equals(this.getWindows(), that.getWindows()) + && Iterables.elementsEqual(this.getWindows(), that.getWindows()) && (this.getClearBit() == that.getClearBit()) && Objects.equals(this.getFireTimestamp(), that.getFireTimestamp()) && Objects.equals(this.getHoldTimestamp(), that.getHoldTimestamp()) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionViews.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionViews.java index 05612cb9fc77..e4d2fe649ff3 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionViews.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionViews.java @@ -2052,6 +2052,7 @@ public Iterator> iterator() { } @Override + @SuppressWarnings("UndefinedEquals") public boolean contains(Object o) { if (!(o instanceof Entry)) { return false; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/WindowedValues.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/WindowedValues.java index 79f4b4a62ac2..ba2720f5e39b 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/WindowedValues.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/WindowedValues.java @@ -52,6 +52,7 @@ import org.apache.beam.sdk.util.common.ElementByteSizeObserver; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables; import org.checkerframework.checker.nullness.qual.MonotonicNonNull; import org.checkerframework.checker.nullness.qual.Nullable; import org.checkerframework.dataflow.qual.Pure; @@ -413,7 +414,7 @@ public static boolean equals( // comparisons are made on their Chronology objects. return left.getTimestamp().isEqual(right.getTimestamp()) && Objects.equals(left.getValue(), right.getValue()) - && Objects.equals(left.getWindows(), right.getWindows()) + && Iterables.elementsEqual(left.getWindows(), right.getWindows()) && Objects.equals(left.getPaneInfo(), right.getPaneInfo()); } @@ -789,6 +790,7 @@ public WindowedValue withValue(NewT newValue) { } @Override + @SuppressWarnings("UndefinedEquals") public boolean equals(@Nullable Object o) { if (o instanceof TimestampedValueInMultipleWindows) { TimestampedValueInMultipleWindows that = (TimestampedValueInMultipleWindows) o; diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/SerializableCoderTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/SerializableCoderTest.java index f3569322a182..a8cb79b1298e 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/SerializableCoderTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/SerializableCoderTest.java @@ -45,6 +45,7 @@ import org.apache.beam.sdk.util.SerializableUtils; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.TypeDescriptor; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.checkerframework.checker.nullness.qual.Nullable; import org.hamcrest.Description; import org.hamcrest.Matchers; @@ -121,7 +122,7 @@ public void testSerializableCoder() throws Exception { byte[] encoded = CoderUtils.encodeToByteArray(coder, records); Iterable decoded = CoderUtils.decodeFromByteArray(coder, encoded); - assertEquals(records, decoded); + assertEquals(records, Lists.newArrayList(decoded)); } @Test diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/JavaBeanSchemaTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/JavaBeanSchemaTest.java index 736cc250a827..df6c9cf18e5a 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/JavaBeanSchemaTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/JavaBeanSchemaTest.java @@ -330,9 +330,9 @@ public void testPrimitiveArrayGetters() throws NoSuchSchemaException { Long[] longArray = {42L, 43L, 44L}; PrimitiveArrayBean bean = new PrimitiveArrayBean(strList, intArray, longArray); Row row = registry.getToRowFunction(PrimitiveArrayBean.class).apply(bean); - assertEquals(strList, row.getArray("strings")); - assertEquals(Ints.asList(intArray), row.getArray("integers")); - assertEquals(Arrays.asList(longArray), row.getArray("longs")); + assertEquals(strList, (List) row.getArray("strings")); + assertEquals(Ints.asList(intArray), (List) row.getArray("integers")); + assertEquals(Arrays.asList(longArray), (List) row.getArray("longs")); // Ensure that list caching works. assertSame(row.getArray("strings"), row.getArray("strings")); @@ -350,9 +350,9 @@ public void testPrimitiveArraySetters() throws NoSuchSchemaException { .addArray(42L, 43L, 44L, 45L) .build(); PrimitiveArrayBean bean = registry.getFromRowFunction(PrimitiveArrayBean.class).apply(row); - assertEquals(row.getArray("strings"), bean.getStrings()); - assertEquals(row.getArray("integers"), Ints.asList(bean.getIntegers())); - assertEquals(row.getArray("longs"), Arrays.asList(bean.getLongs())); + assertEquals((List) row.getArray("strings"), bean.getStrings()); + assertEquals((List) row.getArray("integers"), Ints.asList(bean.getIntegers())); + assertEquals((List) row.getArray("longs"), Arrays.asList(bean.getLongs())); } @Test @@ -402,7 +402,7 @@ public void testNestedArraysGetters() throws NoSuchSchemaException { Lists.newArrayList("g", "h", "i")); NestedArraysBean bean = new NestedArraysBean(listOfLists); Row row = registry.getToRowFunction(NestedArraysBean.class).apply(bean); - assertEquals(listOfLists, row.getArray("lists")); + assertEquals(listOfLists, (List) row.getArray("lists")); } @Test diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/JavaFieldSchemaTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/JavaFieldSchemaTest.java index 66794d5a512e..7a66decde017 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/JavaFieldSchemaTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/JavaFieldSchemaTest.java @@ -355,9 +355,9 @@ public void testPrimitiveArrayGetters() throws NoSuchSchemaException { Long[] longArray = {42L, 43L, 44L}; PrimitiveArrayPOJO pojo = new PrimitiveArrayPOJO(strList, intArray, longArray); Row row = registry.getToRowFunction(PrimitiveArrayPOJO.class).apply(pojo); - assertEquals(strList, row.getArray("strings")); - assertEquals(Ints.asList(intArray), row.getArray("integers")); - assertEquals(Arrays.asList(longArray), row.getArray("longs")); + assertEquals(strList, (List) row.getArray("strings")); + assertEquals(Ints.asList(intArray), (List) row.getArray("integers")); + assertEquals(Arrays.asList(longArray), (List) row.getArray("longs")); // Ensure that list caching works. assertSame(row.getArray("strings"), row.getArray("strings")); @@ -375,9 +375,9 @@ public void testPrimitiveArraySetters() throws NoSuchSchemaException { .addArray(42L, 43L, 44L, 45L) .build(); PrimitiveArrayPOJO pojo = registry.getFromRowFunction(PrimitiveArrayPOJO.class).apply(row); - assertEquals(row.getArray("strings"), pojo.strings); - assertEquals(row.getArray("integers"), Ints.asList(pojo.integers)); - assertEquals(row.getArray("longs"), Arrays.asList(pojo.longs)); + assertEquals((List) row.getArray("strings"), pojo.strings); + assertEquals((List) row.getArray("integers"), Ints.asList(pojo.integers)); + assertEquals((List) row.getArray("longs"), Arrays.asList(pojo.longs)); } @Test @@ -428,7 +428,7 @@ public void testNestedArraysGetters() throws NoSuchSchemaException { Lists.newArrayList("g", "h", "i")); NestedArraysPOJO pojo = new NestedArraysPOJO(listOfLists); Row row = registry.getToRowFunction(NestedArraysPOJO.class).apply(pojo); - assertEquals(listOfLists, row.getArray("lists")); + assertEquals(listOfLists, (List) row.getArray("lists")); } @Test diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/TestJavaBeans.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/TestJavaBeans.java index 738099de5df7..694db05b0918 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/TestJavaBeans.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/TestJavaBeans.java @@ -34,6 +34,7 @@ import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber; import org.apache.beam.sdk.schemas.annotations.SchemaIgnore; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.CaseFormat; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables; import org.checkerframework.checker.nullness.qual.Nullable; import org.joda.time.DateTime; import org.joda.time.Instant; @@ -905,7 +906,7 @@ public boolean equals(@Nullable Object o) { } NestedCollectionBean that = (NestedCollectionBean) o; return Objects.equals(simples, that.simples) - && Objects.equals(iterableSimples, that.iterableSimples); + && Iterables.elementsEqual(iterableSimples, that.iterableSimples); } @Override @@ -1178,7 +1179,7 @@ public boolean equals(@Nullable Object o) { return false; } IterableBean that = (IterableBean) o; - return Objects.equals(strings, that.strings); + return Iterables.elementsEqual(strings, that.strings); } @Override diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/TestPOJOs.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/TestPOJOs.java index eca6b1099864..0d9338654ad9 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/TestPOJOs.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/TestPOJOs.java @@ -35,6 +35,7 @@ import org.apache.beam.sdk.schemas.annotations.SchemaIgnore; import org.apache.beam.sdk.schemas.logicaltypes.EnumerationType; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.CaseFormat; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables; import org.checkerframework.checker.nullness.qual.Nullable; import org.joda.time.DateTime; import org.joda.time.Instant; @@ -664,7 +665,10 @@ public boolean equals(@Nullable Object o) { } NestedCollectionPOJO that = (NestedCollectionPOJO) o; return Objects.equals(simples, that.simples) - && Objects.equals(iterableSimples, that.iterableSimples); + && (iterableSimples == that.iterableSimples + || (iterableSimples != null + && that.iterableSimples != null + && Iterables.elementsEqual(iterableSimples, that.iterableSimples))); } @Override @@ -891,7 +895,10 @@ public boolean equals(@Nullable Object o) { return false; } PojoWithIterable that = (PojoWithIterable) o; - return Objects.equals(strings, that.strings); + return strings == that.strings + || (strings != null + && that.strings != null + && Iterables.elementsEqual(strings, that.strings)); } @Override diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataTest.java index 65372573fcd3..9f9f15e25c6b 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataTest.java @@ -60,6 +60,7 @@ import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMultimap; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Multimap; import org.checkerframework.checker.nullness.qual.Nullable; import org.hamcrest.CustomTypeSafeMatcher; @@ -268,7 +269,7 @@ public void populateDisplayData(DisplayData.Builder builder) { Map map = data.asMap(); assertEquals(1, map.size()); assertThat(data, hasDisplayItem("foo", "bar")); - assertEquals(map.values(), data.items()); + assertEquals(Lists.newArrayList(map.values()), Lists.newArrayList(data.items())); } @Test diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/TimerTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/TimerTest.java index 06f43350f5ae..23534c300e8b 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/TimerTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/TimerTest.java @@ -22,6 +22,7 @@ import static org.junit.Assert.assertTrue; import java.util.Collections; +import java.util.Set; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.testing.CoderProperties; @@ -46,7 +47,7 @@ public void testClearTimer() { assertTrue(clearedTimer.getClearBit()); assertEquals("timer", clearedTimer.getUserKey()); assertEquals("tag", clearedTimer.getDynamicTimerTag()); - assertEquals(Collections.singleton(GlobalWindow.INSTANCE), clearedTimer.getWindows()); + assertEquals(Collections.singleton(GlobalWindow.INSTANCE), (Set) clearedTimer.getWindows()); } @Test @@ -64,7 +65,7 @@ public void testTimer() { assertEquals("tag", timer.getDynamicTimerTag()); assertEquals(FIRE_TIME, timer.getFireTimestamp()); assertEquals(HOLD_TIME, timer.getHoldTimestamp()); - assertEquals(Collections.singleton(GlobalWindow.INSTANCE), timer.getWindows()); + assertEquals(Collections.singleton(GlobalWindow.INSTANCE), (Set) timer.getWindows()); assertEquals(PaneInfo.NO_FIRING, timer.getPaneInfo()); assertFalse(timer.getClearBit()); } diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java index 37583f0ceedc..95d43aed1703 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java @@ -189,7 +189,7 @@ public void testCreatesArray() { Stream.of(Schema.Field.of("array", Schema.FieldType.array(Schema.FieldType.INT32))) .collect(toSchema()); Row row = Row.withSchema(type).addArray(data).build(); - assertEquals(data, row.getArray("array")); + assertEquals(data, (List) row.getArray("array")); } @Test @@ -199,7 +199,7 @@ public void testCreatesIterable() { Stream.of(Schema.Field.of("iter", Schema.FieldType.iterable(Schema.FieldType.INT32))) .collect(toSchema()); Row row = Row.withSchema(type).addIterable(data).build(); - assertEquals(data, row.getIterable("iter")); + assertEquals(data, Lists.newArrayList(row.getIterable("iter"))); } @Test @@ -209,7 +209,7 @@ public void testCreatesAndComparesNullArray() { Stream.of(Schema.Field.nullable("array", Schema.FieldType.array(Schema.FieldType.INT32))) .collect(toSchema()); Row row = Row.withSchema(type).addArray(data).build(); - assertEquals(data, row.getArray("array")); + assertEquals(data, (List) row.getArray("array")); Row otherNonNull = Row.withSchema(type).addValue(ImmutableList.of(1, 2, 3)).build(); Row otherNull = Row.withSchema(type).addValue(null).build(); @@ -224,7 +224,7 @@ public void testCreatesAndComparesNullIterable() { Stream.of(Schema.Field.nullable("iter", Schema.FieldType.iterable(Schema.FieldType.INT32))) .collect(toSchema()); Row row = Row.withSchema(type).addIterable(data).build(); - assertEquals(data, row.getArray("iter")); + assertEquals(data, (List) row.getArray("iter")); Row otherNonNull = Row.withSchema(type).addValue(ImmutableList.of(1, 2, 3)).build(); Row otherNull = Row.withSchema(type).addValue(null).build(); @@ -241,7 +241,7 @@ public void testCreatesArrayWithNullElement() { "array", Schema.FieldType.array(Schema.FieldType.INT32.withNullable(true)))) .collect(toSchema()); Row row = Row.withSchema(type).addArray(data).build(); - assertEquals(data, row.getArray("array")); + assertEquals(data, (List) row.getArray("array")); } @Test @@ -253,7 +253,7 @@ public void testCreatesIterableWithNullElement() { "iter", Schema.FieldType.iterable(Schema.FieldType.INT32.withNullable(true)))) .collect(toSchema()); Row row = Row.withSchema(type).addIterable(data).build(); - assertEquals(data, row.getIterable("iter")); + assertEquals(data, Lists.newArrayList(row.getIterable("iter"))); } @Test @@ -269,7 +269,7 @@ public void testCreatesRowArray() { Stream.of(Schema.Field.of("array", FieldType.array(FieldType.row(nestedType)))) .collect(toSchema()); Row row = Row.withSchema(type).addArray(data).build(); - assertEquals(data, row.getArray("array")); + assertEquals(data, (List) row.getArray("array")); } @Test @@ -285,7 +285,7 @@ public void testCreatesRowIterable() { Stream.of(Schema.Field.of("iter", FieldType.iterable(FieldType.row(nestedType)))) .collect(toSchema()); Row row = Row.withSchema(type).addIterable(data).build(); - assertEquals(data, row.getIterable("iter")); + assertEquals(data, Lists.newArrayList(row.getIterable("iter"))); } @Test @@ -295,7 +295,7 @@ public void testCreatesArrayArray() { Stream.of(Schema.Field.of("array", FieldType.array(FieldType.array(FieldType.INT32)))) .collect(toSchema()); Row row = Row.withSchema(type).addArray(data).build(); - assertEquals(data, row.getArray("array")); + assertEquals(data, (List) row.getArray("array")); } @Test @@ -305,7 +305,7 @@ public void testCreatesIterableArray() { Stream.of(Schema.Field.of("iter", FieldType.iterable(FieldType.array(FieldType.INT32)))) .collect(toSchema()); Row row = Row.withSchema(type).addIterable(data).build(); - assertEquals(data, row.getIterable("iter")); + assertEquals(data, Lists.newArrayList(row.getIterable("iter"))); } @Test @@ -321,7 +321,7 @@ public void testCreatesArrayArrayWithNullElement() { .withNullable(true))) .collect(toSchema()); Row row = Row.withSchema(type).addArray(data).build(); - assertEquals(data, row.getArray("array")); + assertEquals(data, (List) row.getArray("array")); } @Test @@ -338,7 +338,7 @@ public void testCreatesIterableIterableWithNullElement() { .withNullable(true))) .collect(toSchema()); Row row = Row.withSchema(type).addIterable(data).build(); - assertEquals(data, row.getIterable("iter")); + assertEquals(data, Lists.newArrayList(row.getIterable("iter"))); } @Test @@ -354,7 +354,7 @@ public void testCreatesArrayOfMap() { "array", FieldType.array(FieldType.map(FieldType.INT32, FieldType.STRING)))) .collect(toSchema()); Row row = Row.withSchema(type).addArray(data).build(); - assertEquals(data, row.getArray("array")); + assertEquals(data, (List) row.getArray("array")); } @Test @@ -370,7 +370,7 @@ public void testCreatesIterableOfMap() { "iter", FieldType.iterable(FieldType.map(FieldType.INT32, FieldType.STRING)))) .collect(toSchema()); Row row = Row.withSchema(type).addIterable(data).build(); - assertEquals(data, row.getIterable("iter")); + assertEquals(data, Lists.newArrayList(row.getIterable("iter"))); } @Test @@ -396,7 +396,7 @@ public void testCreateAndCompareNullMap() { Stream.of(Schema.Field.nullable("map", FieldType.map(FieldType.INT32, FieldType.STRING))) .collect(toSchema()); Row row = Row.withSchema(type).addValue(data).build(); - assertEquals(data, row.getArray("map")); + assertEquals(data, (List) row.getArray("map")); Row otherNonNull = Row.withSchema(type).addValue(ImmutableMap.of(1, "value1")).build(); Row otherNull = Row.withSchema(type).addValue(null).build(); diff --git a/sdks/java/extensions/sorter/src/test/java/org/apache/beam/sdk/extensions/sorter/BufferedExternalSorterTest.java b/sdks/java/extensions/sorter/src/test/java/org/apache/beam/sdk/extensions/sorter/BufferedExternalSorterTest.java index b6d56dc41fab..5d6e54c6b689 100644 --- a/sdks/java/extensions/sorter/src/test/java/org/apache/beam/sdk/extensions/sorter/BufferedExternalSorterTest.java +++ b/sdks/java/extensions/sorter/src/test/java/org/apache/beam/sdk/extensions/sorter/BufferedExternalSorterTest.java @@ -110,7 +110,7 @@ public void testNoFallback() throws Exception { testSorter.add(kvs[1]); testSorter.add(kvs[2]); - assertEquals(Arrays.asList(kvs[0], kvs[1], kvs[2]), testSorter.sort()); + assertEquals(Arrays.asList(kvs[0], kvs[1], kvs[2]), (java.util.List) testSorter.sort()); // Verify external sorter was never called verify(mockExternalSorter, never()).add(any(KV.class)); @@ -142,7 +142,7 @@ public void testFallback() throws Exception { testSorter.add(kvs[1]); testSorter.add(kvs[2]); - assertEquals(Arrays.asList(kvs[0], kvs[1], kvs[2]), testSorter.sort()); + assertEquals(Arrays.asList(kvs[0], kvs[1], kvs[2]), (java.util.List) testSorter.sort()); verify(mockExternalSorter, times(1)).add(kvs[0]); verify(mockExternalSorter, times(1)).add(kvs[1]); diff --git a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/PrecombineGroupingTable.java b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/PrecombineGroupingTable.java index d6827e821782..4929c128dbf7 100644 --- a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/PrecombineGroupingTable.java +++ b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/PrecombineGroupingTable.java @@ -41,6 +41,7 @@ import org.apache.beam.sdk.values.WindowedValue; import org.apache.beam.sdk.values.WindowedValues; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables; import org.joda.time.Instant; /** @@ -219,7 +220,8 @@ public boolean equals(Object o) { return false; } WindowedGroupingTableKey that = (WindowedGroupingTableKey) o; - return structuralKey.equals(that.structuralKey) && windows.equals(that.windows); + return structuralKey.equals(that.structuralKey) + && Iterables.elementsEqual(windows, that.windows); } @Override diff --git a/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/WindowMergingFnRunnerTest.java b/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/WindowMergingFnRunnerTest.java index 2deaa602c252..de7fa2edbcab 100644 --- a/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/WindowMergingFnRunnerTest.java +++ b/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/WindowMergingFnRunnerTest.java @@ -100,7 +100,9 @@ public void testWindowMergingWithMergingWindowFn() throws Exception { KV, Iterable>>>> output = mapFunction.apply(input); assertEquals(input.getKey(), output.getKey()); - assertEquals(expectedToBeUnmerged, output.getValue().getKey()); + assertThat( + output.getValue().getKey(), + containsInAnyOrder(Iterables.toArray(expectedToBeUnmerged, BoundedWindow.class))); KV> mergedOutput = Iterables.getOnlyElement(output.getValue().getValue()); assertEquals(new IntervalWindow(new Instant(7L), new Instant(11L)), mergedOutput.getKey()); @@ -123,7 +125,9 @@ public void testWindowMergingWithMergingWindowFn() throws Exception { output = mapFunction.apply(input); assertEquals(input.getKey(), output.getKey()); - assertEquals(expectedToBeUnmerged, output.getValue().getKey()); + assertThat( + output.getValue().getKey(), + containsInAnyOrder(Iterables.toArray(expectedToBeUnmerged, BoundedWindow.class))); mergedOutput = Iterables.getOnlyElement(output.getValue().getValue()); assertEquals(new IntervalWindow(new Instant(15L), new Instant(18L)), mergedOutput.getKey()); assertThat(mergedOutput.getValue(), containsInAnyOrder(expectedToBeMergedGroup2)); diff --git a/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/state/LazyCachingIteratorToIterableTest.java b/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/state/LazyCachingIteratorToIterableTest.java index fd701099a778..a39341661391 100644 --- a/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/state/LazyCachingIteratorToIterableTest.java +++ b/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/state/LazyCachingIteratorToIterableTest.java @@ -81,12 +81,14 @@ public void testInterleavedIteration() { @Test public void testEqualsAndHashCode() { - Iterable iterA = + LazyCachingIteratorToIterable iterA = new LazyCachingIteratorToIterable<>(PrefetchableIterators.fromArray("A", "B", "C")); - Iterable iterB = + LazyCachingIteratorToIterable iterB = new LazyCachingIteratorToIterable<>(PrefetchableIterators.fromArray("A", "B", "C")); - Iterable iterC = new LazyCachingIteratorToIterable<>(PrefetchableIterators.fromArray()); - Iterable iterD = new LazyCachingIteratorToIterable<>(PrefetchableIterators.fromArray()); + LazyCachingIteratorToIterable iterC = + new LazyCachingIteratorToIterable<>(PrefetchableIterators.fromArray()); + LazyCachingIteratorToIterable iterD = + new LazyCachingIteratorToIterable<>(PrefetchableIterators.fromArray()); assertEquals(iterA, iterB); assertEquals(iterC, iterD); assertNotEquals(iterA, iterC); diff --git a/sdks/java/io/file-schema-transform/src/test/java/org/apache/beam/sdk/io/fileschematransform/XmlRowValueTest.java b/sdks/java/io/file-schema-transform/src/test/java/org/apache/beam/sdk/io/fileschematransform/XmlRowValueTest.java index 9d930f50f4fb..b4b8e3f15a96 100644 --- a/sdks/java/io/file-schema-transform/src/test/java/org/apache/beam/sdk/io/fileschematransform/XmlRowValueTest.java +++ b/sdks/java/io/file-schema-transform/src/test/java/org/apache/beam/sdk/io/fileschematransform/XmlRowValueTest.java @@ -133,7 +133,7 @@ public void arrayPrimitiveDataTypes() { assertTrue(booleans, booleansValueList.isPresent()); assertEquals( booleans, - row.getArray(booleans), + (List) row.getArray(booleans), booleansValueList.get().stream() .map(XmlRowValue::getPrimitiveValue) .collect(Collectors.toList())); @@ -145,7 +145,7 @@ public void arrayPrimitiveDataTypes() { assertTrue(doubles, doublesValueList.isPresent()); assertEquals( doubles, - row.getArray(doubles), + (List) row.getArray(doubles), doublesValueList.get().stream() .map(XmlRowValue::getPrimitiveValue) .collect(Collectors.toList())); @@ -157,7 +157,7 @@ public void arrayPrimitiveDataTypes() { assertTrue(floats, floatsValueList.isPresent()); assertEquals( floats, - row.getArray(floats), + (List) row.getArray(floats), floatsValueList.get().stream() .map(XmlRowValue::getPrimitiveValue) .collect(Collectors.toList())); @@ -169,7 +169,7 @@ public void arrayPrimitiveDataTypes() { assertTrue(integers, integersValueList.isPresent()); assertEquals( integers, - row.getArray(integers), + (List) row.getArray(integers), integersValueList.get().stream() .map(XmlRowValue::getPrimitiveValue) .collect(Collectors.toList())); @@ -181,7 +181,7 @@ public void arrayPrimitiveDataTypes() { assertTrue(longs, longsValueList.isPresent()); assertEquals( longs, - row.getArray(longs), + (List) row.getArray(longs), longsValueList.get().stream() .map(XmlRowValue::getPrimitiveValue) .collect(Collectors.toList())); @@ -193,7 +193,7 @@ public void arrayPrimitiveDataTypes() { assertTrue(strings, stringsValueList.isPresent()); assertEquals( strings, - row.getArray(strings), + (List) row.getArray(strings), stringsValueList.get().stream() .map(XmlRowValue::getPrimitiveValue) .collect(Collectors.toList())); @@ -244,7 +244,7 @@ public void byteType() { byteListValue.setValue(byteList, row); assertEquals( singleByte, - row.getArray(byteList), + (List) row.getArray(byteList), byteListValue.getValueList().stream() .map(XmlRowValue::getPrimitiveValue) .collect(Collectors.toList())); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RecordWriterManagerTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RecordWriterManagerTest.java index 2672ac70c088..db6ac943b8d1 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RecordWriterManagerTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RecordWriterManagerTest.java @@ -341,7 +341,7 @@ public void testRequireClosingBeforeFetchingDataFiles() { /** DataFile doesn't implement a .equals() method. Check equality manually. */ private static void checkDataFileEquality(DataFile d1, DataFile d2) { - assertEquals(d1.path(), d2.path()); + assertEquals(d1.path().toString(), d2.path().toString()); assertEquals(d1.format(), d2.format()); assertEquals(d1.recordCount(), d2.recordCount()); assertEquals(d1.partition(), d2.partition()); diff --git a/sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcSchemaIOProviderTest.java b/sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcSchemaIOProviderTest.java index 3a8b83b992e2..98d21507643d 100644 --- a/sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcSchemaIOProviderTest.java +++ b/sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcSchemaIOProviderTest.java @@ -35,6 +35,7 @@ import org.apache.beam.sdk.transforms.Count; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.Row; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.junit.BeforeClass; import org.junit.Rule; import org.junit.Test; @@ -233,7 +234,7 @@ public void testAbleToReadDataSourceConfiguration() { "connectionProp", Objects.requireNonNull(dataSourceConf.getConnectionProperties()).get()); assertEquals( new ArrayList<>(Collections.singleton("initSql")), - Objects.requireNonNull(dataSourceConf.getConnectionInitSqls()).get()); + Lists.newArrayList(Objects.requireNonNull(dataSourceConf.getConnectionInitSqls()).get())); assertEquals(3, (int) dataSourceConf.getMaxConnections().get()); assertEquals("test.jar", Objects.requireNonNull(dataSourceConf.getDriverJars()).get()); assertEquals(10L, schemaIO.config.getInt64("writeBatchSize").longValue()); diff --git a/sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcUtilTest.java b/sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcUtilTest.java index da0faad46bc6..3bfda67f919c 100644 --- a/sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcUtilTest.java +++ b/sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcUtilTest.java @@ -327,7 +327,7 @@ public void testFqnFromHikariDataSourcePostgreSql() { assertEquals("cloudsql_postgresql", components.getScheme()); assertEquals( ImmutableList.of("example.com:project", "some-region", "instance-name", "postgres"), - components.getSegments()); + Lists.newArrayList(components.getSegments())); } @Test @@ -341,7 +341,7 @@ public void testFqnFromBasicDataSourcePostgreSql() { assertEquals("cloudsql_postgresql", components.getScheme()); assertEquals( ImmutableList.of("example.com:project", "some-region", "instance-name", "postgres"), - components.getSegments()); + Lists.newArrayList(components.getSegments())); } @Test @@ -358,7 +358,8 @@ public void testFqnFromHikariDataSourceMySql() { JdbcUtil.FQNComponents components = JdbcUtil.FQNComponents.of(dataSource); assertEquals("cloudsql_mysql", components.getScheme()); assertEquals( - ImmutableList.of("some-project", "US", "instance-name", "db"), components.getSegments()); + ImmutableList.of("some-project", "US", "instance-name", "db"), + Lists.newArrayList(components.getSegments())); } @Test diff --git a/sdks/java/io/rabbitmq/src/main/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqMessage.java b/sdks/java/io/rabbitmq/src/main/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqMessage.java index 9936d9a9c6f8..7f561b72b523 100644 --- a/sdks/java/io/rabbitmq/src/main/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqMessage.java +++ b/sdks/java/io/rabbitmq/src/main/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqMessage.java @@ -327,7 +327,9 @@ public boolean equals(@Nullable Object obj) { && Objects.equals(replyTo, other.replyTo) && Objects.equals(expiration, other.expiration) && Objects.equals(messageId, other.messageId) - && Objects.equals(timestamp, other.timestamp) + && Objects.equals( + timestamp == null ? null : timestamp.toInstant(), + other.timestamp == null ? null : other.timestamp.toInstant()) && Objects.equals(type, other.type) && Objects.equals(userId, other.userId) && Objects.equals(appId, other.appId) diff --git a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/SolaceCheckpointMark.java b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/SolaceCheckpointMark.java index a913fd6133ea..83aed07374b3 100644 --- a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/SolaceCheckpointMark.java +++ b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/SolaceCheckpointMark.java @@ -25,6 +25,7 @@ import org.apache.beam.sdk.extensions.avro.coders.AvroCoder; import org.apache.beam.sdk.io.UnboundedSource; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables; import org.checkerframework.checker.nullness.qual.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -80,7 +81,10 @@ public boolean equals(@Nullable Object o) { return false; } SolaceCheckpointMark that = (SolaceCheckpointMark) o; - return Objects.equals(safeToAck, that.safeToAck); + return safeToAck == that.safeToAck + || (safeToAck != null + && that.safeToAck != null + && Iterables.elementsEqual(safeToAck, that.safeToAck)); } @Override