[Improve][E2E] improve oracle e2e

hawk9821 · Dec 16, 2024 · 095a8c0 · 095a8c0
1 parent 8e29ecf
commit 095a8c0
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 9 deletions.
diff --git a/...rg/apache/seatunnel/connectors/seatunnel/cdc/oracle/config/OracleSourceConfigFactory.java b/...rg/apache/seatunnel/connectors/seatunnel/cdc/oracle/config/OracleSourceConfigFactory.java
@@ -21,6 +21,7 @@
 import org.apache.seatunnel.connectors.cdc.debezium.EmbeddedDatabaseHistory;
 
 import io.debezium.connector.oracle.OracleConnector;
+import lombok.extern.slf4j.Slf4j;
 
 import java.util.List;
 import java.util.Properties;
@@ -30,6 +31,7 @@
 import static org.apache.seatunnel.shade.com.google.common.base.Preconditions.checkNotNull;
 
 /** A factory to initialize {@link OracleSourceConfig}. */
+@Slf4j
 public class OracleSourceConfigFactory extends JdbcSourceConfigFactory {
 
     private static final long serialVersionUID = 1L;

diff --git a/...c/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/AbstractOracleCDCIT.java b/...c/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/AbstractOracleCDCIT.java
@@ -179,14 +179,12 @@ private static String formatValue(Object value, String columnName, Connection co
             }
             LocalDateTime localDateTime = zonedDateTime.toLocalDateTime();
             DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSS");
-            String t = localDateTime.format(formatter);
-            return t;
+            return localDateTime.format(formatter);
         }
 
         if (value instanceof LocalDateTime) {
             DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSS");
-            String t = ((LocalDateTime) value).format(formatter);
-            return t;
+            return ((LocalDateTime) value).format(formatter);
         }
         if (columnName.equalsIgnoreCase("VAL_NUMBER_1") && value instanceof BigDecimal) {
             BigDecimal bdValue = (BigDecimal) value;
@@ -206,9 +204,17 @@ private static String formatValue(Object value, String columnName, Connection co
         return value.toString();
     }
 
-    protected static void dropTable(String jdbcUrl, String tableName) {
-        try (Connection connection = getJdbcConnection(jdbcUrl, CONNECTOR_USER, CONNECTOR_PWD)) {
-            connection.createStatement().execute(String.format("DROP TABLE %s", tableName));
+    protected static void dropTable(String jdbcUrl, String schemaName, String tableName) {
+        try (Connection connection = getJdbcConnection(jdbcUrl, CONNECTOR_USER, CONNECTOR_PWD);
+                Statement statement = connection.createStatement()) {
+            ResultSet resultSet =
+                    statement.executeQuery(
+                            String.format(
+                                    "SELECT * FROM ALL_TABLES WHERE OWNER='%s' AND TABLE_NAME='%s'",
+                                    schemaName, tableName));
+            if (resultSet.next()) {
+                statement.execute(String.format("DROP TABLE %s.%s", schemaName, tableName));
+            }
         } catch (SQLException e) {
             throw new RuntimeException(e);
         }

diff --git a/...ava/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCWithSchemaChangeIT.java b/...ava/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCWithSchemaChangeIT.java
@@ -189,8 +189,8 @@ public void testOracleCdc2OracleWithSchemaEvolutionCase(TestContainer container)
     @TestTemplate
     public void testOracleCdc2MysqlWithSchemaEvolutionCase(TestContainer container)
             throws Exception {
-        dropTable(ORACLE_CONTAINER.getJdbcUrl(), SCEHMA_NAME + "." + SOURCE_TABLE1);
-        dropTable(ORACLE_CONTAINER.getJdbcUrl(), SCEHMA_NAME + "." + SOURCE_TABLE1 + "_SINK");
+        dropTable(ORACLE_CONTAINER.getJdbcUrl(), SCEHMA_NAME, SOURCE_TABLE1);
+        dropTable(ORACLE_CONTAINER.getJdbcUrl(), SCEHMA_NAME, SOURCE_TABLE1 + "_SINK");
         createAndInitialize("full_types", ADMIN_USER, ADMIN_PWD);
         CompletableFuture.runAsync(
                 () -> {