feat: use same server info for all map modes

pkg/batchmapper/examples/batchmap-flatmap/README.md

@@ -2,24 +2,7 @@
 
 An example User Defined Function that demonstrates how to write a batch map based `flatmap` User Defined Function.
 
-
-To start a vertex in batch map mode we need to add the annotations as following
-```yaml
-    - name: go-split
-      metadata:
-        annotations:
-          numaflow.numaproj.io/batch-map: "true"
-      scale:
-        min: 1
-      udf:
-        container:
-          # Split input message into an array with comma, see https://github.com/numaproj/numaflow-go/tree/main/pkg/batchmapper/examples/batchmap-flatmap
-          image: quay.io/numaio/numaflow-go/batch-map-flatmap:stable
-          imagePullPolicy: Always
-```
-
-
-Some considerations for batch map are as follows
+Some important considerations for batch map are as follows
 
 - The user will have to ensure that the BatchResponse is tagged with the correct request ID as this will be used by Numaflow for populating information required for system correctness like MessageID for the ISB deduplication.
 

pkg/batchmapper/examples/batchmap-flatmap/pipeline.yaml

@@ -8,9 +8,6 @@ spec:
       source:
         http: {}
     - name: go-split
-      metadata:
-        annotations:
-          numaflow.numaproj.io/batch-map: "true"
       scale:
         min: 1
       udf:

pkg/batchmapper/server.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/numaproj/numaflow-go/pkg"
 	batchmappb "github.com/numaproj/numaflow-go/pkg/apis/proto/batchmap/v1"
+	"github.com/numaproj/numaflow-go/pkg/info"
 	"github.com/numaproj/numaflow-go/pkg/shared"
 )
 
@@ -36,9 +37,13 @@ func (m *server) Start(ctx context.Context) error {
 	ctxWithSignal, stop := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM)
 	defer stop()
 
-	// write server info to the file
+	// create a server info to the file, we need to add metadata to ensure selection of the
+	// correct map mode, in this case batch map
+	serverInfo := info.GetDefaultServerInfo()
+	serverInfo.Metadata = map[string]string{info.MapModeKey: string(info.BatchMap)}
+
 	// start listening on unix domain socket
-	lis, err := shared.PrepareServer(m.opts.sockAddr, m.opts.serverInfoFilePath)
+	lis, err := shared.PrepareServer(m.opts.sockAddr, m.opts.serverInfoFilePath, serverInfo)
 	if err != nil {
 		return fmt.Errorf("failed to execute net.Listen(%q, %q): %v", uds, address, err)
 	}

pkg/batchmapper/service.go

@@ -19,7 +19,7 @@ const (
 	uds                   = "unix"
 	address               = "/var/run/numaflow/batchmap.sock"
 	defaultMaxMessageSize = 1024 * 1024 * 64
-	serverInfoFilePath    = "/var/run/numaflow/batchmapper-server-info"
+	serverInfoFilePath    = "/var/run/numaflow/mapper-server-info"
 )
 
 // Service implements the proto gen server interface and contains the map operation

pkg/info/server_info.go

@@ -13,7 +13,7 @@ import (
 
 var END = fmt.Sprintf("%U__END__", '\\') // U+005C__END__
 
-func GetSDKVersion() string {
+func getSDKVersion() string {
 	version := ""
 	info, ok := debug.ReadBuildInfo()
 	if !ok {
@@ -109,3 +109,11 @@ func Read(opts ...Option) (*ServerInfo, error) {
 	}
 	return info, nil
 }
+
+// GetDefaultServerInfo returns a ServerInfo object with the default fields populated for Go-SDK
+func GetDefaultServerInfo() *ServerInfo {
+	serverInfo := &ServerInfo{Protocol: UDS, Language: Go, MinimumNumaflowVersion: MinimumNumaflowVersion,
+		Version: getSDKVersion()}
+	return serverInfo
+
+}

pkg/info/server_info_test.go

@@ -22,8 +22,8 @@ func Test_getSDKVersion(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if got := GetSDKVersion(); got != tt.want {
-				t.Errorf("GetSDKVersion() = %v, want %v", got, tt.want)
+			if got := getSDKVersion(); got != tt.want {
+				t.Errorf("getSDKVersion() = %v, want %v", got, tt.want)
 			}
 		})
 	}

pkg/info/types.go

@@ -15,6 +15,17 @@ const (
 	Java   Language = "java"
 )
 
+type MapMode string
+
+const (
+	UnaryMap  MapMode = "unary-map"
+	StreamMap MapMode = "stream-map"
+	BatchMap  MapMode = "batch-map"
+)
+
+// MapModeKey is the key used in the server info metadata map to indicate which map mode is enabled.
+const MapModeKey = "MAP_MODE"
+
 // MinimumNumaflowVersion specifies the minimum Numaflow version required by the current SDK version
 const MinimumNumaflowVersion = "1.2.0-rc4"
 

pkg/mapper/server.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/numaproj/numaflow-go/pkg"
 	mappb "github.com/numaproj/numaflow-go/pkg/apis/proto/map/v1"
+	"github.com/numaproj/numaflow-go/pkg/info"
 	"github.com/numaproj/numaflow-go/pkg/shared"
 )
 
@@ -36,9 +37,13 @@ func (m *server) Start(ctx context.Context) error {
 	ctxWithSignal, stop := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM)
 	defer stop()
 
-	// write server info to the file
+	// create a server info to the file, we need to add metadata to ensure selection of the
+	// correct map mode, in this case unary map
+	serverInfo := info.GetDefaultServerInfo()
+	serverInfo.Metadata = map[string]string{info.MapModeKey: string(info.UnaryMap)}
+
 	// start listening on unix domain socket
-	lis, err := shared.PrepareServer(m.opts.sockAddr, m.opts.serverInfoFilePath)
+	lis, err := shared.PrepareServer(m.opts.sockAddr, m.opts.serverInfoFilePath, serverInfo)
 	if err != nil {
 		return fmt.Errorf("failed to execute net.Listen(%q, %q): %v", uds, address, err)
 	}

pkg/mapstreamer/server.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/numaproj/numaflow-go/pkg"
 	mapstreampb "github.com/numaproj/numaflow-go/pkg/apis/proto/mapstream/v1"
+	"github.com/numaproj/numaflow-go/pkg/info"
 	"github.com/numaproj/numaflow-go/pkg/shared"
 )
 
@@ -35,9 +36,13 @@ func (m *server) Start(ctx context.Context) error {
 	ctxWithSignal, stop := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM)
 	defer stop()
 
-	// write server info to the file
+	// create a server info to the file, we need to add metadata to ensure selection of the
+	// correct map mode, in this case streaming map
+	serverInfo := info.GetDefaultServerInfo()
+	serverInfo.Metadata = map[string]string{info.MapModeKey: string(info.StreamMap)}
+
 	// start listening on unix domain socket
-	lis, err := shared.PrepareServer(m.opts.sockAddr, m.opts.serverInfoFilePath)
+	lis, err := shared.PrepareServer(m.opts.sockAddr, m.opts.serverInfoFilePath, serverInfo)
 	if err != nil {
 		return fmt.Errorf("failed to execute net.Listen(%q, %q): %v", uds, address, err)
 	}

pkg/mapstreamer/service.go

@@ -12,7 +12,7 @@ const (
 	uds                   = "unix"
 	defaultMaxMessageSize = 1024 * 1024 * 64
 	address               = "/var/run/numaflow/mapstream.sock"
-	serverInfoFilePath    = "/var/run/numaflow/mapstreamer-server-info"
+	serverInfoFilePath    = "/var/run/numaflow/mapper-server-info"
 )
 
 // Service implements the proto gen server interface and contains the map

pkg/reducer/server.go

@@ -8,6 +8,7 @@ import (
 
 	numaflow "github.com/numaproj/numaflow-go/pkg"
 	reducepb "github.com/numaproj/numaflow-go/pkg/apis/proto/reduce/v1"
+	"github.com/numaproj/numaflow-go/pkg/info"
 	"github.com/numaproj/numaflow-go/pkg/shared"
 )
 
@@ -37,7 +38,7 @@ func (r *server) Start(ctx context.Context) error {
 
 	// write server info to the file
 	// start listening on unix domain socket
-	lis, err := shared.PrepareServer(r.opts.sockAddr, r.opts.serverInfoFilePath)
+	lis, err := shared.PrepareServer(r.opts.sockAddr, r.opts.serverInfoFilePath, info.GetDefaultServerInfo())
 	if err != nil {
 		return fmt.Errorf("failed to execute net.Listen(%q, %q): %v", uds, address, err)
 	}

pkg/reducestreamer/server.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/numaproj/numaflow-go/pkg"
 	reducepb "github.com/numaproj/numaflow-go/pkg/apis/proto/reduce/v1"
+	"github.com/numaproj/numaflow-go/pkg/info"
 	"github.com/numaproj/numaflow-go/pkg/shared"
 )
 
@@ -37,7 +38,7 @@ func (r *server) Start(ctx context.Context) error {
 
 	// write server info to the file
 	// start listening on unix domain socket
-	lis, err := shared.PrepareServer(r.opts.sockAddr, r.opts.serverInfoFilePath)
+	lis, err := shared.PrepareServer(r.opts.sockAddr, r.opts.serverInfoFilePath, info.GetDefaultServerInfo())
 	if err != nil {
 		return fmt.Errorf("failed to execute net.Listen(%q, %q): %v", uds, address, err)
 	}

pkg/sessionreducer/server.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/numaproj/numaflow-go/pkg"
 	sessionreducepb "github.com/numaproj/numaflow-go/pkg/apis/proto/sessionreduce/v1"
+	"github.com/numaproj/numaflow-go/pkg/info"
 	"github.com/numaproj/numaflow-go/pkg/shared"
 )
 
@@ -37,7 +38,7 @@ func (r *server) Start(ctx context.Context) error {
 
 	// write server info to the file
 	// start listening on unix domain socket
-	lis, err := shared.PrepareServer(r.opts.sockAddr, r.opts.serverInfoFilePath)
+	lis, err := shared.PrepareServer(r.opts.sockAddr, r.opts.serverInfoFilePath, info.GetDefaultServerInfo())
 	if err != nil {
 		return fmt.Errorf("failed to execute net.Listen(%q, %q): %v", uds, address, err)
 	}

pkg/shared/util.go

@@ -16,10 +16,13 @@ const (
 	uds = "unix"
 )
 
-func PrepareServer(sockAddr string, infoFilePath string) (net.Listener, error) {
+func PrepareServer(sockAddr string, infoFilePath string, serverInfo *info.ServerInfo) (net.Listener, error) {
+	// If serverInfo is not provided, then create a default server info instance.
+	if serverInfo == nil {
+		serverInfo = info.GetDefaultServerInfo()
+	}
 	// If infoFilePath is not empty, write the server info to the file.
 	if infoFilePath != "" {
-		serverInfo := &info.ServerInfo{Protocol: info.UDS, Language: info.Go, MinimumNumaflowVersion: info.MinimumNumaflowVersion, Version: info.GetSDKVersion()}
 		if err := info.Write(serverInfo, info.WithServerInfoFilePath(infoFilePath)); err != nil {
 			return nil, err
 		}

pkg/sideinput/server.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/numaproj/numaflow-go/pkg"
 	sideinputpb "github.com/numaproj/numaflow-go/pkg/apis/proto/sideinput/v1"
+	"github.com/numaproj/numaflow-go/pkg/info"
 	"github.com/numaproj/numaflow-go/pkg/shared"
 )
 
@@ -37,7 +38,7 @@ func (s *server) Start(ctx context.Context) error {
 	defer stop()
 
 	// start listening on unix domain socket
-	lis, err := shared.PrepareServer(s.opts.sockAddr, s.opts.serverInfoFilePath)
+	lis, err := shared.PrepareServer(s.opts.sockAddr, s.opts.serverInfoFilePath, info.GetDefaultServerInfo())
 	if err != nil {
 		return fmt.Errorf("failed to execute net.Listen(%q, %q): %v", uds, address, err)
 	}

pkg/sinker/server.go

@@ -9,6 +9,7 @@ import (
 
 	numaflow "github.com/numaproj/numaflow-go/pkg"
 	sinkpb "github.com/numaproj/numaflow-go/pkg/apis/proto/sink/v1"
+	"github.com/numaproj/numaflow-go/pkg/info"
 	"github.com/numaproj/numaflow-go/pkg/shared"
 )
 
@@ -36,7 +37,7 @@ func (s *sinkServer) Start(ctx context.Context) error {
 
 	// write server info to the file
 	// start listening on unix domain socket
-	lis, err := shared.PrepareServer(s.opts.sockAddr, s.opts.serverInfoFilePath)
+	lis, err := shared.PrepareServer(s.opts.sockAddr, s.opts.serverInfoFilePath, info.GetDefaultServerInfo())
 	if err != nil {
 		return fmt.Errorf("failed to execute net.Listen(%q, %q): %v", uds, address, err)
 	}

pkg/sourcer/server.go

@@ -9,6 +9,7 @@ import (
 
 	numaflow "github.com/numaproj/numaflow-go/pkg"
 	sourcepb "github.com/numaproj/numaflow-go/pkg/apis/proto/source/v1"
+	"github.com/numaproj/numaflow-go/pkg/info"
 	"github.com/numaproj/numaflow-go/pkg/shared"
 )
 
@@ -38,7 +39,7 @@ func (s *server) Start(ctx context.Context) error {
 
 	// write server info to the file
 	// start listening on unix domain socket
-	lis, err := shared.PrepareServer(s.opts.sockAddr, s.opts.serverInfoFilePath)
+	lis, err := shared.PrepareServer(s.opts.sockAddr, s.opts.serverInfoFilePath, info.GetDefaultServerInfo())
 	if err != nil {
 		return fmt.Errorf("failed to execute net.Listen(%q, %q): %v", uds, address, err)
 	}

pkg/sourcetransformer/server.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/numaproj/numaflow-go/pkg"
 	v1 "github.com/numaproj/numaflow-go/pkg/apis/proto/sourcetransform/v1"
+	"github.com/numaproj/numaflow-go/pkg/info"
 	"github.com/numaproj/numaflow-go/pkg/shared"
 )
 
@@ -37,7 +38,7 @@ func (m *server) Start(ctx context.Context) error {
 
 	// write server info to the file
 	// start listening on unix domain socket
-	lis, err := shared.PrepareServer(m.opts.sockAddr, m.opts.serverInfoFilePath)
+	lis, err := shared.PrepareServer(m.opts.sockAddr, m.opts.serverInfoFilePath, info.GetDefaultServerInfo())
 	if err != nil {
 		return fmt.Errorf("failed to execute net.Listen(%q, %q): %v", uds, address, err)
 	}