[source-mongodb-v2] Force state update once a while

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/debezium/internals/DebeziumMessageProducer.kt

@@ -43,9 +43,7 @@ class DebeziumMessageProducer<T>(
     private val offsetManager: AirbyteFileOffsetBackingStore?
     private val targetPosition: CdcTargetPosition<T>
     private val schemaHistoryManager: Optional<AirbyteSchemaHistoryStorage>
-
-    private var shouldEmitStateMessage = false
-
+
     private val eventConverter: DebeziumEventConverter
 
     init {
@@ -68,7 +66,6 @@ class DebeziumMessageProducer<T>(
         previousCheckpointOffset.clear()
         previousCheckpointOffset.putAll(checkpointOffsetToSend)
         checkpointOffsetToSend.clear()
-        shouldEmitStateMessage = false
         return stateMessage
     }
 
@@ -94,12 +91,6 @@ class DebeziumMessageProducer<T>(
             }
         }
 
-        if (checkpointOffsetToSend.size == 1 && !message.isSnapshotEvent) {
-            if (targetPosition.isEventAheadOffset(checkpointOffsetToSend, message)) {
-                shouldEmitStateMessage = true
-            }
-        }
-
         return eventConverter.toAirbyteMessage(message)
     }
 
@@ -119,7 +110,7 @@ class DebeziumMessageProducer<T>(
     }
 
     override fun shouldEmitStateMessage(stream: ConfiguredAirbyteStream?): Boolean {
-        return shouldEmitStateMessage
+        return true
     }
 
     /**

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/source/relationaldb/state/SourceStateIterator.kt

@@ -27,7 +27,26 @@ open class SourceStateIterator<T>(
     private var lastCheckpoint: Instant = Instant.now()
 
     override fun computeNext(): AirbyteMessage? {
-        var iteratorHasNextValue: Boolean
+        var shouldForceUpdateState = shouldEmitStateMessage() &&
+            sourceStateMessageProducer.shouldEmitStateMessage(stream);
+        // check shouldForceUpdateState in case the sync is stuck; therefore we should send out
+        // state messages even if there is no records update. This could be useful to advance DBZ
+        // pointer on a less updated table.
+        if (shouldForceUpdateState) {
+            val stateMessage =
+                sourceStateMessageProducer.generateStateMessageAtCheckpoint(stream)
+            stateMessage!!.withSourceStats(
+                AirbyteStateStats().withRecordCount(recordCount.toDouble())
+            )
+            LOGGER.info { "sending state message, with count per stream: $streamRecordCount " }
+
+            recordCount = 0L
+            streamRecordCount.clear()
+
+            lastCheckpoint = Instant.now()
+            return AirbyteMessage().withType(AirbyteMessage.Type.STATE).withState(stateMessage)
+        }
+        val iteratorHasNextValue: Boolean
         try {
             iteratorHasNextValue = messageIterator.hasNext()
         } catch (ex: Exception) {
@@ -38,23 +57,6 @@ open class SourceStateIterator<T>(
             throw FailedRecordIteratorException(ex)
         }
         if (iteratorHasNextValue) {
-            if (
-                shouldEmitStateMessage() &&
-                    sourceStateMessageProducer.shouldEmitStateMessage(stream)
-            ) {
-                val stateMessage =
-                    sourceStateMessageProducer.generateStateMessageAtCheckpoint(stream)
-                stateMessage!!.withSourceStats(
-                    AirbyteStateStats().withRecordCount(recordCount.toDouble())
-                )
-                LOGGER.info { "sending state message, with count per stream: $streamRecordCount " }
-
-                recordCount = 0L
-                streamRecordCount.clear()
-
-                lastCheckpoint = Instant.now()
-                return AirbyteMessage().withType(AirbyteMessage.Type.STATE).withState(stateMessage)
-            }
             // Use try-catch to catch Exception that could occur when connection to the database
             // fails
             try {

airbyte-integrations/connectors/source-mongodb-v2/build.gradle

@@ -5,7 +5,7 @@ plugins {
 airbyteJavaConnector {
     cdkVersionRequired = '0.45.1'
     features = ['db-sources', 'datastore-mongo']
-    useLocalCdk = false
+    useLocalCdk = true
 }
 
 application {

airbyte-integrations/connectors/source-postgres/build.gradle

@@ -14,7 +14,7 @@ java {
 airbyteJavaConnector {
     cdkVersionRequired = '0.48.2'
     features = ['db-sources', 'datastore-postgres']
-    useLocalCdk = false
+    useLocalCdk = true
 }
 
 application {