That is a good point..
I cannot think of a better option than documentation and warning, and also
given that we'd probably better not reusing the function name `until` for
close time.
Guozhang
On Tue, Jul 10, 2018 at 3:31 PM, John Roesler <j...@confluent.io> wrote:
> I had some opportunity to reflect on the default for close time today...
>
> Note that the current "close time" is equal to the retention time, and
> therefore "close" today shares the default retention of 24h.
>
> It would definitely break any application that today specifies a retention
> time to set close shorter than that time. It's also likely to break apps if
> they *don't* set the retention time and rely on the 24h default. So it's
> unfortunate, but I think if "close" isn't set, we should use the retention
> time instead of a fixed default.
>
> When we ultimately remove the retention time parameter ("until"), we will
> have to set "close" to a default of 24h.
>
> Of course, this has a negative impact on the user of "final results", since
> they won't see any output at all for retentionTime/24h, and may find this
> confusing. What can we do about this except document it well? Maybe log a
> warning if we see that close wasn't explicitly set while using "final
> results"?
>
> Thanks,
> -John
>
> On Tue, Jul 10, 2018 at 10:46 AM John Roesler <j...@confluent.io> wrote:
>
> > Hi Guozhang,
> >
> > That sounds good to me. I'll include that in the KIP.
> >
> > Thanks,
> > -John
> >
> > On Mon, Jul 9, 2018 at 6:33 PM Guozhang Wang <wangg...@gmail.com> wrote:
> >
> >> Let me clarify a bit on what I meant about moving `retentionPeriod` to
> >> WindowStoreBuilder:
> >>
> >> In another discussion we had around KIP-319 / 330, that the "retention
> >> period" should not really be a window spec, but only a window store
> spec,
> >> as it only affects how long to retain each window to be queryable along
> >> with the storage cost.
> >>
> >> More specifically, today the "maintainMs" returned from Windows is used
> in
> >> three places:
> >>
> >> 1) for windowed aggregations, they are passed in directly into
> >> `Stores.persistentWindows()` as the retention period parameters. For
> this
> >> use case we should just let the WindowStoreBuilder to specify this value
> >> itself.
> >>
> >> NOTE: It is also returned in the KStreamWindowAggregate processor, to
> >> determine if a received record should be dropped due to its lateness. We
> >> may need to think of another way to get this value inside the processor
> >>
> >> 2) for windowed stream-stream join, it is used as the join range
> parameter
> >> but only to check that "windowSizeMs <= retentionPeriodMs". We can do
> this
> >> check at the store builder lever instead of at the processor level.
> >>
> >>
> >> If we can remove its usage in both 1) and 2), then we should be able to
> >> safely remove this from the `Windows` spec.
> >>
> >>
> >> Guozhang
> >>
> >>
> >> On Mon, Jul 9, 2018 at 3:53 PM, John Roesler <j...@confluent.io> wrote:
> >>
> >> > Thanks for the reply, Guozhang,
> >> >
> >> > Good! I agree, that is also a good reason, and I actually made use of
> >> that
> >> > in my tests. I'll update the KIP.
> >> >
> >> > By the way, I chose "allowedLateness" as I was trying to pick a better
> >> name
> >> > than "close", but I think it's actually the wrong name. We don't want
> to
> >> > bound the lateness of events in general, only with respect to the end
> of
> >> > their window.
> >> >
> >> > If we have a window [0,10), with "allowedLateness" of 5, then if we
> get
> >> an
> >> > event with timestamp 3 at time 9, the name implies we'd reject it,
> which
> >> > seems silly. Really, we'd only want to start rejecting that event at
> >> stream
> >> > time 15.
> >> >
> >> > What I meant was more like "allowedLatenessAfterWindowEnd", but
> that's
> >> too
> >> > verbose. I think that "close" + some documentation about what it means
> >> will
> >> > be better.
> >> >
> >> > 1: "Close" would be measured from the end of the window, so a
> reasonable
> >> > default would be "0". Recall that "close" really only needs to be
> >> specified
> >> > for final results, and a default of 0 would produce the most intuitive
> >> > results. If folks later discover that they are missing some late
> events,
> >> > they can adjust the parameter accordingly. IMHO, any other value would
> >> just
> >> > be a guess on our part.
> >> >
> >> > 2a:
> >> > I think you're saying to re-use "until" instead of adding "close" to
> the
> >> > window.
> >> >
> >> > The downside here would be that the semantic change could be more
> >> confusing
> >> > than deprecating "until" and introducing window "close" and a
> >> > "retentionTime" on the store builder. The deprecation is a good,
> >> controlled
> >> > way for us to make sure people are getting the semantics they think
> >> they're
> >> > getting, as well as giving us an opportunity to link people to the API
> >> they
> >> > should use instead.
> >> >
> >> > I didn't fully understand the second part, but it sounds like you're
> >> > suggesting to add a new "retentionTime" setter to Windows to bridge
> the
> >> gap
> >> > until we add it to the store builder? That seems kind of roundabout to
> >> me,
> >> > if that's what you meant. We could just immediately add it to the
> store
> >> > builders in the same PR.
> >> >
> >> > 2b: Sounds good to me!
> >> >
> >> > Thanks again,
> >> > -John
> >> >
> >> >
> >> > On Mon, Jul 9, 2018 at 4:55 PM Guozhang Wang <wangg...@gmail.com>
> >> wrote:
> >> >
> >> > > John,
> >> > >
> >> > > Thanks for your replies. As for the two options of the API, I think
> >> I'm
> >> > > slightly inclined to the first option as well. My motivation is a
> bit
> >> > > different, as I think of the first one maybe more flexible, for
> >> example:
> >> > >
> >> > > KTable<Windowed<..>> table = ... count();
> >> > >
> >> > > table.toStream().peek(..); // want to peek at the changelog
> stream,
> >> do
> >> > > not care about final results.
> >> > >
> >> > > table.suppress().toStream().to("topic"); // sending to a topic,
> >> want
> >> > to
> >> > > only send the final results.
> >> > >
> >> > > --------------
> >> > >
> >> > > Besides that, I have a few more minor questions:
> >> > >
> >> > > 1. For "allowedLateness", what should be the default value? I.e. if
> >> user
> >> > do
> >> > > not specify "allowedLateness" in TimeWindows, what value should we
> >> set?
> >> > >
> >> > > 2. For API names, some personal suggestions here:
> >> > >
> >> > > 2.a) "allowedLateness" -> "until" (semantics changed, and also
> value
> >> is
> >> > > defined as delta on top of window length), where "until" ->
> >> > > "retentionPeriod", and the latter will be removed from `Windows` to
> `
> >> > > WindowStoreBuilder` in the future.
> >> > >
> >> > > 2.b) "BufferConfig" -> "Buffered" ?
> >> > >
> >> > >
> >> > >
> >> > > Guozhang
> >> > >
> >> > >
> >> > > On Mon, Jul 9, 2018 at 2:09 PM, John Roesler <j...@confluent.io>
> >> wrote:
> >> > >
> >> > > > Hey Matthias and Guozhang,
> >> > > >
> >> > > > Sorry for the slow reply. I was mulling about your feedback and
> >> > weighing
> >> > > > some ideas in a sketchbook PR: https://github.com/apache/
> >> > kafka/pull/5337
> >> > > .
> >> > > >
> >> > > > Your thought about keeping suppression independent of business
> logic
> >> > is a
> >> > > > very good one. I agree that it would make more sense to add some
> >> kind
> >> > of
> >> > > > "window close" concept to the window definition.
> >> > > >
> >> > > > In fact, doing that immediately solves the inconsistency problem
> >> > Guozhang
> >> > > > brought up. There's no need to add a "final results" or "emission"
> >> > option
> >> > > > to the windowed aggregation.
> >> > > >
> >> > > > What do you think about an API more like this:
> >> > > >
> >> > > > final StreamsBuilder builder = new StreamsBuilder();
> >> > > >
> >> > > > builder
> >> > > > .stream("input", Consumed.with(STRING_SERDE, STRING_SERDE))
> >> > > > .groupBy(
> >> > > > (String k1, String v1) -> k1,
> >> > > > Serialized.with(STRING_SERDE, STRING_SERDE)
> >> > > > )
> >> > > > .windowedBy(TimeWindows
> >> > > > .of(scaledTime(2L))
> >> > > > .until(scaledTime(3L))
> >> > > > .allowedLateness(scaledTime(1L))
> >> > > > )
> >> > > > .count(Materialized.as("counts"))
> >> > > > .suppress(
> >> > > > emitFinalResultsOnly(
> >> > > > BufferConfig.withBufferKeys(10_000L).bufferFullStrategy(
> >> > SHUT_DOWN)
> >> > > > )
> >> > > > )
> >> > > > .toStream()
> >> > > > .to("output-suppressed", Produced.with(STRING_SERDE,
> LONG_SERDE));
> >> > > >
> >> > > > Note that:
> >> > > > * "emitFinalResultsOnly" is available *only* on windowed tables
> >> > > (enforced
> >> > > > by the type system at compile time), and it determines the time to
> >> wait
> >> > > by
> >> > > > looking at "allowedLateness" on the TimeWindows config.
> >> > > > * querying "counts" will produce results (eventually) consistent
> >> with
> >> > > > what's observable in "output-suppressed".
> >> > > > * in all cases, "suppress" has no effect on business logic, just
> on
> >> > > event
> >> > > > suppression.
> >> > > >
> >> > > > Is this API straightforward? Or do you still prefer the version
> that
> >> > both
> >> > > > proposed:
> >> > > >
> >> > > > ...
> >> > > > .windowedBy(TimeWindows
> >> > > > .of(scaledTime(2L))
> >> > > > .until(scaledTime(3L))
> >> > > > .allowedLateness(scaledTime(1L))
> >> > > > )
> >> > > > .count(
> >> > > > Materialized.as("counts"),
> >> > > > emitFinalResultsOnly(
> >> > > > BufferConfig.withBufferKeys(10_000L).bufferFullStrategy(
> >> > SHUT_DOWN)
> >> > > > )
> >> > > > )
> >> > > > ...
> >> > > >
> >> > > > To me, these two are practically identical, and I still vaguely
> >> prefer
> >> > > the
> >> > > > first one.
> >> > > >
> >> > > > The prototype has made clearer to me that users of "final results
> >> for
> >> > > > windows" and users of "suppression for table events" both need to
> >> > > configure
> >> > > > the suppression buffer.
> >> > > >
> >> > > > This buffer configuration consists of:
> >> > > > 1. how many keys or bytes to keep in memory
> >> > > > 2. what to do if memory runs out (shut down, start using disk,
> ...)
> >> > > >
> >> > > > So it's not as simple as setting a "final results" flag. We'll
> >> either
> >> > > have
> >> > > > an "Emit" config object on the windowed aggregators that takes the
> >> same
> >> > > > BufferConfig that the "Suppress" config on the suppression
> >> operator, or
> >> > > we
> >> > > > just use the suppression operator for both.
> >> > > >
> >> > > > Perhaps it would sweeten the deal a little to point out that we
> >> have 2
> >> > > > overloads already for each windowed aggregator (with and without
> >> > > > Materialized). Adding "Emitted" or something would mean that we'd
> >> add a
> >> > > new
> >> > > > overload for each one, taking us up to 4 overloads each for
> "count",
> >> > > > "aggregate" and "reduce". Using "suppress" means that we don't add
> >> any
> >> > > new
> >> > > > overloads.
> >> > > >
> >> > > > Thanks again for helping to hash this out,
> >> > > > -John
> >> > > >
> >> > > > On Fri, Jul 6, 2018 at 6:20 PM Guozhang Wang <wangg...@gmail.com>
> >> > wrote:
> >> > > >
> >> > > > > I think I agree with Matthias for having dedicated APIs for
> >> windowed
> >> > > > > operation final output scenario, PLUS separating the window
> close
> >> > which
> >> > > > the
> >> > > > > "final output" would rely on, from the window retention time
> >> itself
> >> > > > > (admittedly it would make this KIP effort larger, but if we
> >> believe
> >> > we
> >> > > > need
> >> > > > > to do this separation anyways we could just do it now).
> >> > > > >
> >> > > > > And then we can have the `KTable#suppress()` for
> >> > > intermediate-suppression
> >> > > > > only, not for late-record-suppression, until we've seen that
> >> becomes
> >> > a
> >> > > > > common feature request because our current design still allows
> to
> >> be
> >> > > > > extended for that purpose.
> >> > > > >
> >> > > > >
> >> > > > > Guozhang
> >> > > > >
> >> > > > > On Wed, Jul 4, 2018 at 12:53 PM, Matthias J. Sax <
> >> > > matth...@confluent.io>
> >> > > > > wrote:
> >> > > > >
> >> > > > > > Thanks for the discussion. I am just catching up.
> >> > > > > >
> >> > > > > > In general, I think we have different uses cases and
> >> non-windowed
> >> > and
> >> > > > > > windowed is quite different. For the non-windowed case,
> >> suppress()
> >> > > has
> >> > > > > > no (useful) close or retention time, no final semantics, and
> >> also
> >> > no
> >> > > > > > business logic impact.
> >> > > > > >
> >> > > > > > On the other hand, for windowed aggregations, close time and
> >> final
> >> > > > > > result do have a meaning. IMHO, `close()` is part of business
> >> logic
> >> > > > > > while retention time is not. Also, suppression of intermediate
> >> > result
> >> > > > is
> >> > > > > > not a business rule and there might be use case for which
> either
> >> > > "early
> >> > > > > > intermediate" (before window end time) are suppressed only, or
> >> all
> >> > > > > > intermediates are suppressed (maybe also something in the
> >> middle,
> >> > ie,
> >> > > > > > just reduce the load of intermediate updates). Thus,
> >> > > window-suppression
> >> > > > > > is much richer.
> >> > > > > >
> >> > > > > > IMHO, a generic `suppress()` operator that can be inserted
> into
> >> the
> >> > > > data
> >> > > > > > flow at any point is useful. Maybe we should keep is as
> generic
> >> as
> >> > > > > > possible. However, it might be difficult to use with regard to
> >> > > > > > windowing, as the mental effort to use it is high.
> >> > > > > >
> >> > > > > > With regard to Guozhang's comment:
> >> > > > > >
> >> > > > > > > we will actually
> >> > > > > > > process data as old as 30 days as well, while most of the
> late
> >> > > > updates
> >> > > > > > > beyond 5 minutes would be discarded anyways.
> >> > > > > >
> >> > > > > > If we use `suppress()` as a standalone operator, this is
> correct
> >> > and
> >> > > > > > intended IMHO. To address the issue if the behavior is
> >> unwanted, I
> >> > > > would
> >> > > > > > suggest to add a "suppress option" directly to
> >> > > > > > `count()/reduce()/aggregate()` window operator similar to
> >> > > > > > `Materialized`. This would be an "embedded suppress" and avoid
> >> the
> >> > > > > > issue. It would also address the issue about mental effort for
> >> > > "single
> >> > > > > > final window result" use case.
> >> > > > > >
> >> > > > > > I also think that a shorter close-time than retention time is
> >> > useful
> >> > > > for
> >> > > > > > window aggregation. If we add close() to the window definition
> >> and
> >> > > > > > until() to `Materialized`, we can separate both correctly
> IMHO.
> >> > > > > >
> >> > > > > > About setting `close = min(close,retention)` I am not sure. We
> >> > might
> >> > > > > > rather throw an exception than reducing the close time
> >> > automatically.
> >> > > > > > Otherwise, I see many user question about "I set close to X
> but
> >> it
> >> > > does
> >> > > > > > not get updated for some data that is with delay of X".
> >> > > > > >
> >> > > > > > The tricky question might be to design the API in a backward
> >> > > compatible
> >> > > > > > way though.
> >> > > > > >
> >> > > > > >
> >> > > > > >
> >> > > > > > -Matthias
> >> > > > > >
> >> > > > > > On 7/3/18 5:38 AM, John Roesler wrote:
> >> > > > > > > Hi Guozhang,
> >> > > > > > >
> >> > > > > > > I see. It seems like if we want to decouple 1) and 2), we
> >> need to
> >> > > > alter
> >> > > > > > the
> >> > > > > > > definition of the window. Do you think it would close the
> gap
> >> if
> >> > we
> >> > > > > > added a
> >> > > > > > > "window close" time to the window definition?
> >> > > > > > >
> >> > > > > > > Such as:
> >> > > > > > >
> >> > > > > > > builder.stream("input")
> >> > > > > > > .groupByKey()
> >> > > > > > > .windowedBy(
> >> > > > > > > TimeWindows
> >> > > > > > > .of(60_000)
> >> > > > > > > .closeAfter(10 * 60)
> >> > > > > > > .until(30L * 24 * 60 * 60 * 1000)
> >> > > > > > > )
> >> > > > > > > .count()
> >> > > > > > > .suppress(Suppression.finalResultsOnly());
> >> > > > > > >
> >> > > > > > > Possibly called "finalResultsAtWindowClose" or something?
> >> > > > > > >
> >> > > > > > > Thanks,
> >> > > > > > > -John
> >> > > > > > >
> >> > > > > > > On Mon, Jul 2, 2018 at 6:50 PM Guozhang Wang <
> >> wangg...@gmail.com
> >> > >
> >> > > > > wrote:
> >> > > > > > >
> >> > > > > > >> Hey John,
> >> > > > > > >>
> >> > > > > > >> Obviously I'm too lazy on email replying diligence compared
> >> with
> >> > > you
> >> > > > > :)
> >> > > > > > >> Will try to reply them separately:
> >> > > > > > >>
> >> > > > > > >>
> >> > > > > > >> ------------------------------
> ------------------------------
> >> > > > > > -----------------
> >> > > > > > >>
> >> > > > > > >> To reply your email on "Mon, Jul 2, 2018 at 8:23 AM":
> >> > > > > > >>
> >> > > > > > >> I'm aware of this use case, but again, the concern is that,
> >> in
> >> > > this
> >> > > > > > setting
> >> > > > > > >> in order to let the window be queryable for 30 days, we
> will
> >> > > > actually
> >> > > > > > >> process data as old as 30 days as well, while most of the
> >> late
> >> > > > updates
> >> > > > > > >> beyond 5 minutes would be discarded anyways. Personally I
> >> think
> >> > > for
> >> > > > > the
> >> > > > > > >> final update scenario, the ideal situation users would want
> >> is
> >> > > that
> >> > > > > "do
> >> > > > > > not
> >> > > > > > >> process any data that is less than 5 minutes, and of course
> >> no
> >> > > > update
> >> > > > > > >> records to the downstream later than 5 minutes either; but
> >> > retain
> >> > > > the
> >> > > > > > >> window to be queryable for 30 days". And by doing that the
> >> final
> >> > > > > window
> >> > > > > > >> snapshot would also be aligned with the update stream as
> >> well.
> >> > In
> >> > > > > other
> >> > > > > > >> words, among these three periods:
> >> > > > > > >>
> >> > > > > > >> 1) the retention length of the window / table.
> >> > > > > > >> 2) the late records acceptance for updating the window.
> >> > > > > > >> 3) the late records update to be sent downstream.
> >> > > > > > >>
> >> > > > > > >> Final update use cases would naturally want 2) = 3), while
> 1)
> >> > may
> >> > > be
> >> > > > > > >> different and larger, while what we provide now is that 1)
> =
> >> 2),
> >> > > > which
> >> > > > > > >> could be different and in practice larger than 3), hence
> not
> >> the
> >> > > > most
> >> > > > > > >> intuitive for their needs.
> >> > > > > > >>
> >> > > > > > >>
> >> > > > > > >>
> >> > > > > > >> ------------------------------
> ------------------------------
> >> > > > > > -----------------
> >> > > > > > >>
> >> > > > > > >> To reply your email on "Mon, Jul 2, 2018 at 10:27 AM":
> >> > > > > > >>
> >> > > > > > >> I'd like option 2) over option 1) better as well from
> >> > programming
> >> > > > pov.
> >> > > > > > But
> >> > > > > > >> I'm wondering if option 2) would provide the above
> semantics
> >> or
> >> > it
> >> > > > is
> >> > > > > > still
> >> > > > > > >> coupling 1) with 2) as well ?
> >> > > > > > >>
> >> > > > > > >>
> >> > > > > > >>
> >> > > > > > >> Guozhang
> >> > > > > > >>
> >> > > > > > >>
> >> > > > > > >>
> >> > > > > > >>
> >> > > > > > >> On Mon, Jul 2, 2018 at 1:08 PM, John Roesler <
> >> j...@confluent.io
> >> > >
> >> > > > > wrote:
> >> > > > > > >>
> >> > > > > > >>> In fact, to push the idea further (which IIRC is what
> >> Matthias
> >> > > > > > originally
> >> > > > > > >>> proposed), if we can accept "Suppression#finalResultsOnly"
> >> in
> >> > my
> >> > > > last
> >> > > > > > >>> email, then we could also consider whether to eliminate
> >> > > > > > >>> "suppressLateEvents" entirely.
> >> > > > > > >>>
> >> > > > > > >>> We could always add it later, but you've both expressed
> >> doubt
> >> > > that
> >> > > > > > there
> >> > > > > > >>> are practical use cases for it outside of final-results.
> >> > > > > > >>>
> >> > > > > > >>> -John
> >> > > > > > >>>
> >> > > > > > >>> On Mon, Jul 2, 2018 at 12:27 PM John Roesler <
> >> > j...@confluent.io>
> >> > > > > > wrote:
> >> > > > > > >>>
> >> > > > > > >>>> Hi again, Guozhang ;) Here's the second part of my
> >> response...
> >> > > > > > >>>>
> >> > > > > > >>>> It seems like your main concern is: "if I'm a user who
> >> wants
> >> > > final
> >> > > > > > >> update
> >> > > > > > >>>> semantics, how complicated is it for me to get it?"
> >> > > > > > >>>>
> >> > > > > > >>>> I think we have to assume that people don't always have
> >> time
> >> > to
> >> > > > > become
> >> > > > > > >>>> deeply familiar with all the nuances of a programming
> >> > > environment
> >> > > > > > >> before
> >> > > > > > >>>> they use it. Especially if they're evaluating several
> >> > frameworks
> >> > > > for
> >> > > > > > >>> their
> >> > > > > > >>>> use case, it's very valuable to make it as obvious as
> >> possible
> >> > > how
> >> > > > > to
> >> > > > > > >>>> accomplish various computations with Streams.
> >> > > > > > >>>>
> >> > > > > > >>>> To me the biggest question is whether with a fresh
> >> > perspective,
> >> > > > > people
> >> > > > > > >>>> would say "oh, I get it, I have to bound my lateness and
> >> > > suppress
> >> > > > > > >>>> intermediate updates, and of course I'll get only the
> final
> >> > > > > result!",
> >> > > > > > >> or
> >> > > > > > >>> if
> >> > > > > > >>>> it's more like "wtf? all I want is the final result, what
> >> are
> >> > > all
> >> > > > > > these
> >> > > > > > >>>> parameters?".
> >> > > > > > >>>>
> >> > > > > > >>>> I was talking with Matthias a while back, and he had an
> >> idea
> >> > > that
> >> > > > I
> >> > > > > > >> think
> >> > > > > > >>>> can help, which is to essentially set up a final-result
> >> recipe
> >> > > in
> >> > > > > > >>> addition
> >> > > > > > >>>> to the raw parameters. I previously thought that it
> >> wouldn't
> >> > be
> >> > > > > > >> possible
> >> > > > > > >>> to
> >> > > > > > >>>> restrict its usage to Windowed KTables, but thinking
> about
> >> it
> >> > > > again
> >> > > > > > >> this
> >> > > > > > >>>> weekend, I have a couple of ideas:
> >> > > > > > >>>>
> >> > > > > > >>>> ================
> >> > > > > > >>>> = 1. Static Wrapper =
> >> > > > > > >>>> ================
> >> > > > > > >>>> We can define an extra static function that "wraps" a
> >> KTable
> >> > > with
> >> > > > > > >>>> final-result semantics.
> >> > > > > > >>>>
> >> > > > > > >>>> public static <K extends Windowed, V> KTable<K, V>
> >> > > > finalResultsOnly(
> >> > > > > > >>>> final KTable<K, V> windowedKTable,
> >> > > > > > >>>> final Duration maxAllowedLateness,
> >> > > > > > >>>> final Suppression.BufferFullStrategy
> bufferFullStrategy)
> >> {
> >> > > > > > >>>> return windowedKTable.suppress(
> >> > > > > > >>>> Suppression.suppressLateEvents(
> maxAllowedLateness)
> >> > > > > > >>>> .suppressIntermediateEvents(
> >> > > > > > >>>> IntermediateSuppression
> >> > > > > > >>>> .emitAfter(maxAllowedLateness)
> >> > > > > > >>>> .bufferFullStrategy(
> >> > bufferFullStrategy)
> >> > > > > > >>>> )
> >> > > > > > >>>> );
> >> > > > > > >>>> }
> >> > > > > > >>>>
> >> > > > > > >>>> Because windowedKTable is a parameter, the static
> function
> >> can
> >> > > > > easily
> >> > > > > > >>>> impose an extra bound on the key type, that it extends
> >> > Windowed.
> >> > > > > This
> >> > > > > > >>> would
> >> > > > > > >>>> make "final results only" only available on windowed
> >> ktables.
> >> > > > > > >>>>
> >> > > > > > >>>> Here's how it would look to use:
> >> > > > > > >>>>
> >> > > > > > >>>> final KTable<Windowed<Integer>, Long> windowCounts = ...
> >> > > > > > >>>> final KTable<Windowed<Integer>, Long> finalCounts =
> >> > > > > > >>>> finalResultsOnly(
> >> > > > > > >>>> windowCounts,
> >> > > > > > >>>> Duration.ofMinutes(10),
> >> > > > > > >>>> Suppression.BufferFullStrategy.SHUT_DOWN
> >> > > > > > >>>> );
> >> > > > > > >>>>
> >> > > > > > >>>> Trying to use it on a non-windowed KTable yields:
> >> > > > > > >>>>
> >> > > > > > >>>>> Error:(129, 35) java: method finalResultsOnly in class
> >> > > > > > >>>>> org.apache.kafka.streams.kstream.internals.
> >> > KTableAggregateTest
> >> > > > > > cannot
> >> > > > > > >>> be
> >> > > > > > >>>>> applied to given types;
> >> > > > > > >>>>> required:
> >> > > > > > >>>>> org.apache.kafka.streams.kstream.KTable<K,V>,java.time.
> >> > > > > > >>> Duration,org.apache.kafka.streams.kstream.Suppression.
> >> > > > > > BufferFullStrategy
> >> > > > > > >>>>> found:
> >> > > > > > >>>>> org.apache.kafka.streams.kstream.KTable<java.lang.
> >> > > > > > >>> String,java.lang.String>,java.time.Duration,org.apache.
> >> > > > > > >>> kafka.streams.kstream.Suppression.BufferFullStrategy
> >> > > > > > >>>>> reason: inference variable K has incompatible bounds
> >> > > > > > >>>>> equality constraints: java.lang.String
> >> > > > > > >>>>> upper bounds:
> >> org.apache.kafka.streams.kstream.Windowed
> >> > > > > > >>>>
> >> > > > > > >>>>
> >> > > > > > >>>>
> >> > > > > > >>>> =================================================
> >> > > > > > >>>> = 2. Add <K,V> parameters and recipe method to
> Suppression
> >> =
> >> > > > > > >>>> =================================================
> >> > > > > > >>>>
> >> > > > > > >>>> By adding K,V parameters to Suppression, we can provide a
> >> > > > similarly
> >> > > > > > >>>> bounded config method directly on the Suppression class:
> >> > > > > > >>>>
> >> > > > > > >>>> public static <K extends Windowed, V> Suppression<K, V>
> >> > > > > > >>>> finalResultsOnly(final Duration maxAllowedLateness, final
> >> > > > > > >>>> BufferFullStrategy bufferFullStrategy) {
> >> > > > > > >>>> return Suppression
> >> > > > > > >>>> .<K, V>suppressLateEvents(maxAllowedLateness)
> >> > > > > > >>>> .suppressIntermediateEvents(
> IntermediateSuppression
> >> > > > > > >>>> .emitAfter(maxAllowedLateness)
> >> > > > > > >>>> .bufferFullStrategy(bufferFullStrategy)
> >> > > > > > >>>> );
> >> > > > > > >>>> }
> >> > > > > > >>>>
> >> > > > > > >>>> Then, here's how it would look to use it:
> >> > > > > > >>>>
> >> > > > > > >>>> final KTable<Windowed<Integer>, Long> windowCounts = ...
> >> > > > > > >>>> final KTable<Windowed<Integer>, Long> finalCounts =
> >> > > > > > >>>> windowCounts.suppress(
> >> > > > > > >>>> Suppression.finalResultsOnly(
> >> > > > > > >>>> Duration.ofMinutes(10)
> >> > > > > > >>>> Suppression.BufferFullStrategy.SHUT_DOWN
> >> > > > > > >>>> )
> >> > > > > > >>>> );
> >> > > > > > >>>>
> >> > > > > > >>>> Trying to use it on a non-windowed ktable yields:
> >> > > > > > >>>>
> >> > > > > > >>>>> Error:(127, 35) java: method finalResultsOnly in class
> >> > > > > > >>>>> org.apache.kafka.streams.kstream.Suppression<K,V>
> cannot
> >> be
> >> > > > applied
> >> > > > > > to
> >> > > > > > >>>>> given types;
> >> > > > > > >>>>> required:
> >> > > > > > >>>>> java.time.Duration,org.apache.kafka.streams.kstream.
> >> > > > > > >>> Suppression.BufferFullStrategy
> >> > > > > > >>>>> found:
> >> > > > > > >>>>> java.time.Duration,org.apache.kafka.streams.kstream.
> >> > > > > > >>> Suppression.BufferFullStrategy
> >> > > > > > >>>>> reason: explicit type argument java.lang.String does
> not
> >> > > > conform
> >> > > > > to
> >> > > > > > >>>>> declared bound(s)
> >> org.apache.kafka.streams.kstream.Windowed
> >> > > > > > >>>>
> >> > > > > > >>>>
> >> > > > > > >>>>
> >> > > > > > >>>> ============
> >> > > > > > >>>> = Downsides =
> >> > > > > > >>>> ============
> >> > > > > > >>>>
> >> > > > > > >>>> Of course, there's a downside either way:
> >> > > > > > >>>> * for 1: this "wrapper" interaction would be the first
> in
> >> the
> >> > > > DSL.
> >> > > > > Is
> >> > > > > > >> it
> >> > > > > > >>>> too strange, and how discoverable would it be?
> >> > > > > > >>>> * for 2: adding those type parameters to Suppression will
> >> > force
> >> > > > all
> >> > > > > > >>>> callers to provide them in the event of a chained
> >> construction
> >> > > > > because
> >> > > > > > >>> Java
> >> > > > > > >>>> doesn't do RHS recursive type inference. This is already
> >> > visible
> >> > > > in
> >> > > > > > >> other
> >> > > > > > >>>> parts of the Streams DSL. For example, often calls to
> >> > > Materialized
> >> > > > > > >>> builders
> >> > > > > > >>>> have to provide seemingly obvious type bounds.
> >> > > > > > >>>>
> >> > > > > > >>>> ============
> >> > > > > > >>>> = Conclusion =
> >> > > > > > >>>> ============
> >> > > > > > >>>>
> >> > > > > > >>>> I think option 2 is more "normal" and discoverable. It
> does
> >> > > have a
> >> > > > > > >>>> downside, but it's one that's pre-existing elsewhere in
> the
> >> > DSL.
> >> > > > > > >>>>
> >> > > > > > >>>> WDYT? Would the addition of this "recipe" method to
> >> > Suppression
> >> > > > > > resolve
> >> > > > > > >>>> your concern?
> >> > > > > > >>>>
> >> > > > > > >>>> Thanks again,
> >> > > > > > >>>> -John
> >> > > > > > >>>>
> >> > > > > > >>>> On Sun, Jul 1, 2018 at 11:24 PM Guozhang Wang <
> >> > > wangg...@gmail.com
> >> > > > >
> >> > > > > > >>> wrote:
> >> > > > > > >>>>
> >> > > > > > >>>>> Hi John,
> >> > > > > > >>>>>
> >> > > > > > >>>>> Regarding the metrics: yeah I think I'm with you that
> the
> >> > > dropped
> >> > > > > > >>> records
> >> > > > > > >>>>> due to window retention or emit suppression policies
> >> should
> >> > be
> >> > > > > > >> recorded
> >> > > > > > >>>>> differently, and using this KIP's proposed metric would
> be
> >> > > fine.
> >> > > > If
> >> > > > > > >> you
> >> > > > > > >>>>> also think we can use this KIP's proposed metrics to
> cover
> >> > the
> >> > > > > window
> >> > > > > > >>>>> retention cased skipping records, then we can include
> the
> >> > > changes
> >> > > > > in
> >> > > > > > >>> this
> >> > > > > > >>>>> KIP as well.
> >> > > > > > >>>>>
> >> > > > > > >>>>> Regarding the current proposal, I'm actually not too
> >> worried
> >> > > > about
> >> > > > > > the
> >> > > > > > >>>>> inconsistency between query semantics and downstream
> emit
> >> > > > > semantics.
> >> > > > > > >> For
> >> > > > > > >>>>> queries, we will always return the current running
> >> results of
> >> > > the
> >> > > > > > >>> windows,
> >> > > > > > >>>>> being it partial or final results depending on the
> window
> >> > > > retention
> >> > > > > > >> time
> >> > > > > > >>>>> anyways, which has nothing to do whether the emitted
> >> stream
> >> > > > should
> >> > > > > be
> >> > > > > > >>> one
> >> > > > > > >>>>> final output per key or not. I also agree that having a
> >> > unified
> >> > > > > > >>> operation
> >> > > > > > >>>>> is generally better for users to focus on leveraging
> that
> >> one
> >> > > > only
> >> > > > > > >> than
> >> > > > > > >>>>> learning about two set of operations. The only question
> I
> >> had
> >> > > is,
> >> > > > > for
> >> > > > > > >>>>> final
> >> > > > > > >>>>> updates of window stores, if it is a bit awkward to
> >> > understand
> >> > > > the
> >> > > > > > >>>>> configuration combo. Thinking about this more, I think
> my
> >> > root
> >> > > > > worry
> >> > > > > > >> in
> >> > > > > > >>>>> the
> >> > > > > > >>>>> "suppressLateEvents" call for windowed tables, since
> from
> >> a
> >> > > user
> >> > > > > > >>>>> perspective: if my retention time is X which means "pay
> >> the
> >> > > cost
> >> > > > to
> >> > > > > > >>> allow
> >> > > > > > >>>>> late records up to X to still be applied updating the
> >> > tables",
> >> > > > why
> >> > > > > > >>> would I
> >> > > > > > >>>>> ever want to suppressLateEvents by Y ( < X), to say "do
> >> not
> >> > > send
> >> > > > > the
> >> > > > > > >>>>> updates up to Y, which means the downstream operator or
> >> sink
> >> > > > topic
> >> > > > > > for
> >> > > > > > >>>>> this
> >> > > > > > >>>>> stream would actually see a truncated update stream
> while
> >> > I've
> >> > > > paid
> >> > > > > > >>> larger
> >> > > > > > >>>>> cost for that"; and of course, Y > X would not make
> sense
> >> > > either
> >> > > > as
> >> > > > > > >> you
> >> > > > > > >>>>> would not see any updates later than X anyways. So in
> >> all, my
> >> > > > > feeling
> >> > > > > > >> is
> >> > > > > > >>>>> that it makes less sense for windowed table's
> >> > > > "suppressLateEvents"
> >> > > > > > >> with
> >> > > > > > >>> a
> >> > > > > > >>>>> parameter that is not equal to the window retention, and
> >> > > opening
> >> > > > > the
> >> > > > > > >>> door
> >> > > > > > >>>>> in the current proposal may confuse people with that.
> >> > > > > > >>>>>
> >> > > > > > >>>>> Again, above is just a subjective opinion and probably
> we
> >> can
> >> > > > also
> >> > > > > > >> bring
> >> > > > > > >>>>> up
> >> > > > > > >>>>> some scenarios that users does want to set X != Y.. but
> >> > > > personally
> >> > > > > I
> >> > > > > > >>> feel
> >> > > > > > >>>>> that even if the semantics for this scenario if
> intuitive
> >> for
> >> > > > user
> >> > > > > to
> >> > > > > > >>>>> understand, doe that really make sense and should we
> >> really
> >> > > open
> >> > > > > the
> >> > > > > > >>> door
> >> > > > > > >>>>> for it. So I think maybe separating the final update in
> a
> >> > > > separate
> >> > > > > > >> API's
> >> > > > > > >>>>> benefits may overwhelm the advantage of having one
> uniform
> >> > > > > > definition.
> >> > > > > > >>> And
> >> > > > > > >>>>> for my alternative proposal, the rationale was from both
> >> my
> >> > > > concern
> >> > > > > > >>> about
> >> > > > > > >>>>> "suppressLateEvents" for windowed store, and Matthias'
> >> > question
> >> > > > > about
> >> > > > > > >>>>> "suppressLateEvents" for non-windowed stores, that if it
> >> is
> >> > > less
> >> > > > > > >>>>> meaningful
> >> > > > > > >>>>> for both, we can consider removing it completely and
> only
> >> do
> >> > > > > > >>>>> "IntermediateSuppression" in Suppress instead.
> >> > > > > > >>>>>
> >> > > > > > >>>>> So I'd summarize my thoughts in the following questions:
> >> > > > > > >>>>>
> >> > > > > > >>>>> 1. Does "suppressLateEvents" with parameter Y != X
> (window
> >> > > > > retention
> >> > > > > > >>> time)
> >> > > > > > >>>>> for windowed stores make sense in practice?
> >> > > > > > >>>>> 2. Does "suppressLateEvents" with any parameter Y for
> >> > > > non-windowed
> >> > > > > > >>> stores
> >> > > > > > >>>>> make sense in practice?
> >> > > > > > >>>>>
> >> > > > > > >>>>>
> >> > > > > > >>>>>
> >> > > > > > >>>>> Guozhang
> >> > > > > > >>>>>
> >> > > > > > >>>>>
> >> > > > > > >>>>> On Fri, Jun 29, 2018 at 2:26 PM, Bill Bejeck <
> >> > > bbej...@gmail.com>
> >> > > > > > >> wrote:
> >> > > > > > >>>>>
> >> > > > > > >>>>>> Thanks for the explanation, that does make sense. I
> have
> >> > some
> >> > > > > > >>>>> questions on
> >> > > > > > >>>>>> operations, but I'll just wait for the PR and tests.
> >> > > > > > >>>>>>
> >> > > > > > >>>>>> Thanks,
> >> > > > > > >>>>>> Bill
> >> > > > > > >>>>>>
> >> > > > > > >>>>>> On Wed, Jun 27, 2018 at 8:14 PM John Roesler <
> >> > > j...@confluent.io
> >> > > > >
> >> > > > > > >>> wrote:
> >> > > > > > >>>>>>
> >> > > > > > >>>>>>> Hi Bill,
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> Thanks for the review!
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> Your question is very much applicable to the KIP and
> >> not at
> >> > > all
> >> > > > > an
> >> > > > > > >>>>>>> implementation detail. Thanks for bringing it up.
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> I'm proposing not to change the existing caches and
> >> > > > > configurations
> >> > > > > > >>> at
> >> > > > > > >>>>> all
> >> > > > > > >>>>>>> (for now).
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> Imagine you have a topology like this:
> >> > > > > > >>>>>>> commit.interval.ms = 100
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> (ktable1 (cached)) -> (suppress emitAfter 200)
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> The first ktable (ktable1) will respect the commit
> >> interval
> >> > > and
> >> > > > > > >>> buffer
> >> > > > > > >>>>>>> events for 100ms before logging, storing, or
> forwarding
> >> > them
> >> > > > > > >> (IIRC).
> >> > > > > > >>>>>>> Therefore, the second ktable (suppress) will only see
> >> the
> >> > > > events
> >> > > > > > >> at
> >> > > > > > >>> a
> >> > > > > > >>>>>> rate
> >> > > > > > >>>>>>> of once per 100ms. It will apply its own buffering,
> and
> >> > emit
> >> > > > once
> >> > > > > > >>> per
> >> > > > > > >>>>>> 200ms
> >> > > > > > >>>>>>> This case is pretty trivial because the suppress time
> >> is a
> >> > > > > > >> multiple
> >> > > > > > >>> of
> >> > > > > > >>>>>> the
> >> > > > > > >>>>>>> commit interval.
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> When it's not an integer multiple, you'll get behavior
> >> like
> >> > > in
> >> > > > > > >> this
> >> > > > > > >>>>>> marble
> >> > > > > > >>>>>>> diagram:
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> <-(k:1)--(k:2)--(k:3)--(k:4)--(k:5)--(k:6)->
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> [ KTable caching with commit interval = 2 ]
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> <--------(k:2)---------(k:4)---------(k:6)->
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> [ suppress with emitAfter = 3 ]
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> <---------------(k:2)----------------(k:6)->
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> If this behavior isn't desired (for example, if you
> >> wanted
> >> > to
> >> > > > > emit
> >> > > > > > >>>>> (k:3)
> >> > > > > > >>>>>> at
> >> > > > > > >>>>>>> time 3, I'd recommend setting the
> >> > "cache.max.bytes.buffering"
> >> > > > to
> >> > > > > 0
> >> > > > > > >>> or
> >> > > > > > >>>>>>> modifying the topology to disable caching. Then, the
> >> > behavior
> >> > > > is
> >> > > > > > >>> more
> >> > > > > > >>>>>>> simply determined just by the suppress operator.
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> Does that seem right to you?
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> Regarding the changelogs, because the suppression
> >> operator
> >> > > > hangs
> >> > > > > > >>> onto
> >> > > > > > >>>>>>> events for a while, it will need its own changelog.
> The
> >> > > > changelog
> >> > > > > > >>>>>>> should represent the current state of the buffer at
> all
> >> > > times.
> >> > > > So
> >> > > > > > >>> when
> >> > > > > > >>>>>> the
> >> > > > > > >>>>>>> suppress operator sees (k:2), for example, it will log
> >> > (k:2).
> >> > > > > When
> >> > > > > > >>> it
> >> > > > > > >>>>>>> later gets to time 3, it's time to emit (k:2)
> >> downstream.
> >> > > > Because
> >> > > > > > >> k
> >> > > > > > >>>>> is no
> >> > > > > > >>>>>>> longer buffered, the suppress operator will log
> >> (k:null).
> >> > > Thus,
> >> > > > > > >> when
> >> > > > > > >>>>>>> recovering,
> >> > > > > > >>>>>>> it can rebuild the buffer by reading its changelog.
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> What do you think about this?
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> Thanks,
> >> > > > > > >>>>>>> -John
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>> On Wed, Jun 27, 2018 at 4:16 PM Bill Bejeck <
> >> > > bbej...@gmail.com
> >> > > > >
> >> > > > > > >>>>> wrote:
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>>> Hi John, thanks for the KIP.
> >> > > > > > >>>>>>>>
> >> > > > > > >>>>>>>> Early on in the KIP, you mention the current
> approaches
> >> > for
> >> > > > > > >>>>> controlling
> >> > > > > > >>>>>>> the
> >> > > > > > >>>>>>>> rate of downstream records from a KTable, cache size
> >> > > > > > >> configuration
> >> > > > > > >>>>> and
> >> > > > > > >>>>>>>> commit time.
> >> > > > > > >>>>>>>>
> >> > > > > > >>>>>>>> Will these configuration parameters still be in
> effect
> >> for
> >> > > > > > >> tables
> >> > > > > > >>>>> that
> >> > > > > > >>>>>>>> don't use suppression? For tables taking advantage
> of
> >> > > > > > >>> suppression,
> >> > > > > > >>>>>> will
> >> > > > > > >>>>>>>> these configurations have no impact?
> >> > > > > > >>>>>>>> This last question may be to implementation specific
> >> but
> >> > if
> >> > > > the
> >> > > > > > >>>>>> requested
> >> > > > > > >>>>>>>> suppression time is longer than the specified commit
> >> time,
> >> > > > will
> >> > > > > > >>> the
> >> > > > > > >>>>>>> latest
> >> > > > > > >>>>>>>> record in the suppression buffer get stored in a
> >> > changelog?
> >> > > > > > >>>>>>>>
> >> > > > > > >>>>>>>> Thanks,
> >> > > > > > >>>>>>>> Bill
> >> > > > > > >>>>>>>>
> >> > > > > > >>>>>>>> On Wed, Jun 27, 2018 at 3:04 PM John Roesler <
> >> > > > j...@confluent.io
> >> > > > > > >>>
> >> > > > > > >>>>>> wrote:
> >> > > > > > >>>>>>>>
> >> > > > > > >>>>>>>>> Thanks for the feedback, Matthias,
> >> > > > > > >>>>>>>>>
> >> > > > > > >>>>>>>>> It seems like in straightforward relational
> processing
> >> > > cases,
> >> > > > > > >> it
> >> > > > > > >>>>>> would
> >> > > > > > >>>>>>>> not
> >> > > > > > >>>>>>>>> make sense to bound the lateness of KTables. In
> >> general,
> >> > it
> >> > > > > > >>> seems
> >> > > > > > >>>>>>> better
> >> > > > > > >>>>>>>> to
> >> > > > > > >>>>>>>>> have "guard rails" in place that make it easier to
> >> write
> >> > > > > > >>> sensible
> >> > > > > > >>>>>>>> programs
> >> > > > > > >>>>>>>>> than insensible ones.
> >> > > > > > >>>>>>>>>
> >> > > > > > >>>>>>>>> But I'm still going to argue in favor of keeping it
> >> for
> >> > all
> >> > > > > > >>>>> KTables
> >> > > > > > >>>>>> ;)
> >> > > > > > >>>>>>>>>
> >> > > > > > >>>>>>>>> 1. I believe it is simpler to understand the
> operator
> >> if
> >> > it
> >> > > > > > >> has
> >> > > > > > >>>>> one
> >> > > > > > >>>>>>>> uniform
> >> > > > > > >>>>>>>>> definition, regardless of context. It's well defined
> >> and
> >> > > > > > >>> intuitive
> >> > > > > > >>>>>> what
> >> > > > > > >>>>>>>>> will happen when you use late-event suppression on a
> >> > > KTable,
> >> > > > > > >> so
> >> > > > > > >>> I
> >> > > > > > >>>>>> think
> >> > > > > > >>>>>>>>> nothing surprising or dangerous will happen in that
> >> case.
> >> > > > From
> >> > > > > > >>> my
> >> > > > > > >>>>>>>>> perspective, having two sets of allowed operations
> is
> >> > > > actually
> >> > > > > > >>> an
> >> > > > > > >>>>>>>> increase
> >> > > > > > >>>>>>>>> in cognitive complexity.
> >> > > > > > >>>>>>>>>
> >> > > > > > >>>>>>>>> 2. To me, it's not crazy to use the operator this
> way.
> >> > For
> >> > > > > > >>>>> example,
> >> > > > > > >>>>>> in
> >> > > > > > >>>>>>>> lieu
> >> > > > > > >>>>>>>>> of full-featured timestamp semantics, I can
> implement
> >> > MVCC
> >> > > > > > >>>>> behavior
> >> > > > > > >>>>>>> when
> >> > > > > > >>>>>>>>> building a KTable by
> >> "suppressLateEvents(Duration.ZERO)".
> >> > I
> >> > > > > > >>>>> suspect
> >> > > > > > >>>>>>> that
> >> > > > > > >>>>>>>>> there are other, non-obvious applications of
> >> suppressing
> >> > > late
> >> > > > > > >>>>> events
> >> > > > > > >>>>>> on
> >> > > > > > >>>>>>>>> KTables.
> >> > > > > > >>>>>>>>>
> >> > > > > > >>>>>>>>> 3. Not to get too much into implementation details
> in
> >> a
> >> > KIP
> >> > > > > > >>>>>> discussion,
> >> > > > > > >>>>>>>> but
> >> > > > > > >>>>>>>>> if we did want to make late-event suppression
> >> available
> >> > > only
> >> > > > > > >> on
> >> > > > > > >>>>>>> windowed
> >> > > > > > >>>>>>>>> KTables, we have two enforcement options:
> >> > > > > > >>>>>>>>> a. check when we build the topology - this would
> be
> >> > > simple
> >> > > > > > >> to
> >> > > > > > >>>>>>>> implement,
> >> > > > > > >>>>>>>>> but would be a runtime check. Hopefully, people
> write
> >> > tests
> >> > > > > > >> for
> >> > > > > > >>>>> their
> >> > > > > > >>>>>>>>> topology before deploying them, so the feedback loop
> >> > isn't
> >> > > > > > >>>>>>> instantaneous,
> >> > > > > > >>>>>>>>> but it's not too long either.
> >> > > > > > >>>>>>>>> b. add a new WindowedKTable type - this would be a
> >> > > compile
> >> > > > > > >>> time
> >> > > > > > >>>>>>> check,
> >> > > > > > >>>>>>>>> but would also be substantial increase of both
> >> interface
> >> > > and
> >> > > > > > >>> code
> >> > > > > > >>>>>>>>> complexity.
> >> > > > > > >>>>>>>>>
> >> > > > > > >>>>>>>>> We should definitely strive to have guard rails
> >> > protecting
> >> > > > > > >>> against
> >> > > > > > >>>>>>>>> surprising or dangerous behavior. Protecting against
> >> > > programs
> >> > > > > > >>>>> that we
> >> > > > > > >>>>>>>> don't
> >> > > > > > >>>>>>>>> currently predict is a lesser benefit, and I think
> we
> >> can
> >> > > put
> >> > > > > > >> up
> >> > > > > > >>>>>> guard
> >> > > > > > >>>>>>>>> rails on a case-by-case basis for that. It seems
> like
> >> the
> >> > > > > > >>>>> increase in
> >> > > > > > >>>>>>>>> cognitive (and potentially code and interface)
> >> complexity
> >> > > > > > >> makes
> >> > > > > > >>> me
> >> > > > > > >>>>>>> think
> >> > > > > > >>>>>>>> we
> >> > > > > > >>>>>>>>> should skip this case.
> >> > > > > > >>>>>>>>>
> >> > > > > > >>>>>>>>> What do you think?
> >> > > > > > >>>>>>>>>
> >> > > > > > >>>>>>>>> Thanks,
> >> > > > > > >>>>>>>>> -John
> >> > > > > > >>>>>>>>>
> >> > > > > > >>>>>>>>> On Wed, Jun 27, 2018 at 11:59 AM Matthias J. Sax <
> >> > > > > > >>>>>>> matth...@confluent.io>
> >> > > > > > >>>>>>>>> wrote:
> >> > > > > > >>>>>>>>>
> >> > > > > > >>>>>>>>>> Thanks for the KIP John.
> >> > > > > > >>>>>>>>>>
> >> > > > > > >>>>>>>>>> One initial comments about the last example
> "Bounded
> >> > > > > > >>> lateness":
> >> > > > > > >>>>>> For a
> >> > > > > > >>>>>>>>>> non-windowed KTable bounding the lateness does not
> >> > really
> >> > > > > > >> make
> >> > > > > > >>>>>> sense,
> >> > > > > > >>>>>>>>>> does it?
> >> > > > > > >>>>>>>>>>
> >> > > > > > >>>>>>>>>> Thus, I am wondering if we should allow
> >> > > > > > >> `suppressLateEvents()`
> >> > > > > > >>>>> for
> >> > > > > > >>>>>>> this
> >> > > > > > >>>>>>>>>> case? It seems to be better to only allow it for
> >> > > > > > >>>>> windowed-KTables.
> >> > > > > > >>>>>>>>>>
> >> > > > > > >>>>>>>>>>
> >> > > > > > >>>>>>>>>> -Matthias
> >> > > > > > >>>>>>>>>>
> >> > > > > > >>>>>>>>>>
> >> > > > > > >>>>>>>>>> On 6/27/18 8:53 AM, Ted Yu wrote:
> >> > > > > > >>>>>>>>>>> I noticed this (lack of primary parameter) as
> well.
> >> > > > > > >>>>>>>>>>>
> >> > > > > > >>>>>>>>>>> What you gave as new example is semantically the
> >> same
> >> > as
> >> > > > > > >>> what
> >> > > > > > >>>>> I
> >> > > > > > >>>>>>>>>> suggested.
> >> > > > > > >>>>>>>>>>> So it is good by me.
> >> > > > > > >>>>>>>>>>>
> >> > > > > > >>>>>>>>>>> Thanks
> >> > > > > > >>>>>>>>>>>
> >> > > > > > >>>>>>>>>>> On Wed, Jun 27, 2018 at 7:31 AM, John Roesler <
> >> > > > > > >>>>> j...@confluent.io
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>>>> wrote:
> >> > > > > > >>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>> Thanks for taking look, Ted,
> >> > > > > > >>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>> I agree this is a departure from the conventions
> of
> >> > > > > > >> Streams
> >> > > > > > >>>>> DSL.
> >> > > > > > >>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>> Most of our config objects have one or two
> >> "required"
> >> > > > > > >>>>>> parameters,
> >> > > > > > >>>>>>>>> which
> >> > > > > > >>>>>>>>>> fit
> >> > > > > > >>>>>>>>>>>> naturally with the static factory method
> approach.
> >> > > > > > >>>>> TimeWindow,
> >> > > > > > >>>>>> for
> >> > > > > > >>>>>>>>>> example,
> >> > > > > > >>>>>>>>>>>> requires a size parameter, so we can naturally
> say
> >> > > > > > >>>>>>>>> TimeWindows.of(size).
> >> > > > > > >>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>> I think in the case of a suppression, there's
> >> really
> >> > no
> >> > > > > > >>>>> "core"
> >> > > > > > >>>>>>>>>> parameter,
> >> > > > > > >>>>>>>>>>>> and "Suppression.of()" seems sillier than "new
> >> > > > > > >>>>> Suppression()". I
> >> > > > > > >>>>>>>> think
> >> > > > > > >>>>>>>>>> that
> >> > > > > > >>>>>>>>>>>> Suppression.of(duration) would be ambiguous,
> since
> >> > there
> >> > > > > > >>> are
> >> > > > > > >>>>>> many
> >> > > > > > >>>>>>>>>> durations
> >> > > > > > >>>>>>>>>>>> that we can configure.
> >> > > > > > >>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>> However, thinking about it again, I suppose that
> I
> >> can
> >> > > > > > >> give
> >> > > > > > >>>>> each
> >> > > > > > >>>>>>>>>>>> configuration method a static version, which
> would
> >> let
> >> > > > > > >> you
> >> > > > > > >>>>>> replace
> >> > > > > > >>>>>>>>> "new
> >> > > > > > >>>>>>>>>>>> Suppression()." with "Suppression." in all the
> >> > examples.
> >> > > > > > >>>>>>> Basically,
> >> > > > > > >>>>>>>>>> instead
> >> > > > > > >>>>>>>>>>>> of "of()", we'd support any of the methods I
> >> listed.
> >> > > > > > >>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>> For example:
> >> > > > > > >>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>> windowCounts
> >> > > > > > >>>>>>>>>>>> .suppress(
> >> > > > > > >>>>>>>>>>>> Suppression
> >> > > > > > >>>>>>>>>>>> .suppressLateEvents(Duration.
> >> > ofMinutes(10))
> >> > > > > > >>>>>>>>>>>> .suppressIntermediateEvents(
> >> > > > > > >>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>
> >> IntermediateSuppression.emitAfter(Duration.ofMinutes(
> >> > 10))
> >> > > > > > >>>>>>>>>>>> )
> >> > > > > > >>>>>>>>>>>> );
> >> > > > > > >>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>> Does that seem better?
> >> > > > > > >>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>> Thanks,
> >> > > > > > >>>>>>>>>>>> -John
> >> > > > > > >>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>> On Wed, Jun 27, 2018 at 12:44 AM Ted Yu <
> >> > > > > > >>> yuzhih...@gmail.com
> >> > > > > > >>>>>>
> >> > > > > > >>>>>>>> wrote:
> >> > > > > > >>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>> I started to read this KIP which contains a lot
> of
> >> > > > > > >>>>> materials.
> >> > > > > > >>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>> One suggestion:
> >> > > > > > >>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>> .suppress(
> >> > > > > > >>>>>>>>>>>>> new Suppression()
> >> > > > > > >>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>> Do you think it would be more consistent with
> the
> >> > rest
> >> > > > > > >> of
> >> > > > > > >>>>>> Streams
> >> > > > > > >>>>>>>>> data
> >> > > > > > >>>>>>>>>>>>> structures by supporting `of` ?
> >> > > > > > >>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>> Suppression.of(Duration.ofMinutes(10))
> >> > > > > > >>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>> Cheers
> >> > > > > > >>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>> On Tue, Jun 26, 2018 at 1:11 PM, John Roesler <
> >> > > > > > >>>>>> j...@confluent.io
> >> > > > > > >>>>>>>>
> >> > > > > > >>>>>>>>>> wrote:
> >> > > > > > >>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>>> Hello devs and users,
> >> > > > > > >>>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>>> Please take some time to consider this proposal
> >> for
> >> > > > > > >> Kafka
> >> > > > > > >>>>>>> Streams:
> >> > > > > > >>>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>>> KIP-328: Ability to suppress updates for
> KTables
> >> > > > > > >>>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>>> link:
> >> https://cwiki.apache.org/confluence/x/sQU0BQ
> >> > > > > > >>>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>>> The basic idea is to provide:
> >> > > > > > >>>>>>>>>>>>>> * more usable control over update rate (vs the
> >> > current
> >> > > > > > >>>>> state
> >> > > > > > >>>>>>> store
> >> > > > > > >>>>>>>>>>>>> caches)
> >> > > > > > >>>>>>>>>>>>>> * the final-result-for-windowed-computations
> >> > feature
> >> > > > > > >>> which
> >> > > > > > >>>>>>> several
> >> > > > > > >>>>>>>>>>>> people
> >> > > > > > >>>>>>>>>>>>>> have requested
> >> > > > > > >>>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>>> I look forward to your feedback!
> >> > > > > > >>>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>>> Thanks,
> >> > > > > > >>>>>>>>>>>>>> -John
> >> > > > > > >>>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>>
> >> > > > > > >>>>>>>>>>>
> >> > > > > > >>>>>>>>>>
> >> > > > > > >>>>>>>>>>
> >> > > > > > >>>>>>>>>
> >> > > > > > >>>>>>>>
> >> > > > > > >>>>>>>
> >> > > > > > >>>>>>
> >> > > > > > >>>>>
> >> > > > > > >>>>>
> >> > > > > > >>>>>
> >> > > > > > >>>>> --
> >> > > > > > >>>>> -- Guozhang
> >> > > > > > >>>>>
> >> > > > > > >>>>
> >> > > > > > >>>
> >> > > > > > >>
> >> > > > > > >>
> >> > > > > > >>
> >> > > > > > >> --
> >> > > > > > >> -- Guozhang
> >> > > > > > >>
> >> > > > > > >
> >> > > > > >
> >> > > > > >
> >> > > > >
> >> > > > >
> >> > > > > --
> >> > > > > -- Guozhang
> >> > > > >
> >> > > >
> >> > >
> >> > >
> >> > >
> >> > > --
> >> > > -- Guozhang
> >> > >
> >> >
> >>
> >>
> >>
> >> --
> >> -- Guozhang
> >>
> >
>
--
-- Guozhang
