Prevent dropping late messages in accumulator

[th0ger]

We use a streamsorter accumulator to sort events.
The source even_times can only be slightly unordered, so we have maxDelay: 10s on the pipeline spec.

The accumulator is dropping late events with DEBUG Late message detected, dropping, and we see the forwarder_drop_metric grow.
We figured out that we were missing the allowedLateness | API setting on the accumulator. So we just set it to a high value allowedLateness: 10m.

But, that caused us to now get a different ERROR Old message popped up, Watermark has progressed past event time, and forwarder_drop_metric still high.

The code in question should be (unaligned/reducer.rs):

// Handle late messages
if msg.is_late {
    if self.current_watermark.timestamp_millis() == -1 {
        // this can happen when the pipeline is just started, and we receive a late message
        // before we receive any watermark.
        warn!(event_time = ?msg.event_time.timestamp_millis(), watermark = ?self.current_watermark.timestamp_millis(), "Late message detected, but watermark is -1, dropping message");
        increment_late_message_drop_metric("no-valid-wm-late-message");
        return;
    }

    // Drop the late message if it is outside the allowed lateness window.
    if msg.event_time < self.current_watermark.sub(self.allowed_lateness) {
        debug!(event_time = ?msg.event_time.timestamp_millis(), watermark = ?self.current_watermark.timestamp_millis(), "Late message detected, dropping");
        increment_late_message_drop_metric("late-message");
        return;
    }
}

// Validate message event time against current watermark (must not have progressed past this message)
if self.current_watermark > msg.event_time {
    error!(
        current_watermark = ?self.current_watermark.timestamp_millis(),
        message_event_time = ?msg.event_time.timestamp_millis(),
        is_late = ?msg.is_late,
        offset = ?msg.offset,
        "Old message popped up, Watermark has progressed past event time"
    );
    increment_late_message_drop_metric("old-message-popped-up");
    return;
}

The intent of the first block is the drop is_late events, unless they are still within the allowed_lateness range.

But, the second blok still catches any message self.current_watermark > msg.event_timewith (including is_late messages), despite the events being inside the allowedLatenessrange.
So, the allowed_lateness setting seems to have no effect for late messages, except for a different log message.

[th0ger]

As a small side note, this code seems counter intuitive:

error!(
    ...
    is_late = ?msg.is_late,
    ...
);
increment_late_message_drop_metric("old-message-popped-up");

This part of the code can be reached for both is_late == true/false, and the error parameter emphasizes that.
But the function name increment_late_message_drop_metric indicates that it's used to hande late messages only.
The taget metric forwarder_drop_total indicates a more general metric, where drops could occur due to is_late or not.

Maybe the helper function increment_late_message_drop_metric should be called increment_forwarder_drop_metric?

[yhl25]

Looks like a bug to me. Do you see is_late being set to true in the logs for these dropped messages?

[th0ger]

Yes is_late is true.

[yhl25]

We should avoid old message check for late messages, this PR should fix it. We will track it as part of v1.9.1 release. Thanks for reporting!

[th0ger]

Excellent, I was thinking of the same solution of wrapping the old-message-popped-up in an else block.

We're about to upgrade to v1.9.0, but will postpone to v1.9.1 then. Hope it will be out soon?