[Improve][Oracle-CDC] Support ReadOnlyLogWriterFlushStrategy (#8912)

Author: hailin0

Diff for: seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/io/debezium/connector/oracle/logminer/LogMinerStreamingChangeEventSource.java

@@ -17,6 +17,8 @@
 
 package io.debezium.connector.oracle.logminer;
 
+import org.apache.seatunnel.connectors.seatunnel.cdc.oracle.config.OracleSourceConfigFactory;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -33,6 +35,7 @@
 import io.debezium.connector.oracle.logminer.logwriter.CommitLogWriterFlushStrategy;
 import io.debezium.connector.oracle.logminer.logwriter.LogWriterFlushStrategy;
 import io.debezium.connector.oracle.logminer.logwriter.RacCommitLogWriterFlushStrategy;
+import io.debezium.connector.oracle.logminer.logwriter.ReadOnlyLogWriterFlushStrategy;
 import io.debezium.connector.oracle.logminer.processor.LogMinerEventProcessor;
 import io.debezium.jdbc.JdbcConfiguration;
 import io.debezium.pipeline.ErrorHandler;
@@ -1012,7 +1015,12 @@ private boolean isTableAllColumnsSupplementalLoggingEnabled(
      *
      * @return the strategy to be used to flush Oracle's LGWR process, never {@code null}.
      */
-    private LogWriterFlushStrategy resolveFlushStrategy() {
+    public LogWriterFlushStrategy resolveFlushStrategy() {
+        if (connectorConfig
+                .getConfig()
+                .getBoolean(OracleSourceConfigFactory.LOG_MINING_READONLY_KEY, false)) {
+            return new ReadOnlyLogWriterFlushStrategy();
+        }
         if (connectorConfig.isRacSystem()) {
             return new RacCommitLogWriterFlushStrategy(
                     connectorConfig, jdbcConfiguration, streamingMetrics);

Diff for: seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/io/debezium/connector/oracle/logminer/logwriter/ReadOnlyLogWriterFlushStrategy.java

@@ -0,0 +1,38 @@
+/*
+ * 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 io.debezium.connector.oracle.logminer.logwriter;
+
+import io.debezium.DebeziumException;
+import io.debezium.connector.oracle.Scn;
+
+public class ReadOnlyLogWriterFlushStrategy implements LogWriterFlushStrategy {
+    @Override
+    public String getHost() {
+        throw new DebeziumException("Not applicable when using read-only flushing strategy");
+    }
+
+    @Override
+    public void flush(Scn currentScn) throws InterruptedException {
+        // no operation
+    }
+
+    @Override
+    public void close() throws Exception {
+        // no operation
+    }
+}

Diff for: seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/config/OracleSourceConfigFactory.java

@@ -41,6 +41,7 @@ public class OracleSourceConfigFactory extends JdbcSourceConfigFactory {
     public static final String SCHEMA_CHANGE_KEY = "include.schema.changes";
     public static final String LOG_MINING_STRATEGY_KEY = "log.mining.strategy";
     public static final String LOG_MINING_STRATEGY_DEFAULT = "online_catalog";
+    public static final String LOG_MINING_READONLY_KEY = "log.mining.read.only";
 
     private List<String> schemaList;
 
@@ -106,6 +107,7 @@ public OracleSourceConfig create(int subtask) {
         props.setProperty("connect.timeout.ms", String.valueOf(connectTimeoutMillis));
         // disable tombstones
         props.setProperty("tombstones.on.delete", String.valueOf(false));
+        props.setProperty(LOG_MINING_READONLY_KEY, "true");
 
         if (originUrl != null) {
             props.setProperty("database.url", originUrl);

Diff for: seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/test/java/io/debezium/connector/oracle/logminer/logwriter/ReadOnlyLogWriterFlushStrategyTest.java

@@ -0,0 +1,54 @@
+/*
+ * 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 io.debezium.connector.oracle.logminer.logwriter;
+
+import org.apache.seatunnel.connectors.seatunnel.cdc.oracle.config.OracleSourceConfigFactory;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import io.debezium.DebeziumException;
+import io.debezium.config.Configuration;
+import io.debezium.connector.oracle.OracleConnectorConfig;
+import io.debezium.connector.oracle.logminer.LogMinerStreamingChangeEventSource;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ReadOnlyLogWriterFlushStrategyTest {
+
+    @Test
+    void returnsReadOnlyLogWriterFlushStrategyWhenReadOnlyKeyIsTrue() throws Exception {
+        OracleConnectorConfig config = mock(OracleConnectorConfig.class);
+        Configuration configuration = mock(Configuration.class);
+        when(config.getConfig()).thenReturn(configuration);
+        when(configuration.getBoolean(OracleSourceConfigFactory.LOG_MINING_READONLY_KEY, false))
+                .thenReturn(true);
+
+        LogMinerStreamingChangeEventSource source =
+                new LogMinerStreamingChangeEventSource(
+                        config, null, null, null, null, null, null, null);
+        LogWriterFlushStrategy strategy = source.resolveFlushStrategy();
+        assertTrue(strategy instanceof ReadOnlyLogWriterFlushStrategy);
+
+        Assertions.assertThrows(DebeziumException.class, () -> strategy.getHost());
+        strategy.flush(null);
+        strategy.close();
+    }
+}