Hi, I'm very sorry to participate in this discussion so late. But I have a little question. I understand the goal of this FLIP is to make `Writer` support asynchronous. But my question is: why not let `Committer` support asynchronization? If only `Writer` supports asynchronization, ExactlyOnce is impossible. Since I am not an expert in Kineses, I would like everyone to point out if I missed anything. Best, Guowei
On Sat, Jul 17, 2021 at 12:27 AM Till Rohrmann <trohrm...@apache.org> wrote: > Sure, thanks for the pointers. > > Cheers, > Till > > On Fri, Jul 16, 2021 at 6:19 PM Hausmann, Steffen > <shau...@amazon.de.invalid> > wrote: > > > Hi Till, > > > > You are right, I’ve left out some implementation details, which have > > actually changed a couple of time as part of the ongoing discussion. You > > can find our current prototype here [1] and a sample implementation of > the > > KPL free Kinesis sink here [2]. > > > > I plan to update the FLIP. But I think would it be make sense to wait > > until the implementation has stabilized enough before we update the FLIP > to > > the final state. > > > > Does that make sense? > > > > Cheers, Steffen > > > > [1] > > > https://github.com/sthm/flink/tree/flip-171-177/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink > > [2] > > > https://github.com/sthm/flink/blob/flip-171-177/flink-connectors/flink-connector-kinesis-171/src/main/java/software/amazon/flink/connectors/AmazonKinesisDataStreamSink.java > > > > From: Till Rohrmann <trohrm...@apache.org> > > Date: Friday, 16. July 2021 at 18:10 > > To: Piotr Nowojski <pnowoj...@apache.org> > > Cc: Steffen Hausmann <shau...@amazon.de>, "dev@flink.apache.org" < > > dev@flink.apache.org>, Arvid Heise <ar...@apache.org> > > Subject: RE: [EXTERNAL] [DISCUSS] FLIP-171: Async Sink > > > > > > CAUTION: This email originated from outside of the organization. Do not > > click links or open attachments unless you can confirm the sender and > know > > the content is safe. > > > > > > Hi Steffen, > > > > I've taken another look at the FLIP and I stumbled across a couple of > > inconsistencies. I think it is mainly because of the lacking code. For > > example, it is not fully clear to me based on the current FLIP how we > > ensure that there are no in-flight requests when > > AsyncSinkWriter.snapshotState is called. Also the concrete implementation > > of the AsyncSinkCommitter could be helpful for understanding how the > > AsyncSinkWriter works in the end. Do you plan to update the FLIP > > accordingly? > > > > Cheers, > > Till > > > > On Wed, Jun 30, 2021 at 8:36 AM Piotr Nowojski <pnowoj...@apache.org > > <mailto:pnowoj...@apache.org>> wrote: > > Thanks for addressing this issue :) > > > > Best, Piotrek > > > > wt., 29 cze 2021 o 17:58 Hausmann, Steffen <shau...@amazon.de<mailto: > > shau...@amazon.de>> napisał(a): > > Hey Poitr, > > > > I've just adapted the FLIP and changed the signature for the > > `submitRequestEntries` method: > > > > protected abstract void submitRequestEntries(List<RequestEntryT> > > requestEntries, ResultFuture<?> requestResult); > > > > In addition, we are likely to use an AtomicLong to track the number of > > outstanding requests, as you have proposed in 2b). I've already indicated > > this in the FLIP, but it's not fully fleshed out. But as you have said, > > that seems to be an implementation detail and the important part is the > > change of the `submitRequestEntries` signature. > > > > Thanks for your feedback! > > > > Cheers, Steffen > > > > > > On 25.06.21, 17:05, "Hausmann, Steffen" <shau...@amazon.de.INVALID> > wrote: > > > > CAUTION: This email originated from outside of the organization. Do > > not click links or open attachments unless you can confirm the sender and > > know the content is safe. > > > > > > > > Hi Piotr, > > > > I’m happy to take your guidance on this. I need to think through your > > proposals and I’ll follow-up on Monday with some more context so that we > > can close the discussion on these details. But for now, I’ll close the > vote. > > > > Thanks, Steffen > > > > From: Piotr Nowojski <pnowoj...@apache.org<mailto: > pnowoj...@apache.org > > >> > > Date: Friday, 25. June 2021 at 14:48 > > To: Till Rohrmann <trohrm...@apache.org<mailto:trohrm...@apache.org > >> > > Cc: Steffen Hausmann <shau...@amazon.de<mailto:shau...@amazon.de>>, > " > > dev@flink.apache.org<mailto:dev@flink.apache.org>" <dev@flink.apache.org > > <mailto:dev@flink.apache.org>>, Arvid Heise <ar...@apache.org<mailto: > > ar...@apache.org>> > > Subject: RE: [EXTERNAL] [DISCUSS] FLIP-171: Async Sink > > > > > > CAUTION: This email originated from outside of the organization. Do > > not click links or open attachments unless you can confirm the sender and > > know the content is safe. > > > > > > Hey, > > > > I've just synced with Arvid about a couple of more remarks from my > > side and he shared mine concerns. > > > > 1. I would very strongly recommend ditching `CompletableFuture<?> ` > > from the `protected abstract CompletableFuture<?> > > submitRequestEntries(List<RequestEntryT> requestEntries);` in favor of > > something like > > `org.apache.flink.streaming.api.functions.async.ResultFuture` interface. > > `CompletableFuture<?>` would partially make the threading model of the > > `AsyncSincWriter` part of the public API and it would tie our hands. > > Regardless how `CompletableFuture<?>` is used, it imposes performance > > overhead because it's synchronisation/volatile inside of it. On the other > > hand something like: > > > > protected abstract void submitRequestEntries(List<RequestEntryT> > > requestEntries, ResultFuture<?> requestResult); > > > > Would allow us to implement the threading model as we wish. > > `ResultFuture` could be backed via `CompletableFuture<?>` underneath, but > > it could also be something more efficient. I will explain what I have in > > mind in a second. > > > > 2. It looks to me that proposed `AsyncSinkWriter` Internals are not > > very efficient and maybe the threading model hasn't been thought through? > > Especially private fields: > > > > private final BlockingDeque<RequestEntryT> bufferedRequestEntries; > > private BlockingDeque<CompletableFuture<?>> inFlightRequests; > > > > are a bit strange to me. Why do we need two separate thread safe > > collections? Why do we need a `BlockingDeque` of `CompletableFuture<?>`s? > > If we are already using a fully synchronised collection, there should be > no > > need for another layer of thread safe `CompletableFuture<?>`. > > > > As I understand, the threading model of the `AsyncSinkWriter` is very > > similar to that of the `AsyncWaitOperator`, with very similar > requirements > > for inducing backpressure. How I would see it implemented is for example: > > > > a) Having a single lock, that would encompass the whole > > `AsyncSinkWriter#flush()` method. `flush()` would be called from the task > > thread (mailbox). To induce backpressure, `#flush()` would just call > > `lock.wait()`. `ResultFuture#complete(...)` called from an async thread, > > would also synchronize on the same lock, and mark some of the inflight > > requests as completed and call `lock.notify()`. > > > > b) More efficient solution. On the hot path we would have for example > > only `AtomicLong numberOfInFlightRequests`. Task thread would be bumping > > it, `ResultFuture#complete()` would be decreasing it. If the task thread > > when bumping `numberOfInFlightRequests` exceeds a threshold, he goes to > > sleep/wait on a lock or some `CompletableFuture`. If > > `ResultFuture#complete()` when decreasing the count goes below the > > threshold, it would wake up the task thread. Compared to the option a), > > on the hot path, option b) would have only AtomicLong.increment overhead > > > > c) We could use mailbox, the same way as AsyncWaitOperator is doing. > > In this case `ResultFuture#complete()` would be enquing mailbox action, > > which is thread safe on it's own. > > > > Either of those options would be more efficient and simpler (from the > > threading model perspective) than having two `BlockingQueues` and > > `CompletableFuture<?>`. Also as you can see, neither of those solutions > > require the overhead of ` CompletableFuture<?> > > submitRequestEntries(List<RequestEntryT> requestEntries)`. Each one of > > those could use a more efficient and custom implementation of > > `ResultFuture.complete(...)`. > > > > > > Whether we use a), b) or c) I think should be an implementation > > detail. But to allow this to truly be an implementation detail, we would > > need to agree on 1. Nevertheless I think that the change I proposed in 1. > > is small enough that I think there is no need to cancel the current vote > on > > the FLIP. > > > > WDYT? > > > > Piotrek > > > > > > wt., 22 cze 2021 o 11:42 Till Rohrmann <trohrm...@apache.org<mailto: > > trohrm...@apache.org><mailto:trohrm...@apache.org<mailto: > > trohrm...@apache.org>>> napisał(a): > > Adding the InterruptException to the write method would make it > > explicit that the write call can block but must react to interruptions > > (e.g. when Flink wants to cancel the operation). I think this makes the > > contract a bit clearer. > > > > I think starting simple and then extending the API as we see the need > > is a good idea. > > > > Cheers, > > Till > > > > On Tue, Jun 22, 2021 at 11:20 AM Hausmann, Steffen < > shau...@amazon.de > > <mailto:shau...@amazon.de><mailto:shau...@amazon.de<mailto: > > shau...@amazon.de>>> wrote: > > Hey, > > > > Agreed on starting with a blocking `write`. I've adapted the FLIP > > accordingly. > > > > For now I've chosen to add the `InterruptedException` to the `write` > > method signature as I'm not fully understanding the implications of > > swallowing the exception. Depending on the details of the code that is > > calling the write method, it may cause event loss. But this seems more of > > an implementation detail, that we can revisit once we are actually > > implementing the sink. > > > > Unless there are additional comments, does it make sense to start the > > voting process in the next day or two? > > > > Cheers, Steffen > > > > > > On 21.06.21, 14:51, "Piotr Nowojski" <pnowoj...@apache.org<mailto: > > pnowoj...@apache.org><mailto:pnowoj...@apache.org<mailto: > > pnowoj...@apache.org>>> wrote: > > > > CAUTION: This email originated from outside of the organization. > > Do not click links or open attachments unless you can confirm the sender > > and know the content is safe. > > > > > > > > Hi, > > > > Thanks Steffen for the explanations. I think it makes sense to > me. > > > > Re Arvid/Steffen: > > > > - Keep in mind that even if we choose to provide a non blocking > > API using > > the `isAvailable()`/`getAvailableFuture()` method, we would still > > need to > > support blocking inside the sinks. For example at the very least, > > emitting > > many records at once (`flatMap`) or firing timers are scenarios > > when output > > availability would be ignored at the moment by the runtime. Also > I > > would > > imagine writing very large (like 1GB) records would be blocking > on > > something as well. > > - Secondly, exposing availability to the API level might not be > > that > > easy/trivial. The availability pattern as defined in > > `AvailabilityProvider` > > class is quite complicated and not that easy to implement by a > > user. > > > > Both of those combined with lack of a clear motivation for adding > > `AvailabilityProvider` to the sinks/operators/functions, I would > > vote on > > just starting with blocking `write` calls. This can always be > > extended in > > the future with availability if needed/motivated properly. > > > > That would be aligned with either Arvid's option 1 or 2. I don't > > know what > > are the best practices with `InterruptedException`, but I'm > always > > afraid > > of it, so I would feel personally safer with option 2. > > > > I'm not sure what problem option 3 is helping to solve? Adding > > `wakeUp()` > > would sound strange to me. > > > > Best, > > Piotrek > > > > pon., 21 cze 2021 o 12:15 Arvid Heise <ar...@apache.org<mailto: > > ar...@apache.org><mailto:ar...@apache.org<mailto:ar...@apache.org>>> > > napisał(a): > > > > > Hi Piotr, > > > > > > to pick up this discussion thread again: > > > - This FLIP is about providing some base implementation for > > FLIP-143 sinks > > > that make adding new implementations easier, similar to the > > > SourceReaderBase. > > > - The whole availability topic will most likely be a separate > > FLIP. The > > > basic issue just popped up here because we currently have no > way > > to signal > > > backpressure in sinks except by blocking `write`. This feels > > quite natural > > > in sinks with sync communication but quite unnatural in async > > sinks. > > > > > > Now we have a couple of options. In all cases, we would have > > some WIP > > > limit on the number of records/requests being able to be > > processed in > > > parallel asynchronously (similar to asyncIO). > > > 1. We use some blocking queue in `write`, then we need to > handle > > > interruptions. In the easiest case, we extend `write` to throw > > the > > > `InterruptedException`, which is a small API change. > > > 2. We use a blocking queue, but handle interrupts and > > swallow/translate > > > them. No API change. > > > Both solutions block the task thread, so any RPC message / > > unaligned > > > checkpoint would be processed only after the backpressure is > > temporarily > > > lifted. That's similar to the discussions that you linked. > > Cancellation may > > > also be a tad harder on 2. > > > 3. We could also add some `wakeUp` to the `SinkWriter` similar > to > > > `SplitFetcher` [1]. Basically, you use a normal queue with a > > completeable > > > future on which you block. Wakeup would be a clean way to > > complete it next > > > to the natural completion through finished requests. > > > 4. We add availability to the sink. However, this API change > > also requires > > > that we allow operators to be available so it may be a bigger > > change with > > > undesired side-effects. On the other hand, we could also use > the > > same > > > mechanism for asyncIO. > > > > > > For users of FLIP-171, none of the options are exposed. So we > > could also > > > start with a simple solution (add `InterruptedException`) and > > later try to > > > add availability. Option 1+2 would also not require an > > additional FLIP; we > > > could add it as part of this FLIP. > > > > > > Best, > > > > > > Arvid > > > > > > [1] > > > > > > https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcher.java#L258-L258 > > > On Thu, Jun 10, 2021 at 10:09 AM Hausmann, Steffen > > > <shau...@amazon.de.invalid> wrote: > > > > > >> Hey Piotrek, > > >> > > >> Thanks for your comments on the FLIP. I'll address your second > > question > > >> first, as I think it's more central to this FLIP. Just looking > > at the AWS > > >> ecosystem, there are several sinks with overlapping > > functionality. I've > > >> chosen AWS sinks here because I'm most familiar with those, > but > > a similar > > >> argument applies more generically for destination that support > > async ingest. > > >> > > >> There is, for instance, a sink for Amazon Kinesis Data Streams > > that is > > >> part of Apache Flink [1], a sink for Amazon Kinesis Data > > Firehose [2], a > > >> sink for Amazon DynamoDB [3], and a sink for Amazon Timestream > > [4]. All > > >> these sinks have implemented their own mechanisms for > batching, > > persisting, > > >> and retrying events. And I'm not sure if all of them properly > > participate > > >> in checkpointing. [3] even seems to closely mirror [1] as it > > contains > > >> references to the Kinesis Producer Library, which is unrelated > > to Amazon > > >> DynamoDB. > > >> > > >> These sinks predate FLIP-143. But as batching, persisting, and > > retrying > > >> capabilities do not seem to be part of FLIP-143, I'd argue > that > > we would > > >> end up with similar duplication, even if these sinks were > > rewritten today > > >> based on FLIP-143. And that's the idea of FLIP-171: abstract > > away these > > >> commonly required capabilities so that it becomes easy to > > create support > > >> for a wide range of destination without having to think about > > batching, > > >> retries, checkpointing, etc. I've included an example in the > > FLIP [5] that > > >> shows that it only takes a couple of lines of code to > implement > > a sink with > > >> exactly-once semantics. To be fair, the example is lacking > > robust failure > > >> handling and some more advanced capabilities of [1], but I > > think it still > > >> supports this point. > > >> > > >> Regarding your point on the isAvailable pattern. We need some > > way for the > > >> sink to propagate backpressure and we would also like to > > support time based > > >> buffering hints. There are two options I currently see and > > would need > > >> additional input on which one is the better or more desirable > > one. The > > >> first option is to use the non-blocking isAvailable pattern. > > Internally, > > >> the sink persists buffered events in the snapshot state which > > avoids having > > >> to flush buffered record on a checkpoint. This seems to align > > well with the > > >> non-blocking isAvailable pattern. The second option is to make > > calls to > > >> `write` blocking and leverage an internal thread to trigger > > flushes based > > >> on time based buffering hints. We've discussed these options > > with Arvid and > > >> suggested to assumed that the `isAvailable` pattern will > become > > available > > >> for sinks through and additional FLIP. > > >> > > >> I think it is an important discussion to have. My > understanding > > of the > > >> implications for Flink in general are very naïve, so I'd be > > happy to get > > >> further guidance. However, I don't want to make this > discussion > > part of > > >> FLIP-171. For FLIP-171 we'll use whatever is available. > > >> > > >> Does that make sense? > > >> > > >> Cheers, Steffen > > >> > > >> > > >> [1] > > >> > > > https://github.com/apache/flink/tree/master/flink-connectors/flink-connector-kinesis > > >> [2] > > https://github.com/aws/aws-kinesisanalytics-flink-connectors > > >> [3] > > https://github.com/klarna-incubator/flink-connector-dynamodb > > >> [4] https://github.com/awslabs/amazon-timestream-tools/ > > >> [5] > > >> > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink#FLIP171:AsyncSink-SimplifiedAsyncSinkWriterforKinesisDataStreams > > >> > > >> > > >> On 09.06.21, 19:44, "Piotr Nowojski" <pnowoj...@apache.org > > <mailto:pnowoj...@apache.org><mailto:pnowoj...@apache.org<mailto: > > pnowoj...@apache.org>>> wrote: > > >> > > >> CAUTION: This email originated from outside of the > > organization. Do > > >> not click links or open attachments unless you can confirm the > > sender and > > >> know the content is safe. > > >> > > >> > > >> > > >> Hi Steffen, > > >> > > >> Thanks for writing down the proposal. Back when the new > > Sink API was > > >> being > > >> discussed, I was proposing to add our usual > > `CompletableFuture<Void> > > >> isAvailable()` pattern to make sinks non-blocking. You can > > see the > > >> discussion starting here [1], and continuing for a couple > > of more > > >> posts > > >> until here [2]. Back then, the outcome was that it would > > give very > > >> little > > >> benefit, at the expense of making the API more > complicated. > > Could you > > >> maybe > > >> relate your proposal to that discussion from last year? > > >> > > >> I see that your proposal is going much further than just > > adding the > > >> availability method, could you also motivate this a bit > > further? > > >> Could you > > >> maybe reference/show some sinks that: > > >> 1. are already implemented using FLIP-143 > > >> 2. that have some code duplication... > > >> 3. ...this duplication would be solved by FLIP-171 > > >> > > >> Best, > > >> Piotrek > > >> > > >> [1] > > >> > > >> > > > http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-FLIP-143-Unified-Sink-API-tp44602p44872.html > > >> [2] > > >> > > >> > > > http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-FLIP-143-Unified-Sink-API-tp44602p44930.html > > >> > > >> śr., 9 cze 2021 o 09:49 Hausmann, Steffen > > <shau...@amazon.de.invalid> > > >> napisał(a): > > >> > > >> > Hi there, > > >> > > > >> > We would like to start a discussion thread on "FLIP-171: > > Async > > >> Sink" [1], > > >> > where we propose to create a common abstraction for > > destinations > > >> that > > >> > support async requests. This abstraction will make it > > easier to add > > >> > destinations to Flink by implementing a lightweight > shim, > > while it > > >> avoids > > >> > maintaining dozens of independent sinks. > > >> > > > >> > Looking forward to your feedback. > > >> > > > >> > Cheers, Steffen > > >> > > > >> > [1] > > >> > > > >> > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink > > >> > > > >> > > > >> > > > >> > Amazon Web Services EMEA SARL > > >> > 38 avenue John F. Kennedy, L-1855 Luxembourg > > >> > Sitz der Gesellschaft: L-1855 Luxemburg > > >> > eingetragen im Luxemburgischen Handelsregister unter > > R.C.S. B186284 > > >> > > > >> > Amazon Web Services EMEA SARL, Niederlassung Deutschland > > >> > Marcel-Breuer-Str. 12, D-80807 Muenchen > > >> > Sitz der Zweigniederlassung: Muenchen > > >> > eingetragen im Handelsregister des Amtsgerichts Muenchen > > unter HRB > > >> 242240, > > >> > USt-ID DE317013094 > > >> > > > >> > > > >> > > > >> > > > >> > > >> > > >> > > >> > > >> Amazon Web Services EMEA SARL > > >> 38 avenue John F. Kennedy, L-1855 Luxembourg > > >> Sitz der Gesellschaft: L-1855 Luxemburg > > >> eingetragen im Luxemburgischen Handelsregister unter R.C.S. > > B186284 > > >> > > >> Amazon Web Services EMEA SARL, Niederlassung Deutschland > > >> Marcel-Breuer-Str. 12, D-80807 Muenchen > > >> Sitz der Zweigniederlassung: Muenchen > > >> eingetragen im Handelsregister des Amtsgerichts Muenchen unter > > HRB > > >> 242240, USt-ID DE317013094 > > >> > > >> > > >> > > >> > > > > > > > > > > Amazon Web Services EMEA SARL > > 38 avenue John F. Kennedy, L-1855 Luxembourg > > Sitz der Gesellschaft: L-1855 Luxemburg > > eingetragen im Luxemburgischen Handelsregister unter R.C.S. B186284 > > > > Amazon Web Services EMEA SARL, Niederlassung Deutschland > > Marcel-Breuer-Str. 12, D-80807 Muenchen > > Sitz der Zweigniederlassung: Muenchen > > eingetragen im Handelsregister des Amtsgerichts Muenchen unter HRB > > 242240, USt-ID DE317013094 > > > > > > > > > > > > Amazon Web Services EMEA SARL > > 38 avenue John F. Kennedy, L-1855 Luxembourg > > Sitz der Gesellschaft: L-1855 Luxemburg > > eingetragen im Luxemburgischen Handelsregister unter R.C.S. B186284 > > > > Amazon Web Services EMEA SARL, Niederlassung Deutschland > > Marcel-Breuer-Str. 12, D-80807 Muenchen > > Sitz der Zweigniederlassung: Muenchen > > eingetragen im Handelsregister des Amtsgerichts Muenchen unter HRB > > 242240, USt-ID DE317013094 > > > > > > > > > > > > > > > > Amazon Web Services EMEA SARL > > 38 avenue John F. Kennedy, L-1855 Luxembourg > > Sitz der Gesellschaft: L-1855 Luxemburg > > eingetragen im Luxemburgischen Handelsregister unter R.C.S. B186284 > > > > Amazon Web Services EMEA SARL, Niederlassung Deutschland > > Marcel-Breuer-Str. 12, D-80807 Muenchen > > Sitz der Zweigniederlassung: Muenchen > > eingetragen im Handelsregister des Amtsgerichts Muenchen unter HRB > 242240, > > USt-ID DE317013094 > > > > > > > > > > > > Amazon Web Services EMEA SARL > > 38 avenue John F. Kennedy, L-1855 Luxembourg > > Sitz der Gesellschaft: L-1855 Luxemburg > > eingetragen im Luxemburgischen Handelsregister unter R.C.S. B186284 > > > > Amazon Web Services EMEA SARL, Niederlassung Deutschland > > Marcel-Breuer-Str. 12, D-80807 Muenchen > > Sitz der Zweigniederlassung: Muenchen > > eingetragen im Handelsregister des Amtsgerichts Muenchen unter HRB > 242240, > > USt-ID DE317013094 > > > > > > > > >
