debezium/dbz#1850 Add snapshot lock and query SPI implementations

- Add NoSnapshotLock and SharedSnapshotLock for snapshot locking
- Add SelectAllSnapshotQuery for snapshot query
- Register SPI services in META-INF/services
- Update AbstractStreamingAdapter and YashanDbConnectorTask
- Update pom.xml dependencies

Signed-off-by: Tlinian <2629731238@qq.com>

Author: Tlinian

pom.xml

@@ -167,6 +167,10 @@
             <groupId>io.debezium</groupId>
             <artifactId>debezium-util</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.debezium</groupId>
+            <artifactId>debezium-storage-kafka</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.debezium</groupId>
             <artifactId>debezium-connect-plugins</artifactId>

src/main/java/io/debezium/connector/yashandb/AbstractStreamingAdapter.java

@@ -70,8 +70,8 @@ protected boolean areSameTimestamp(Scn scn1, Scn scn2, YashanDbConnection connec
 
         final String query = "SELECT 1 FROM DUAL WHERE SCN_TO_TIMESTAMP(?)=SCN_TO_TIMESTAMP(?)";
         try (PreparedStatement ps = connection.connection().prepareStatement(query)) {
-            ps.setString(1, scn1.toString());
-            ps.setString(2, scn2.toString());
+            ps.setLong(1, scn1.longValue());
+            ps.setLong(2, scn2.longValue());
             ResultSet rs = ps.executeQuery();
             return rs.next();
         }

src/main/java/io/debezium/connector/yashandb/YashanDbConnectorTask.java

@@ -114,13 +114,12 @@ public ChangeEventSourceCoordinator<YashanDbPartition, YashanDbOffsetContext> st
         CustomConverterRegistry customConverterRegistry = connectorConfig.getServiceRegistry().tryGetService(CustomConverterRegistry.class);
         this.schema = new YashanDbDatabaseSchema(connectorConfig, valueConverters, defaultValueConverter, schemaNameAdjuster,
                 topicNamingStrategy, tableNameCaseSensitivity, customConverterRegistry, taskContext);
-        final SnapshotterService snapshotterService = connectorConfig.getServiceRegistry().tryGetService(SnapshotterService.class);
 
         Offsets<YashanDbPartition, YashanDbOffsetContext> previousOffsets = getPreviousOffsets(new YashanDbPartition.Provider(connectorConfig),
                 connectorConfig.getAdapter().getOffsetContextLoader());
 
         beanRegistryJdbcConnection = connectionFactory.newConnection();
-        // Manual Bean Registration
+        // Manual Bean Registration - must be done before any service that requires it is created (e.g. SnapshotterService)
         connectorConfig.getBeanRegistry().add(StandardBeanNames.CONFIGURATION, config);
         connectorConfig.getBeanRegistry().add(StandardBeanNames.CONNECTOR_CONFIG, connectorConfig);
         connectorConfig.getBeanRegistry().add(StandardBeanNames.DATABASE_SCHEMA, schema);
@@ -129,6 +128,8 @@ public ChangeEventSourceCoordinator<YashanDbPartition, YashanDbOffsetContext> st
         connectorConfig.getBeanRegistry().add(StandardBeanNames.OFFSETS, previousOffsets);
         connectorConfig.getBeanRegistry().add(StandardBeanNames.CDC_SOURCE_TASK_CONTEXT, taskContext);
 
+        final SnapshotterService snapshotterService = connectorConfig.getServiceRegistry().tryGetService(SnapshotterService.class);
+
         YashanDbPartition partition = previousOffsets.getTheOnlyPartition();
         YashanDbOffsetContext previousOffset = previousOffsets.getTheOnlyOffset();
 

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.connector.yashandb.snapshot.lock;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Optional;
+
+import io.debezium.annotation.ConnectorSpecific;
+import io.debezium.connector.yashandb.YashanDbConnector;
+import io.debezium.connector.yashandb.YashanDbConnectorConfig;
+import io.debezium.snapshot.spi.SnapshotLock;
+
+@ConnectorSpecific(connector = YashanDbConnector.class)
+public class NoSnapshotLock implements SnapshotLock {
+
+    @Override
+    public String name() {
+        return YashanDbConnectorConfig.SnapshotLockingMode.NONE.getValue();
+    }
+
+    @Override
+    public void configure(Map<String, ?> properties) {
+
+    }
+
+    @Override
+    public Optional<String> tableLockingStatement(Duration lockTimeout, String tableId) {
+
+        return Optional.empty();
+    }
+}

src/main/java/io/debezium/connector/yashandb/snapshot/lock/SharedSnapshotLock.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.connector.yashandb.snapshot.lock;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Optional;
+
+import io.debezium.annotation.ConnectorSpecific;
+import io.debezium.connector.yashandb.YashanDbConnector;
+import io.debezium.connector.yashandb.YashanDbConnectorConfig;
+import io.debezium.snapshot.spi.SnapshotLock;
+
+@ConnectorSpecific(connector = YashanDbConnector.class)
+public class SharedSnapshotLock implements SnapshotLock {
+
+    @Override
+    public String name() {
+        return YashanDbConnectorConfig.SnapshotLockingMode.SHARED.getValue();
+    }
+
+    @Override
+    public void configure(Map<String, ?> properties) {
+
+    }
+
+    @Override
+    public Optional<String> tableLockingStatement(Duration lockTimeout, String tableId) {
+
+        return Optional.of(String.format("LOCK TABLE %s IN SHARE MODE", tableId));
+    }
+}

src/main/java/io/debezium/connector/yashandb/snapshot/query/SelectAllSnapshotQuery.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.connector.yashandb.snapshot.query;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import io.debezium.annotation.ConnectorSpecific;
+import io.debezium.bean.StandardBeanNames;
+import io.debezium.bean.spi.BeanRegistry;
+import io.debezium.bean.spi.BeanRegistryAware;
+import io.debezium.config.CommonConnectorConfig;
+import io.debezium.connector.yashandb.YashanDbConnector;
+import io.debezium.connector.yashandb.YashanDbOffsetContext;
+import io.debezium.relational.RelationalSnapshotChangeEventSource;
+import io.debezium.snapshot.spi.SnapshotQuery;
+
+@ConnectorSpecific(connector = YashanDbConnector.class)
+public class SelectAllSnapshotQuery implements SnapshotQuery, BeanRegistryAware {
+
+    private BeanRegistry beanRegistry;
+
+    @Override
+    public String name() {
+        return CommonConnectorConfig.SnapshotQueryMode.SELECT_ALL.getValue();
+    }
+
+    @Override
+    public void configure(Map<String, ?> properties) {
+
+    }
+
+    @Override
+    public void injectBeanRegistry(BeanRegistry beanRegistry) {
+        this.beanRegistry = beanRegistry;
+    }
+
+    @Override
+    public Optional<String> snapshotQuery(String tableId, List<String> snapshotSelectColumns) {
+
+        final RelationalSnapshotChangeEventSource.RelationalSnapshotContext snapshotContext = beanRegistry.lookupByName(StandardBeanNames.SNAPSHOT_CONTEXT,
+                RelationalSnapshotChangeEventSource.RelationalSnapshotContext.class);
+        final YashanDbOffsetContext offset = (YashanDbOffsetContext) snapshotContext.offset;
+
+        final String snapshotOffset = offset.getScn().toString();
+        String columns = String.join(", ", snapshotSelectColumns);
+        assert snapshotOffset != null;
+        return Optional.of(String.format("SELECT %s FROM %s AS OF SCN %s", columns, tableId, snapshotOffset));
+
+    }
+}

src/main/resources/META-INF/services/io.debezium.converters.spi.CloudEventsProvider

@@ -0,0 +1 @@
+io.debezium.connector.yashandb.converters.YashanDbCloudEventsProvider

src/main/resources/META-INF/services/io.debezium.pipeline.source.spi.EventMetadataProvider

@@ -0,0 +1 @@
+io.debezium.connector.yashandb.YashanDbEventMetadataProvider

src/main/resources/META-INF/services/io.debezium.snapshot.spi.SnapshotLock

@@ -0,0 +1,2 @@
+io.debezium.connector.yashandb.snapshot.lock.NoSnapshotLock
+io.debezium.connector.yashandb.snapshot.lock.SharedSnapshotLock

src/main/resources/META-INF/services/io.debezium.snapshot.spi.SnapshotQuery

@@ -0,0 +1 @@
+io.debezium.connector.yashandb.snapshot.query.SelectAllSnapshotQuery