Ways to enforce event ordering/sorting

[th0ger]

While numaflow does not guarantee ordering (yet), how can I restrict pipeline/image configurations to ensure that an ordered source keeps ordered?

Let's take a build-in kafka source as an example where the topic is time ordered.

I've tried to limit pod replicas:
'vertices.scale.max: 1'
https://numaflow.numaproj.io/user-guide/reference/autoscaling/

And disable threading in source UDF:
'max_threads=1'
https://github.com/numaproj/numaflow-python/blob/main/pynumaflow%2Fsourcer%2Fasync_server.py

But, it still seems that my events occasionally become unordered.

Can I expect the input to be ordered?What did I miss?

[vigith]

Could you set read-batch-size to 1? Please share the pipeline spec?

[th0ger]

Thanks @vigith. Adding readBatchSize: 1 seems to have restored the ordering for now.
But I'm not 100%, because I can't really make tests of low-level threading/async operations that could kick in on production workloads.

I'm puzzled why readBatchSize>1 could cause shuffling inside the batch. The reads from kafka/NATS preserves order, I'll presume. Is numaflow shuffling by design?

Currently I play it extra safe safe with all options:

• readBatchSize: 1 (on pipeline)
• scale.max: 1 (for each vertex)
• partitions: 1 (for map vertex)
• max_threads=1 (for each udf grpc's)

Are they sufficient to keep ordering? Can some be left out?

---

apiVersion: numaflow.numaproj.io/v1alpha1
kind: Pipeline
spec:
  limits:
    readBatchSize: 1
  watermark:
    maxDelay: 10s
  vertices:
    - name: input
      scale:
        min: 1
        max: 1
      source:
        kafka: ...
        transformer:
          container:
            image: mysourcetransformer
    - name: mymap
      partitions: 1
      scale:
        min: 1
        max: 1
      udf:
        container:
          image: mymap
    - name: output
      scale:
        min: 1
        max: 1
      sink:
        blackhole: {}
  edges:
    - from: in
      to: map
    - from: map
      to: output

from pynumaflow.sourcetransformer import SourceTransformServer
from .mysourcetransformer import MySourceTransformer
grpc_server = SourceTransformServer(source_transform_instance=MySourceTransformer(), max_threads=1)
grpc_server.start()

from pynumaflow.mapper import MapServer
from .mymapper import MyMapper
grpc_server = MapServer(mapper_instance=MyMapper(), max_threads=1)
grpc_server.start()

The MyMapper goes something like this (heavily simplified, left out de/serialisation)

from pynumaflow.mapper import Mapper, Messages, Message, Datum

class MyMapper(Mapper):
    def __init__(self) -> None:
        """Mapper class able to hold previous state in memory."""
        self.exec_count: int = 0
        self.prev_datum: Datum | None = None
        super().__init__()

    def handler(self, keys: list[str], datum: Datum) -> Messages:
        # Some calculation with current datum and prev state
        if self.prev_datum:
            delta_value = datum.value.someval - self.prev_datum.someval
        else:

        # Update state
        self.exec_count += 1
        self.prev_datum = datum

        message = Message(value=delta_value, keys=keys, tags=[])
        return Messages(message)

[yhl25]

When the read batch size is greater than 1, we invoke udf concurrently, which may lead to out-of-order responses. Setting the read batch size to 1 and the max pod count to 1 should be good enough to maintain the order.

[th0ger]

Thanks @yhl25. What about max_threads, how does that affect ordering?

[yhl25]

max_threads doesn’t affect ordering when the read batch size is set to 1 because we invoke the UDF sequentially, processing one message at a time.