chore: skip publishing the last empty struct when generator stops

pkg/shared/util/struct.go

@@ -0,0 +1,15 @@
+package util
+
+import "reflect"
+
+// IsZeroStruct checks if a struct is zero value.
+func IsZeroStruct(v interface{}) bool {
+	val := reflect.ValueOf(v)
+	for i := 0; i < val.NumField(); i++ {
+		field := val.Field(i)
+		if !field.IsZero() {
+			return false
+		}
+	}
+	return true
+}

pkg/shared/util/struct_test.go

@@ -0,0 +1,41 @@
+package util
+
+import "testing"
+
+func TestIsZeroStruct(t *testing.T) {
+	type args struct {
+		v interface{}
+	}
+	tests := []struct {
+		name string
+		args args
+		want bool
+	}{
+		{
+			name: "zero_struct",
+			args: args{v: struct {
+				Name string
+				Age  int
+			}{}},
+			want: true,
+		},
+		{
+			name: "non_zero_struct",
+			args: args{v: struct {
+				Name string
+				Age  int
+			}{
+				Name: "John",
+				Age:  25,
+			}},
+			want: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := IsZeroStruct(tt.args.v); got != tt.want {
+				t.Errorf("IsZeroStruct() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

pkg/sources/generator/tickgen.go

@@ -32,6 +32,7 @@ import (
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
+	"github.com/numaproj/numaflow/pkg/shared/util"
 	"github.com/numaproj/numaflow/pkg/sources/sourcer"
 )
 
@@ -198,6 +199,13 @@ loop:
 		// we implement Read With Wait semantics
 		select {
 		case r := <-mg.srcChan:
+			// when the channel is closed and empty, go will still read from the channel once and return the zero value.
+			// exclude the zero value from being passed to the next vertex,
+			// ensuring all messages produced by generator follow the same data schema.
+			if util.IsZeroStruct(r) {
+				mg.logger.Info("Zero value read from the generator channel, skipping...")
+				continue
+			}
 			msgs = append(msgs, mg.newReadMessage(r.key, r.data, r.offset, r.ts))
 		case <-timeout:
 			break loop