Update StartedMongoSourceTask.java

fix if statement
mongodb · Oct 11, 2024 · 0e7c7f9 · 0e7c7f9
1 parent a0c81bf
commit 0e7c7f9
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/src/main/java/com/mongodb/kafka/connect/source/StartedMongoSourceTask.java b/src/main/java/com/mongodb/kafka/connect/source/StartedMongoSourceTask.java
@@ -683,7 +683,7 @@ private BsonDocument getResumeToken(final MongoSourceConfig sourceConfig) {
       invalidatedCursor = false;
     } else {
       Map<String, Object> offset = getOffset(sourceTaskContextAccessor.get(), sourceConfig);
-      if (offset != null && offset.containsKey(ID_FIELD) && !offset.containsKey(COPY_KEY)) {
+      if (offset != null && offset.containsKey(ID_FIELD) && !(offset.containsKey(COPY_KEY) && Boolean.parseBoolean(offset.get(COPY_KEY).toString())))) {
         resumeToken = BsonDocument.parse((String) offset.get(ID_FIELD));
         if (offset.containsKey(HEARTBEAT_KEY)) {
           LOGGER.info("Resume token from heartbeat: {}", resumeToken);