adjust to review comments by @Abacn

Author: tkaymak

.gitignore

@@ -34,9 +34,6 @@ sdks/java/maven-archetypes/examples-java8/src/main/resources/archetype-resources
 sdks/java/maven-archetypes/gcp-bom-examples/src/main/resources/archetype-resources/src/
 sdks/java/maven-archetypes/examples/src/main/resources/archetype-resources/sample.txt
 
-# Ignore generated debezium data
-sdks/java/io/debezium/data/
-
 # Ignore files generated by the Python build process.
 **/*.pyc
 **/*.pyo

CHANGES.md

@@ -31,7 +31,7 @@
 
 ## I/Os
 
-* Debezium IO upgraded to 3.1.1 requires Java 17 (Java) ([#34747](https://github.com/apache/beam/issues/34747)).
+* Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 
 ## New Features / Improvements
 
@@ -75,6 +75,7 @@
 ## I/Os
 
 * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
+* Debezium IO upgraded to 3.1.1 requires Java 17 (Java) ([#34747](https://github.com/apache/beam/issues/34747)).
 
 ## New Features / Improvements
 * Adding Google Storage Requests Pays feature (Golang)([#30747](https://github.com/apache/beam/issues/30747)).
@@ -196,7 +197,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)).
-* Debezium IO (Java) has been upgraded from depending on version 1.3.1.Final of io.debezium to 3.1.1.Final. This may cause some breaking changes since the libraries do not maintain full compatibility ([#33526](https://github.com/apache/beam/issues/33526)).
 
 ## Bugfixes
 

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -674,8 +674,8 @@ class BeamModulePlugin implements Plugin<Project> {
         activemq_junit                              : "org.apache.activemq.tooling:activemq-junit:$activemq_version",
         activemq_kahadb_store                       : "org.apache.activemq:activemq-kahadb-store:$activemq_version",
         activemq_mqtt                               : "org.apache.activemq:activemq-mqtt:$activemq_version",
-        antlr                                       : "org.antlr:antlr4:4.10",
-        antlr_runtime                               : "org.antlr:antlr4-runtime:4.10",
+        antlr                                       : "org.antlr:antlr4:4.7",
+        antlr_runtime                               : "org.antlr:antlr4-runtime:4.7",
         args4j                                      : "args4j:args4j:2.33",
         auto_value_annotations                      : "com.google.auto.value:auto-value-annotations:$autovalue_version",
         // TODO: https://github.com/apache/beam/issues/34993 after stopping supporting Java 8

sdks/java/io/debezium/build.gradle

@@ -19,23 +19,13 @@ import groovy.json.JsonOutput
 
 plugins { id 'org.apache.beam.module' }
 
-// The order is intended here - Debezium 3 requires Java 17, or later
-// Overwrite javaVersion global property if corresponding project property specified
-if (project.hasProperty('java17Home')) {
-    javaVersion = "1.17"
-} else if (project.hasProperty('java21Home')) {
-    javaVersion = "1.21"
-} else if (JavaVersion.VERSION_1_8.compareTo(JavaVersion.current()) < 0) {
-    // Otherwise, compile the project with java11 spec
-    javaVersion = "1.17"
-}
-
 applyJavaNature(
         automaticModuleName: 'org.apache.beam.sdk.io.debezium',
         mavenRepositories: [
                 [id: 'io.confluent', url: 'https://packages.confluent.io/maven/']
         ],
         enableSpotbugs: false,
+        requireJavaVersion: JavaVersion.VERSION_17,
 )
 provideIntegrationTestingDependencies()
 
@@ -53,8 +43,6 @@ dependencies {
 
     // Kafka connect dependencies
     implementation "org.apache.kafka:connect-api:3.9.0"
-    implementation "org.apache.kafka:connect-json:3.9.0"
-    permitUnusedDeclared "org.apache.kafka:connect-json:3.9.0"
 
     // Debezium dependencies
     implementation group: 'io.debezium', name: 'debezium-core', version: '3.1.1.Final'
@@ -80,7 +68,15 @@ dependencies {
     testImplementation group: 'io.debezium', name: 'debezium-connector-postgres', version: '3.1.1.Final'
 }
 
+// TODO: Remove pin after Beam has unpinned it
+// Pin the Antlr version
+configurations.all {
+    resolutionStrategy {
+        force 'org.antlr:antlr4:4.10', 'org.antlr:antlr4-runtime:4.10'
+    }
+}
 
+// TODO: Remove pin after upgrading Beam's Jackson version
 // Force Jackson versions for the test runtime classpath
 configurations.named("testRuntimeClasspath") {
     resolutionStrategy.force (
@@ -139,23 +135,6 @@ task integrationTest(type: Test, dependsOn: processTestResources) {
   }
 }
 
-// The order is intended here - Debezium 3 requires Java 17, or later
-if (project.hasProperty('java17Home')) {
-    project.tasks.withType(JavaCompile) {
-        setJavaVerOptions(options, '17')
-    }
-    project.tasks.withType(Javadoc) {
-        executable = project.findProperty('java17Home') + '/bin/javadoc'
-    }
-} else if (project.hasProperty('java21Home')) {
-    project.tasks.withType(JavaCompile) {
-        setJavaVerOptions(options, '21')
-    }
-    project.tasks.withType(Javadoc) {
-        executable = project.findProperty('java21Home') + '/bin/javadoc'
-    }
-}
-
 project.tasks.each {
     it.onlyIf {project.hasProperty('java17Home') || project.hasProperty('java21Home')
                 || JavaVersion.VERSION_17.compareTo(JavaVersion.current()) <= 0

sdks/java/io/debezium/expansion-service/build.gradle

@@ -20,22 +20,12 @@ apply plugin: 'org.apache.beam.module'
 apply plugin: 'application'
 mainClassName = "org.apache.beam.sdk.expansion.service.ExpansionService"
 
-// The order is intended here - Debezium 3 requires Java 17, or later
-// Overwrite javaVersion global property if corresponding project property specified
-if (project.hasProperty('java17Home')) {
-    javaVersion = "1.17"
-} else if (project.hasProperty('java21Home')) {
-    javaVersion = "1.21"
-} else if (JavaVersion.VERSION_1_8.compareTo(JavaVersion.current()) < 0) {
-    // Otherwise, compile the project with java11 spec
-    javaVersion = "1.17"
-}
-
 applyJavaNature(
         automaticModuleName: 'org.apache.beam.sdk.io.debezium.expansion.service',
         exportJavadoc: false,
         validateShadowJar: false,
         shadowClosure: {},
+        requireJavaVersion: JavaVersion.VERSION_17,
 )
 
 description = "Apache Beam :: SDKs :: Java :: IO :: Debezium :: Expansion Service"
@@ -57,23 +47,6 @@ dependencies {
     runtimeOnly group: 'io.debezium', name: 'debezium-connector-db2', version: debezium_version
 }
 
-// The order is intended here - Debezium 3 requires Java 17, or later
-if (project.hasProperty('java17Home')) {
-    project.tasks.withType(JavaCompile) {
-        setJavaVerOptions(options, '17')
-    }
-    project.tasks.withType(Javadoc) {
-        executable = project.findProperty('java17Home') + '/bin/javadoc'
-    }
-} else if (project.hasProperty('java21Home')) {
-    project.tasks.withType(JavaCompile) {
-        setJavaVerOptions(options, '21')
-    }
-    project.tasks.withType(Javadoc) {
-        executable = project.findProperty('java21Home') + '/bin/javadoc'
-    }
-}
-
 // Module classes requires JDK > 8
 project.tasks.each {
     it.onlyIf {project.hasProperty('java17Home') || project.hasProperty('java21Home')

sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOMySqlConnectorIT.java

@@ -21,6 +21,7 @@
 import static org.apache.beam.sdk.testing.SerializableMatchers.hasItem;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
+import static org.junit.jupiter.api.Assertions.fail;
 
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
@@ -110,7 +111,7 @@ public static DataSource getMysqlDatasource(Void unused) {
     return new HikariDataSource(hikariConfig);
   }
 
-  private void monitorEssentialMetrics() {
+  private void monitorEssentialMetrics() throws SQLException {
     DataSource ds = getMysqlDatasource(null);
     try {
       Connection conn = ds.getConnection();
@@ -128,6 +129,7 @@ private void monitorEssentialMetrics() {
       }
     } catch (SQLException ex) {
       LOG.error("SQL error in monitoring thread. Shutting down.", ex);
+      throw (ex);
     } catch (InterruptedException ex) {
       LOG.info("Monitoring thread interrupted. Shutting down.");
       Thread.currentThread().interrupt();
@@ -207,7 +209,16 @@ public void testDebeziumSchemaTransformMysqlRead() throws InterruptedException {
               return null;
             });
     Thread writeThread = new Thread(() -> writePipeline.run().waitUntilFinish());
-    Thread monitorThread = new Thread(this::monitorEssentialMetrics);
+    Thread monitorThread =
+        new Thread(
+            () -> {
+              try {
+                monitorEssentialMetrics();
+              } catch (SQLException e) {
+                e.printStackTrace();
+                fail("Failed because of SQLException in monitorEssentialMetrics!");
+              }
+            });
     monitorThread.start();
     writeThread.start();
 

sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformTest.java

@@ -33,6 +33,7 @@
 import org.hamcrest.Matchers;
 import org.junit.ClassRule;
 import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.testcontainers.containers.Container;
@@ -44,6 +45,8 @@
 @RunWith(Parameterized.class)
 public class DebeziumReadSchemaTransformTest {
 
+  @ClassRule public static TemporaryFolder tempFolder = new TemporaryFolder();
+
   @ClassRule
   public static final PostgreSQLContainer<?> POSTGRES_SQL_CONTAINER =
       new PostgreSQLContainer<>(
@@ -109,7 +112,9 @@ private PTransform<PCollectionRowTuple, PCollectionRowTuple> makePtransform(
                     Lists.newArrayList(
                         "database.server.id=579676",
                         "schema.history.internal=io.debezium.storage.file.history.FileSchemaHistory",
-                        "schema.history.internal.file.filename=data/schema_history.dat"))
+                        String.format(
+                            "schema.history.internal.file.filename=%s",
+                            tempFolder.getRoot().toPath().resolve("schema_history.dat"))))
                 .build());
   }