fix: support DiscardNew policy for Jetstream streams

api/json-schema/schema.json

@@ -18651,7 +18651,7 @@
           "type": "boolean"
         },
         "bufferConfig": {
-          "description": "Optional configuration for the streams, consumers and buckets to be created in this JetStream service, if specified, it will be merged with the default configuration in numaflow-controller-config. It accepts a YAML format configuration, it may include 4 sections, \"stream\", \"consumer\", \"otBucket\" and \"procBucket\". Available fields under \"stream\" include \"retention\" (e.g. interest, limits, workerQueue), \"maxMsgs\", \"maxAge\" (e.g. 72h), \"replicas\" (1, 3, 5), \"duplicates\" (e.g. 5m). Available fields under \"consumer\" include \"ackWait\" (e.g. 60s) Available fields under \"otBucket\" include \"maxValueSize\", \"history\", \"ttl\" (e.g. 72h), \"maxBytes\", \"replicas\" (1, 3, 5). Available fields under \"procBucket\" include \"maxValueSize\", \"history\", \"ttl\" (e.g. 72h), \"maxBytes\", \"replicas\" (1, 3, 5).",
+          "description": "Optional configuration for the streams, consumers and buckets to be created in this JetStream service, if specified, it will be merged with the default configuration in numaflow-controller-config. It accepts a YAML format configuration, it may include 4 sections, \"stream\", \"consumer\", \"otBucket\" and \"procBucket\". Available fields under \"stream\" include \"maxMsgs\", \"maxAge\" (e.g. 72h), \"replicas\" (1, 3, 5), \"duplicates\" (e.g. 5m). Available fields under \"consumer\" include \"ackWait\" (e.g. 60s) Available fields under \"otBucket\" include \"maxValueSize\", \"history\", \"ttl\" (e.g. 72h), \"maxBytes\", \"replicas\" (1, 3, 5). Available fields under \"procBucket\" include \"maxValueSize\", \"history\", \"ttl\" (e.g. 72h), \"maxBytes\", \"replicas\" (1, 3, 5).",
           "type": "string"
         },
         "containerTemplate": {

api/openapi-spec/swagger.json

@@ -18647,7 +18647,7 @@
           "type": "boolean"
         },
         "bufferConfig": {
-          "description": "Optional configuration for the streams, consumers and buckets to be created in this JetStream service, if specified, it will be merged with the default configuration in numaflow-controller-config. It accepts a YAML format configuration, it may include 4 sections, \"stream\", \"consumer\", \"otBucket\" and \"procBucket\". Available fields under \"stream\" include \"retention\" (e.g. interest, limits, workerQueue), \"maxMsgs\", \"maxAge\" (e.g. 72h), \"replicas\" (1, 3, 5), \"duplicates\" (e.g. 5m). Available fields under \"consumer\" include \"ackWait\" (e.g. 60s) Available fields under \"otBucket\" include \"maxValueSize\", \"history\", \"ttl\" (e.g. 72h), \"maxBytes\", \"replicas\" (1, 3, 5). Available fields under \"procBucket\" include \"maxValueSize\", \"history\", \"ttl\" (e.g. 72h), \"maxBytes\", \"replicas\" (1, 3, 5).",
+          "description": "Optional configuration for the streams, consumers and buckets to be created in this JetStream service, if specified, it will be merged with the default configuration in numaflow-controller-config. It accepts a YAML format configuration, it may include 4 sections, \"stream\", \"consumer\", \"otBucket\" and \"procBucket\". Available fields under \"stream\" include \"maxMsgs\", \"maxAge\" (e.g. 72h), \"replicas\" (1, 3, 5), \"duplicates\" (e.g. 5m). Available fields under \"consumer\" include \"ackWait\" (e.g. 60s) Available fields under \"otBucket\" include \"maxValueSize\", \"history\", \"ttl\" (e.g. 72h), \"maxBytes\", \"replicas\" (1, 3, 5). Available fields under \"procBucket\" include \"maxValueSize\", \"history\", \"ttl\" (e.g. 72h), \"maxBytes\", \"replicas\" (1, 3, 5).",
           "type": "string"
         },
         "containerTemplate": {

config/advanced-install/namespaced-controller-wo-crds.yaml

@@ -189,8 +189,6 @@ data:
         bufferConfig: |
           # The default properties of the buffers (streams) to be created in this JetStream service
           stream:
-            # 0: Limits, 1: Interest, 2: WorkQueue
-            retention: 0
             maxMsgs: 100000
             maxAge: 72h
             maxBytes: -1

config/base/controller-manager/numaflow-controller-config.yaml

@@ -60,8 +60,6 @@ data:
         bufferConfig: |
           # The default properties of the buffers (streams) to be created in this JetStream service
           stream:
-            # 0: Limits, 1: Interest, 2: WorkQueue
-            retention: 0
             maxMsgs: 100000
             maxAge: 72h
             maxBytes: -1

config/install.yaml

@@ -18320,8 +18320,6 @@ data:
         bufferConfig: |
           # The default properties of the buffers (streams) to be created in this JetStream service
           stream:
-            # 0: Limits, 1: Interest, 2: WorkQueue
-            retention: 0
             maxMsgs: 100000
             maxAge: 72h
             maxBytes: -1

config/namespace-install.yaml

@@ -18223,8 +18223,6 @@ data:
         bufferConfig: |
           # The default properties of the buffers (streams) to be created in this JetStream service
           stream:
-            # 0: Limits, 1: Interest, 2: WorkQueue
-            retention: 0
             maxMsgs: 100000
             maxAge: 72h
             maxBytes: -1

docs/APIs.md

@@ -4337,13 +4337,12 @@ created in this JetStream service, if specified, it will be merged with
 the default configuration in numaflow-controller-config. It accepts a
 YAML format configuration, it may include 4 sections, “stream”,
 “consumer”, “otBucket” and “procBucket”. Available fields under “stream”
-include “retention” (e.g. interest, limits, workerQueue), “maxMsgs”,
-“maxAge” (e.g. 72h), “replicas” (1, 3, 5), “duplicates” (e.g. 5m).
-Available fields under “consumer” include “ackWait” (e.g. 60s) Available
-fields under “otBucket” include “maxValueSize”, “history”, “ttl”
-(e.g. 72h), “maxBytes”, “replicas” (1, 3, 5). Available fields under
-“procBucket” include “maxValueSize”, “history”, “ttl” (e.g. 72h),
-“maxBytes”, “replicas” (1, 3, 5).
+include “maxMsgs”, “maxAge” (e.g. 72h), “replicas” (1, 3, 5),
+“duplicates” (e.g. 5m). Available fields under “consumer” include
+“ackWait” (e.g. 60s) Available fields under “otBucket” include
+“maxValueSize”, “history”, “ttl” (e.g. 72h), “maxBytes”, “replicas” (1,
+3, 5). Available fields under “procBucket” include “maxValueSize”,
+“history”, “ttl” (e.g. 72h), “maxBytes”, “replicas” (1, 3, 5).
 </p>
 
 </td>

docs/core-concepts/inter-step-buffer-service.md

@@ -124,8 +124,6 @@ Both these 2 places expect a YAML format configuration like below:
 bufferConfig: |
   # The properties of the buffers (streams) to be created in this JetStream service
   stream:
-    # 0: Limits, 1: Interest, 2: WorkQueue
-    retention: 1
     maxMsgs: 30000
     maxAge: 168h
     maxBytes: -1

docs/operations/numaflow-controller-config.yaml

@@ -56,8 +56,6 @@ data:
         bufferConfig: |
           # The default properties of the buffers (streams) to be created in this JetStream service
           stream:
-            # 0: Limits, 1: Interest, 2: WorkQueue
-            retention: 0
             maxMsgs: 2000000
             maxAge: 168h
             maxBytes: -1

docs/user-guide/reference/configuration/max-message-size.md

@@ -22,4 +22,4 @@ spec:
 
 It's not recommended to use values over `8388608` (8MB) but `max_payload` can be set up to `67108864` (64MB).
 
-Please be aware that if you increase the max message size of the `InterStepBufferService`, you probably will also need to change some other limits. For example, if the size of each messages is as large as 8MB, then 100 messages flowing in the pipeline will make each of the Inter-Step Buffer need at least 800MB of disk space to store the messages, and the memory consumption will also be high, that will probably cause the Inter-Step Buffer Service to crash. In that case, you might need to update the retention policy in the Inter-Step Buffer Service to make sure the messages are not stored for too long. Check out the [Inter-Step Buffer Service](../../../core-concepts/inter-step-buffer-service.md#buffer-configuration) for more details.
+Please be aware that if you increase the max message size of the `InterStepBufferService`, you probably will also need to change some other limits. For example, if the size of each messages is as large as 8MB, then 100 messages flowing in the pipeline will make each of the Inter-Step Buffer need at least 800MB of storage to store the messages, and the memory consumption overhead will also be high, which could cause the Inter-Step Buffer Service to crash. In that eventuality, Check out the [Inter-Step Buffer Service](../../../core-concepts/inter-step-buffer-service.md#buffer-configuration) for more details.

pkg/apis/numaflow/v1alpha1/jetstream_buffer_service.go

@@ -58,7 +58,7 @@ type JetStreamBufferService struct {
 	StartArgs []string `json:"startArgs,omitempty" protobuf:"bytes,9,rep,name=startArgs"`
 	// Optional configuration for the streams, consumers and buckets to be created in this JetStream service, if specified, it will be merged with the default configuration in numaflow-controller-config.
 	// It accepts a YAML format configuration, it may include 4 sections, "stream", "consumer", "otBucket" and "procBucket".
-	// Available fields under "stream" include "retention" (e.g. interest, limits, workerQueue), "maxMsgs", "maxAge" (e.g. 72h), "replicas" (1, 3, 5), "duplicates" (e.g. 5m).
+	// Available fields under "stream" include "maxMsgs", "maxAge" (e.g. 72h), "replicas" (1, 3, 5), "duplicates" (e.g. 5m).
 	// Available fields under "consumer" include "ackWait" (e.g. 60s)
 	// Available fields under "otBucket" include "maxValueSize", "history", "ttl" (e.g. 72h), "maxBytes", "replicas" (1, 3, 5).
 	// Available fields under "procBucket" include "maxValueSize", "history", "ttl" (e.g. 72h), "maxBytes", "replicas" (1, 3, 5).

pkg/isb/stores/jetstream/metrics.go

@@ -56,7 +56,7 @@ var isbSoftUsage = promauto.NewGaugeVec(prometheus.GaugeOpts{
 	Help:      "Percentage of buffer soft usage",
 }, []string{"buffer"})
 
-// isbSolidUsage is used to indicate of buffer that is used up, it is calculated based on the messages remain in the stream (if it's not Limits retention policy)
+// isbSolidUsage is used to indicate buffer usage, it is calculated based on the messages remaining in the stream.
 var isbSolidUsage = promauto.NewGaugeVec(prometheus.GaugeOpts{
 	Subsystem: "isb_jetstream",
 	Name:      "buffer_solid_usage",

pkg/isb/stores/jetstream/reader_test.go

@@ -187,7 +187,7 @@ func addStream(t *testing.T, js nats.JetStreamContext, streamName string) {
 		Name:       streamName,
 		Subjects:   []string{streamName},
 		Retention:  nats.WorkQueuePolicy,
-		Discard:    nats.DiscardOld,
+		Discard:    nats.DiscardNew,
 		MaxMsgs:    100, //
 		Storage:    nats.FileStorage,
 		Duplicates: 2 * 60 * time.Second,

pkg/isb/stores/jetstream/writer.go

@@ -101,11 +101,7 @@ func (jw *jetStreamWriter) runStatusChecker(ctx context.Context) {
 		}
 		var solidUsage, softUsage float64
 		softUsage = (float64(c.NumPending) + float64(c.NumAckPending)) / float64(jw.opts.maxLength)
-		if s.Config.Retention == nats.LimitsPolicy {
-			solidUsage = softUsage
-		} else {
-			solidUsage = float64(s.State.Msgs) / float64(jw.opts.maxLength)
-		}
+		solidUsage = float64(s.State.Msgs) / float64(jw.opts.maxLength)
 		// TODO: solid usage calculation might be incorrect due to incorrect JetStream metadata issue caused by pod migration, need to revisit this later.
 		// We should set up alerts with sort of rules to detect the metadata issue. For example, solidUsage is too much greater than softUsage for a while.
 		if solidUsage >= jw.opts.bufferUsageLimit && softUsage >= jw.opts.bufferUsageLimit {

pkg/isbsvc/jetstream_service.go

@@ -123,8 +123,8 @@ func (jss *jetStreamSvc) CreateBuffersAndBuckets(ctx context.Context, buffers, b
 			if _, err := jss.js.AddStream(&nats.StreamConfig{
 				Name:       streamName,
 				Subjects:   []string{streamName}, // Use the stream name as the only subject
-				Retention:  nats.RetentionPolicy(v.GetInt("stream.retention")),
-				Discard:    nats.DiscardOld,
+				Retention:  nats.WorkQueuePolicy,
+				Discard:    nats.DiscardNew,
 				MaxMsgs:    v.GetInt64("stream.maxMsgs"),
 				MaxAge:     v.GetDuration("stream.maxAge"),
 				MaxBytes:   v.GetInt64("stream.maxBytes"),
@@ -285,9 +285,6 @@ func (jss *jetStreamSvc) GetBufferInfo(ctx context.Context, buffer string) (*Buf
 		return nil, fmt.Errorf("failed to get consumer information of stream %q", streamName)
 	}
 	totalMessages := int64(stream.State.Msgs)
-	if stream.Config.Retention == nats.LimitsPolicy {
-		totalMessages = int64(consumer.NumPending) + int64(consumer.NumAckPending)
-	}
 	bufferInfo := &BufferInfo{
 		Name:            buffer,
 		PendingCount:    int64(consumer.NumPending),

ui/src/components/pages/Namespace/index.test.tsx

@@ -197,7 +197,7 @@ const mockISBRawData = {
               },
             },
             streamConfig:
-              "consumer:\n  ackwait: 60s\n  maxackpending: 25000\notbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 3h\nprocbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 72h\nstream:\n  duplicates: 60s\n  maxage: 72h\n  maxbytes: -1\n  maxmsgs: 100000\n  replicas: 3\n  retention: 0\n  storage: 0\n",
+              "consumer:\n  ackwait: 60s\n  maxackpending: 25000\notbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 3h\nprocbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 72h\nstream:\n  duplicates: 60s\n  maxage: 72h\n  maxbytes: -1\n  maxmsgs: 100000\n  replicas: 3\n  storage: 0\n",
           },
         },
         type: "jetstream",

ui/src/components/pages/Namespace/partials/NamespaceListingWrapper/ISBServiceCard/ISBServiceCard.test.tsx

@@ -159,7 +159,7 @@ const mockData = {
             },
           },
           streamConfig:
-            "consumer:\n  ackwait: 60s\n  maxackpending: 25000\notbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 3h\nprocbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 72h\nstream:\n  duplicates: 60s\n  maxage: 72h\n  maxbytes: -1\n  maxmsgs: 100000\n  replicas: 3\n  retention: 0\n  storage: 0\n",
+            "consumer:\n  ackwait: 60s\n  maxackpending: 25000\notbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 3h\nprocbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 72h\nstream:\n  duplicates: 60s\n  maxage: 72h\n  maxbytes: -1\n  maxmsgs: 100000\n  replicas: 3\n storage: 0\n",
         },
       },
       type: "jetstream",

ui/src/components/pages/Namespace/partials/NamespaceListingWrapper/index.test.tsx

@@ -213,7 +213,7 @@ const mockISBRawData = {
               },
             },
             streamConfig:
-              "consumer:\n  ackwait: 60s\n  maxackpending: 25000\notbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 3h\nprocbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 72h\nstream:\n  duplicates: 60s\n  maxage: 72h\n  maxbytes: -1\n  maxmsgs: 100000\n  replicas: 3\n  retention: 0\n  storage: 0\n",
+              "consumer:\n  ackwait: 60s\n  maxackpending: 25000\notbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 3h\nprocbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 72h\nstream:\n  duplicates: 60s\n  maxage: 72h\n  maxbytes: -1\n  maxmsgs: 100000\n  replicas: 3\n  storage: 0\n",
           },
         },
         type: "jetstream",

ui/src/components/pages/Namespace/partials/PipelineCard/index.test.tsx

@@ -516,7 +516,7 @@ const mockISBData = {
             },
           },
           streamConfig:
-            "consumer:\n  ackwait: 60s\n  maxackpending: 25000\notbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 3h\nprocbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 72h\nstream:\n  duplicates: 60s\n  maxage: 72h\n  maxbytes: -1\n  maxmsgs: 100000\n  replicas: 3\n  retention: 0\n  storage: 0\n",
+            "consumer:\n  ackwait: 60s\n  maxackpending: 25000\notbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 3h\nprocbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 72h\nstream:\n  duplicates: 60s\n  maxage: 72h\n  maxbytes: -1\n  maxmsgs: 100000\n  replicas: 3\n  storage: 0\n",
         },
       },
       type: "jetstream",

ui/src/components/pages/Pipeline/partials/PipelineISBStatus/index.test.tsx

@@ -151,7 +151,7 @@ const mockData = {
             },
           },
           streamConfig:
-            "consumer:\n  ackwait: 60s\n  maxackpending: 25000\notbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 3h\nprocbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 72h\nstream:\n  duplicates: 60s\n  maxage: 72h\n  maxbytes: -1\n  maxmsgs: 100000\n  replicas: 3\n  retention: 0\n  storage: 0\n",
+            "consumer:\n  ackwait: 60s\n  maxackpending: 25000\notbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 3h\nprocbucket:\n  history: 1\n  maxbytes: 0\n  maxvaluesize: 0\n  replicas: 3\n  storage: 0\n  ttl: 72h\nstream:\n  duplicates: 60s\n  maxage: 72h\n  maxbytes: -1\n  maxmsgs: 100000\n  replicas: 3\n storage: 0\n",
         },
       },
       type: "jetstream",