Merge upstream/master into my branch

Author: haperski

Diff for: .github/trigger_files/IO_Iceberg_Integration_Tests.json

@@ -1,4 +1,4 @@
 {
     "comment": "Modify this file in a trivial way to cause this test suite to run.",
-    "modification": 3
+    "modification": 4
 }

Diff for: CHANGES.md

@@ -54,18 +54,53 @@
 * ([#X](https://github.com/apache/beam/issues/X)).
 -->
 
-# [2.64.0] - Unreleased
+# [2.65.0] - Unreleased
 
 ## Highlights
 
-* Managed API for [Java](https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/managed/Managed.html) and [Python](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.managed.html#module-apache_beam.transforms.managed) supports [key I/O connectors](https://beam.apache.org/documentation/io/connectors/) Iceberg, Kafka, and BigQuery.
 * New highly anticipated feature X added to Python SDK ([#X](https://github.com/apache/beam/issues/X)).
 * New highly anticipated feature Y added to Java SDK ([#Y](https://github.com/apache/beam/issues/Y)).
 
 ## I/Os
 
 * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
+
+## New Features / Improvements
+
+* X feature added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
+
+## Breaking Changes
+
+* X behavior was changed ([#X](https://github.com/apache/beam/issues/X)).
+
+## Deprecations
+
+* X behavior is deprecated and will be removed in X versions ([#X](https://github.com/apache/beam/issues/X)).
+
+## Bugfixes
+
+* Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
+
+## Security Fixes
+* Fixed (CVE-YYYY-NNNN)[https://www.cve.org/CVERecord?id=CVE-YYYY-NNNN] (Java/Python/Go) ([#X](https://github.com/apache/beam/issues/X)).
+
+## Known Issues
+
+[comment]: # ( When updating known issues after release, make sure also update website blog in website/www/site/content/blog.)
+* ([#X](https://github.com/apache/beam/issues/X)).
+
+# [2.64.0] - Ongoing Release
+
+## Highlights
+
+* Managed API for [Java](https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/managed/Managed.html) and [Python](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.managed.html#module-apache_beam.transforms.managed) supports [key I/O connectors](https://beam.apache.org/documentation/io/connectors/) Iceberg, Kafka, and BigQuery.
+* New highly anticipated feature X added to Python SDK ([#X](https://github.com/apache/beam/issues/X)).
+* New highly anticipated feature Y added to Java SDK ([#Y](https://github.com/apache/beam/issues/Y)).
+
+## I/Os
+
 * [Java] Use API compatible with both com.google.cloud.bigdataoss:util 2.x and 3.x in BatchLoads ([#34105](https://github.com/apache/beam/pull/34105))
+* [IcebergIO] Added new CDC source for batch and streaming, available as `Managed.ICEBERG_CDC` ([#33504](https://github.com/apache/beam/pull/33504))
 * [IcebergIO] Address edge case where bundle retry following a successful data commit results in data duplication ([#34264](https://github.com/apache/beam/pull/34264))
 
 ## New Features / Improvements
@@ -83,11 +118,6 @@
 * [Python] Reshuffle now correctly respects user-specified type hints, fixing a previous bug where it might use FastPrimitivesCoder wrongly. This change could break pipelines with incorrect type hints in Reshuffle. If you have issues after upgrading, temporarily set update_compatibility_version to a previous Beam version to use the old behavior. The recommended solution is to fix the type hints in your code. ([#33932](https://github.com/apache/beam/pull/33932))
 * [Java] SparkReceiver 2 has been moved to SparkReceiver 3 that supports Spark 3.x. ([#33574](https://github.com/apache/beam/pull/33574))
 * [Python] Correct parsing of `collections.abc.Sequence` type hints was added, which can lead to pipelines failing type hint checks that were previously passing erroneously. These issues will be most commonly seen trying to consume a PCollection with a `Sequence` type hint after a GroupByKey or a CoGroupByKey. ([#33999](https://github.com/apache/beam/pull/33999).
-* X behavior was changed ([#X](https://github.com/apache/beam/issues/X)).
-
-## Deprecations
-
-* X behavior is deprecated and will be removed in X versions ([#X](https://github.com/apache/beam/issues/X)).
 
 ## Bugfixes
 
@@ -98,15 +128,6 @@
 * Fixed checkpoint recovery and streaming behavior in Spark Classic and Portable runner's Flatten transform by replacing queueStream with SingleEmitInputDStream ([#34080](https://github.com/apache/beam/pull/34080), [#18144](https://github.com/apache/beam/issues/18144), [#20426](https://github.com/apache/beam/issues/20426))
 * (Java) Fixed Read caching of UnboundedReader objects to effectively cache across multiple DoFns and avoid checkpointing unstarted reader. [#34146](https://github.com/apache/beam/pull/34146) [#33901](https://github.com/apache/beam/pull/33901)
 
-
-## Security Fixes
-* Fixed (CVE-YYYY-NNNN)[https://www.cve.org/CVERecord?id=CVE-YYYY-NNNN] (Java/Python/Go) ([#X](https://github.com/apache/beam/issues/X)).
-
-## Known Issues
-
-[comment]: # ( When updating known issues after release, make sure also update website blog in website/www/site/content/blog.)
-* ([#X](https://github.com/apache/beam/issues/X)).
-
 # [2.63.0] - 2025-02-18
 
 ## I/Os

Diff for: contributor-docs/discussion-docs/2025.md

@@ -15,5 +15,9 @@ limitations under the License.
 # List Of Documents Submitted To [email protected] In 2025
 | No. | Author | Subject | Date (UTC) |
 |---|---|---|---|
-| 1 | Danny McCormick | [Beam Python Dependency Extras](https://docs.google.com/document/d/1c84Gc-cZRCfrU8f7kWGsNR2o8oSRjCM-dGHO9KvPWPw) | 2025-01-27 17:50:00 |
-| 2 | Danny McCormick | [How vLLM Model Handler Works (Plus a Summary of Model Memory Management in Beam ML)](https://docs.google.com/document/d/1UB4umrtnp1Eg45fiUB3iLS7kPK3BE6pcf0YRDkA289Q) | 2025-01-31 17:50:00 |
+| 1 | Kenneth Knowles | [Apache Beam Release Acceptance Criteria  - Google Sheets](https://docs.google.com/spreadsheets/d/1qk-N5vjXvbcEk68GjbkSZTR8AGqyNUM-oLFo_ZXBpJw) | 2025-01-13 10:54:22 |
+| 2 | Danny McCormick | [Apache Beam Vendored Dependencies Release Guide](https://s.apache.org/beam-release-vendored-artifacts) | 2025-01-13 15:00:51 |
+| 3 | Danny McCormick | [Beam Python & ML Dependency Extras](https://docs.google.com/document/d/1c84Gc-cZRCfrU8f7kWGsNR2o8oSRjCM-dGHO9KvPWPw) | 2025-01-27 15:33:36 |
+| 4 | Danny McCormick | [How vLLM Model Handler Works (Plus a Summary of Model Memory Management in Beam ML)](https://docs.google.com/document/d/1UB4umrtnp1Eg45fiUB3iLS7kPK3BE6pcf0YRDkA289Q) | 2025-01-31 11:56:59 |
+| 5 | Shunping Huang | [Improve Logging Dependencies in Beam Java SDK](https://docs.google.com/document/d/1IkbiM4m8D-aB3NYI1aErFZHt6M7BQ-8eCULh284Davs) | 2025-02-04 15:13:14 |
+| 6 | Ahmed Abualsaud | [Iceberg Incremental Source design](https://s.apache.org/beam-iceberg-incremental-source) | 2025-03-03 14:52:42 |

Diff for: model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto

@@ -74,6 +74,8 @@ message ManagedTransforms {
       "beam:schematransform:org.apache.beam:bigquery_storage_read:v1"];
     BIGQUERY_WRITE = 5 [(org.apache.beam.model.pipeline.v1.beam_urn) =
       "beam:schematransform:org.apache.beam:bigquery_write:v1"];
+    ICEBERG_CDC_READ = 6 [(org.apache.beam.model.pipeline.v1.beam_urn) =
+      "beam:schematransform:org.apache.beam:iceberg_cdc_read:v1"];
   }
 }
 

Diff for: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerImpl.java

@@ -321,9 +321,7 @@ public MetricUpdates getUpdates() {
 
     // Add any metricKey labels to the monitoringInfoLabels.
     if (!metricName.getLabels().isEmpty()) {
-      for (Map.Entry<String, String> entry : metricName.getLabels().entrySet()) {
-        builder.setLabel(entry.getKey(), entry.getValue());
-      }
+      builder.setLabels(metricName.getLabels());
     }
     return builder;
   }

Diff for: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MonitoringInfoEncodings.java

@@ -182,12 +182,12 @@ public static double decodeDoubleCounter(ByteString payload) {
     }
   }
 
-  /** Encodes to {@link MonitoringInfoConstants.TypeUrns#PER_WORKER_HISTOGRAM}. */
+  /** Encodes to {@link MonitoringInfoConstants.TypeUrns#HISTOGRAM}. */
   public static ByteString encodeInt64Histogram(HistogramData inputHistogram) {
     return inputHistogram.toProto().toByteString();
   }
 
-  /** Decodes to {@link MonitoringInfoConstants.TypeUrns#PER_WORKER_HISTOGRAM}. */
+  /** Decodes to {@link MonitoringInfoConstants.TypeUrns#HISTOGRAM}. */
   public static HistogramData decodeInt64Histogram(ByteString payload) {
     try {
       return new HistogramData(HistogramValue.parseFrom(payload));

Diff for: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaTable.java

@@ -161,6 +161,10 @@ public List<String> getTopics() {
     return topics;
   }
 
+  public Map<String, Object> getConfigUpdates() {
+    return configUpdates;
+  }
+
   @Override
   public BeamTableStatistics getTableStatistics(PipelineOptions options) {
     if (rowCountStatistics == null) {

Diff for: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/KafkaTableProvider.java

@@ -21,10 +21,14 @@
 import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull;
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.auto.service.AutoService;
+import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
+import java.util.Map.Entry;
 import java.util.Optional;
 import org.apache.beam.sdk.extensions.sql.TableUtils;
 import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTable;
@@ -112,6 +116,8 @@ public BeamSqlTable buildBeamSqlTable(Table table) {
         properties.has("format")
             ? Optional.of(properties.get("format").asText())
             : Optional.empty();
+
+    BeamKafkaTable kafkaTable = null;
     if (Schemas.isNestedSchema(schema)) {
       Optional<PayloadSerializer> serializer =
           payloadFormat.map(
@@ -120,7 +126,7 @@ public BeamSqlTable buildBeamSqlTable(Table table) {
                       format,
                       checkArgumentNotNull(schema.getField(PAYLOAD_FIELD).getType().getRowSchema()),
                       TableUtils.convertNode2Map(properties)));
-      return new NestedPayloadKafkaTable(schema, bootstrapServers, topics, serializer);
+      kafkaTable = new NestedPayloadKafkaTable(schema, bootstrapServers, topics, serializer);
     } else {
       /*
        * CSV is handled separately because multiple rows can be produced from a single message, which
@@ -129,13 +135,30 @@ public BeamSqlTable buildBeamSqlTable(Table table) {
        * rows.
        */
       if (payloadFormat.orElse("csv").equals("csv")) {
-        return new BeamKafkaCSVTable(schema, bootstrapServers, topics);
+        kafkaTable = new BeamKafkaCSVTable(schema, bootstrapServers, topics);
+      } else {
+        PayloadSerializer serializer =
+            PayloadSerializers.getSerializer(
+                payloadFormat.get(), schema, TableUtils.convertNode2Map(properties));
+        kafkaTable = new PayloadSerializerKafkaTable(schema, bootstrapServers, topics, serializer);
+      }
+    }
+
+    // Get Consumer Properties from Table properties
+    HashMap<String, Object> configUpdates = new HashMap<String, Object>();
+    Iterator<Entry<String, JsonNode>> tableProperties = properties.fields();
+    while (tableProperties.hasNext()) {
+      Entry<String, JsonNode> field = tableProperties.next();
+      if (field.getKey().startsWith("properties.")) {
+        configUpdates.put(field.getKey().replace("properties.", ""), field.getValue().textValue());
       }
-      PayloadSerializer serializer =
-          PayloadSerializers.getSerializer(
-              payloadFormat.get(), schema, TableUtils.convertNode2Map(properties));
-      return new PayloadSerializerKafkaTable(schema, bootstrapServers, topics, serializer);
     }
+
+    if (!configUpdates.isEmpty()) {
+      kafkaTable.updateConsumerProperties(configUpdates);
+    }
+
+    return kafkaTable;
   }
 
   @Override

Diff for: sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/KafkaTableProviderTest.java

@@ -24,13 +24,15 @@
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import java.util.List;
+import java.util.Map;
 import org.apache.beam.sdk.extensions.protobuf.PayloadMessages;
 import org.apache.beam.sdk.extensions.sql.TableUtils;
 import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTable;
 import org.apache.beam.sdk.extensions.sql.meta.Table;
 import org.apache.beam.sdk.io.thrift.payloads.SimpleThriftMessage;
 import org.apache.beam.sdk.schemas.Schema;
 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.ImmutableMap;
 import org.apache.thrift.TBase;
 import org.apache.thrift.protocol.TCompactProtocol;
 import org.apache.thrift.protocol.TProtocolFactory;
@@ -84,6 +86,32 @@ public void testBuildWithExtraTopics() {
     assertEquals(ImmutableList.of(LOCATION_TOPIC, "topic2", "topic3"), kafkaTable.getTopics());
   }
 
+  @Test
+  public void testBuildWithExtraProperties() {
+    Table table =
+        mockTableWithExtraProperties(
+            "hello",
+            ImmutableMap.of(
+                "properties.ssl.truststore.location",
+                "/path/to/kafka.client.truststore.jks",
+                "properties.security.protocol",
+                "SASL_SSL"));
+    BeamSqlTable sqlTable = provider.buildBeamSqlTable(table);
+
+    assertNotNull(sqlTable);
+    assertTrue(sqlTable instanceof BeamKafkaCSVTable);
+
+    BeamKafkaCSVTable kafkaTable = (BeamKafkaCSVTable) sqlTable;
+    assertEquals(LOCATION_BROKER, kafkaTable.getBootstrapServers());
+    assertEquals(
+        ImmutableMap.of(
+            "ssl.truststore.location",
+            "/path/to/kafka.client.truststore.jks",
+            "security.protocol",
+            "SASL_SSL"),
+        kafkaTable.getConfigUpdates());
+  }
+
   @Test
   public void testBuildBeamSqlAvroTable() {
     Table table = mockTable("hello", "avro");
@@ -157,50 +185,56 @@ public void testGetTableType() {
   }
 
   private static Table mockTable(String name) {
-    return mockTable(name, false, null, null, null, null, null, null);
+    return mockTable(name, false, null, null, null, null, null, null, null);
   }
 
   private static Table mockTableWithExtraServers(String name, List<String> extraBootstrapServers) {
-    return mockTable(name, false, extraBootstrapServers, null, null, null, null, null);
+    return mockTable(name, false, extraBootstrapServers, null, null, null, null, null, null);
   }
 
   private static Table mockTableWithExtraTopics(String name, List<String> extraTopics) {
-    return mockTable(name, false, null, extraTopics, null, null, null, null);
+    return mockTable(name, false, null, extraTopics, null, null, null, null, null);
+  }
+
+  private static Table mockTableWithExtraProperties(
+      String name, Map<String, String> extraProperties) {
+    return mockTable(name, false, null, null, extraProperties, null, null, null, null);
   }
 
   private static Table mockTable(String name, String payloadFormat) {
-    return mockTable(name, false, null, null, payloadFormat, null, null, null);
+    return mockTable(name, false, null, null, null, payloadFormat, null, null, null);
   }
 
   private static Table mockProtoTable(String name, Class<?> protoClass) {
-    return mockTable(name, false, null, null, "proto", protoClass, null, null);
+    return mockTable(name, false, null, null, null, "proto", protoClass, null, null);
   }
 
   private static Table mockThriftTable(
       String name,
       Class<? extends TBase<?, ?>> thriftClass,
       Class<? extends TProtocolFactory> thriftProtocolFactoryClass) {
     return mockTable(
-        name, false, null, null, "thrift", null, thriftClass, thriftProtocolFactoryClass);
+        name, false, null, null, null, "thrift", null, thriftClass, thriftProtocolFactoryClass);
   }
 
   private static Table mockNestedBytesTable(String name) {
-    return mockTable(name, true, null, null, null, null, null, null);
+    return mockTable(name, true, null, null, null, null, null, null, null);
   }
 
   private static Table mockNestedThriftTable(
       String name,
       Class<? extends TBase<?, ?>> thriftClass,
       Class<? extends TProtocolFactory> thriftProtocolFactoryClass) {
     return mockTable(
-        name, true, null, null, "thrift", null, thriftClass, thriftProtocolFactoryClass);
+        name, true, null, null, null, "thrift", null, thriftClass, thriftProtocolFactoryClass);
   }
 
   private static Table mockTable(
       String name,
       boolean isNested,
       @Nullable List<String> extraBootstrapServers,
       @Nullable List<String> extraTopics,
+      @Nullable Map<String, String> extraProperties,
       @Nullable String payloadFormat,
       @Nullable Class<?> protoClass,
       @Nullable Class<? extends TBase<?, ?>> thriftClass,
@@ -222,6 +256,12 @@ private static Table mockTable(
       properties.put("topics", topics);
     }
 
+    if (extraProperties != null) {
+      for (Map.Entry<String, String> property : extraProperties.entrySet()) {
+        properties.put(property.getKey(), property.getValue());
+      }
+    }
+
     if (payloadFormat != null) {
       properties.put("format", payloadFormat);
     }

Diff for: sdks/java/io/iceberg/build.gradle

@@ -50,6 +50,7 @@ dependencies {
     implementation library.java.slf4j_api
     implementation library.java.joda_time
     implementation "org.apache.parquet:parquet-column:$parquet_version"
+    implementation "org.apache.parquet:parquet-hadoop:$parquet_version"
     implementation "org.apache.orc:orc-core:$orc_version"
     implementation "org.apache.iceberg:iceberg-core:$iceberg_version"
     implementation "org.apache.iceberg:iceberg-api:$iceberg_version"
@@ -88,7 +89,7 @@ dependencies {
     testImplementation library.java.google_api_services_bigquery
 
     testRuntimeOnly library.java.slf4j_jdk14
-    testRuntimeOnly project(path: ":runners:direct-java", configuration: "shadow")
+    testImplementation project(path: ":runners:direct-java", configuration: "shadow")
     testRuntimeOnly project(path: ":runners:google-cloud-dataflow-java")
     hadoopVersions.each {kv ->
         "hadoopVersion$kv.key" "org.apache.hadoop:hadoop-client:$kv.value"

Diff for: sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java

@@ -82,7 +82,7 @@ public String apply(FileWriteResult input) {
         .apply(
             "Append metadata updates to tables",
             ParDo.of(new AppendFilesToTablesDoFn(catalogConfig, manifestFilePrefix)))
-        .setCoder(KvCoder.of(StringUtf8Coder.of(), SnapshotInfo.CODER));
+        .setCoder(KvCoder.of(StringUtf8Coder.of(), SnapshotInfo.getCoder()));
   }
 
   private static class AppendFilesToTablesDoFn