Eliminate the use of testRuntimeMigration for sdks:java:io:common

sdks/java/io/amazon-web-services/build.gradle

@@ -52,7 +52,7 @@ dependencies {
   runtimeOnly "org.apache.httpcomponents:httpclient:4.5.12"
   testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
   testImplementation project(path: ":sdks:java:extensions:avro", configuration: "testRuntimeMigration")
-  testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+  testImplementation project(path: ":sdks:java:io:common")
   testImplementation "io.findify:s3mock_2.12:0.2.6"
   testImplementation library.java.commons_lang3
   testImplementation library.java.hamcrest

sdks/java/io/amazon-web-services2/build.gradle

@@ -78,7 +78,7 @@ dependencies {
   implementation library.java.commons_codec
   testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
   testImplementation project(path: ":sdks:java:extensions:avro", configuration: "testRuntimeMigration")
-  testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+  testImplementation project(path: ":sdks:java:io:common")
   testImplementation "io.findify:s3mock_2.12:0.2.6"
   testImplementation 'org.elasticmq:elasticmq-rest-sqs_2.12:1.3.5'
   testImplementation library.java.mockito_core

sdks/java/io/bigquery-io-perf-tests/build.gradle

@@ -30,7 +30,7 @@ dependencies {
     testImplementation project(path: ":sdks:java:testing:test-utils", configuration: "testRuntimeMigration")
     testImplementation project(path: ":sdks:java:extensions:google-cloud-platform-core", configuration: "testRuntimeMigration")
     testImplementation project(path: ":sdks:java:testing:test-utils", configuration: "testRuntimeMigration")
-    testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+    testImplementation project(path: ":sdks:java:io:common")
     testImplementation project(path: ":sdks:java:io:google-cloud-platform", configuration: "testRuntimeMigration")
     testImplementation project(":sdks:java:io:synthetic")
     testImplementation library.java.junit

sdks/java/io/cassandra/build.gradle

@@ -45,7 +45,7 @@ dependencies {
   implementation library.java.slf4j_api
   implementation library.java.cassandra_driver_core
   implementation library.java.cassandra_driver_mapping
-  testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+  testImplementation project(path: ":sdks:java:io:common")
   testImplementation library.java.junit
   testImplementation library.java.hamcrest
   testImplementation library.java.mockito_core

sdks/java/io/cdap/build.gradle

@@ -72,5 +72,5 @@ dependencies {
     testImplementation project(path: ":sdks:java:extensions:avro", configuration: "testRuntimeMigration")
     testImplementation project(path: ":sdks:java:testing:test-utils", configuration: "testRuntimeMigration")
     testImplementation project(path: ":runners:direct-java", configuration: "shadow")
-    testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+    testImplementation project(path: ":sdks:java:io:common")
 }

sdks/java/io/common/build.gradle

@@ -20,15 +20,17 @@ plugins { id 'org.apache.beam.module' }
 applyJavaNature( exportJavadoc: false, automaticModuleName: 'org.apache.beam.sdk.io.common')
 
 description = "Apache Beam :: SDKs :: Java :: IO :: Common"
-ext.summary = "Code used by all Beam IOs"
+ext.summary = "Common test fixture used by Beam IOs"
 
 dependencies {
-  testImplementation library.java.vendored_guava_32_1_2_jre
+  implementation library.java.vendored_guava_32_1_2_jre
+  implementation library.java.joda_time
+  implementation library.java.postgres
+  implementation library.java.slf4j_api
+  implementation library.java.testcontainers_jdbc
+  implementation 'com.zaxxer:HikariCP-java6:2.3.13'
+  implementation project(path: ":sdks:java:core", configuration: "shadow")
+
   testImplementation library.java.junit
   testImplementation library.java.hamcrest
-  testImplementation library.java.postgres
-  testImplementation library.java.testcontainers_mysql
-  testImplementation library.java.testcontainers_postgresql
-  testImplementation 'com.zaxxer:HikariCP-java6:2.3.13'
-  testImplementation project(path: ":sdks:java:core", configuration: "shadow")
 }

sdks/java/io/common/src/test/java/org/apache/beam/sdk/io/common/DatabaseTestHelper.java → sdks/java/io/common/src/main/java/org/apache/beam/sdk/io/common/DatabaseTestHelper.java

@@ -17,10 +17,12 @@
  */
 package org.apache.beam.sdk.io.common;
 
-import static org.junit.Assert.assertEquals;
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
 
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.sql.Connection;
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -45,32 +47,28 @@
 import org.testcontainers.containers.JdbcDatabaseContainer;
 
 /** This class contains helper methods to ease database usage in tests. */
+@SuppressFBWarnings(
+    value = {"OBL_UNSATISFIED_OBLIGATION", "OBL_UNSATISFIED_OBLIGATION_EXCEPTION_EDGE"},
+    justification = "https://github.com/spotbugs/spotbugs/issues/493")
 public class DatabaseTestHelper {
   private static Map<String, DataSource> hikariSources = new HashMap<>();
 
-  public static ResultSet performQuery(JdbcDatabaseContainer<?> container, String sql)
-      throws SQLException {
-    DataSource ds = getDataSourceForContainer(container);
-    Statement statement = ds.getConnection().createStatement();
-    statement.execute(sql);
-    ResultSet resultSet = statement.getResultSet();
-
-    resultSet.next();
-    return resultSet;
-  }
-
   public static DataSource getDataSourceForContainer(JdbcDatabaseContainer<?> container) {
-    if (hikariSources.get(container.getJdbcUrl()) != null) {
-      return hikariSources.get(container.getJdbcUrl());
+    String jdbcUrl = checkNotNull(container.getJdbcUrl());
+    DataSource existingSource = hikariSources.get(jdbcUrl);
+    if (existingSource != null) {
+      return existingSource;
     }
     HikariConfig hikariConfig = new HikariConfig();
     // Keeping a small connection pool to a testContainer to avoid overwhelming it.
     hikariConfig.setMaximumPoolSize(2);
-    hikariConfig.setJdbcUrl(container.getJdbcUrl());
+    hikariConfig.setJdbcUrl(jdbcUrl);
     hikariConfig.setUsername(container.getUsername());
     hikariConfig.setPassword(container.getPassword());
     hikariConfig.setDriverClassName(container.getDriverClassName());
-    return hikariSources.put(container.getJdbcUrl(), new HikariDataSource(hikariConfig));
+    existingSource = new HikariDataSource(hikariConfig);
+    hikariSources.put(jdbcUrl, existingSource);
+    return existingSource;
   }
 
   public static PGSimpleDataSource getPostgresDataSource(PostgresIOTestPipelineOptions options) {
@@ -102,19 +100,13 @@ public static void createTable(
     while (true) {
       // This is not implemented as try-with-resources because it appears that try-with-resources is
       // not correctly catching the PSQLException thrown by dataSource.getConnection()
-      Connection connection = null;
-      try {
-        connection = dataSource.getConnection();
-        try (Statement statement = connection.createStatement()) {
-          statement.execute(String.format("create table %s (%s)", tableName, fieldsList));
-          return;
-        }
+      try (Connection connection = dataSource.getConnection();
+          Statement statement = connection.createStatement()) {
+        statement.execute(String.format("create table %s (%s)", tableName, fieldsList));
+        return;
+
       } catch (SQLException e) {
         exception = e;
-      } finally {
-        if (connection != null) {
-          connection.close();
-        }
       }
       boolean hasNext;
       try {
@@ -176,10 +168,9 @@ public static Optional<Long> getPostgresTableSize(DataSource dataSource, String
 
   public static void createTableWithStatement(DataSource dataSource, String stmt)
       throws SQLException {
-    try (Connection connection = dataSource.getConnection()) {
-      try (Statement statement = connection.createStatement()) {
-        statement.execute(stmt);
-      }
+    try (Connection connection = dataSource.getConnection();
+        Statement statement = connection.createStatement()) {
+      statement.execute(stmt);
     }
   }
 
@@ -194,14 +185,15 @@ public static ArrayList<KV<Integer, String>> getTestDataToWrite(long rowsToAdd)
 
   public static void assertRowCount(DataSource dataSource, String tableName, int expectedRowCount)
       throws SQLException {
-    try (Connection connection = dataSource.getConnection()) {
-      try (Statement statement = connection.createStatement()) {
-        try (ResultSet resultSet = statement.executeQuery("select count(*) from " + tableName)) {
-          resultSet.next();
-          int count = resultSet.getInt(1);
-          assertEquals(expectedRowCount, count);
-        }
-      }
+    int count;
+    try (Connection connection = dataSource.getConnection();
+        Statement statement = connection.createStatement();
+        ResultSet resultSet = statement.executeQuery("select count(*) from " + tableName)) {
+      resultSet.next();
+      count = resultSet.getInt(1);
     }
+    // check state outside the try close to enable auto-close resources
+    checkState(
+        expectedRowCount == count, "Expected count: %d, actual: %d", expectedRowCount, count);
   }
 }

sdks/java/io/common/src/test/java/org/apache/beam/sdk/io/common/HashingFn.java → sdks/java/io/common/src/main/java/org/apache/beam/sdk/io/common/HashingFn.java

@@ -38,9 +38,9 @@ public class HashingFn extends CombineFn<String, HashingFn.Accum, String> {
 
   /** Serializable Class to store the HashCode of input String. */
   public static class Accum implements Serializable {
-    HashCode hashCode = null;
+    @Nullable HashCode hashCode;
 
-    public Accum(HashCode value) {
+    public Accum(@Nullable HashCode value) {
       this.hashCode = value;
     }
 

sdks/java/io/common/src/test/java/org/apache/beam/sdk/io/common/SchemaAwareJavaBeans.java → sdks/java/io/common/src/main/java/org/apache/beam/sdk/io/common/SchemaAwareJavaBeans.java

@@ -398,17 +398,17 @@ public abstract static class NullableAllPrimitiveDataTypes implements Serializab
     @AutoValue.Builder
     public abstract static class Builder {
 
-      public abstract Builder setABoolean(Boolean value);
+      public abstract Builder setABoolean(@Nullable Boolean value);
 
-      public abstract Builder setADouble(Double value);
+      public abstract Builder setADouble(@Nullable Double value);
 
-      public abstract Builder setAFloat(Float value);
+      public abstract Builder setAFloat(@Nullable Float value);
 
-      public abstract Builder setAnInteger(Integer value);
+      public abstract Builder setAnInteger(@Nullable Integer value);
 
-      public abstract Builder setALong(Long value);
+      public abstract Builder setALong(@Nullable Long value);
 
-      public abstract Builder setAString(String value);
+      public abstract Builder setAString(@Nullable String value);
 
       public abstract NullableAllPrimitiveDataTypes build();
     }

sdks/java/io/common/src/test/java/org/apache/beam/sdk/io/common/TestRow.java → sdks/java/io/common/src/main/java/org/apache/beam/sdk/io/common/TestRow.java

@@ -43,7 +43,7 @@ public int compareTo(TestRow other) {
     return id().compareTo(other.id());
   }
 
-  /** Creates a {@link org.apache.beam.sdk.io.common.TestRow} from the seed value. */
+  /** Creates a {@link TestRow} from the seed value. */
   public static TestRow fromSeed(Integer seed) {
     return create(seed, getNameForSeed(seed));
   }
@@ -54,8 +54,8 @@ public static String getNameForSeed(Integer seed) {
   }
 
   /**
-   * Returns a range of {@link org.apache.beam.sdk.io.common.TestRow}s for seed values between
-   * rangeStart (inclusive) and rangeEnd (exclusive).
+   * Returns a range of {@link TestRow}s for seed values between rangeStart (inclusive) and rangeEnd
+   * (exclusive).
    */
   public static Iterable<TestRow> getExpectedValues(int rangeStart, int rangeEnd) {
     List<TestRow> ret = new ArrayList<>(rangeEnd - rangeStart + 1);
@@ -65,17 +65,15 @@ public static Iterable<TestRow> getExpectedValues(int rangeStart, int rangeEnd)
     return ret;
   }
 
-  /**
-   * Uses the input Long values as seeds to produce {@link org.apache.beam.sdk.io.common.TestRow}s.
-   */
+  /** Uses the input Long values as seeds to produce {@link TestRow}s. */
   public static class DeterministicallyConstructTestRowFn extends DoFn<Long, TestRow> {
     @ProcessElement
     public void processElement(ProcessContext c) {
       c.output(fromSeed(c.element().intValue()));
     }
   }
 
-  /** Outputs just the name stored in the {@link org.apache.beam.sdk.io.common.TestRow}. */
+  /** Outputs just the name stored in the {@link TestRow}. */
   public static class SelectNameFn extends DoFn<TestRow, String> {
     @ProcessElement
     public void processElement(ProcessContext c) {
@@ -95,9 +93,8 @@ public void processElement(ProcessContext c) {
           5_000_000, "c44f8a5648cd9207c9c6f77395a998dc");
 
   /**
-   * Returns the hash value that {@link org.apache.beam.sdk.io.common.HashingFn} will return when it
-   * is run on {@link org.apache.beam.sdk.io.common.TestRow}s produced by getExpectedValues(0,
-   * rowCount).
+   * Returns the hash value that {@link HashingFn} will return when it is run on {@link TestRow}s
+   * produced by getExpectedValues(0, rowCount).
    */
   public static String getExpectedHashForRowCount(int rowCount)
       throws UnsupportedOperationException {

sdks/java/io/common/src/main/java/org/apache/beam/sdk/io/common/package-info.java

@@ -0,0 +1,19 @@
+/*
+ * 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.
+ */
+/** Common test fixture for Beam IOs. */
+package org.apache.beam.sdk.io.common;

sdks/java/io/csv/build.gradle

@@ -31,5 +31,5 @@ dependencies {
     testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
     testImplementation library.java.junit
     testRuntimeOnly project(path: ":runners:direct-java", configuration: "shadow")
-    testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+    testImplementation project(path: ":sdks:java:io:common")
 }

sdks/java/io/debezium/build.gradle

@@ -39,7 +39,7 @@ dependencies {
     provided library.java.jackson_dataformat_csv
     permitUnusedDeclared library.java.jackson_dataformat_csv
     testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
-    testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+    testImplementation project(path: ":sdks:java:io:common")
 
     // Test dependencies
     testImplementation library.java.junit

sdks/java/io/elasticsearch/build.gradle

@@ -36,5 +36,5 @@ dependencies {
   implementation "org.apache.httpcomponents:httpasyncclient:4.1.4"
   implementation "org.apache.httpcomponents:httpcore-nio:4.4.12"
   implementation "org.elasticsearch.client:elasticsearch-rest-client:7.17.22"
-  testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+  testImplementation project(path: ":sdks:java:io:common")
 }

sdks/java/io/file-based-io-tests/build.gradle

@@ -27,7 +27,7 @@ ext.summary = "Integration tests for reading/writing using file-based sources/si
 dependencies {
   testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
   testImplementation project(path: ":sdks:java:extensions:avro", configuration: "testRuntimeMigration")
-  testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+  testImplementation project(path: ":sdks:java:io:common")
   testImplementation project(path: ":sdks:java:io:xml", configuration: "testRuntimeMigration")
   testImplementation project(path: ":sdks:java:io:parquet", configuration: "testRuntimeMigration")
   testImplementation project(path: ":sdks:java:testing:test-utils", configuration: "testRuntimeMigration")

sdks/java/io/file-schema-transform/build.gradle

@@ -53,7 +53,7 @@ dependencies {
 
     testImplementation library.java.junit
     testImplementation project(path: ":sdks:java:core", configuration: "shadow")
-    testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+    testImplementation project(path: ":sdks:java:io:common")
     testImplementation project(path: ":sdks:java:io:parquet")
     testImplementation project(path: ":sdks:java:io:xml")
     testRuntimeOnly project(path: ":runners:direct-java", configuration: "shadow")

sdks/java/io/google-ads/build.gradle

@@ -36,7 +36,7 @@ dependencies {
   implementation library.java.joda_time
   implementation library.java.vendored_guava_32_1_2_jre
   testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
-  testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+  testImplementation project(path: ":sdks:java:io:common")
   testImplementation library.java.mockito_core
   testImplementation library.java.junit
   testRuntimeOnly project(path: ":runners:direct-java", configuration: "shadow")

sdks/java/io/google-cloud-platform/build.gradle

@@ -164,7 +164,7 @@ dependencies {
   testImplementation project(path: ":sdks:java:extensions:google-cloud-platform-core", configuration: "testRuntimeMigration")
   testImplementation project(path: ":sdks:java:extensions:protobuf", configuration: "testRuntimeMigration")
   testImplementation project(path: ":runners:direct-java", configuration: "shadow")
-  testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+  testImplementation project(path: ":sdks:java:io:common")
   testImplementation project(path: ":sdks:java:testing:test-utils", configuration: "testRuntimeMigration")
   testImplementation library.java.mockito_core
   testImplementation library.java.powermock

sdks/java/io/hadoop-format/build.gradle

@@ -64,7 +64,7 @@ dependencies {
   provided library.java.hadoop_mapreduce_client_core
   testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
   testImplementation project(path: ":sdks:java:extensions:avro", configuration: "testRuntimeMigration")
-  testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+  testImplementation project(path: ":sdks:java:io:common")
   testImplementation project(path: ":sdks:java:testing:test-utils", configuration: "testRuntimeMigration")
   testImplementation project(":sdks:java:io:jdbc")
   testImplementation "org.elasticsearch.plugin:transport-netty4-client:$elastic_search_version"

sdks/java/io/hbase/build.gradle

@@ -42,7 +42,7 @@ dependencies {
   implementation project(":sdks:java:io:hadoop-common")
   implementation library.java.slf4j_api
   provided "org.apache.hbase:hbase-shaded-client:$hbase_version"
-  testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+  testImplementation project(path: ":sdks:java:io:common")
   testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
   testImplementation library.java.junit
   testImplementation library.java.hamcrest

sdks/java/io/hcatalog/build.gradle

@@ -67,7 +67,6 @@ dependencies {
     exclude group: "org.apache.hive", module: "hive-exec"
     exclude group: "com.google.protobuf", module: "protobuf-java"
   }
-  testImplementation project(":sdks:java:io:common").sourceSets.test.output
   testImplementation library.java.commons_io
   testImplementation library.java.junit
   testImplementation library.java.hamcrest

sdks/java/io/influxdb/build.gradle

@@ -35,5 +35,5 @@ dependencies {
   testImplementation library.java.mockito_core
   testRuntimeOnly project(path: ":runners:direct-java", configuration: "shadow")
   testImplementation project(path: ":sdks:java:extensions:avro", configuration: "testRuntimeMigration")
-  testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+  testImplementation project(path: ":sdks:java:io:common")
 }

sdks/java/io/jdbc/build.gradle

@@ -38,7 +38,7 @@ dependencies {
   testImplementation "org.apache.derby:derbynet:10.14.2.0"
   testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
   testImplementation project(path: ":sdks:java:extensions:avro", configuration: "testRuntimeMigration")
-  testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+  testImplementation project(path: ":sdks:java:io:common")
   testImplementation project(path: ":sdks:java:testing:test-utils", configuration: "testRuntimeMigration")
   testImplementation library.java.junit
   testImplementation library.java.slf4j_api