fix test and ordered processing doc

Signed-off-by: Yashash Lokesh <yashashhl25@gmail.com>

Author: yhl25

docs/user-guide/reference/ordered-processing.md

@@ -58,8 +58,6 @@ kind: Pipeline
 metadata:
   name: my-pipeline
 spec:
-  limits:
-    readBatchSize: 1        # recommended for strict ordering
   ordered:
     enabled: true           # enable order-preserving processing pipeline-wide
   vertices:
@@ -118,8 +116,9 @@ case.
 - **Source vertices** always preserve input order regardless of the `ordered` setting.
 - **Key-based routing**: ordering is guaranteed per key. Messages with different keys may still be interleaved across
   partitions. Ensure your UDF or SDK sets meaningful message keys to leverage per-key ordering.
-- **`readBatchSize: 1`** is required for strict ordering. With a larger batch size, multiple messages may
-  be in-flight simultaneously within a single pod.
+- **In-flight cap is forced to 1**: when ordered processing is enabled on a Map or Sink vertex, the controller
+  automatically sets `limits.concurrency = 1` so each pod processes one message at a time. You don't need to set
+  `concurrency` or `readBatchSize` yourself — any explicit `concurrency` you set is overridden for ordered vertices.
 - **Throughput trade-off**: ordered processing limits parallelism within a partition. Consider the number of partitions
   carefully to balance ordering guarantees with throughput requirements.
 - **Join vertices (multiple input edges)**: When a vertex receives messages from multiple upstream vertices
@@ -169,8 +168,6 @@ kind: Pipeline
 metadata:
   name: ordered-pipeline
 spec:
-  limits:
-    readBatchSize: 1
   ordered:
     enabled: true
   vertices:
@@ -202,8 +199,8 @@ spec:
 In the example above:
 
 - `ordered.enabled: true` enables order-preserving processing pipeline-wide.
-- `limits.readBatchSize: 1` is required so that each pod processes one message at a time, which is essential for
-strict in-order guarantees.
+- For each ordered Map/Sink vertex the controller automatically forces `limits.concurrency = 1` so each pod processes
+  one message at a time, which is essential for strict in-order guarantees.
 - The `cat` (Map) and `out` (Sink) vertices each have `partitions: 3`, so they will run with exactly 3 replicas.
 - Source vertices (`in-1`, `in-2`) always preserve input order and require no extra configuration.
 - Because `in-1` and `in-2` both feed into `cat` (a join), messages from the two sources are interleaved at `cat`.

pkg/reconciler/pipeline/controller.go

@@ -659,6 +659,18 @@ func buildVertices(pl *dfv1.Pipeline) map[string]dfv1.Vertex {
 		// Resolve and set the effective ordered config on the vertex
 		// This merges pipeline-level and vertex-level ordered config so the vertex is self-contained
 		resolveOrderedConfig(pl, vCopy)
+		// When ordered processing is enabled on a Map/Sink vertex, force the in-flight cap to 1 so
+		// each partition processes one message at a time. Source and Reduce vertices preserve order
+		// by other means (single-stream-per-replica routing for sources, partition-by-key for
+		// reduce), so we leave their concurrency alone. We override even if the user set
+		// concurrency explicitly: ordered processing without concurrency=1 would silently violate
+		// the FIFO contract.
+		if vCopy.IsOrdered() && (vCopy.IsMapUDF() || vCopy.IsASink()) {
+			if vCopy.Limits == nil {
+				vCopy.Limits = &dfv1.VertexLimits{}
+			}
+			vCopy.Limits.Concurrency = ptr.To[uint64](1)
+		}
 		replicas := int32(1)
 		// If the desired phase is paused, or we are in the middle of pausing we should not start any vertex replicas
 		if isLifecycleChange(pl) {

rust/numaflow-core/src/source.rs

@@ -1067,9 +1067,12 @@ mod tests {
         .unwrap();
 
         let tracker = Tracker::new(None, CancellationToken::new());
+        // Concurrency is large enough to comfortably hold 50 unacked messages plus read-ahead
+        // batches; matches the previous behavior that depended on the now-removed 10000-message
+        // MAX_ACK_PENDING constant.
         let source: Source<crate::typ::WithoutRateLimiter> = Source::new(
             5,
-            5,
+            10000,
             SourceType::UserDefinedSource(Box::new(src_read), Box::new(src_ack), lag_reader),
             tracker.clone(),
             true,