Add commit_time to the checkpoint in case of consistent streaming (#…

…250)
yugabyte · Jul 19, 2023 · ff44f79 · ff44f79
1 parent ad82d71
commit ff44f79
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 3 deletions.
diff --git a/src/main/java/io/debezium/connector/yugabytedb/YugabyteDBConsistentStreamingSource.java b/src/main/java/io/debezium/connector/yugabytedb/YugabyteDBConsistentStreamingSource.java
@@ -5,6 +5,7 @@
 import io.debezium.connector.yugabytedb.connection.OpId;
 import io.debezium.connector.yugabytedb.connection.ReplicationConnection;
 import io.debezium.connector.yugabytedb.connection.ReplicationMessage;
+import io.debezium.connector.yugabytedb.connection.ReplicationMessage.Operation;
 import io.debezium.connector.yugabytedb.connection.YugabyteDBConnection;
 import io.debezium.connector.yugabytedb.connection.pgproto.YbProtoReplicationMessage;
 import io.debezium.connector.yugabytedb.consistent.Merger;
@@ -239,7 +240,7 @@ protected void getChanges2(ChangeEventSourceContext context, YBPartition ybParti
                                         response.getIndex(),
                                         response.getKey(),
                                         response.getWriteId(),
-                                        response.getSnapshotTime());
+                                        response.getResp().getSafeHybridTime());
                                 offsetContext.updateWalPosition(part, finalOpid);
                                 offsetContext.updateWalSegmentIndex(part, response.getWalSegmentIndex());
                                 LOGGER.debug("The final opid for tablet {} is {}", part.getTabletId(), finalOpid);
@@ -323,7 +324,7 @@ private void dispatchMessage(YugabyteDBOffsetContext offsetContext, Map<String,
                 record.getFromOpId().getIndex(),
                 record.getFromOpId().getWriteIdKey().toByteArray(),
                 record.getFromOpId().getWriteId(),
-                snapshotTime);
+                record.getRowMessage().getCommitTime() - 1);
 
         if (message.isLastEventForLsn()) {
             lastCompletelyProcessedLsn = lsn;

diff --git a/src/main/java/io/debezium/connector/yugabytedb/YugabyteDBStreamingChangeEventSource.java b/src/main/java/io/debezium/connector/yugabytedb/YugabyteDBStreamingChangeEventSource.java
@@ -540,7 +540,7 @@ protected void getChanges2(ChangeEventSourceContext context,
                                     record.getFromOpId().getIndex(),
                                     record.getFromOpId().getWriteIdKey().toByteArray(),
                                     record.getFromOpId().getWriteId(),
-                                    record.getRowMessage().getCommitTime());
+                                    record.getRowMessage().getCommitTime() - 1);
 
                             if (message.isLastEventForLsn()) {
                                 lastCompletelyProcessedLsn = lsn;