[pull] master from apache:master

CHANGES.md

@@ -76,6 +76,9 @@
 * (Java/Python) `Watch` can bound its deduplication state by event time, retiring an output key once the greatest emitted timestamp has moved more than the allowed lateness past it. Java adds `Watch.growthOf(...).withTimestampCursor()`. Python adds `allowed_lateness` for the existing `timestamp_cursor` option ([#18459](https://github.com/apache/beam/issues/18459)).
 * (Java) Spark Structured Streaming runner: stateful ParDo with state, timers, `@RequiresTimeSortedInput` and tagged outputs is now supported in batch mode ([#39779](https://github.com/apache/beam/issues/39779)).
 * (Python) Added support for Vertex AI Model Monitoring V2 in RunInference ([#39738](https://github.com/apache/beam/issues/39738)).
+* [Flink Runner] Added opt-in static round-robin split assignment for small bounded sources via the new `sourceStaticSplitThresholdMb` pipeline option. The default of 0 keeps the existing lazy pull-based assignment ([#39873](https://github.com/apache/beam/issues/39873)).
+* Added automatic caching of bounded, single-pane side-input views for classic Java Flink DataStream execution ([#39866](https://github.com/apache/beam/issues/39866)).
+* (Python) Added `Sample.Any`, the Python equivalent of Java's `Sample.any`, which returns up to n arbitrary elements from a PCollection ([#18552](https://github.com/apache/beam/issues/18552)).
 
 ## Breaking Changes
 
@@ -89,7 +92,6 @@
 
 ## Bugfixes
 
-* (Java) Restored binary compatibility for `CoderTranslatorRegistrar` implementations compiled against Beam 2.76 and earlier ([#38714](https://github.com/apache/beam/issues/38714)).
 * (Java) Fixed the Spark runner firing processing-time timers in reverse timestamp order ([#39824](https://github.com/apache/beam/issues/39824)).
 * (Python) Fixed incorrect profiler options handling on portable runners ([#39613](https://github.com/apache/beam/issues/39613)).
 * (Java) KafkaIO dynamic reads no longer require the obsolete `beam_fn_api` experiment ([#29998](https://github.com/apache/beam/issues/29998)).
@@ -124,8 +126,6 @@
 
 ## New Features / Improvements
 
-* [Flink Runner] Added opt-in static round-robin split assignment for small bounded sources via the new `sourceStaticSplitThresholdMb` pipeline option. The default of 0 keeps the existing lazy pull-based assignment ([#39873](https://github.com/apache/beam/issues/39873)).
-* Added automatic caching of bounded, single-pane side-input views for classic Java Flink DataStream execution ([#39866](https://github.com/apache/beam/issues/39866)).
 * Added `GroupIntoBatches` transform and the standard
   `beam:coder:sharded_key:v1` coder to the Go SDK, along with
   `beam.Coder.IsDeterministic`, `beam.PCollection.WindowingStrategy`,
@@ -146,7 +146,6 @@
 * (Python) Added `Watch`, a transform that polls a growing set of outputs for each input element, deduplicates outputs across poll rounds, and stops per a user-supplied termination condition
   ([#21521](https://github.com/apache/beam/issues/21521)).
 * (Python) Added support to analyze core dumps created after python worker segmentation faults with `pystack` (or `gdb` if installed) using the `--profiler_agent=coredump` pipeline option. ([#39484](https://github.com/apache/beam/issues/39484)).
-* (Python) Added `Sample.Any`, the Python equivalent of Java's `Sample.any`, which returns up to n arbitrary elements from a PCollection ([#18552](https://github.com/apache/beam/issues/18552)).
 
 ## Breaking Changes
 
@@ -161,9 +160,6 @@
   Use pipeline option `--updateCompatibilityVersion=2.75.0` (or any older version) to keep the old behavior ([#39344](https://github.com/apache/beam/issues/39344)).
 * `DoFn.process` returning a `str`, `bytes`, or `dict` (instead of an iterable wrapping one) now raises a `TypeError` rather than silently iterating per-character/byte/key (Python) ([#18712](https://github.com/apache/beam/issues/18712)).
 * (Java) Added `DRAINING` and `DRAINED` states to `PipelineResult`, including runner state mappings and Dataflow update handling ([#39020](https://github.com/apache/beam/issues/39020)).
-* (Python) Typehints of dataclass fields are honored during type inferences. To restore the behavior of fallback-to-any,
-  use pipeline option `--exclude_infer_dataclass_field_type` ([#38797](https://github.com/apache/beam/issues/38797)).
-  However fixing forward is recommended.
 * (Java) IcebergIO and projects that use it must now be built with Java 17 or later as a result of Iceberg 1.11.0 upgrade ([#38925](https://github.com/apache/beam/issues/38925)).
 
 ## Bugfixes

runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/MetricsAccumulator.java

@@ -17,6 +17,7 @@
  */
 package org.apache.beam.runners.spark.metrics;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.io.IOException;
 import org.apache.beam.runners.core.metrics.MetricsContainerStepMap;
 import org.apache.beam.runners.spark.SparkPipelineOptions;
@@ -82,6 +83,10 @@ public static void init(SparkPipelineOptions opts, JavaSparkContext jsc) {
     }
   }
 
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "Spark merges only the accumulator instance the driver registered. A copy would collect metrics that nothing reports.")
   public static MetricsContainerStepMapAccumulator getInstance() {
     if (instance == null) {
       throw new IllegalStateException("Metrics accumulator has not been instantiated");

runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/MetricsAccumulator.java

@@ -17,6 +17,7 @@
  */
 package org.apache.beam.runners.spark.structuredstreaming.metrics;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import org.apache.beam.runners.core.metrics.MetricsContainerStepMap;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 import org.apache.spark.sql.SparkSession;
@@ -85,6 +86,10 @@ public MetricsContainerStepMap value() {
    * Get the {@link MetricsAccumulator} on this driver. If there's no such accumulator yet, it will
    * be created and registered using the provided {@link SparkSession}.
    */
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "Spark merges only the accumulator instance the driver registered. A copy would collect metrics that nothing reports.")
   public static MetricsAccumulator getInstance(SparkSession session) {
     MetricsAccumulator current = instance;
     if (current != null) {

sdks/java/build-tools/src/main/resources/beam/spotbugs-filter.xml

@@ -57,7 +57,6 @@
 
   <!-- TODO(https://github.com/apache/beam/issues/35312) resolve findings-->
   <Bug pattern="CT_CONSTRUCTOR_THROW"/>
-  <Bug pattern="MS_EXPOSE_REP"/>
 
   <!--
     Many test classes are captured by lambdas and marked `implements Serializable`. They are not

sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistry.java

@@ -52,6 +52,7 @@
 import org.apache.beam.sdk.values.TimestampedValue;
 import org.apache.beam.sdk.values.TypeDescriptor;
 import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.sdk.values.ValueKind;
 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.HashMultimap;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
@@ -137,6 +138,8 @@ private CommonTypes() {
           TimestampedValue.class,
           CoderProviders.fromStaticMethods(
               TimestampedValue.class, TimestampedValue.TimestampedValueCoder.class));
+      builder.put(
+          ValueKind.class, CoderProviders.fromStaticMethods(ValueKind.class, ValueKindCoder.class));
       builder.put(Void.class, CoderProviders.fromStaticMethods(Void.class, VoidCoder.class));
       builder.put(
           byte[].class, CoderProviders.fromStaticMethods(byte[].class, ByteArrayCoder.class));

sdks/java/core/src/main/java/org/apache/beam/sdk/coders/ValueKindCoder.java

@@ -0,0 +1,88 @@
+/*
+ * 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.beam.sdk.coders;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.Elements;
+import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.ValueKind;
+import org.apache.beam.sdk.values.ValueKindUtil;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * A {@link Coder} for {@link ValueKind}, encoded in 1 byte as the matching {@link
+ * Elements.ValueKind.Enum} number, so the wire format stays stable if the enum is reordered and
+ * matches the portability representation.
+ */
+public class ValueKindCoder extends AtomicCoder<ValueKind> {
+
+  public static ValueKindCoder of() {
+    return INSTANCE;
+  }
+
+  private static final ValueKindCoder INSTANCE = new ValueKindCoder();
+  private static final TypeDescriptor<ValueKind> TYPE_DESCRIPTOR =
+      TypeDescriptor.of(ValueKind.class);
+
+  private ValueKindCoder() {}
+
+  @Override
+  public void encode(ValueKind value, OutputStream outStream) throws IOException, CoderException {
+    if (value == null) {
+      throw new CoderException("cannot encode a null ValueKind");
+    }
+    outStream.write(ValueKindUtil.toProto(value).getNumber());
+  }
+
+  @Override
+  public ValueKind decode(InputStream inStream) throws IOException, CoderException {
+    int number = inStream.read();
+    if (number == -1) {
+      throw new CoderException(new EOFException("EOF encountered decoding a ValueKind"));
+    }
+    Elements.ValueKind.@Nullable Enum proto = Elements.ValueKind.Enum.forNumber(number);
+    if (proto == null) {
+      throw new CoderException("Unknown ValueKind number: " + number);
+    }
+
+    return ValueKindUtil.fromProto(proto);
+  }
+
+  @Override
+  public boolean consistentWithEquals() {
+    return true;
+  }
+
+  @Override
+  public boolean isRegisterByteSizeObserverCheap(ValueKind value) {
+    return true;
+  }
+
+  @Override
+  protected long getEncodedElementByteSize(ValueKind value) {
+    return 1;
+  }
+
+  @Override
+  public TypeDescriptor<ValueKind> getEncodedTypeDescriptor() {
+    return TYPE_DESCRIPTOR;
+  }
+}

sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Lineage.java

@@ -19,6 +19,7 @@
 
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -122,6 +123,10 @@ private static Lineage createLineage(PipelineOptions options, LineageDirection d
   }
 
   /** {@link Lineage} representing sources and optionally side inputs. */
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "Every reporter writes into the same metric cell, so all callers need the one shared instance. A copy would drop the lineage it records.")
   public static Lineage getSources() {
     Lineage localSources = sources;
     if (localSources == null) {
@@ -131,6 +136,10 @@ public static Lineage getSources() {
   }
 
   /** {@link Lineage} representing sinks. */
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "Every reporter writes into the same metric cell, so all callers need the one shared instance. A copy would drop the lineage it records.")
   public static Lineage getSinks() {
     Lineage localSinks = sinks;
     if (localSinks == null) {

sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/SourceMetrics.java

@@ -17,6 +17,7 @@
  */
 package org.apache.beam.sdk.metrics;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Joiner;
 
 /** Standard {@link org.apache.beam.sdk.io.Source} Metrics. */
@@ -69,6 +70,10 @@ public static Counter bytesReadBySplit(String splitId) {
   }
 
   /** Gauge for source backlog in bytes. */
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "A Gauge is a handle onto one metric cell, not a value. A copy would send the reading nowhere.")
   public static Gauge backlogBytes() {
     return BACKLOG_BYTES_GAUGE;
   }
@@ -84,6 +89,10 @@ public static Gauge backlogBytesOfSplit(String splitId) {
   }
 
   /** Gauge for source backlog in elements. */
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "A Gauge is a handle onto one metric cell, not a value. A copy would send the reading nowhere.")
   public static Gauge backlogElements() {
     return BACKLOG_ELEMENTS_GAUGE;
   }

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

@@ -22,6 +22,7 @@
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
 
 import com.google.auto.value.AutoValue;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.util.Set;
 import org.apache.beam.model.pipeline.v1.RunnerApi.Coder;
 import org.apache.beam.model.pipeline.v1.RunnerApi.FunctionSpec;
@@ -97,6 +98,12 @@ private ModelCoders() {}
           SHARDED_KEY_CODER_URN,
           NULLABLE_CODER_URN);
 
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "Returns a Guava ImmutableSet."
+              + " Spotbugs matches its known-immutable list by fully qualified name, so it cannot recognise collections relocated into org.apache.beam.vendor.guava."
+              + " See https://github.com/spotbugs/spotbugs/issues/1601.")
   public static Set<String> urns() {
     return MODEL_CODER_URNS;
   }

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

@@ -22,6 +22,7 @@
 import static org.apache.beam.sdk.util.construction.BeamUrns.getUrn;
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Collections;
@@ -417,6 +418,12 @@ public RunnerApi.PTransform translate(
       knownPayloadTranslators;
 
   @Internal
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "Returns a Guava ImmutableMap."
+              + " Spotbugs matches its known-immutable list by fully qualified name, so it cannot recognise collections relocated into org.apache.beam.vendor.guava."
+              + " See https://github.com/spotbugs/spotbugs/issues/1601.")
   public static Map<Class<? extends PTransform>, TransformPayloadTranslator>
       getKnownPayloadTranslators() {
     if (knownPayloadTranslators == null) {

sdks/java/core/src/test/java/org/apache/beam/sdk/coders/ValueKindCoderTest.java

@@ -0,0 +1,81 @@
+/*
+ * 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.beam.sdk.coders;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Arrays;
+import java.util.List;
+import org.apache.beam.sdk.testing.CoderProperties;
+import org.apache.beam.sdk.util.CoderUtils;
+import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.ValueKind;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link ValueKindCoder}. */
+@RunWith(JUnit4.class)
+public class ValueKindCoderTest {
+
+  private static final Coder<ValueKind> TEST_CODER = ValueKindCoder.of();
+
+  private static final List<ValueKind> TEST_VALUES =
+      Arrays.asList(
+          ValueKind.INSERT, ValueKind.UPDATE_BEFORE, ValueKind.UPDATE_AFTER, ValueKind.DELETE);
+
+  /** One byte per value, holding the proto enum number. */
+  private static final List<String> TEST_ENCODINGS = Arrays.asList("AQ", "Ag", "Aw", "BA");
+
+  @Rule public ExpectedException thrown = ExpectedException.none();
+
+  @Test
+  public void testDecodeEncodeEqual() throws Exception {
+    for (ValueKind value : TEST_VALUES) {
+      CoderProperties.coderDecodeEncodeEqual(TEST_CODER, value);
+    }
+  }
+
+  @Test
+  public void testWireFormatEncode() throws Exception {
+    CoderProperties.coderEncodesBase64(TEST_CODER, TEST_VALUES, TEST_ENCODINGS);
+  }
+
+  /** VALUE_KIND_UNSPECIFIED (0) means INSERT, for backwards compatibility. */
+  @Test
+  public void testDecodeUnspecified() throws Exception {
+    assertEquals(ValueKind.INSERT, CoderUtils.decodeFromBase64(TEST_CODER, "AA"));
+  }
+
+  @Test
+  public void testDecodeUnknownNumberThrows() throws Exception {
+    thrown.expect(CoderException.class);
+    thrown.expectMessage("Unknown ValueKind number: 42");
+
+    CoderUtils.decodeFromBase64(TEST_CODER, "Kg");
+  }
+
+  @Test
+  public void testCoderRegistryResolvesValueKind() throws Exception {
+    assertEquals(
+        ValueKindCoder.of(),
+        CoderRegistry.createDefault().getCoder(TypeDescriptor.of(ValueKind.class)));
+  }
+}

sdks/java/io/iceberg/build.gradle

@@ -49,6 +49,7 @@ dependencies {
     implementation library.java.avro
     implementation library.java.slf4j_api
     implementation library.java.joda_time
+    implementation library.java.guava
     implementation "org.apache.parquet:parquet-column:$parquet_version"
     implementation "org.apache.parquet:parquet-hadoop:$parquet_version"
     implementation "org.apache.parquet:parquet-common:$parquet_version"