From 29fbdc03bfa03bfb9ef04d42dd318e6027a95bd1 Mon Sep 17 00:00:00 2001
From: Jun Zhao <jun.zhao1@brex.com>
Date: Thu, 16 Jun 2022 14:44:19 -0700
Subject: [PATCH] DBZ-5031 Stop skipping message if last received lsn is higher
 or equal to the startStreamingLsn in WalPositionLocation

---
 .../connector/postgresql/connection/WalPositionLocator.java     | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/connection/WalPositionLocator.java b/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/connection/WalPositionLocator.java
index 68525f6ea..1bb505025 100644
--- a/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/connection/WalPositionLocator.java
+++ b/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/connection/WalPositionLocator.java
@@ -130,7 +130,7 @@ public boolean skipMessage(Lsn lsn) {
         if (passMessages) {
             return false;
         }
-        if (startStreamingLsn == null || startStreamingLsn.equals(lsn)) {
+        if (startStreamingLsn == null || startStreamingLsn.compareTo(lsn) <= 0) {
             LOGGER.info("Message with LSN '{}' arrived, switching off the filtering", lsn);
             passMessages = true;
             return false;