Repository: numaproj/numaflow
Pull request #153 (2022-08-30 20:40:59)

chore: use string as the type of key
Signed-off-by: Derek Wang <whynowy@gmail.com>

Explain what this PR does.

<!--

Before you push your changes:

* Run `make pre-push -B` to fix codegen and lint problems.

Then, you MUST:

* Sign-off your commit (otherwise the DCO check will fail).
* Use [a conventional commit message](https://www.conventionalcommits.org/en/v1.0.0/) (otherwise the commit message check will fail).

If you did not do this, reset all your commit and replace them with a single commit:

```
git reset HEAD~1 ;# change 1 to how many commits you made
git commit --signoff -m 'feat: my feat. Fixes #1234'
```

When creating your PR: 

* Make sure that "Fixes #" or "Closes #" is in both the PR title (for release notes) and description (to automatically link and close the issue).
* Say how you tested your changes. If you changed the UI, attach screenshots.
* Set your PR as a draft initially.
* Your PR needs to pass the required checks before it can be approved. 
* Once required tests have passed, mark your PR "Ready for review".

If changes were requested, once you've made them, you MUST dismiss the review to get it reviewed again.

-->



Patch:
diff --git a/pkg/isb/format.go b/pkg/isb/format.go
index 6e9cee37..5970c0d2 100644
--- a/pkg/isb/format.go
+++ b/pkg/isb/format.go
@@ -21,7 +21,7 @@ type Header struct {
 	// ID is used for exactly-once-semantics. ID is usually populated from the offset, if offset is available.
 	ID string
 	// Key is (key,value) in the map-reduce paradigm which will be used for conditional forwarding.
-	Key []byte
+	Key string
 }
 
 // Body is the body of the message
diff --git a/pkg/isb/forward/interfaces.go b/pkg/isb/forward/interfaces.go
index f16a0e05..b98cfc2b 100644
--- a/pkg/isb/forward/interfaces.go
+++ b/pkg/isb/forward/interfaces.go
@@ -5,8 +5,8 @@ import (
 )
 
 var (
-	All  = GoWhere(func([]byte) ([]string, error) { return []string{dfv1.MessageKeyAll}, nil })
-	Drop = GoWhere(func([]byte) ([]string, error) { return []string{dfv1.MessageKeyDrop}, nil })
+	All  = GoWhere(func(string) ([]string, error) { return []string{dfv1.MessageKeyAll}, nil })
+	Drop = GoWhere(func(string) ([]string, error) { return []string{dfv1.MessageKeyDrop}, nil })
 )
 
 // ToWhichStepDecider decides which step to forward after applying the WhereTo function.
@@ -14,14 +14,14 @@ type ToWhichStepDecider interface {
 	// WhereTo decides where to forward the result to based on the name of the step it returns.
 	// It supports 2 addition keywords which need not be a step name. They are "ALL" and "DROP"
 	// where former means, forward to all the neighbouring steps and latter means do not forward anywhere.
-	WhereTo([]byte) ([]string, error)
+	WhereTo(string) ([]string, error)
 }
 
 // GoWhere is the step decider on where it needs to go
-type GoWhere func([]byte) ([]string, error)
+type GoWhere func(string) ([]string, error)
 
 // WhereTo decides where the data goes to.
-func (gw GoWhere) WhereTo(b []byte) ([]string, error) {
+func (gw GoWhere) WhereTo(b string) ([]string, error) {
 	return gw(b)
 }
 
diff --git a/pkg/isb/jetstream/reader.go b/pkg/isb/jetstream/reader.go
index 98412ef9..2c49a16f 100644
--- a/pkg/isb/jetstream/reader.go
+++ b/pkg/isb/jetstream/reader.go
@@ -262,7 +262,7 @@ func convert2IsbMsgHeader(header nats.Header) isb.Header {
 		r.ID = x
 	}
 	if x := header.Get(_key); x != "" {
-		r.Key = []byte(x)
+		r.Key = x
 	}
 	if x := header.Get(_eventTime); x != "" {
 		i, _ := strconv.ParseInt(x, 10, 64)
diff --git a/pkg/sources/kafka/reader.go b/pkg/sources/kafka/reader.go
index 2e24b441..98400d31 100644
--- a/pkg/sources/kafka/reader.go
+++ b/pkg/sources/kafka/reader.go
@@ -325,7 +325,7 @@ func toReadMessage(m *sarama.ConsumerMessage) *isb.ReadMessage {
 		Header: isb.Header{
 			PaneInfo: isb.PaneInfo{EventTime: m.Timestamp},
 			ID:       offset,
-			Key:      m.Key,
+			Key:      string(m.Key),
 		},
 		Body: isb.Body{Payload: m.Value},
 	}
diff --git a/pkg/udf/applier/uds_http.go b/pkg/udf/applier/uds_http.go
index 87809a18..cdd42613 100644
--- a/pkg/udf/applier/uds_http.go
+++ b/pkg/udf/applier/uds_http.go
@@ -166,7 +166,8 @@ retry:
 			Header: isb.Header{
 				PaneInfo: parentPaneInfo,
 				ID:       fmt.Sprintf("%s-%d", offset.String(), i),
-				Key:      key,
+				// TODO: change to use string directly after SDK upgrade
+				Key: string(key),
 			},
 			Body: isb.Body{
 				Payload: m.Value,
diff --git a/pkg/udf/udf.go b/pkg/udf/udf.go
index 312237e5..2d62c8bd 100644
--- a/pkg/udf/udf.go
+++ b/pkg/udf/udf.go
@@ -3,10 +3,11 @@ package udf
 import (
 	"context"
 	"fmt"
-	"github.com/numaproj/numaflow/pkg/watermark/generic/jetstream"
 	"sync"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/watermark/generic/jetstream"
+
 	"go.uber.org/zap"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
@@ -103,7 +104,7 @@ func (u *UDFProcessor) Start(ctx context.Context) error {
 		return fmt.Errorf("unrecognized isbs type %q", u.ISBSvcType)
 	}
 
-	conditionalForwarder := forward.GoWhere(func(key []byte) ([]string, error) {
+	conditionalForwarder := forward.GoWhere(func(key string) ([]string, error) {
 		result := []string{}
 		_key := string(key)
 		if _key == dfv1.MessageKeyAll || _key == dfv1.MessageKeyDrop {


Test patch:
diff --git a/pkg/isb/forward/forward_test.go b/pkg/isb/forward/forward_test.go
index 81e79caa..c75cc1f7 100644
--- a/pkg/isb/forward/forward_test.go
+++ b/pkg/isb/forward/forward_test.go
@@ -3,11 +3,12 @@ package forward
 import (
 	"context"
 	"fmt"
-	"github.com/numaproj/numaflow/pkg/watermark/generic"
 	"strings"
 	"testing"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/watermark/generic"
+
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/prometheus/client_golang/prometheus/testutil"
 
@@ -27,7 +28,7 @@ var (
 type myForwardTest struct {
 }
 
-func (f myForwardTest) WhereTo(_ []byte) ([]string, error) {
+func (f myForwardTest) WhereTo(_ string) ([]string, error) {
 	return []string{"to1"}, nil
 }
 
@@ -89,7 +90,7 @@ func TestNewInterStepDataForward(t *testing.T) {
 type myForwardDropTest struct {
 }
 
-func (f myForwardDropTest) WhereTo(_ []byte) ([]string, error) {
+func (f myForwardDropTest) WhereTo(_ string) ([]string, error) {
 	return []string{"__DROP__"}, nil
 }
 
@@ -142,7 +143,7 @@ func TestNewInterStepDataForward_drop(t *testing.T) {
 type myForwardApplyErrTest struct {
 }
 
-func (f myForwardApplyErrTest) WhereTo(_ []byte) ([]string, error) {
+func (f myForwardApplyErrTest) WhereTo(_ string) ([]string, error) {
 	return []string{"to1"}, nil
 }
 
@@ -194,7 +195,7 @@ func TestNewInterStepDataForward_WithInternalError(t *testing.T) {
 type myForwardApplyWhereToErrTest struct {
 }
 
-func (f myForwardApplyWhereToErrTest) WhereTo(_ []byte) ([]string, error) {
+func (f myForwardApplyWhereToErrTest) WhereTo(_ string) ([]string, error) {
 	return []string{"to1"}, fmt.Errorf("whereToStep failed")
 }
 
@@ -241,7 +242,7 @@ func TestNewInterStepDataForward_WhereToError(t *testing.T) {
 type myForwardApplyUDFErrTest struct {
 }
 
-func (f myForwardApplyUDFErrTest) WhereTo(_ []byte) ([]string, error) {
+func (f myForwardApplyUDFErrTest) WhereTo(_ string) ([]string, error) {
 	return []string{"to1"}, nil
 }
 
@@ -290,7 +291,7 @@ func TestNewInterStepDataForward_UDFError(t *testing.T) {
 type myForwardToAllTest struct {
 }
 
-func (f myForwardToAllTest) WhereTo(_ []byte) ([]string, error) {
+func (f myForwardToAllTest) WhereTo(_ string) ([]string, error) {
 	return []string{dfv1.MessageKeyAll}, nil
 }
 
diff --git a/pkg/isb/forward/shutdown_test.go b/pkg/isb/forward/shutdown_test.go
index d338c129..c69f042f 100644
--- a/pkg/isb/forward/shutdown_test.go
+++ b/pkg/isb/forward/shutdown_test.go
@@ -2,10 +2,11 @@ package forward
 
 import (
 	"context"
-	"github.com/numaproj/numaflow/pkg/watermark/generic"
 	"testing"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/watermark/generic"
+
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 
 	"github.com/numaproj/numaflow/pkg/isb"
@@ -17,7 +18,7 @@ import (
 type myShutdownTest struct {
 }
 
-func (s myShutdownTest) WhereTo(_ []byte) ([]string, error) {
+func (s myShutdownTest) WhereTo(_ string) ([]string, error) {
 	return []string{dfv1.MessageKeyAll}, nil
 }
 
diff --git a/pkg/isb/jetstream/writer_test.go b/pkg/isb/jetstream/writer_test.go
index c4231a3d..08ae3809 100644
--- a/pkg/isb/jetstream/writer_test.go
+++ b/pkg/isb/jetstream/writer_test.go
@@ -4,10 +4,11 @@ package jetstream
 
 import (
 	"context"
-	"github.com/numaproj/numaflow/pkg/watermark/generic"
 	"testing"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/watermark/generic"
+
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 
@@ -21,7 +22,7 @@ import (
 type myForwardJetStreamTest struct {
 }
 
-func (f myForwardJetStreamTest) WhereTo(_ []byte) ([]string, error) {
+func (f myForwardJetStreamTest) WhereTo(_ string) ([]string, error) {
 	return []string{"to1"}, nil
 }
 
diff --git a/pkg/isb/redis/read_test.go b/pkg/isb/redis/read_test.go
index 3426d694..1a99ee4c 100644
--- a/pkg/isb/redis/read_test.go
+++ b/pkg/isb/redis/read_test.go
@@ -6,11 +6,12 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/numaproj/numaflow/pkg/watermark/generic"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/watermark/generic"
+
 	"github.com/go-redis/redis/v8"
 	"github.com/montanaflynn/stats"
 	"github.com/stretchr/testify/assert"
@@ -275,7 +276,7 @@ type ReadWritePerformance struct {
 type forwardReadWritePerformance struct {
 }
 
-func (f forwardReadWritePerformance) WhereTo(_ []byte) ([]string, error) {
+func (f forwardReadWritePerformance) WhereTo(_ string) ([]string, error) {
 	return []string{"to1"}, nil
 }
 
diff --git a/pkg/isb/redis/write_test.go b/pkg/isb/redis/write_test.go
index 51f213ef..f552a12f 100644
--- a/pkg/isb/redis/write_test.go
+++ b/pkg/isb/redis/write_test.go
@@ -5,10 +5,11 @@ package redis
 import (
 	"context"
 	"fmt"
-	"github.com/numaproj/numaflow/pkg/watermark/generic"
 	"testing"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/watermark/generic"
+
 	"github.com/go-redis/redis/v8"
 	"github.com/stretchr/testify/assert"
 
@@ -285,7 +286,7 @@ func Test_GetRefreshFullError(t *testing.T) {
 type myForwardRedisTest struct {
 }
 
-func (f myForwardRedisTest) WhereTo(_ []byte) ([]string, error) {
+func (f myForwardRedisTest) WhereTo(_ string) ([]string, error) {
 	return []string{"to1"}, nil
 }
 
diff --git a/pkg/isb/testutils/udf.go b/pkg/isb/testutils/udf.go
index 52238bdb..59790ee8 100644
--- a/pkg/isb/testutils/udf.go
+++ b/pkg/isb/testutils/udf.go
@@ -17,7 +17,7 @@ func CopyUDFTestApply(ctx context.Context, readMessage *isb.ReadMessage) ([]*isb
 	_ = payload
 	// copy the payload
 	result := payload
-	var key []byte
+	var key string
 
 	writeMessage := &isb.Message{
 		Header: isb.Header{
diff --git a/pkg/sinks/kafka/kafka_test.go b/pkg/sinks/kafka/kafka_test.go
index db7a375a..60fd741a 100644
--- a/pkg/sinks/kafka/kafka_test.go
+++ b/pkg/sinks/kafka/kafka_test.go
@@ -3,9 +3,10 @@ package kafka
 import (
 	"context"
 	"fmt"
-	"github.com/numaproj/numaflow/pkg/watermark/generic"
 	"testing"
 
+	"github.com/numaproj/numaflow/pkg/watermark/generic"
+
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 
 	mock "github.com/Shopify/sarama/mocks"
@@ -51,14 +52,14 @@ func TestWriteSuccessToKafka(t *testing.T) {
 		{
 			Header: isb.Header{
 				PaneInfo: isb.PaneInfo{},
-				Key:      []byte("key1"),
+				Key:      "key1",
 			},
 			Body: isb.Body{Payload: []byte("welcome1")},
 		},
 		{
 			Header: isb.Header{
 				PaneInfo: isb.PaneInfo{},
-				Key:      []byte("key1"),
+				Key:      "key1",
 			},
 			Body: isb.Body{Payload: []byte("welcome1")},
 		},
@@ -103,14 +104,14 @@ func TestWriteFailureToKafka(t *testing.T) {
 		{
 			Header: isb.Header{
 				PaneInfo: isb.PaneInfo{},
-				Key:      []byte("key1"),
+				Key:      "key1",
 			},
 			Body: isb.Body{Payload: []byte("welcome1")},
 		},
 		{
 			Header: isb.Header{
 				PaneInfo: isb.PaneInfo{},
-				Key:      []byte("key1"),
+				Key:      "key1",
 			},
 			Body: isb.Body{Payload: []byte("welcome1")},
 		},
diff --git a/pkg/sources/kafka/handler_test.go b/pkg/sources/kafka/handler_test.go
index adb951a7..bddde457 100644
--- a/pkg/sources/kafka/handler_test.go
+++ b/pkg/sources/kafka/handler_test.go
@@ -62,6 +62,6 @@ func TestMessageHandling(t *testing.T) {
 	readmsg := readmsgs[0]
 	assert.Equal(t, expectedoffset, readmsg.ID)
 	assert.Equal(t, []byte(value), readmsg.Body.Payload)
-	assert.Equal(t, []byte(key), readmsg.Header.Key)
+	assert.Equal(t, key, readmsg.Header.Key)
 	assert.Equal(t, expectedoffset, readmsg.ReadOffset.String())
 }