[Batch][1.19] Order of elements within the same key in KeyedProcessFunction without event time

[lec ssmi <sh...@gmail.com>]

Hello Flink comminity,

When using a KeyedProcessFunction<K, IN, OUT>, what is the *processing
order* of the records *belonging to the same key* within a single batch?

Concretely:

   1. With *no event-time* defined, and running in *batch mode*, does
   KeyedProcessFunction process all elements of the same key in a
*deterministic
   order*?
   2. Is that order tied to the *upstream partition/file order*, or to
any *internal
   shuffle/sort* guarantees in batch execution (e.g., sort-shuffle /
   hash-shuffle), or is it *unspecified*?
   3. Is the per-key order *stable across runs* (same input & parallelism),
   or can it vary?
   4. If a stable/ascending order is required (e.g., by a field like ts or
   seq), what is the *recommended way* to enforce it before/inside
   KeyedProcessFunction in batch mode? (e.g., explicit sortPartition,
   pre-aggregation, or other operator/setting)

*Minimal sketch:*

env.setRuntimeMode(RuntimeExecutionMode.BATCH);
DataStream<MyEvent> input = env.fromSource(...); // no timestamps/watermarks

input
  .keyBy(e -> e.userId)
  .process(new KeyedProcessFunction<String, MyEvent, MyOut>() {
    @Override
    public void processElement(MyEvent value, Context ctx,
Collector<MyOut> out) {
      // Question: in what order will events with the same userId arrive here?
    }
  });

Any pointers to *docs / source code locations* (for 1.19 batch) that
describe the per-key ordering semantics would be highly appreciated. Thanks!

Best!

Re: [Batch][1.19] Order of elements within the same key in KeyedProcessFunction without event time

[Gabor Somogyi <ga...@gmail.com>]

Hi Lec,

In short the KeyedProcessFunction will see all elements of a key
back-to-back, but the sequence among those elements is unspecified and not
guaranteed stable.
The order can change across runs so any feature built at top of this is
discouraged.

In order to overcome this one option is to assign event timestamps and rely
on BATCH event-time processing.

BR,
G


On Tue, Aug 26, 2025 at 5:18 PM lec ssmi <sh...@gmail.com> wrote:

> Hello Flink comminity,
>
> When using a KeyedProcessFunction<K, IN, OUT>, what is the *processing
> order* of the records *belonging to the same key* within a single batch?
>
> Concretely:
>
>    1. With *no event-time* defined, and running in *batch mode*, does
>    KeyedProcessFunction process all elements of the same key in a *deterministic
>    order*?
>    2. Is that order tied to the *upstream partition/file order*, or to
>    any *internal shuffle/sort* guarantees in batch execution (e.g.,
>    sort-shuffle / hash-shuffle), or is it *unspecified*?
>    3. Is the per-key order *stable across runs* (same input &
>    parallelism), or can it vary?
>    4. If a stable/ascending order is required (e.g., by a field like ts
>    or seq), what is the *recommended way* to enforce it before/inside
>    KeyedProcessFunction in batch mode? (e.g., explicit sortPartition,
>    pre-aggregation, or other operator/setting)
>
> *Minimal sketch:*
>
> env.setRuntimeMode(RuntimeExecutionMode.BATCH);
> DataStream<MyEvent> input = env.fromSource(...); // no timestamps/watermarks
>
> input
>   .keyBy(e -> e.userId)
>   .process(new KeyedProcessFunction<String, MyEvent, MyOut>() {
>     @Override
>     public void processElement(MyEvent value, Context ctx, Collector<MyOut> out) {
>       // Question: in what order will events with the same userId arrive here?
>     }
>   });
>
> Any pointers to *docs / source code locations* (for 1.19 batch) that
> describe the per-key ordering semantics would be highly appreciated. Thanks!
>
> Best!
>