Expose helper to construct dropped messages in accumulator (`Message::to_drop`)

[yhl25]

Summary

Expose a helper on Message to construct a dropped message from an AccumulatorRequest, 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 API:

impl Message {
    /// Builds a Message from the given AccumulatorRequest with drop tags set,
    /// so the message is not forwarded to the next vertex but still allows the
    /// accumulator to advance the watermark and release tracked state.
    pub fn to_drop(request: AccumulatorRequest) -> Self;
}

This complements the existing Message::from_accumulator_request(request) helper used in examples/stream-sorter/src/main.rs.

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: numaproj/numaflow-python#356

Example usage

async fn accumulate(
    &self,
    mut input: mpsc::Receiver<AccumulatorRequest>,
    output: mpsc::Sender<Message>,
) {
    while let Some(request) = input.recv().await {
        if should_drop(&request) {
            let _ = output.send(Message::to_drop(request)).await;
            continue;
        }
        let _ = output.send(Message::from_accumulator_request(request)).await;
    }
}

Acceptance criteria

• Message::to_drop(request) constructs a Message from an AccumulatorRequest 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.

[vaibhavtiwari33]

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