Add oracle live ITs

.github/workflows/spanner-pr.yml

@@ -173,7 +173,10 @@ jobs:
         --it-project="span-cloud-teleport-testing" \
         --it-artifact-bucket="span-cloud-teleport-testing-it-gitactions" \
         --it-private-connectivity="datastream-connect-2" \
-        --it-cloud-proxy-host="10.128.0.16"
+        --it-cloud-proxy-host="10.128.0.16" \
+        --it-oracle-host="10.128.0.108" \
+        --it-oracle-sys-password="TestPassword123" \
+        --it-oracle-password="TestPassword123"
     - name: Upload Smoke Tests Report
       uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
       if: always() # always run even if the previous step fails
@@ -217,6 +220,9 @@ jobs:
         --it-artifact-bucket="span-cloud-teleport-testing-it-gitactions" \
         --it-private-connectivity="datastream-connect-2" \
         --it-cloud-proxy-host="10.128.0.16" \
+        --it-oracle-host="10.128.0.108" \
+        --it-oracle-sys-password="TestPassword123" \
+        --it-oracle-password="TestPassword123" \
         --test="$SPECIFIC_TEST"
     - name: Upload Integration Tests Report
       uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1

cicd/cmd/run-it-smoke-tests/main.go

@@ -69,6 +69,7 @@ func main() {
 		flags.RetryFailures(),
 		flags.StaticOracleHost(),
 		flags.StaticOracleSysPassword(),
+		flags.StaticOraclePassword(),
 		flags.CloudProxyHost(),
 		flags.CloudProxyMySqlPort(),
 		flags.CloudProxyPostgresPort(),

cicd/cmd/run-it-tests/main.go

@@ -70,6 +70,7 @@ func main() {
 		flags.RetryFailures(),
 		flags.StaticOracleHost(),
 		flags.StaticOracleSysPassword(),
+		flags.StaticOraclePassword(),
 		flags.CloudProxyHost(),
 		flags.CloudProxyMySqlPort(),
 		flags.CloudProxyPostgresPort(),

cicd/internal/flags/it-flags.go

@@ -39,6 +39,7 @@ var (
 	dCloudProxyPassword                 string
 	dOracleHost                         string
 	dCloudOracleSysPassword             string
+	dCloudOraclePassword                string
 	dUnifiedWorkerHarnessContainerImage string
 	dIntegrationTestParallelism         string
 	dThreadCount                        string
@@ -61,6 +62,7 @@ func RegisterItFlags() {
 	flag.StringVar(&dCloudProxyPassword, "it-cloud-proxy-password", "t>5xl%J(&qTK6?FaZ", "Password of static Cloud Auth Proxy")
 	flag.StringVar(&dOracleHost, "it-oracle-host", "10.128.0.90", "Hostname or IP address of static Oracle DB")
 	flag.StringVar(&dCloudOracleSysPassword, "it-oracle-sys-password", "oracle", "sys password of static Oracle DB")
+	flag.StringVar(&dCloudOraclePassword, "it-oracle-password", "oracle", "app password of static Oracle DB")
 	flag.StringVar(&dUnifiedWorkerHarnessContainerImage, "it-unified-worker-harness-container-image", "", "Runner harness image to run tests against")
 	flag.StringVar(&dIntegrationTestParallelism, "it-integration-test-parallelism", "3", "The level of parallelism for integration tests")
 	flag.StringVar(&dThreadCount, "it-thread-count", "4", "The IT thread count to use for maven, which is the number of threads per core")
@@ -149,6 +151,10 @@ func StaticOracleSysPassword() string {
 	return "-DcloudOracleSysPassword=" + dCloudOracleSysPassword
 }
 
+func StaticOraclePassword() string {
+	return "-DcloudOraclePassword=" + dCloudOraclePassword
+}
+
 func UnifiedWorkerHarnessContainerImage() string {
 	if dUnifiedWorkerHarnessContainerImage != "" {
 		return "-DunifiedWorkerHarnessContainerImage=" + dUnifiedWorkerHarnessContainerImage

v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToSpannerIT.java

@@ -24,6 +24,7 @@
 import com.google.cloud.spanner.Dialect;
 import com.google.cloud.teleport.metadata.SkipDirectRunnerTest;
 import com.google.cloud.teleport.metadata.TemplateIntegrationTest;
+import com.google.cloud.teleport.v2.templates.oracle.SpannerOracleResourceManager;
 import com.google.common.io.Resources;
 import com.google.pubsub.v1.SubscriptionName;
 import com.google.pubsub.v1.TopicName;
@@ -225,7 +226,8 @@ private void simpleJdbcToSpannerTest(
     cloudSqlResourceManager =
         jdbcType.equals(JDBCType.MYSQL)
             ? CloudMySQLResourceManager.builder(testName).build()
-            : CloudOracleResourceManager.builder(testName).build();
+            : new SpannerOracleResourceManager(
+                (CloudOracleResourceManager.Builder) CloudOracleResourceManager.builder(testName));
 
     // Create Spanner Resource Manager
     SpannerResourceManager.Builder spannerResourceManagerBuilder =
@@ -507,7 +509,8 @@ protected CheckResult check() {
         // Force log file archive - needed so Datastream can see changes which are read from
         // archived log files.
         if (isOracle) {
-          cloudSqlResourceManager.runSQLUpdate("ALTER SYSTEM SWITCH LOGFILE");
+          ((org.apache.beam.it.gcp.cloudsql.CloudOracleResourceManager) cloudSqlResourceManager)
+              .runSQLUpdate("ALTER SYSTEM SWITCH LOGFILE");
         }
         return new CheckResult(success, "Sent " + String.join(", ", messages) + ".");
       }
@@ -578,7 +581,8 @@ protected CheckResult check() {
         // Force log file archive - needed so Datastream can see changes which are read from
         // archived log files.
         if (isOracle) {
-          cloudSqlResourceManager.runSQLUpdate("ALTER SYSTEM SWITCH LOGFILE");
+          ((org.apache.beam.it.gcp.cloudsql.CloudOracleResourceManager) cloudSqlResourceManager)
+              .runSQLUpdate("ALTER SYSTEM SWITCH LOGFILE");
         }
         return new CheckResult(true, "Sent " + String.join(", ", messages) + ".");
       }

v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToSpannerITBase.java

@@ -90,6 +90,45 @@ public SpannerResourceManager setUpPGDialectSpannerResourceManager() {
         .build();
   }
 
+  public com.google.cloud.teleport.v2.templates.oracle.SpannerOracleResourceManager
+      setUpOracleResourceManager() {
+    org.apache.beam.it.gcp.cloudsql.CloudOracleResourceManager.Builder builder =
+        org.apache.beam.it.gcp.cloudsql.CloudOracleResourceManager.builder(testName);
+    builder.maybeUseStaticInstance();
+    if (System.getProperty("cloudOracleHost") != null) {
+      org.apache.beam.it.gcp.cloudsql.CloudOracleResourceManager sysdba =
+          (org.apache.beam.it.gcp.cloudsql.CloudOracleResourceManager)
+              org.apache.beam.it.gcp.cloudsql.CloudOracleResourceManager.builder(testName)
+                  .setUsername("sys as sysdba")
+                  .setPassword(System.getProperty("cloudOraclePassword", "TestPassword123"))
+                  .setDatabaseName("XEPDB1")
+                  .setHost(System.getProperty("cloudOracleHost"))
+                  .setPort(1521)
+                  .build();
+
+      String isoUser =
+          "U_" + org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric(5).toUpperCase();
+      String isoPassword =
+          "P_" + org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric(5).toUpperCase();
+
+      sysdba.runSQLUpdate(
+          String.format("CREATE USER %s IDENTIFIED BY %s CONTAINER=ALL", isoUser, isoPassword));
+      sysdba.runSQLUpdate(String.format("GRANT DBA TO %s CONTAINER=ALL", isoUser));
+      sysdba.runSQLUpdate(
+          String.format("GRANT EXECUTE ON SYS.DBMS_LOGMNR TO %s CONTAINER=ALL", isoUser));
+      sysdba.runSQLUpdate(
+          String.format("ALTER USER %s QUOTA 50m ON SYSTEM CONTAINER=ALL", isoUser));
+
+      builder.setPassword(isoPassword);
+      builder.setHost(System.getProperty("cloudOracleHost"));
+      builder.setPort(1521);
+      builder.setUsername(isoUser);
+      builder.setSystemIdentifier(System.getProperty("cloudOracleSid", "XE"));
+      builder.setDatabaseName("XEPDB1");
+    }
+    return new com.google.cloud.teleport.v2.templates.oracle.SpannerOracleResourceManager(builder);
+  }
+
   public String generateSessionFile(
       int numOfTables, String srcDb, String spannerDb, List<String> tableNames, String sessionFile)
       throws IOException {
@@ -614,4 +653,37 @@ protected ConditionCheck buildBaseConditionCheck(
     }
     return combinedCondition;
   }
+
+  public void flushOracleRedoLogs(
+      org.apache.beam.it.gcp.cloudsql.CloudOracleResourceManager oracleResourceManager) {
+    boolean success = false;
+    if (oracleResourceManager != null) {
+      try {
+        oracleResourceManager.runSQLUpdate("ALTER SYSTEM SWITCH LOGFILE");
+        org.slf4j.LoggerFactory.getLogger(DataStreamToSpannerITBase.class)
+            .info("Successfully flushed Oracle redo logs natively.");
+        success = true;
+      } catch (Exception e) {
+        org.slf4j.LoggerFactory.getLogger(DataStreamToSpannerITBase.class)
+            .warn(
+                "Failed to switch Oracle log via ResourceManager, attempting raw JDBC fallback...",
+                e);
+      }
+    }
+
+    if (!success && System.getProperty("cloudOracleHost") != null) {
+      String url = "jdbc:oracle:thin:@//" + System.getProperty("cloudOracleHost") + ":1521/XE";
+      String user = System.getProperty("cloudOracleUsername", "system");
+      String pass = System.getProperty("cloudOraclePassword", "TestPassword123");
+      try (java.sql.Connection conn = java.sql.DriverManager.getConnection(url, user, pass);
+          java.sql.Statement stmt = conn.createStatement()) {
+        stmt.execute("ALTER SYSTEM SWITCH LOGFILE");
+        org.slf4j.LoggerFactory.getLogger(DataStreamToSpannerITBase.class)
+            .info("Successfully flushed Oracle redo logs via raw JDBC fallback.");
+      } catch (Exception ex) {
+        org.slf4j.LoggerFactory.getLogger(DataStreamToSpannerITBase.class)
+            .error("Raw JDBC fallback log flush also failed.", ex);
+      }
+    }
+  }
 }