chore: skip publishing the last empty struct when generator stops (#2327

)
numaproj · Jan 13, 2025 · 2922173 · 2922173
1 parent ade70d8
commit 2922173
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/pkg/sources/generator/tickgen.go b/pkg/sources/generator/tickgen.go
@@ -197,7 +197,11 @@ loop:
 		// since the Read call is blocking, and runs in an infinite loop,
 		// we implement Read With Wait semantics
 		select {
-		case r := <-mg.srcChan:
+		case r, ok := <-mg.srcChan:
+			if !ok {
+				mg.logger.Info("All the messages have been read. returning.")
+				break loop
+			}
 			msgs = append(msgs, mg.newReadMessage(r.key, r.data, r.offset, r.ts))
 		case <-timeout:
 			break loop