From e809937590117b636b7015d03874a46d789a7772 Mon Sep 17 00:00:00 2001
From: mfvitale <mvitale@redhat.com>
Date: Tue, 26 Mar 2024 11:51:17 +0100
Subject: [PATCH] DBZ-7497 Add a configuration based snapshot modes
 configurable via connector properties

---
 .../io/debezium/connector/db2/Db2ConnectorConfig.java     | 5 +++++
 .../java/io/debezium/connector/db2/Db2ConnectorIT.java    | 8 ++++----
 2 files changed, 9 insertions(+), 4 deletions(-)

diff --git a/src/main/java/io/debezium/connector/db2/Db2ConnectorConfig.java b/src/main/java/io/debezium/connector/db2/Db2ConnectorConfig.java
index d31604a..ee981a0 100644
--- a/src/main/java/io/debezium/connector/db2/Db2ConnectorConfig.java
+++ b/src/main/java/io/debezium/connector/db2/Db2ConnectorConfig.java
@@ -86,6 +86,11 @@ public enum SnapshotMode implements EnumeratedValue {
          */
         WHEN_NEEDED("when_needed"),
 
+        /**
+         * Allows over snapshots by setting connectors properties prefixed with 'snapshot.mode.configuration.based'.
+         */
+        CONFIGURATION_BASED("configuration_based"),
+
         /**
          * Inject a custom snapshotter, which allows for more control over snapshots.
          */
diff --git a/src/test/java/io/debezium/connector/db2/Db2ConnectorIT.java b/src/test/java/io/debezium/connector/db2/Db2ConnectorIT.java
index b5154b2..7407317 100644
--- a/src/test/java/io/debezium/connector/db2/Db2ConnectorIT.java
+++ b/src/test/java/io/debezium/connector/db2/Db2ConnectorIT.java
@@ -472,7 +472,7 @@ public void testTableIncludeList() throws Exception {
         final int TABLES = 1;
         final int ID_START = 10;
         final Configuration config = TestHelper.defaultConfig()
-                .with(Db2ConnectorConfig.SNAPSHOT_MODE, SnapshotMode.SCHEMA_ONLY)
+                .with(Db2ConnectorConfig.SNAPSHOT_MODE, SnapshotMode.NO_DATA)
                 .with(Db2ConnectorConfig.TABLE_INCLUDE_LIST, "db2inst1.tableb")
                 .build();
         connection.execute(
@@ -737,7 +737,7 @@ public void testEmptySchemaWarningAfterApplyingFilters() throws Exception {
     @FixFor("DBZ-775")
     public void shouldConsumeEventsWithMaskedAndTruncatedColumns() throws Exception {
         final Configuration config = TestHelper.defaultConfig()
-                .with(Db2ConnectorConfig.SNAPSHOT_MODE, SnapshotMode.SCHEMA_ONLY)
+                .with(Db2ConnectorConfig.SNAPSHOT_MODE, SnapshotMode.NO_DATA)
                 .with("column.mask.with.12.chars", "DB2INST1.MASKED_HASHED_COLUMN_TABLE.NAME")
                 .with("column.mask.hash.SHA-256.with.salt.CzQMA0cB5K", "DB2INST1.MASKED_HASHED_COLUMN_TABLE.NAME2,DB2INST1.MASKED_HASHED_COLUMN_TABLE.NAME3")
                 .with("column.truncate.to.4.chars", "DB2INST1.TRUNCATED_COLUMN_TABLE.NAME")
@@ -792,7 +792,7 @@ record = tableB.get(0);
     @FixFor("DBZ-775")
     public void shouldRewriteIdentityKey() throws Exception {
         final Configuration config = TestHelper.defaultConfig()
-                .with(Db2ConnectorConfig.SNAPSHOT_MODE, SnapshotMode.SCHEMA_ONLY)
+                .with(Db2ConnectorConfig.SNAPSHOT_MODE, SnapshotMode.NO_DATA)
                 .with(Db2ConnectorConfig.MSG_KEY_COLUMNS, "(.*).tablea:id,cola")
                 .build();
 
@@ -823,7 +823,7 @@ public void shouldRewriteIdentityKey() throws Exception {
     @FixFor({ "DBZ-1916", "DBZ-1830" })
     public void shouldPropagateSourceTypeByDatatype() throws Exception {
         final Configuration config = TestHelper.defaultConfig()
-                .with(Db2ConnectorConfig.SNAPSHOT_MODE, SnapshotMode.SCHEMA_ONLY)
+                .with(Db2ConnectorConfig.SNAPSHOT_MODE, SnapshotMode.NO_DATA)
                 .with("datatype.propagate.source.type", ".+\\.NUMERIC,.+\\.VARCHAR,.+\\.DECIMAL,.+\\.REAL")
                 .build();