feat: Watermark Implementation for Async Data Movement (#2376)

Signed-off-by: Yashash H L <[email protected]> Signed-off-by: Vigith Maurice <[email protected]> Co-authored-by: Vigith Maurice <[email protected]>
numaproj · Jan 31, 2025 · 084be89 · 084be89
1 parent d4e5859
commit 084be89
Show file tree

Hide file tree

Showing 65 changed files with 5,553 additions and 1,028 deletions.
diff --git a/Makefile b/Makefile
@@ -365,4 +365,4 @@ update-manifests-version:
 	cat config/extensions/webhook/kustomization.yaml | sed 's/newTag: .*/newTag: $(VERSION)/' | sed 's@value: quay.io/numaproj/numaflow:.*@value: quay.io/numaproj/numaflow:$(VERSION)@' > /tmp/tmp_kustomization.yaml
 	mv /tmp/tmp_kustomization.yaml config/extensions/webhook/kustomization.yaml
 	cat Makefile | sed 's/^VERSION?=.*/VERSION?=$(VERSION)/' | sed 's/^BASE_VERSION:=.*/BASE_VERSION:=$(VERSION)/' > /tmp/ae_makefile
-	mv /tmp/ae_makefile Makefile
+	mv /tmp/ae_makefile Makefile
diff --git a/hack/generate-proto.sh b/hack/generate-proto.sh
@@ -61,5 +61,5 @@ gen-protoc pkg/apis/proto/mvtxdaemon/mvtxdaemon.proto
 
 gen-protoc pkg/apis/proto/isb/message.proto
 
-gen-protoc pkg/apis/proto/wmb/wmb.proto
+gen-protoc pkg/apis/proto/watermark/watermark.proto
 
diff --git a/pkg/apis/proto/watermark/watermark.pb.go b/pkg/apis/proto/watermark/watermark.pb.go
diff --git a/pkg/apis/proto/wmb/wmb.proto → pkg/apis/proto/watermark/watermark.proto b/pkg/apis/proto/wmb/wmb.proto → pkg/apis/proto/watermark/watermark.proto
@@ -17,7 +17,7 @@ limitations under the License.
 syntax = "proto3";
 option go_package = "github.com/numaproj/numaflow/pkg/apis/proto/isb";
 
-package wmb;
+package watermark;
 
 // WMB is used in the KV offset timeline bucket as the value for the given processor entity key.
 message WMB {
@@ -38,4 +38,10 @@ message WMB {
 
   // Partition to identify the partition to which the watermark belongs.
   int32 partition = 4;
+}
+
+// Heartbeat is used to track the active processors
+message Heartbeat {
+  // Heartbeat(current time in millis) published by the active processors.
+  int64 heartbeat = 1;
 }
Original file line number	Diff line number	Diff line change
Expand Up		@@ -61,5 +61,5 @@ gen-protoc pkg/apis/proto/mvtxdaemon/mvtxdaemon.proto

		gen-protoc pkg/apis/proto/isb/message.proto

		gen-protoc pkg/apis/proto/wmb/wmb.proto
		gen-protoc pkg/apis/proto/watermark/watermark.proto