[oracle] close the oracle connection after the backfill task finished

Author: ruanhang1993

flink-connector-oracle-cdc/src/main/java/com/ververica/cdc/connectors/oracle/source/reader/fetch/OracleScanFetchTask.java

@@ -37,7 +37,6 @@
 import io.debezium.pipeline.source.spi.ChangeEventSource;
 import io.debezium.pipeline.source.spi.SnapshotProgressListener;
 import io.debezium.pipeline.spi.ChangeRecordEmitter;
-import io.debezium.pipeline.spi.OffsetContext;
 import io.debezium.pipeline.spi.SnapshotResult;
 import io.debezium.relational.RelationalSnapshotChangeEventSource;
 import io.debezium.relational.SnapshotChangeRecordEmitter;
@@ -130,19 +129,23 @@ public void execute(Context context) throws Exception {
         }
         // execute redoLog read task
         if (snapshotResult.isCompletedOrSkipped()) {
-            final RedoLogSplitReadTask backfillBinlogReadTask =
-                    createBackfillRedoLogReadTask(backfillBinlogSplit, sourceFetchContext);
-
-            final LogMinerOracleOffsetContextLoader loader =
-                    new LogMinerOracleOffsetContextLoader(
-                            ((OracleSourceFetchTaskContext) context).getDbzConnectorConfig());
-            final OracleOffsetContext oracleOffsetContext =
-                    loader.load(backfillBinlogSplit.getStartingOffset().getOffset());
-            backfillBinlogReadTask.execute(
-                    new SnapshotBinlogSplitChangeEventSourceContext(),
-                    sourceFetchContext.getPartition(),
-                    oracleOffsetContext);
-            taskRunning = false;
+            try (OracleConnection connection =
+                    createOracleConnection(
+                            sourceFetchContext.getSourceConfig().getDbzConfiguration())) {
+                final RedoLogSplitReadTask backfillBinlogReadTask =
+                        createBackfillRedoLogReadTask(
+                                backfillBinlogSplit, sourceFetchContext, connection);
+                final LogMinerOracleOffsetContextLoader loader =
+                        new LogMinerOracleOffsetContextLoader(
+                                ((OracleSourceFetchTaskContext) context).getDbzConnectorConfig());
+                final OracleOffsetContext oracleOffsetContext =
+                        loader.load(backfillBinlogSplit.getStartingOffset().getOffset());
+                backfillBinlogReadTask.execute(
+                        new SnapshotBinlogSplitChangeEventSourceContext(),
+                        sourceFetchContext.getPartition(),
+                        oracleOffsetContext);
+                taskRunning = false;
+            }
         } else {
             taskRunning = false;
             throw new IllegalStateException(
@@ -162,11 +165,9 @@ private StreamSplit createBackfillRedoLogSplit(
     }
 
     private RedoLogSplitReadTask createBackfillRedoLogReadTask(
-            StreamSplit backfillBinlogSplit, OracleSourceFetchTaskContext context) {
-        OracleConnectorConfig oracleConnectorConfig =
-                context.getSourceConfig().getDbzConnectorConfig();
-        final OffsetContext.Loader<OracleOffsetContext> loader =
-                new LogMinerOracleOffsetContextLoader(oracleConnectorConfig);
+            StreamSplit backfillBinlogSplit,
+            OracleSourceFetchTaskContext context,
+            OracleConnection connection) {
         // we should only capture events for the current table,
         // otherwise, we may can't find corresponding schema
         Configuration dezConf =
@@ -180,7 +181,7 @@ private RedoLogSplitReadTask createBackfillRedoLogReadTask(
         // task to read binlog and backfill for current split
         return new RedoLogSplitReadTask(
                 new OracleConnectorConfig(dezConf),
-                createOracleConnection(context.getSourceConfig().getDbzConfiguration()),
+                connection,
                 context.getDispatcher(),
                 context.getErrorHandler(),
                 context.getDatabaseSchema(),