Cherry pick #35981

.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
 }

.github/trigger_files/IO_Iceberg_Integration_Tests_Dataflow.json

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

.github/trigger_files/IO_Iceberg_Managed_Integration_Tests_Dataflow.json

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

.github/trigger_files/beam_PostCommit_SQL.json

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

CHANGES.md

@@ -70,6 +70,7 @@
 ## I/Os
 
 * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
+* Upgraded Iceberg dependency to 1.9.2 ([#35981](https://github.com/apache/beam/pull/35981))
 
 ## New Features / Improvements
 
@@ -98,6 +99,7 @@
   improves support for BigQuery and other SQL dialects. Note: Minor behavior changes are observed such as output
   significant digits related to casting.
 * (Python) Prism runner now enabled by default for most Python pipelines using the direct runner ([#34612](https://github.com/apache/beam/pull/34612)). This may break some tests, see https://github.com/apache/beam/pull/34612 for details on how to handle issues.
+* Dropped Java 8 support for [IO expansion-service](https://central.sonatype.com/artifact/org.apache.beam/beam-sdks-java-io-expansion-service). Cross-language pipelines using this expansion service will need a Java11+ runtime ([#35981](https://github.com/apache/beam/pull/35981).
 
 ## Deprecations
 

build.gradle.kts

@@ -253,6 +253,7 @@ tasks.register("javaPreCommit") {
   dependsOn(":examples:java:sql:preCommit")
   dependsOn(":examples:java:twitter:build")
   dependsOn(":examples:java:twitter:preCommit")
+  dependsOn(":examples:java:iceberg:build")
   dependsOn(":examples:multi-language:build")
   dependsOn(":model:fn-execution:build")
   dependsOn(":model:job-management:build")
@@ -380,6 +381,7 @@ tasks.register("sqlPreCommit") {
   dependsOn(":sdks:java:extensions:sql:datacatalog:build")
   dependsOn(":sdks:java:extensions:sql:expansion-service:build")
   dependsOn(":sdks:java:extensions:sql:hcatalog:build")
+  dependsOn(":sdks:java:extensions:sql:iceberg:build")
   dependsOn(":sdks:java:extensions:sql:jdbc:build")
   dependsOn(":sdks:java:extensions:sql:jdbc:preCommit")
   dependsOn(":sdks:java:extensions:sql:perf-tests:build")
@@ -426,6 +428,7 @@ tasks.register("sqlPostCommit") {
   dependsOn(":sdks:java:extensions:sql:postCommit")
   dependsOn(":sdks:java:extensions:sql:jdbc:postCommit")
   dependsOn(":sdks:java:extensions:sql:datacatalog:postCommit")
+  dependsOn(":sdks:java:extensions:sql:iceberg:integrationTest")
   dependsOn(":sdks:java:extensions:sql:hadoopVersionsTest")
 }
 

examples/java/build.gradle

@@ -71,9 +71,6 @@ dependencies {
   implementation project(":sdks:java:extensions:python")
   implementation project(":sdks:java:io:google-cloud-platform")
   implementation project(":sdks:java:io:kafka")
-  runtimeOnly project(":sdks:java:io:iceberg")
-  runtimeOnly project(":sdks:java:io:iceberg:bqms")
-  implementation project(":sdks:java:managed")
   implementation project(":sdks:java:extensions:ml")
   implementation library.java.avro
   implementation library.java.bigdataoss_util

examples/java/iceberg/build.gradle

@@ -0,0 +1,89 @@
+/*
+ * 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.
+ */
+
+
+plugins {
+  id 'java'
+  id 'org.apache.beam.module'
+  id 'com.gradleup.shadow'
+}
+
+applyJavaNature(
+  exportJavadoc: false,
+  automaticModuleName: 'org.apache.beam.examples.iceberg',
+  // iceberg requires Java11+
+  requireJavaVersion: JavaVersion.VERSION_11
+)
+
+description = "Apache Beam :: Examples :: Java :: Iceberg"
+ext.summary = """Apache Beam Java SDK examples using IcebergIO."""
+
+/** Define the list of runners which execute a precommit test.
+ * Some runners are run from separate projects, see the preCommit task below
+ * for details.
+ */
+def preCommitRunners = ["directRunner", "flinkRunner", "sparkRunner"]
+// The following runners have configuration created but not added to preCommit
+def nonPreCommitRunners = ["dataflowRunner", "prismRunner"]
+for (String runner : preCommitRunners) {
+  configurations.create(runner + "PreCommit")
+}
+for (String runner: nonPreCommitRunners) {
+  configurations.create(runner + "PreCommit")
+}
+configurations.sparkRunnerPreCommit {
+  // Ban certain dependencies to prevent a StackOverflow within Spark
+  // because JUL -> SLF4J -> JUL, and similarly JDK14 -> SLF4J -> JDK14
+  exclude group: "org.slf4j", module: "jul-to-slf4j"
+  exclude group: "org.slf4j", module: "slf4j-jdk14"
+}
+
+dependencies {
+  implementation enforcedPlatform(library.java.google_cloud_platform_libraries_bom)
+  runtimeOnly project(":sdks:java:io:iceberg")
+  runtimeOnly project(":sdks:java:io:iceberg:bqms")
+  implementation project(path: ":sdks:java:core", configuration: "shadow")
+  implementation project(":sdks:java:extensions:google-cloud-platform-core")
+  implementation project(":sdks:java:io:google-cloud-platform")
+  implementation project(":sdks:java:managed")
+  implementation library.java.google_auth_library_oauth2_http
+  implementation library.java.joda_time
+  runtimeOnly project(path: ":runners:direct-java", configuration: "shadow")
+  implementation library.java.vendored_guava_32_1_2_jre
+  runtimeOnly library.java.hadoop_client
+  runtimeOnly library.java.bigdataoss_gcs_connector
+
+  // Add dependencies for the PreCommit configurations
+  // For each runner a project level dependency on the examples project.
+  for (String runner : preCommitRunners) {
+    delegate.add(runner + "PreCommit", project(path: ":examples:java", configuration: "testRuntimeMigration"))
+  }
+  directRunnerPreCommit project(path: ":runners:direct-java", configuration: "shadow")
+  flinkRunnerPreCommit project(":runners:flink:${project.ext.latestFlinkVersion}")
+  sparkRunnerPreCommit project(":runners:spark:3")
+  sparkRunnerPreCommit project(":sdks:java:io:hadoop-file-system")
+  dataflowRunnerPreCommit project(":runners:google-cloud-dataflow-java")
+  dataflowRunnerPreCommit project(":runners:google-cloud-dataflow-java:worker") // v2 worker
+  dataflowRunnerPreCommit project(":sdks:java:harness") // v2 worker
+  prismRunnerPreCommit project(":runners:prism:java")
+
+  // Add dependency if requested on command line for runner
+  if (project.hasProperty("runnerDependency")) {
+    runtimeOnly project(path: project.getProperty("runnerDependency"))
+  }
+}

examples/java/src/main/java/org/apache/beam/examples/cookbook/IcebergBatchWriteExample.java → examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/IcebergBatchWriteExample.java

@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.beam.examples.cookbook;
+package org.apache.beam.examples.iceberg;
 
 import java.io.IOException;
 import java.util.Map;

examples/java/src/main/java/org/apache/beam/examples/cookbook/IcebergRestCatalogCDCExample.java → examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/IcebergRestCatalogCDCExample.java

@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.beam.examples.cookbook;
+package org.apache.beam.examples.iceberg;
 
 import static org.apache.beam.sdk.managed.Managed.ICEBERG_CDC;
 

examples/java/src/main/java/org/apache/beam/examples/cookbook/IcebergRestCatalogStreamingWriteExample.java → examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/IcebergRestCatalogStreamingWriteExample.java

@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.beam.examples.cookbook;
+package org.apache.beam.examples.iceberg;
 
 import com.google.auth.oauth2.GoogleCredentials;
 import java.io.IOException;

examples/java/src/main/java/org/apache/beam/examples/cookbook/IcebergTaxiExamples.java → examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/IcebergTaxiExamples.java

@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.beam.examples.cookbook;
+package org.apache.beam.examples.iceberg;
 
 import java.util.Arrays;
 import java.util.Map;

sdks/java/extensions/sql/build.gradle

@@ -76,10 +76,6 @@ dependencies {
   fmppTask "org.freemarker:freemarker:2.3.31"
   fmppTemplates library.java.vendored_calcite_1_40_0
   implementation project(path: ":sdks:java:core", configuration: "shadow")
-  implementation project(":sdks:java:managed")
-  implementation project(":sdks:java:io:iceberg")
-  runtimeOnly project(":sdks:java:io:iceberg:bqms")
-  runtimeOnly project(":sdks:java:io:iceberg:hive")
   implementation project(":sdks:java:extensions:avro")
   implementation project(":sdks:java:extensions:join-library")
   permitUnusedDeclared project(":sdks:java:extensions:join-library") // BEAM-11761
@@ -120,9 +116,6 @@ dependencies {
   permitUnusedDeclared library.java.hadoop_client
   provided library.java.kafka_clients
 
-  implementation "org.apache.iceberg:iceberg-api:1.6.1"
-  permitUnusedDeclared "org.apache.iceberg:iceberg-api:1.6.1" // errorprone crash cannot find this transient dep
-  testImplementation "org.apache.iceberg:iceberg-core:1.6.1"
   testImplementation library.java.vendored_calcite_1_40_0
   testImplementation library.java.vendored_guava_32_1_2_jre
   testImplementation library.java.junit

sdks/java/extensions/sql/iceberg/build.gradle

@@ -0,0 +1,81 @@
+import groovy.json.JsonOutput
+
+/*
+ * 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.
+ */
+
+plugins { id 'org.apache.beam.module' }
+
+applyJavaNature(
+  automaticModuleName: 'org.apache.beam.sdk.extensions.sql.meta.provider.hcatalog',
+  // iceberg requires Java11+
+  requireJavaVersion: JavaVersion.VERSION_11,
+)
+
+dependencies {
+  implementation project(":sdks:java:extensions:sql")
+  implementation project(":sdks:java:core")
+  implementation project(":sdks:java:managed")
+  implementation project(":sdks:java:io:iceberg")
+  runtimeOnly project(":sdks:java:io:iceberg:bqms")
+  runtimeOnly project(":sdks:java:io:iceberg:hive")
+  // TODO(https://github.com/apache/beam/issues/21156): Determine how to build without this dependency
+  provided "org.immutables:value:2.8.8"
+  permitUnusedDeclared "org.immutables:value:2.8.8"
+  implementation library.java.slf4j_api
+  implementation library.java.vendored_guava_32_1_2_jre
+  implementation library.java.vendored_calcite_1_40_0
+  implementation library.java.jackson_databind
+
+  testImplementation library.java.joda_time
+  testImplementation library.java.junit
+  testImplementation library.java.google_api_services_bigquery
+  testImplementation "org.apache.iceberg:iceberg-api:1.9.2"
+  testImplementation "org.apache.iceberg:iceberg-core:1.9.2"
+  testImplementation project(":sdks:java:io:google-cloud-platform")
+  testImplementation project(":sdks:java:extensions:google-cloud-platform-core")
+}
+
+task integrationTest(type: Test) {
+  def gcpProject = project.findProperty('gcpProject') ?: 'apache-beam-testing'
+  def gcsTempRoot = project.findProperty('gcsTempRoot') ?: 'gs://temp-storage-for-end-to-end-tests/'
+
+  // Disable Gradle cache (it should not be used because the IT's won't run).
+  outputs.upToDateWhen { false }
+
+  def pipelineOptions = [
+    "--project=${gcpProject}",
+    "--tempLocation=${gcsTempRoot}",
+    "--blockOnRun=false"]
+
+  systemProperty "beamTestPipelineOptions", JsonOutput.toJson(pipelineOptions)
+
+  include '**/*IT.class'
+
+  maxParallelForks 4
+  classpath = project(":sdks:java:extensions:sql:iceberg")
+    .sourceSets
+    .test
+    .runtimeClasspath
+  testClassesDirs = files(project(":sdks:java:extensions:sql:iceberg").sourceSets.test.output.classesDirs)
+  useJUnit { }
+}
+
+configurations.all {
+  // iceberg-core needs avro:1.12.0
+  resolutionStrategy.force 'org.apache.avro:avro:1.12.0'
+}

sdks/java/extensions/sql/iceberg/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergCatalogRegistrar.java

@@ -0,0 +1,31 @@
+/*
+ * 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.extensions.sql.meta.provider.iceberg;
+
+import com.google.auto.service.AutoService;
+import org.apache.beam.sdk.extensions.sql.meta.catalog.Catalog;
+import org.apache.beam.sdk.extensions.sql.meta.catalog.CatalogRegistrar;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
+
+@AutoService(CatalogRegistrar.class)
+public class IcebergCatalogRegistrar implements CatalogRegistrar {
+  @Override
+  public Iterable<Class<? extends Catalog>> getCatalogs() {
+    return ImmutableList.<Class<? extends Catalog>>builder().add(IcebergCatalog.class).build();
+  }
+}

sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergReadWriteIT.java → sdks/java/extensions/sql/iceberg/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergReadWriteIT.java

@@ -19,7 +19,6 @@
 
 import static java.lang.String.format;
 import static java.util.Arrays.asList;
-import static org.apache.beam.sdk.extensions.sql.utils.DateTimeUtils.parseTimestampWithUTCTimeZone;
 import static org.apache.beam.sdk.schemas.Schema.FieldType.BOOLEAN;
 import static org.apache.beam.sdk.schemas.Schema.FieldType.DOUBLE;
 import static org.apache.beam.sdk.schemas.Schema.FieldType.FLOAT;
@@ -64,6 +63,7 @@
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.joda.time.Duration;
+import org.joda.time.format.DateTimeFormat;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Rule;
@@ -238,7 +238,9 @@ public void runSqlWriteAndRead(boolean withPartitionFields)
                 (float) 1.0,
                 1.0,
                 true,
-                parseTimestampWithUTCTimeZone("2018-05-28 20:17:40.123"),
+                DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss.SSS")
+                    .withZoneUTC()
+                    .parseDateTime("2018-05-28 20:17:40.123"),
                 "varchar",
                 "char",
                 asList("123", "456"),

sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/PubsubToIcebergIT.java → sdks/java/extensions/sql/iceberg/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/PubsubToIcebergIT.java

@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.beam.sdk.extensions.sql;
+package org.apache.beam.sdk.extensions.sql.meta.provider.iceberg;
 
 import static java.lang.String.format;
 import static java.nio.charset.StandardCharsets.UTF_8;
@@ -33,6 +33,7 @@
 import java.util.UUID;
 import java.util.stream.Collectors;
 import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.extensions.sql.SqlTransform;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryUtils;
 import org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage;
 import org.apache.beam.sdk.io.gcp.pubsub.TestPubsub;

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/catalog/InMemoryCatalogRegistrar.java

@@ -18,16 +18,12 @@
 package org.apache.beam.sdk.extensions.sql.meta.catalog;
 
 import com.google.auto.service.AutoService;
-import org.apache.beam.sdk.extensions.sql.meta.provider.iceberg.IcebergCatalog;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
 
 @AutoService(CatalogRegistrar.class)
 public class InMemoryCatalogRegistrar implements CatalogRegistrar {
   @Override
   public Iterable<Class<? extends Catalog>> getCatalogs() {
-    return ImmutableList.<Class<? extends Catalog>>builder()
-        .add(InMemoryCatalog.class)
-        .add(IcebergCatalog.class)
-        .build();
+    return ImmutableList.<Class<? extends Catalog>>builder().add(InMemoryCatalog.class).build();
   }
 }