Repository: numaproj/numaflow
Pull request #456 (2023-01-09 21:59:04)

chore: clean up some comments
Signed-off-by: Keran Yang <yangkr920208@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/forward/forward.go b/pkg/isb/forward/forward.go
index 7be5504c..1126b963 100644
--- a/pkg/isb/forward/forward.go
+++ b/pkg/isb/forward/forward.go
@@ -203,8 +203,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 	// fetch watermark if available
 	// TODO: make it async (concurrent and wait later)
 	// let's track only the first element's watermark. This is important because we reassign the watermark we fetch
-	// to all the elements in the batch. If we were to assign last element's watermark, we will wronly mark on-time data
-	// as πlate date.
+	// to all the elements in the batch. If we were to assign last element's watermark, we will wrongly mark on-time data as late.
 	processorWM := isdf.fetchWatermark.GetWatermark(readMessages[0].ReadOffset)
 	for _, m := range readMessages {
 		readBytesCount.With(map[string]string{metricspkg.LabelVertex: isdf.vertexName, metricspkg.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Add(float64(len(m.Payload)))
@@ -294,7 +293,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		}
 	}
 
-	// let us ack the only if we have successfully forwarded all the messages.
+	// let us ack the chunk only if we have successfully forwarded all the messages.
 	// we need the readOffsets to acknowledge later
 	var readOffsets = make([]isb.Offset, len(readMessages))
 	for idx, m := range udfResults {
@@ -315,7 +314,6 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 
 // ackFromBuffer acknowledges an array of offsets back to fromBuffer and is a blocking call or until shutdown has been initiated.
 func (isdf *InterStepDataForward) ackFromBuffer(ctx context.Context, offsets []isb.Offset) error {
-
 	var ackRetryBackOff = wait.Backoff{
 		Factor:   1,
 		Jitter:   0.1,
diff --git a/pkg/isb/forward/shutdown.go b/pkg/isb/forward/shutdown.go
index 93693198..f83a0c3a 100644
--- a/pkg/isb/forward/shutdown.go
+++ b/pkg/isb/forward/shutdown.go
@@ -66,7 +66,7 @@ func (isdf *InterStepDataForward) Stop() {
 // ForceStop sets up the force shutdown flag.
 func (isdf *InterStepDataForward) ForceStop() {
 	// call stop (what if we have an enthusiastic shutdown that forces first)
-	// eg. I know I have written a wrong UDF shutdown ASAP
+	// e.g. I know I have written a wrong UDF, so shutdown ASAP
 	isdf.Stop()
 	isdf.Shutdown.rwlock.Lock()
 	defer isdf.Shutdown.rwlock.Unlock()
diff --git a/pkg/isb/stores/simplebuffer/buffer.go b/pkg/isb/stores/simplebuffer/buffer.go
index 6833072c..4c7e667a 100644
--- a/pkg/isb/stores/simplebuffer/buffer.go
+++ b/pkg/isb/stores/simplebuffer/buffer.go
@@ -239,7 +239,7 @@ func (b *InMemoryBuffer) Ack(_ context.Context, offsets []isb.Offset) []error {
 			errs[index] = isb.MessageAckErr{
 				Name:    b.name,
 				Message: fmt.Sprintf("given index (%d) >= size of the buffer (%d)", intOffset, b.size),
-				Offset:  isb.Offset(offset),
+				Offset:  offset,
 			}
 			continue
 		}
diff --git a/pkg/sources/generator/tickgen.go b/pkg/sources/generator/tickgen.go
index a6615f24..c17f0e90 100644
--- a/pkg/sources/generator/tickgen.go
+++ b/pkg/sources/generator/tickgen.go
@@ -45,7 +45,7 @@ var log = logging.NewLogger()
 var timeAttr = "Createdts"
 
 // payload generated by the generator function
-// look at newreadmessage function
+// look at newReadMessage function
 type payload struct {
 	Data      []byte
 	Createdts int64
@@ -215,13 +215,6 @@ func (mg *memgen) IsEmpty() bool {
 	return len(mg.srcchan) == 0
 }
 
-// Read reads a chunk of messages and returns at the first occurrence of an error. Error does not indicate that the
-// array of result is empty, the callee should process all the elements in the array even if the error is set. Read
-// will not mark the message in the buffer as "READ" if the read for that index is erring.
-// There is a chance that we have read the message and the container got forcefully terminated before processing. To provide
-// at-least-once semantics for reading, during restart we will have to reprocess all unacknowledged messages.
-//
-// the context passed to read should be different from the lifecycle context that is used by this vertex.
 func (mg *memgen) Read(ctx context.Context, count int64) ([]*isb.ReadMessage, error) {
 	msgs := make([]*isb.ReadMessage, 0, count)
 	// timeout should not be re-triggered for every run of the for loop. it is for the entire Read() call.
@@ -233,7 +226,7 @@ loop:
 		select {
 		case r := <-mg.srcchan:
 			tickgenSourceReadCount.With(map[string]string{metricspkg.LabelVertex: mg.name, metricspkg.LabelPipeline: mg.pipelineName}).Inc()
-			msgs = append(msgs, mg.newreadmessage(r.data, r.offset))
+			msgs = append(msgs, mg.newReadMessage(r.data, r.offset))
 		case <-timeout:
 			mg.logger.Debugw("Timed out waiting for messages to read.", zap.Duration("waited", mg.readTimeout))
 			break loop
@@ -244,7 +237,7 @@ loop:
 		// into the offset timeline store.
 		// Please note that we are inserting the watermark before the data has been persisted into ISB by the forwarder.
 		o := msgs[len(msgs)-1].ReadOffset
-		// use the first eventime as watermark to make it conservative
+		// use the first event time as watermark to make it conservative
 		nanos, _ := msgs[0].ReadOffset.Sequence()
 		// remove the nanosecond precision
 		mg.sourcePublishWM.PublishWatermark(processor.Watermark(time.Unix(0, nanos)), o)
@@ -277,7 +270,7 @@ func (mg *memgen) ForceStop() {
 
 // Start starts reading from the source
 // context is used to control the lifecycle of this component.
-// this context will be used to shutdown the vertex once a os.signal is received.
+// this context will be used to shut down the vertex once an os.signal is received.
 func (mg *memgen) Start() <-chan struct{} {
 	mg.generator(mg.lifecycleCtx, mg.rpu, mg.timeunit)
 	return mg.forwarder.Start()
@@ -293,7 +286,7 @@ func (mg *memgen) generator(ctx context.Context, rate int, timeunit time.Duratio
 		defer ticker.Stop()
 		for {
 			select {
-			// we dont need to wait for ticker to fire to return
+			// we don't need to wait for ticker to fire to return
 			// when context closes
 			case <-ctx.Done():
 				log.Info("Context.Done is called. exiting generator loop.")
@@ -313,7 +306,6 @@ func (mg *memgen) generator(ctx context.Context, rate int, timeunit time.Duratio
 								log.Info("Context.Done is called. returning from the inner function")
 								return
 							case mg.srcchan <- r:
-
 							}
 						}
 					}()
@@ -323,11 +315,11 @@ func (mg *memgen) generator(ctx context.Context, rate int, timeunit time.Duratio
 	}()
 }
 
-func (mg *memgen) newreadmessage(payload []byte, offset int64) *isb.ReadMessage {
+func (mg *memgen) newReadMessage(payload []byte, offset int64) *isb.ReadMessage {
 	msg := isb.Message{
 		Header: isb.Header{
 			// TODO: insert the right time based on the generator
-			PaneInfo: isb.PaneInfo{EventTime: timefromNanos(parseTime(payload))},
+			PaneInfo: isb.PaneInfo{EventTime: timeFromNanos(parseTime(payload))},
 			ID:       strconv.FormatInt(offset, 10) + "-" + strconv.FormatInt(int64(mg.vertexInstance.Replica), 10),
 		},
 		Body: isb.Body{Payload: payload},
@@ -339,8 +331,8 @@ func (mg *memgen) newreadmessage(payload []byte, offset int64) *isb.ReadMessage
 	}
 }
 
-func timefromNanos(etime int64) time.Time {
-	// unparseable json or invalid time format will be substituted with current time.
+func timeFromNanos(etime int64) time.Time {
+	// un-parseable json or invalid time format will be substituted with current time.
 	if etime > 0 {
 		return time.Unix(0, etime)
 	}
@@ -348,7 +340,6 @@ func timefromNanos(etime int64) time.Time {
 }
 
 func parseTime(payload []byte) int64 {
-
 	var anyJson map[string]interface{}
 	unmarshalErr := json.Unmarshal(payload, &anyJson)
 
@@ -357,7 +348,7 @@ func parseTime(payload []byte) int64 {
 		return 0
 	}
 
-	// for now lets pretend that the time unit is nanos and that the time attribute is known
+	// for now let's pretend that the time unit is nanos and that the time attribute is known
 	eventTime := anyJson[timeAttr]
 	if i, ok := eventTime.(float64); ok {
 		return int64(i)
diff --git a/pkg/sources/http/http.go b/pkg/sources/http/http.go
index 39c8ee50..f8f29c46 100644
--- a/pkg/sources/http/http.go
+++ b/pkg/sources/http/http.go
@@ -55,8 +55,8 @@ type httpSource struct {
 	// source watermark publisher
 	sourcePublishWM publish.Publisher
 	// context cancel function
-	cancelfn context.CancelFunc
-	shutdown func(context.Context) error
+	cancelFunc context.CancelFunc
+	shutdown   func(context.Context) error
 }
 
 type Option func(*httpSource) error
@@ -108,34 +108,30 @@ func New(vertexInstance *dfv1.VertexInstance, writers []isb.BufferWriter, fetchW
 		if s, err := sharedutil.GetSecretFromVolume(x.Token); err != nil {
 			return nil, fmt.Errorf("failed to get auth token, %w", err)
 		} else {
-			auth = string(s)
+			auth = s
 		}
 	}
 	mux := http.NewServeMux()
 	mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
 		if !h.ready {
-			w.WriteHeader(503)
-			_, _ = w.Write([]byte("503 not ready\n"))
+			http.Error(w, "http source not ready", http.StatusServiceUnavailable)
 			return
 		}
-		w.WriteHeader(204)
+		w.WriteHeader(http.StatusNoContent)
 	})
 	mux.HandleFunc("/vertices/"+vertexInstance.Vertex.Spec.Name, func(w http.ResponseWriter, r *http.Request) {
 		if auth != "" && r.Header.Get("Authorization") != "Bearer "+auth {
-			w.WriteHeader(403)
-			_, _ = w.Write([]byte("403 forbidden\n"))
+			http.Error(w, "request not authorized", http.StatusForbidden)
 			return
 		}
 		if !h.ready {
-			w.WriteHeader(503)
-			_, _ = w.Write([]byte("503 not ready\n"))
+			http.Error(w, "http source not ready", http.StatusServiceUnavailable)
 			return
 		}
 		msg, err := io.ReadAll(r.Body)
 		_ = r.Body.Close()
 		if err != nil {
-			w.WriteHeader(http.StatusBadRequest)
-			_, _ = w.Write([]byte(err.Error()))
+			http.Error(w, err.Error(), http.StatusBadRequest)
 			return
 		}
 
@@ -147,8 +143,7 @@ func New(vertexInstance *dfv1.VertexInstance, writers []isb.BufferWriter, fetchW
 		if x := r.Header.Get(dfv1.KeyMetaEventTime); x != "" {
 			i, err := strconv.ParseInt(x, 10, 64)
 			if err != nil {
-				w.WriteHeader(http.StatusBadRequest)
-				_, _ = w.Write([]byte(err.Error()))
+				http.Error(w, err.Error(), http.StatusBadRequest)
 				return
 			}
 			eventTime = time.UnixMilli(i)
@@ -204,7 +199,7 @@ func New(vertexInstance *dfv1.VertexInstance, writers []isb.BufferWriter, fetchW
 	}
 	h.forwarder = forwarder
 	ctx, cancel := context.WithCancel(context.Background())
-	h.cancelfn = cancel
+	h.cancelFunc = cancel
 	entityName := fmt.Sprintf("%s-%d", vertexInstance.Vertex.Name, vertexInstance.Replica)
 	processorEntity := processor.NewProcessorEntity(entityName)
 	h.sourcePublishWM = publish.NewPublish(ctx, processorEntity, publishWMStores, publish.IsSource(), publish.WithDelay(vertexInstance.Vertex.Spec.Watermark.GetMaxDelay()))
@@ -246,7 +241,7 @@ func (h *httpSource) Ack(_ context.Context, offsets []isb.Offset) []error {
 
 func (h *httpSource) Close() error {
 	h.logger.Info("Shutting down http source server...")
-	h.cancelfn()
+	h.cancelFunc()
 	close(h.messages)
 	if err := h.shutdown(context.Background()); err != nil {
 		return err
diff --git a/pkg/sources/kafka/reader.go b/pkg/sources/kafka/reader.go
index 0898cd5a..adfa7726 100644
--- a/pkg/sources/kafka/reader.go
+++ b/pkg/sources/kafka/reader.go
@@ -121,11 +121,6 @@ func (r *KafkaSource) GetName() string {
 	return r.name
 }
 
-// Read reads a chunk of messages and returns at the first occurrence of an error. Error does not indicate that the
-// array of result is empty, the callee should process all the elements in the array even if the error is set. Read
-// will not mark the message in the buffer as "READ" if the read for that index is erring.
-// There is a chance that we have read the message and the container got forcefully terminated before processing. To provide
-// at-least-once semantics for reading, during restart we will have to reprocess all unacknowledged messages.
 func (r *KafkaSource) Read(_ context.Context, count int64) ([]*isb.ReadMessage, error) {
 	// It stores latest timestamps for different partitions
 	oldestTimestamps := make(map[int32]time.Time)
diff --git a/pkg/sources/sourcer.go b/pkg/sources/sourcer.go
index 821af274..6ba242a3 100644
--- a/pkg/sources/sourcer.go
+++ b/pkg/sources/sourcer.go
@@ -21,7 +21,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb/forward"
 )
 
-// Sourcer interface provides a isb.BufferReader abstraction over the underlying data source.
+// Sourcer interface provides an isb.BufferReader abstraction over the underlying data source.
 // This is intended to be consumed by a connector like isb.forward
 type Sourcer interface {
 	isb.BufferReader
diff --git a/pkg/udf/applier/map_applier.go b/pkg/udf/applier/map_applier.go
index 805d6e5a..971e1438 100644
--- a/pkg/udf/applier/map_applier.go
+++ b/pkg/udf/applier/map_applier.go
@@ -22,13 +22,13 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb"
 )
 
-// MapApplier applies the HTTPBasedUDF on the read message and gives back a new message. Any UserError will be retried here, while
+// MapApplier applies the UDF on the read message and gives back a new message. Any UserError will be retried here, while
 // InternalErr can be returned and could be retried by the callee.
 type MapApplier interface {
 	ApplyMap(ctx context.Context, message *isb.ReadMessage) ([]*isb.Message, error)
 }
 
-// ApplyFunc utility function used to create an Applier implementation
+// ApplyMapFunc utility function used to create an Applier implementation
 type ApplyMapFunc func(context.Context, *isb.ReadMessage) ([]*isb.Message, error)
 
 func (a ApplyMapFunc) ApplyMap(ctx context.Context, message *isb.ReadMessage) ([]*isb.Message, error) {
diff --git a/pkg/udf/builtin/filter/filter.go b/pkg/udf/builtin/filter/filter.go
index b80de383..ab4f88e2 100644
--- a/pkg/udf/builtin/filter/filter.go
+++ b/pkg/udf/builtin/filter/filter.go
@@ -50,7 +50,6 @@ func New(args map[string]string) (functionsdk.MapFunc, error) {
 }
 
 func (f filter) apply(msg []byte) (functionsdk.Message, error) {
-
 	result, err := expr.EvalBool(f.expression, msg)
 	if err != nil {
 		return functionsdk.MessageToDrop(), err
diff --git a/pkg/udf/function/uds_grpc.go b/pkg/udf/function/uds_grpc.go
index e9152160..d9f38178 100644
--- a/pkg/udf/function/uds_grpc.go
+++ b/pkg/udf/function/uds_grpc.go
@@ -123,7 +123,7 @@ func (u *udsGRPCBasedUDF) ApplyMap(ctx context.Context, readMessage *isb.ReadMes
 
 // should we pass metadata information ?
 
-// Reduce accepts a channel of isbMessages and returns the aggregated result
+// ApplyReduce accepts a channel of isbMessages and returns the aggregated result
 func (u *udsGRPCBasedUDF) ApplyReduce(ctx context.Context, partitionID *partition.ID, messageStream <-chan *isb.ReadMessage) ([]*isb.Message, error) {
 	datumCh := make(chan *functionpb.Datum)
 	var wg sync.WaitGroup


Test patch:
diff --git a/pkg/sources/generator/tickgen_test.go b/pkg/sources/generator/tickgen_test.go
index 40f77c7a..bfbe5c75 100644
--- a/pkg/sources/generator/tickgen_test.go
+++ b/pkg/sources/generator/tickgen_test.go
@@ -165,13 +165,13 @@ func TestUnparseableTime(t *testing.T) {
 
 func TestTimeForValidTime(t *testing.T) {
 	nanotime := time.Now().UnixNano()
-	parsedtime := timefromNanos(nanotime)
+	parsedtime := timeFromNanos(nanotime)
 	assert.Equal(t, nanotime, parsedtime.UnixNano())
 }
 
 func TestTimeForInvalidTime(t *testing.T) {
 	nanotime := int64(-1)
-	parsedtime := timefromNanos(nanotime)
+	parsedtime := timeFromNanos(nanotime)
 	assert.True(t, parsedtime.UnixNano() > 0)
 }