DBZ-7508 Exit from readChunk after createDataEventsForTable if snapshot is not running anymore

debezium-connector-mysql/src/test/java/io/debezium/connector/mysql/ReadOnlyIncrementalSnapshotIT.java

@@ -342,7 +342,6 @@ public void testStopSnapshotKafkaSignal() throws Exception {
                     return logInterceptor.containsMessage(tableRemoveMessage);
                 });
 
-        stopConnector();
     }
 
     @Test

debezium-core/src/main/java/io/debezium/pipeline/source/snapshot/incremental/AbstractIncrementalSnapshotChangeEventSource.java

@@ -388,8 +388,13 @@ protected void readChunk(P partition, OffsetContext offsetContext) throws Interr
                                 context.maximumKey().orElse(new Object[0]));
                     }
                 }
+
                 if (createDataEventsForTable(partition)) {
 
+                    if (!context.snapshotRunning()) { // A stop signal has been processed and window cleared.
+                        return;
+                    }
+
                     if (window.isEmpty()) {
                         LOGGER.info("No data returned by the query, incremental snapshotting of table '{}' finished",
                                 currentTableId);