working cursor based incremental

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/source/jdbc/AbstractJdbcSource.kt

@@ -805,4 +805,26 @@ abstract class AbstractJdbcSource<Datatype>(
             return result
         }
     }
+
+    override fun createReadIterator(
+        database: JdbcDatabase,
+        airbyteStream: ConfiguredAirbyteStream,
+        catalog: ConfiguredAirbyteCatalog?,
+        table: TableInfo<CommonField<Datatype>>,
+        stateManager: StateManager?,
+        emittedAt: Instant
+    ): AutoCloseableIterator<AirbyteMessage> {
+        val iterator = super.createReadIterator(
+            database,
+            airbyteStream,
+            catalog,
+            table,
+            stateManager,
+            emittedAt
+        )
+        return when (airbyteStream.syncMode) {
+            SyncMode.INCREMENTAL -> augmentWithStreamStatus(airbyteStream, iterator)
+            else -> iterator
+        }
+    }
 }

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

@@ -416,7 +416,7 @@ protected constructor(driverClassName: String) :
      * @param emittedAt Time when data was emitted from the Source database
      * @return
      */
-    private fun createReadIterator(
+    protected open fun createReadIterator(
         database: Database,
         airbyteStream: ConfiguredAirbyteStream,
         catalog: ConfiguredAirbyteCatalog?,

airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/MySqlSource.java

@@ -491,6 +491,7 @@ public List<AutoCloseableIterator<AirbyteMessage>> getIncrementalIterators(final
                 tableNameToTable,
                 cursorBasedStateManager, emittedAt));
 
+
         return Stream.of(initialLoadIterator, cursorBasedIterator).flatMap(Collection::stream).collect(Collectors.toList());
       }
     }