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,15 +8,12 @@ spec:
       source:
         http: {}
     - 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
+          image: quay.io/numaio/numaflow-go/map-flatmap:stable
           imagePullPolicy: Always
     - name: go-udsink
       scale:

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,16 @@ 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
+	// write 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.MapModeMetadata: string(info.BatchMap)}
+	if err := info.Write(serverInfo, info.WithServerInfoFilePath(m.opts.serverInfoFilePath)); err != nil {
+		return err
+	}
+
 	// start listening on unix domain socket
-	lis, err := shared.PrepareServer(m.opts.sockAddr, m.opts.serverInfoFilePath)
+	lis, err := shared.PrepareServer(m.opts.sockAddr, "")
 	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

@@ -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/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"
+)
+
+// MapModeMetadata field is used to indicate which map mode is enabled
+const MapModeMetadata = "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,16 @@ 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
+	// write 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.MapModeMetadata: string(info.UnaryMap)}
+	if err := info.Write(serverInfo, info.WithServerInfoFilePath(m.opts.serverInfoFilePath)); err != nil {
+		return err
+	}
+
 	// start listening on unix domain socket
-	lis, err := shared.PrepareServer(m.opts.sockAddr, m.opts.serverInfoFilePath)
+	lis, err := shared.PrepareServer(m.opts.sockAddr, "")
 	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,16 @@ 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
+	// write 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.MapModeMetadata: string(info.StreamMap)}
+	if err := info.Write(serverInfo, info.WithServerInfoFilePath(m.opts.serverInfoFilePath)); err != nil {
+		return err
+	}
+
 	// start listening on unix domain socket
-	lis, err := shared.PrepareServer(m.opts.sockAddr, m.opts.serverInfoFilePath)
+	lis, err := shared.PrepareServer(m.opts.sockAddr, "")
 	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/shared/util.go

@@ -19,7 +19,7 @@ const (
 func PrepareServer(sockAddr string, infoFilePath string) (net.Listener, error) {
 	// 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()}
+		serverInfo := info.GetDefaultServerInfo()
 		if err := info.Write(serverInfo, info.WithServerInfoFilePath(infoFilePath)); err != nil {
 			return nil, err
 		}