fix: include dropped messages in source watermark calculation #1404
Conversation
Signed-off-by: Keran Yang <[email protected]>
| @@ -71,7 +71,7 @@ func New(args map[string]string) (sourcetransformer.SourceTransformFunc, error) | |||
| func (e expressions) apply(et time.Time, payload []byte) (sourcetransformer.Message, error) { | |||
| result, err := expr.EvalBool(e.filterExpr, payload) | |||
| if err != nil { | |||
| return sourcetransformer.MessageToDrop(), err | |||
| return sourcetransformer.MessageToDrop(et), err | |||
There was a problem hiding this comment.
This means the dropped messages use the default event time coming from the original datum?
There was a problem hiding this comment.
Yes. Thanks for raising this question. It seems using the extracted event time should be more accurate. WDYT?
There was a problem hiding this comment.
et is "almost" accurate because we subtract maxDelay before publishing the watermark.
One could also argue that, since this is an "error", perhaps we should stick to -1 watermark.
Let's change this in the future based on the feedback we receive.
There was a problem hiding this comment.
If we use default et (or -1), then why don't we use the same in the original MessageToDrop()?
There was a problem hiding this comment.
This is just for the builtin extractEventTimeFilter, where we choose to assign the original event time to dropped messages. For other transformers, users have to provide a new event time.
For extractEventTimeFilter, I think we should also assign the new event time if it has been successfully extracted from the payload. But I am also ok with making the change in the future after feedbacks.
closes #1357