DBZ-7015 Make slot validation check opt-in

couchbase · Oct 16, 2023 · f9b07e6 · f9b07e6
1 parent 7fd68a5
commit f9b07e6
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 3 deletions.
diff --git a/...ctor-postgres/src/main/java/io/debezium/connector/postgresql/PostgresConnectorConfig.java b/...ctor-postgres/src/main/java/io/debezium/connector/postgresql/PostgresConnectorConfig.java
@@ -523,6 +523,16 @@ public static SchemaRefreshMode parse(String value) {
                     "Whether or not to drop the logical replication slot when the connector finishes orderly. " +
                             "By default the replication is kept so that on restart progress can resume from the last recorded location");
 
+    public static final Field SLOT_SEEK_TO_KNOWN_OFFSET = Field.createInternal("slot.seek.to.known.offset.on.start")
+            .withDisplayName("Seek to last known offset on the replication slot")
+            .withType(Type.BOOLEAN)
+            .withGroup(Field.createGroupEntry(Field.Group.CONNECTION_ADVANCED_REPLICATION, 3))
+            .withDefault(false)
+            .withImportance(Importance.HIGH)
+            .withDescription(
+                    "Whether or not to seek to the last known offset on the replication slot." +
+                            "Enabling this option results in startup failure if the slot is re-created instead of data loss.");
+
     public static final Field PUBLICATION_NAME = Field.create("publication.name")
             .withDisplayName("Publication")
             .withType(Type.STRING)
@@ -952,6 +962,10 @@ protected boolean dropSlotOnStop() {
         return getConfig().getBoolean(DROP_SLOT_ON_STOP);
     }
 
+    public boolean slotSeekToKnownOffsetOnStart() {
+        return getConfig().getBoolean(SLOT_SEEK_TO_KNOWN_OFFSET);
+    }
+
     public String publicationName() {
         return getConfig().getString(PUBLICATION_NAME);
     }

diff --git a/.../main/java/io/debezium/connector/postgresql/connection/PostgresReplicationConnection.java b/.../main/java/io/debezium/connector/postgresql/connection/PostgresReplicationConnection.java
@@ -404,7 +404,9 @@ public ReplicationStream startStreaming(Lsn offset, WalPositionLocator walPositi
         int tryCount = 0;
         while (true) {
             try {
-                validateSlotIsInExpectedState(walPosition);
+                if (connectorConfig.slotSeekToKnownOffsetOnStart()) {
+                    validateSlotIsInExpectedState(walPosition);
+                }
                 return createReplicationStream(lsn, walPosition);
             }
             catch (Exception e) {

diff --git a/...onnector-postgres/src/test/java/io/debezium/connector/postgresql/PostgresConnectorIT.java b/...onnector-postgres/src/test/java/io/debezium/connector/postgresql/PostgresConnectorIT.java
@@ -3317,7 +3317,8 @@ public void shouldStopConnectorOnSlotRecreation() throws InterruptedException {
 
         configBuilder = TestHelper.defaultConfig()
                 .with(PostgresConnectorConfig.SNAPSHOT_MODE, SnapshotMode.NEVER.name())
-                .with(PostgresConnectorConfig.DROP_SLOT_ON_STOP, Boolean.TRUE);
+                .with(PostgresConnectorConfig.DROP_SLOT_ON_STOP, Boolean.TRUE)
+                .with(PostgresConnectorConfig.SLOT_SEEK_TO_KNOWN_OFFSET, Boolean.TRUE);
 
         start(PostgresConnector.class, configBuilder.build());
         Awaitility.await().atMost(TestHelper.waitTimeForRecords() * 5, TimeUnit.SECONDS)
@@ -3349,7 +3350,8 @@ public void shouldSeekToCorrectOffset() throws InterruptedException {
 
         configBuilder = TestHelper.defaultConfig()
                 .with(PostgresConnectorConfig.SNAPSHOT_MODE, SnapshotMode.NEVER.name())
-                .with(PostgresConnectorConfig.DROP_SLOT_ON_STOP, Boolean.TRUE);
+                .with(PostgresConnectorConfig.DROP_SLOT_ON_STOP, Boolean.TRUE)
+                .with(PostgresConnectorConfig.SLOT_SEEK_TO_KNOWN_OFFSET, Boolean.TRUE);
 
         start(PostgresConnector.class, configBuilder.build());
         consumeRecordsByTopic(1);