Expose helper to construct dropped messages in accumulator (Message.to_drop(datum))

Summary

Expose a helper method on Message to construct a dropped message from a Datum, so users can explicitly emit drops from an accumulator (and other UDFs) to allow the watermark to progress and tracked windows/messages to be cleaned up.

Proposed signature:

Message.to_drop(datum)

Internally, this should construct a Message from the given Datum and set the drop tags on it, so the message is not forwarded to the next vertex but still allows the accumulator to advance the watermark and release tracked state.

Motivation

Today there is no clean way for an accumulator to signal "I have handled this datum but don't want to forward it." Users that omit output for some inputs (e.g. blackhole/filter/multiplexer-style accumulators) end up retaining tracked windows and messages, which leads to unbounded memory growth.

Related: #356

Proposed API

def my_accumulator(datums, output):
    async for datum in datums:
        if should_drop(datum):
            await output.put(Message.to_drop(datum))
            continue
        ...

Acceptance criteria

Message.to_drop(datum) constructs a Message from a Datum with drop tags set.
Emitting such a message from an accumulator progresses the watermark and cleans up tracked windows/messages.
Message is not forwarded to the next vertex.
Unit tests covering accumulator drop behavior.
Example/docs updated to show the drop pattern.

Not required anymore: numaproj/numaflow-java#218 (comment)

@yhl25

Hi @yhl25, @vaibhavtiwari33.
Can you explain why this changes is marked not needed?

We actually encounter the case described by OP, i.e. accumulator discarding selected events and output watermark not progressing.
It was fixed in our end by

            await output.put(
                Message(
                    value=b"",
                    keys=[],
                    tags=[DROP],
                    event_time=self.latest_wm,
                    watermark=self.latest_wm,
                )
            )

The syntax would be clearer if accumulator._dtypes.Message.to_drop() received a watermark arg.

Could you please specify the version being used here? The reasoning behind not adding the api can be found here

We're on latest numaflow 1.8.3 and pynumaflow 0.14.0.

The EOF implementation described in pynumaflow-go did not progress our watermarks. They were just stuck, and the pipeline_processing_lag kept growing indefinitely (hours and days).

After adding above DROP code, the watermark progresses.

That would be a bug IMO, would it be possible for you to create a bug report for this to allow us looking into this further?

numaproj/numaflow#3645 (comment)

Navigation Menu

Expose helper to construct dropped messages in accumulator (`Message.to_drop(datum)`) #357

Summary

Motivation

Proposed API

Acceptance criteria

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Expose helper to construct dropped messages in accumulator (Message.to_drop(datum)) #357

Description

Summary

Motivation

Proposed API

Acceptance criteria

Activity

vaibhavtiwari33 commented on Jun 17, 2026

th0ger commented on Sep 3, 2026

vaibhavtiwari33 commented on Sep 3, 2026

th0ger commented on Sep 4, 2026

vaibhavtiwari33 commented on Sep 4, 2026

vaibhavtiwari33 commented on Sep 21, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

Expose helper to construct dropped messages in accumulator (`Message.to_drop(datum)`) #357