fix: incorrect side inputs watch logic

cmd/commands/root.go

@@ -48,5 +48,5 @@ func init() {
 	rootCmd.AddCommand(NewWebhookCommand())
 	rootCmd.AddCommand(NewSideInputsInitCommand())
 	rootCmd.AddCommand(NewSideInputsManagerCommand())
-	rootCmd.AddCommand(NewSideInputsWatcherCommand())
+	rootCmd.AddCommand(NewSideInputsSynchronizerCommand())
 }

cmd/commands/side_inputs_watcher.go → cmd/commands/side_inputs_synchronizer.go

@@ -28,15 +28,15 @@ import (
 	"github.com/numaproj/numaflow/pkg/sideinputs/synchronizer"
 )
 
-func NewSideInputsWatcherCommand() *cobra.Command {
+func NewSideInputsSynchronizerCommand() *cobra.Command {
 	var (
 		isbSvcType      string
 		sideInputsStore string
 		sideInputs      []string
 	)
 	command := &cobra.Command{
-		Use:   "side-inputs-watcher",
-		Short: "Start the Side Inputs Watcher",
+		Use:   "side-inputs-synchronizer",
+		Short: "Start the Side Inputs Synchronizer",
 		RunE: func(cmd *cobra.Command, args []string) error {
 			pipelineName, defined := os.LookupEnv(dfv1.EnvPipelineName)
 			if !defined {
@@ -47,7 +47,7 @@ func NewSideInputsWatcherCommand() *cobra.Command {
 				return fmt.Errorf("no side inputs are defined for this vertex")
 			}
 
-			logger := logging.NewLogger().Named("side-inputs-watcher").With("pipeline", pipelineName)
+			logger := logging.NewLogger().Named("side-inputs-synchronizer").With("pipeline", pipelineName)
 			ctx := logging.WithLogger(signals.SetupSignalHandler(), logger)
 			sideInputsWatcher := synchronizer.NewSideInputsSynchronizer(dfv1.ISBSvcType(isbSvcType), pipelineName, sideInputsStore, sideInputs)
 			return sideInputsWatcher.Start(ctx)

pkg/apis/numaflow/v1alpha1/const.go

@@ -71,7 +71,7 @@ const (
 	CtrUdtransformer     = "transformer"
 	CtrUdSideInput       = "udsi"
 	CtrInitSideInputs    = "init-side-inputs"
-	CtrSideInputsWatcher = "side-inputs-watcher"
+	CtrSideInputsWatcher = "side-inputs-synchronizer"
 
 	// components
 	ComponentISBSvc           = "isbsvc"

pkg/apis/numaflow/v1alpha1/vertex_types.go

@@ -277,7 +277,7 @@ func (v Vertex) GetPodSpec(req GetVertexPodSpecReq) (*corev1.PodSpec, error) {
 			Image:           req.Image,
 			ImagePullPolicy: req.PullPolicy,
 			Resources:       standardResources,
-			Args:            []string{"side-inputs-watcher", "--isbsvc-type=" + string(req.ISBSvcType), "--side-inputs-store=" + req.SideInputsStoreName, "--side-inputs=" + strings.Join(v.Spec.SideInputs, ",")},
+			Args:            []string{"side-inputs-synchronizer", "--isbsvc-type=" + string(req.ISBSvcType), "--side-inputs-store=" + req.SideInputsStoreName, "--side-inputs=" + strings.Join(v.Spec.SideInputs, ",")},
 		}
 		sideInputsWatcher.Env = append(sideInputsWatcher.Env, v.commonEnvs()...)
 		if x := v.Spec.SideInputsContainerTemplate; x != nil {

pkg/shared/kvs/jetstream/kv_store.go

@@ -202,7 +202,7 @@ func (jss *jetStreamStore) Watch(ctx context.Context) (<-chan kvs.KVEntry, <-cha
 					continue
 				}
 				if value == nil {
-					jss.log.Infow("watcher initialization and subscription got nil value")
+					jss.log.Infow("Watcher initialization and subscription got nil value")
 					continue
 				}
 				jss.previousFetchTime = value.Created()

pkg/sideinputs/initializer/initializer.go

@@ -110,9 +110,9 @@ func startSideInputInitializer(ctx context.Context, store kvs.KVStorer, mountPat
 			m[value.Key()] = value.Value()
 			// Wait for the data to be ready in the side input store, and then copy it to the disk
 			if gotAllSideInputVals(sideInputs, m) {
-				for sideInput := range m {
+				for _, sideInput := range sideInputs {
 					p := path.Join(mountPath, sideInput)
-					log.Info("Initializing Side Input data for %q", p)
+					log.Infof("Initializing Side Input data for %q", sideInput)
 					err := utils.UpdateSideInputFile(ctx, p, m[sideInput])
 					if err != nil {
 						return fmt.Errorf("failed to update Side Input value, %w", err)
@@ -132,9 +132,6 @@ func startSideInputInitializer(ctx context.Context, store kvs.KVStorer, mountPat
 // gotAllSideInputVals checks if values for all side-inputs
 // have been received from the KV bucket
 func gotAllSideInputVals(sideInputs []string, m map[string][]byte) bool {
-	if len(sideInputs) != len(m) {
-		return false
-	}
 	for _, sideInput := range sideInputs {
 		if _, ok := m[sideInput]; !ok {
 			return false

pkg/sideinputs/synchronizer/synchronizer.go

@@ -62,7 +62,7 @@ func (sis *sideInputsSynchronizer) Start(ctx context.Context) error {
 	)
 
 	log := logging.FromContext(ctx)
-	log.Infow("Starting Side Inputs Watcher", zap.Strings("sideInputs", sis.sideInputs))
+	log.Infow("Starting Side Inputs Synchronizer", zap.Strings("sideInputs", sis.sideInputs))
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
@@ -98,7 +98,7 @@ func startSideInputSynchronizer(ctx context.Context, watch kvs.KVStorer, mountPa
 	for {
 		select {
 		case <-stopped:
-			log.Info("Side Input watcher stopped")
+			log.Info("Side Input Synchronizer stopped")
 			return
 		case value := <-watchCh:
 			if value == nil {