On Thu, Feb 11, 2021 at 9:38 PM Robert Bradshaw <[email protected]> wrote:
> Of course the right answer is to just implement sink triggers and sidestep > the question altogether :). > > In the meantime, I think leaving AfterSynchronizedProcessingTime in the > model makes the most sense, and runners can choose an implementation > between firing eagerly and waiting some amount of time until they think all > (most?) downstream results are in before firing, depending on how smart the > runner wants to be. As you point out, they're all correct, and we'll have > multiple firings due to the upstream trigger anyway, and this is safer than > it used to be (though still possibly requires work). > Just to clarify, as I got a little confused, is your suggestion: Leave AfterSynchronizedProcessingTime* triggers in the model/proto, let the SDK put them in where they want, and let runners decide how to interpret them? (this SGTM and requires the least/no changes) Kenn *noting that TimeDomain.SYNCHRONIZED_PROCESSING_TIME is not related to this, except in implementation, and should be removed either way. > On Wed, Feb 10, 2021 at 1:37 PM Kenneth Knowles <[email protected]> wrote: > >> Hi all, >> >> TL;DR: >> 1. should we replace "after synchronized processing time" with "after >> count 1"? >> 2. should we remove "continuation trigger" and leave this to runners? >> >> ---- >> >> "AfterSynchronizedProcessingTime" triggers were invented to solve a >> specific problem. They are inconsistent across SDKs today. >> >> - You have an aggregation/GBK with aligned processing time trigger like >> ("output every minute on the minute") >> - You have a downstream aggregation/GBK between that and the sink >> - You expect to have about one output every minute per key+window pair >> >> Any output of the upstream aggregation may contribute to any key+window >> of the downstream aggregation. The AfterSynchronizedProcessingTime trigger >> waits for all the processing time based triggers to fire and commit their >> outputs. The downstream aggregation will output as fast as possible in >> panes consistent with the upstream aggregation. >> >> - The Java SDK behavior is as above, to output "as fast as reasonable". >> - The Python SDK never uses "AfterSynchronizedProcessingTime" triggers >> but simply propagates the same trigger to the next GBK, creating additional >> delay. >> - I don't know what the Go SDK may do, if it supports this at all. >> >> Any behavior could be defined as "correct". A simple option could be to >> have the downstream aggregation "fire always" aka "after element count 1". >> How would this change things? We would potentially see many more outputs. >> >> Why did we do this in the first place? There are (at least) these reasons: >> >> - Previously, triggers could "finish" an aggregation thus dropping all >> further data. In this case, waiting for all outputs is critical or else you >> lose data. Now triggers cannot finish aggregations. >> - Whenever there may be more than one pane, a user has to write logic to >> compensate and deal with it. Changing from guaranteed single pane to >> multi-pane would break things. So if the user configures a single firing, >> all downstream aggregations must respect it. Now that triggers cannot >> finish, I think processing time can only be used in multi-pane contexts >> anyhow. >> - The above example illustrates how the behavior in Java maintains >> something that the user will expect. Or so we think. Maybe users don't care. >> >> How did we get into this inconsistent state? When the user specifies >> triggering it applies to the very nearest aggregation/GBK. The SDK decides >> what triggering to insert downstream. One possibility is to remove this and >> have it unspecified, left to runner behavior. >> >> I think maybe these pieces of complexity are both not helpful and also >> not (necessarily) breaking changes to alter, especially considering we have >> inconsistency in the model. >> >> WDYT? And I wonder what this means for xlang and portability... how does >> continuation triggering even work? (if at all) >> >> Kenn >> >
