DBZ-7308 Align all snapshot mode on all connectors

src/main/java/io/debezium/connector/db2/Db2ConnectorConfig.java

@@ -8,6 +8,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigDef.Importance;
@@ -404,8 +405,8 @@ public SnapshotIsolationMode getSnapshotIsolationMode() {
         return this.snapshotIsolationMode;
     }
 
-    public SnapshotLockingMode getSnapshotLockingMode() {
-        return this.snapshotLockingMode;
+    public Optional<SnapshotLockingMode> getSnapshotLockingMode() {
+        return Optional.of(this.snapshotLockingMode);
     }
 
     public SnapshotMode getSnapshotMode() {

src/main/java/io/debezium/connector/db2/Db2ConnectorTask.java

@@ -21,8 +21,6 @@
 import io.debezium.config.Field;
 import io.debezium.connector.base.ChangeEventQueue;
 import io.debezium.connector.common.BaseSourceTask;
-import io.debezium.connector.db2.snapshot.Db2SnapshotLockProvider;
-import io.debezium.connector.db2.snapshot.Db2SnapshotterServiceProvider;
 import io.debezium.document.DocumentReader;
 import io.debezium.jdbc.DefaultMainConnectionProvidingConnectionFactory;
 import io.debezium.jdbc.MainConnectionProvidingConnectionFactory;
@@ -37,7 +35,6 @@
 import io.debezium.relational.TableId;
 import io.debezium.schema.SchemaFactory;
 import io.debezium.schema.SchemaNameAdjuster;
-import io.debezium.service.spi.ServiceRegistry;
 import io.debezium.snapshot.SnapshotterService;
 import io.debezium.spi.topic.TopicNamingStrategy;
 import io.debezium.util.Clock;
@@ -230,11 +227,4 @@ private static Configuration applyFetchSizeToJdbcConfig(Configuration config) {
         return config;
     }
 
-    @Override
-    protected void registerServiceProviders(ServiceRegistry serviceRegistry) {
-
-        super.registerServiceProviders(serviceRegistry);
-        serviceRegistry.registerServiceProvider(new Db2SnapshotLockProvider());
-        serviceRegistry.registerServiceProvider(new Db2SnapshotterServiceProvider());
-    }
 }

src/main/java/io/debezium/connector/db2/Db2SnapshotChangeEventSource.java

@@ -11,7 +11,6 @@
 import java.sql.Savepoint;
 import java.sql.Statement;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -32,7 +31,6 @@
 import io.debezium.relational.Tables;
 import io.debezium.schema.SchemaChangeEvent;
 import io.debezium.snapshot.SnapshotterService;
-import io.debezium.spi.snapshot.Snapshotter;
 import io.debezium.util.Clock;
 
 public class Db2SnapshotChangeEventSource extends RelationalSnapshotChangeEventSource<Db2Partition, Db2OffsetContext> {
@@ -51,41 +49,6 @@ public Db2SnapshotChangeEventSource(Db2ConnectorConfig connectorConfig, MainConn
         this.jdbcConnection = connectionFactory.mainConnection();
     }
 
-    @Override
-    public SnapshottingTask getSnapshottingTask(Db2Partition partition, Db2OffsetContext previousOffset) {
-
-        final Snapshotter snapshotter = snapshotterService.getSnapshotter();
-
-        List<String> dataCollectionsToBeSnapshotted = connectorConfig.getDataCollectionsToBeSnapshotted();
-        Map<String, String> snapshotSelectOverridesByTable = connectorConfig.getSnapshotSelectOverridesByTable().entrySet().stream()
-                .collect(Collectors.toMap(e -> e.getKey().identifier(), Map.Entry::getValue));
-
-        boolean offsetExists = previousOffset != null;
-        boolean snapshotInProgress = false;
-
-        if (offsetExists) {
-            snapshotInProgress = previousOffset.isSnapshotRunning();
-        }
-
-        if (offsetExists && !previousOffset.isSnapshotRunning()) {
-            LOGGER.info("A previous offset indicating a completed snapshot has been found. Neither schema nor data will be snapshotted.");
-        }
-
-        boolean shouldSnapshotSchema = snapshotter.shouldSnapshotSchema(offsetExists, snapshotInProgress);
-        boolean shouldSnapshotData = snapshotter.shouldSnapshotData(offsetExists, snapshotInProgress);
-
-        if (shouldSnapshotData && shouldSnapshotSchema) {
-            LOGGER.info("According to the connector configuration both schema and data will be snapshot.");
-        }
-        else if (shouldSnapshotSchema) {
-            LOGGER.info("According to the connector configuration only schema will be snapshot.");
-        }
-
-        return new SnapshottingTask(shouldSnapshotSchema, shouldSnapshotData,
-                dataCollectionsToBeSnapshotted, snapshotSelectOverridesByTable,
-                false);
-    }
-
     @Override
     protected SnapshotContext<Db2Partition, Db2OffsetContext> prepare(Db2Partition partition, boolean onDemand) {
         return new Db2SnapshotContext(partition, jdbcConnection.getRealDatabaseName(), onDemand);
@@ -125,7 +88,7 @@ else if (connectorConfig.getSnapshotIsolationMode() == SnapshotIsolationMode.EXC
                     }
 
                     Optional<String> lockingStatement = snapshotterService.getSnapshotLock().tableLockingStatement(connectorConfig.snapshotLockTimeout(),
-                            Set.of(quoteTableName(tableId)));
+                            quoteTableName(tableId));
 
                     if (lockingStatement.isPresent()) {
                         LOGGER.info("Locking table {}", tableId);

src/main/java/io/debezium/connector/db2/Db2StreamingChangeEventSource.java

@@ -115,11 +115,6 @@ public void init(Db2OffsetContext offsetContext) {
     public void execute(ChangeEventSourceContext context, Db2Partition partition, Db2OffsetContext offsetContext)
             throws InterruptedException {
 
-        if (!snapshotterService.getSnapshotter().shouldStream()) {
-            LOGGER.info("Streaming is not enabled in current configuration");
-            return;
-        }
-
         final Metronome metronome = Metronome.sleeper(pollInterval, clock);
         final Queue<Db2ChangeTable> schemaChangeCheckpoints = new PriorityQueue<>((x, y) -> x.getStopLsn().compareTo(y.getStopLsn()));
         try {

src/main/java/io/debezium/connector/db2/snapshot/lock/ExclusiveSnapshotLock.java

@@ -8,7 +8,6 @@
 import java.time.Duration;
 import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 
 import io.debezium.connector.db2.Db2ConnectorConfig;
 import io.debezium.snapshot.spi.SnapshotLock;
@@ -26,9 +25,7 @@ public void configure(Map<String, ?> properties) {
     }
 
     @Override
-    public Optional<String> tableLockingStatement(Duration lockTimeout, Set<String> tableIds) {
-
-        String tableId = tableIds.iterator().next(); // For Db2 we expect just one table at time.
+    public Optional<String> tableLockingStatement(Duration lockTimeout, String tableId) {
 
         return Optional.of("SELECT * FROM " + tableId + " WHERE 0=1 WITH CS");
     }

src/main/java/io/debezium/connector/db2/snapshot/lock/NoSnapshotLock.java

@@ -8,7 +8,6 @@
 import java.time.Duration;
 import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 
 import io.debezium.connector.db2.Db2ConnectorConfig;
 import io.debezium.snapshot.spi.SnapshotLock;
@@ -26,7 +25,7 @@ public void configure(Map<String, ?> properties) {
     }
 
     @Override
-    public Optional<String> tableLockingStatement(Duration lockTimeout, Set<String> tableIds) {
+    public Optional<String> tableLockingStatement(Duration lockTimeout, String tableId) {
 
         return Optional.empty();
     }