Hi Aljoscha, Good point on the potential optimization in the source. One thing to clarify, by "adding a minimumTimestamp()/maximumTimestamp() method pair to the split interface", did you mean "split reader interface"? If so, what should the readers do if they do not have such additional information? I am wondering if it is possible to leave such optimization to the source internal implementation.
@all After reading all the feedback, Biao and I talked a little bit offline. We would like to share some new thoughts with you and see what do you think. When looking at the Source API, we were trying to answer two questions. First of all, how would Flink use this API if someone else implemented it. Secondly, how would the connector contributors implement the interface? How difficult is the implementation. KafkaConsumer is a typical example of a thread-less reader. The idea was to allow different threading model on top of it. It could be a global single thread handles record fetching and processing in an event loop pattern; it could also be one dedicated fetcher thread for each consumer and a separate thread pool for record processing. The API gives the freedom of picking up threading model to the users. To answer the first question, I would love to have such a source reader API so Flink can choose whatever threading model it wants. However, implementing such an interface could be pretty challenging and error prone. On the other hand, having a source reader with a naive blocking socket is probably simple enough in most cases (actually sometimes this might even be the most efficient way). But it does not leave much option to Flink other than creating one thread per reader. Given the above thoughts, it might be reasonable to separate the SplitReader API into two: SyncReader and AsyncReader. The sync reader just has a simple blocking takeNext() API. And the AsyncReader just has a pollNext(Callback) or Future<?> pollNext(). All the other methods are shared by both readers and could be put into a package private parent interface like BaseSplitReader. Having these two readers allows both complicated and simple implementation, depending on the SplitReader writers. From Flink's perspective, it will choose a more efficient threading model if the SplitReader is an AsyncReader. Otherwise, it may have to use the one thread per reader model if the reader is a SyncReader. Users can also choose to implement both interface, in that case, it is up to Flink to choose which interface to use. Admittedly, this solution does have one more interface, but still seems rewarding. Any thoughts? Thanks, Jiangjie (Becket) Qin On Sun, Nov 18, 2018 at 11:33 PM Biao Liu <mmyy1...@gmail.com> wrote: > Hi community, > > Thank you guys for sharing ideas. > > The thing I really concern is about the thread mode. > Actually in Alibaba, we have implemented our "split reader" based source > two years ago. That's based on "SourceFunction", it's just an extension not > a refactoring. It's almost same with the version Thomas and Jamie described > in Google Doc. It really helps in many scenarios. > > However I don't like the thread mode which starts a thread for each split. > Starting extra thread in operator is not an ideal way IMO. Especially > thread count is decided by split count. So I was wondering if there is a > more elegant way. Do we really want these threads in Flink core? > > I agree that blocking interface is more easy to implement. Could we at > least separate the split reader with source function into different > interfaces? Not all sources would like to read all splits concurrently. In > batch scenario, reading splits one by one is more general. And also not all > sources are partitioned, right? > I prefer there is a new source interface with "pull mode" only, no split. > There is a splittable source extended it. And there is one implementation > that starting threads for each split, reading all splits concurrently. > > > Thomas Weise <t...@apache.org> 于2018年11月18日周日 上午3:18写道： > > > @Aljoscha to address your question first: In the case of the Kinesis > > consumer (with current Kinesis consumer API), there would also be N+1 > > threads. I have implemented a prototype similar to what is shown in > Jamie's > > document, where the thread ownership is similar to what you have done for > > Kafka. > > > > The equivalent of split reader manages its own thread and the "source > main > > thread" is responsible for emitting the data. The interface between the N > > reader threads and the 1 emitter is a blocking queue per consumer thread. > > The emitter can now control which queue to consume from based on the > event > > time progress. > > > > This is akin to a "non-blocking" interface *between emitter and split > > reader*. Emitter uses poll to retrieve records from the N queues (which > > requires non-blocking interaction). The emitter is independent of the > split > > reader implementation, that part could live in Flink. > > > > Regarding whether or not to assume that split readers always need a > thread > > and in addition that these reader threads should be managed by Flink: It > > depends on the API of respective external systems and I would not bake > that > > assumption into Flink. Some client libraries manage their own threads > (see > > push based API like JMS and as I understand it may also apply to the new > > fan-out Kinesis API: > > > > > https://docs.aws.amazon.com/streams/latest/dev/building-enhanced-consumers-api.html > > ). > > In such cases it would not make sense to layer another reader thread on > > top. It may instead be better if Flink provides to the split reader the > > queue/buffer to push records to. > > > > The discussion so far has largely ignored the discovery aspect. There are > > some important considerations such as ordering dependency of splits and > > work rebalancing that may affect the split reader interface. Should we > fork > > this into a separate thread? > > > > Thanks, > > Thomas > > > > > > On Fri, Nov 16, 2018 at 8:09 AM Piotr Nowojski <pi...@data-artisans.com> > > wrote: > > > > > Hi Jamie, > > > > > > As it was already covered with my discussion with Becket, there is an > > easy > > > way to provide blocking API on top of non-blocking API. And yes we both > > > agreed that blocking API is easier to implement by users. > > > > > > I also do not agree with respect to usefulness of non blocking API. > > > Actually Kafka connector is the one that could be more efficient thanks > > to > > > the removal of the one layer of threading. > > > > > > Piotrek > > > > > > > On 16 Nov 2018, at 02:21, Jamie Grier <jgr...@lyft.com.INVALID> > wrote: > > > > > > > > Thanks Aljoscha for getting this effort going! > > > > > > > > There's been plenty of discussion here already and I'll add my big +1 > > to > > > > making this interface very simple to implement for a new > > > > Source/SplitReader. Writing a new production quality connector for > > Flink > > > > is very difficult today and requires a lot of detailed knowledge > about > > > > Flink, event time progress, watermarking, idle shard detection, etc > and > > > it > > > > would be good to move almost all of this type of code into Flink > itself > > > and > > > > out of source implementations. I also think this is totally doable > and > > > I'm > > > > really excited to see this happening. > > > > > > > > I do have a couple of thoughts about the API and the implementation.. > > > > > > > > In a perfect world there would be a single thread per Flink source > > > sub-task > > > > and no additional threads for SplitReaders -- but this assumes a > world > > > > where you have true async IO APIs for the upstream systems (like > Kafka > > > and > > > > Kinesis, S3, HDFS, etc). If that world did exist the single thread > > could > > > > just sit in an efficient select() call waiting for new data to arrive > > on > > > > any Split. That'd be awesome.. > > > > > > > > But, that world doesn't exist and given that practical consideration > I > > > > would think the next best implementation is going to be, in practice, > > > > probably a thread per SplitReader that does nothing but call the > source > > > API > > > > and drop whatever it reads into a (blocking) queue -- as Aljoscha > > > mentioned > > > > (calling it N+1) and as we started to describe here: > > > > > > > > > > https://docs.google.com/document/d/13HxfsucoN7aUls1FX202KFAVZ4IMLXEZWyRfZNd6WFM/edit#heading=h.lfi2u4fv9cxa > > > > > > > > I guess my point is that I think we should strive to move as much of > > > > something like the diagram referenced in the above doc into Flink > > itself > > > > and out of sources and simplify the SplitReader API as much as > possible > > > as > > > > well. > > > > > > > > With the above in mind and with regard to the discussion about > > blocking, > > > > etc.. I'm not sure I agree with some of the discussion so far with > > > regard > > > > to this API design. The calls to the upstream systems > (kafka/kinesis) > > > are > > > > in fact going to be blocking calls. So a simple API without the > > > constraint > > > > that the methods must be implemented in a non-blocking way seems > better > > > to > > > > me from the point of view of somebody writing a new source > > > implementation. > > > > My concern is that if you force the implementer of the SplitReader > > > > interface to do so in a non-blocking way you're just going to make it > > > > harder to write those implementations. Those calls to read the next > > bit > > > of > > > > data are going to be blocking calls with most known important sources > > -- > > > at > > > > least Kafka/Kinesis/HDFS -- so I think maybe we should just deal with > > > that > > > > head on and work around it a higher level so the SplitReader > interface > > > > stays super simple to implement. This means we manage all the > > threading > > > in > > > > Flink core, the API stays pull-based, and the implementer is allowed > to > > > > simply block until they have data to return. > > > > > > > > I maybe would change my mind about this if truly asynchronous APIs to > > the > > > > upstream source systems were likely to be available in the near > future > > or > > > > are now and I'm just ignorant of it. But even then the supporting > code > > > in > > > > Flink to drive async and sync sources would be different and in fact > > they > > > > might just have different APIs altogether -- SplitReader vs > > > > AsyncSplitReader maybe. > > > > > > > > In the end I think playing with the implementation, across more than > > one > > > > source, and moving as much common code into Flink itself will reveal > > the > > > > best API of course. > > > > > > > > One other interesting note is that you need to preserve per-partition > > > > ordering so you have to take care with the implementation if it were > to > > > be > > > > based on a thread pool and futures so as not to reorder the reads. > > > > > > > > Anyway, I'm thrilled to see this starting to move forward and I'd > very > > > much > > > > like to help with the implementation wherever I can. We're doing a > > > > simplified internal version of some of this at Lyft for just Kinesis > > > > because we need a solution for event time alignment in the very short > > > term > > > > but we'd like to immediately start helping to do this properly in > Flink > > > > after that. One of the end goals for us is event time alignment > across > > > > heterogeneous sources. Another is making it possible for non-expert > > > users > > > > to have a high probability of being able to write their own, correct, > > > > connectors. > > > > > > > > -Jamie > > > > > > > > On Thu, Nov 15, 2018 at 3:43 AM Aljoscha Krettek < > aljos...@apache.org> > > > > wrote: > > > > > > > >> Hi, > > > >> > > > >> I thought I had sent this mail a while ago but I must have forgotten > > to > > > >> send it. > > > >> > > > >> There is another thing we should consider for splits: the range of > > > >> timestamps that it can contain. For example, the splits of a file > > source > > > >> would know what the minimum and maximum timestamp in the splits is, > > > >> roughly. For infinite splits, such as Kafka partitions, the minimum > > > would > > > >> be meaningful but the maximum would be +Inf. If the splits expose > the > > > >> interval of time that they contain the readers, or the component > that > > > >> manages the readers can make decisions about which splits to forward > > and > > > >> read first. And it can also influence the minimum watermark that a > > > reader > > > >> forwards: it should never emit a watermark if it knows there are > > splits > > > to > > > >> read that have a lower minimum timestamp. I think it should be as > easy > > > as > > > >> adding a minimumTimestamp()/maximumTimestamp() method pair to the > > split > > > >> interface. > > > >> > > > >> Another thing we need to resolve is the actual reader interface. I > see > > > >> there has been some good discussion but I don't know if we have > > > consensus. > > > >> We should try and see how specific sources could be implemented with > > the > > > >> new interface. For example, for Kafka I think we need to have N+1 > > > threads > > > >> per task (where N is the number of splits that a task is reading > > from). > > > On > > > >> thread is responsible for reading from the splits. And each split > has > > > its > > > >> own (internal) thread for reading from Kafka and putting messages in > > an > > > >> internal queue to pull from. This is similar to how the current > Kafka > > > >> source is implemented, which has a separate fetcher thread. The > reason > > > for > > > >> this split is that we always need to try reading from Kafka to keep > > the > > > >> throughput up. In the current implementation the internal queue (or > > > >> handover) limits the read rate of the reader threads. > > > >> > > > >> @Thomas, what do you think this would look like for Kinesis? > > > >> > > > >> Best, > > > >> Aljoscha > > > >> > > > >>> On 15. Nov 2018, at 03:56, Becket Qin <becket....@gmail.com> > wrote: > > > >>> > > > >>> Hi Piotrek, > > > >>> > > > >>> Thanks a lot for the detailed reply. All makes sense to me. > > > >>> > > > >>> WRT the confusion between advance() / getCurrent(), do you think it > > > would > > > >>> help if we combine them and have something like: > > > >>> > > > >>> CompletableFuture<T> getNext(); > > > >>> long getWatermark(); > > > >>> long getCurrentTimestamp(); > > > >>> > > > >>> Cheers, > > > >>> > > > >>> Jiangjie (Becket) Qin > > > >>> > > > >>> On Tue, Nov 13, 2018 at 9:56 PM Piotr Nowojski < > > > pi...@data-artisans.com> > > > >>> wrote: > > > >>> > > > >>>> Hi, > > > >>>> > > > >>>> Thanks again for the detailed answer :) Sorry for responding with > a > > > >> delay. > > > >>>> > > > >>>>> Completely agree that in pattern 2, having a callback is > necessary > > > for > > > >>>> that > > > >>>>> single thread outside of the connectors. And the connectors MUST > > have > > > >>>>> internal threads. > > > >>>> > > > >>>> Yes, this thread will have to exists somewhere. In pattern 2 it > > exists > > > >> in > > > >>>> the connector (at least from the perspective of the Flink > execution > > > >>>> engine). In pattern 1 it exists inside the Flink execution engine. > > > With > > > >>>> completely blocking connectors, like simple reading from files, > both > > > of > > > >>>> those approaches are basically the same. The difference is when > user > > > >>>> implementing Flink source is already working with a non blocking > > code > > > >> with > > > >>>> some internal threads. In this case, pattern 1 would result in > > "double > > > >>>> thread wrapping”, while pattern 2 would allow to skip one layer of > > > >>>> indirection. > > > >>>> > > > >>>>> If we go that way, we should have something like "void > > > >>>>> poll(Callback) / void advance(callback)". I am curious how would > > > >>>>> CompletableFuture work here, though. If 10 readers returns 10 > > > >> completable > > > >>>>> futures, will there be 10 additional threads (so 20 threads in > > total) > > > >>>>> blocking waiting on them? Or will there be a single thread busy > > loop > > > >>>>> checking around? > > > >>>> > > > >>>> To be honest, I haven’t thought this completely through and I > > haven’t > > > >>>> tested/POC’ed it. Having said that, I can think of at least couple > > of > > > >>>> solutions. First is something like this: > > > >>>> > > > >>>> > > > >>>> > > > >> > > > > > > https://github.com/prestodb/presto/blob/master/presto-main/src/main/java/com/facebook/presto/execution/executor/TaskExecutor.java#L481-L507 > > > >>>> < > > > >>>> > > > >> > > > > > > https://github.com/prestodb/presto/blob/master/presto-main/src/main/java/com/facebook/presto/execution/executor/TaskExecutor.java#L481-L507 > > > >>>>> > > > >>>> > > > >>>> Line: > > > >>>> > > > >>>> `blocked = split.process();` > > > >>>> > > > >>>> Is where the execution goes into to the task/sources. This is > where > > > the > > > >>>> returned future is handled: > > > >>>> > > > >>>> blocked.addListener(() -> { > > > >>>> blockedSplits.remove(split); > > > >>>> // reset the level priority to > > > >> prevent > > > >>>> previously-blocked splits from starving existing splits > > > >>>> split.resetLevelPriority(); > > > >>>> waitingSplits.offer(split); > > > >>>> }, executor); > > > >>>> > > > >>>> Fundamentally callbacks and Futures are more or less > interchangeable > > > You > > > >>>> can always wrap one into another (creating a callback that > > completes a > > > >>>> future and attach a callback once future completes). In this case > > the > > > >>>> difference for me is mostly: > > > >>>> - api with passing callback allows the callback to be fired > multiple > > > >> times > > > >>>> and to fire it even if the connector is not blocked. This is what > I > > > >> meant > > > >>>> by saying that api `CompletableFuture<?> isBlocked()` is a bit > > > simpler. > > > >>>> Connector can only return either “I’m not blocked” or “I’m blocked > > > and I > > > >>>> will tell you only once when I’m not blocked anymore”. > > > >>>> > > > >>>> But this is not the most important thing for me here. For me > > important > > > >>>> thing is to try our best to make Flink task’s control and > execution > > > >> single > > > >>>> threaded. For that both callback and future APIs should work the > > same. > > > >>>> > > > >>>>> WRT pattern 1, a single blocking take() API should just work. The > > > good > > > >>>>> thing is that a blocking read API is usually simpler to > implement. > > > >>>> > > > >>>> Yes, they are easier to implement (especially if you are not the > one > > > >> that > > > >>>> have to deal with the additional threading required around them ;) > > ). > > > >> But > > > >>>> to answer this issue, if we choose pattern 2, we can always > provide > > a > > > >>>> proxy/wrapper that would using the internal thread implement the > > > >>>> non-blocking API while exposing blocking API to the user. It would > > > >>>> implement pattern 2 for the user exposing to him pattern 1. In > other > > > >> words > > > >>>> implementing pattern 1 in pattern 2 paradigm, while making it > > possible > > > >> to > > > >>>> implement pure pattern 2 connectors. > > > >>>> > > > >>>>> BTW, one thing I am also trying to avoid is pushing users to > > perform > > > IO > > > >>>> in > > > >>>>> a method like "isBlocked()". If the method is expected to fetch > > > records > > > >>>>> (even if not returning them), naming it something more explicit > > would > > > >>>> help > > > >>>>> avoid confusion. > > > >>>> > > > >>>> If we choose so, we could rework it into something like: > > > >>>> > > > >>>> CompletableFuture<?> advance() > > > >>>> T getCurrent(); > > > >>>> Watermark getCurrentWatermark() > > > >>>> > > > >>>> But as I wrote before, this is more confusing to me for the exact > > > >> reasons > > > >>>> you mentioned :) I would be confused what should be done in > > > `adanvce()` > > > >> and > > > >>>> what in `getCurrent()`. However, again this naming issue is not > that > > > >>>> important to me and probably is matter of taste/personal > > preferences. > > > >>>> > > > >>>> Piotrek > > > >>>> > > > >>>>> On 9 Nov 2018, at 18:37, Becket Qin <becket....@gmail.com> > wrote: > > > >>>>> > > > >>>>> Hi Piotrek, > > > >>>>> > > > >>>>> Thanks for the explanation. We are probably talking about the > same > > > >> thing > > > >>>>> but in different ways. To clarify a little bit, I think there are > > two > > > >>>>> patterns to read from a connector. > > > >>>>> > > > >>>>> Pattern 1: Thread-less connector with a blocking read API. > Outside > > of > > > >> the > > > >>>>> connector, there is one IO thread per reader, doing blocking > read. > > An > > > >>>>> additional thread will interact with all the IO threads. > > > >>>>> Pattern 2: Connector with internal thread(s) and non-blocking > API. > > > >>>> Outside > > > >>>>> of the connector, there is one thread for ALL readers, doing IO > > > relying > > > >>>> on > > > >>>>> notification callbacks in the reader. > > > >>>>> > > > >>>>> In both patterns, there must be at least one thread per > connector, > > > >> either > > > >>>>> inside (created by connector writers) or outside (created by > Flink) > > > of > > > >>>> the > > > >>>>> connector. Ideally there are NUM_CONNECTORS + 1 threads in total, > > to > > > >> make > > > >>>>> sure that 1 thread is fully non-blocking. > > > >>>>> > > > >>>>>> Btw, I don’t know if you understand my point. Having only > `poll()` > > > and > > > >>>>> `take()` is not enough for single threaded task. If our source > > > >> interface > > > >>>>> doesn’t provide `notify()` callback nor >`CompletableFuture<?> > > > >>>>> isBlocked(),`, there is no way to implement single threaded task > > that > > > >>>> both > > > >>>>> reads the data from the source connector and can also react to > > system > > > >>>>> events. Ok, non >blocking `poll()` would allow that, but with > busy > > > >>>> looping. > > > >>>>> > > > >>>>> Completely agree that in pattern 2, having a callback is > necessary > > > for > > > >>>> that > > > >>>>> single thread outside of the connectors. And the connectors MUST > > have > > > >>>>> internal threads. If we go that way, we should have something > like > > > >> "void > > > >>>>> poll(Callback) / void advance(callback)". I am curious how would > > > >>>>> CompletableFuture work here, though. If 10 readers returns 10 > > > >> completable > > > >>>>> futures, will there be 10 additional threads (so 20 threads in > > total) > > > >>>>> blocking waiting on them? Or will there be a single thread busy > > loop > > > >>>>> checking around? > > > >>>>> > > > >>>>> WRT pattern 1, a single blocking take() API should just work. The > > > good > > > >>>>> thing is that a blocking read API is usually simpler to > implement. > > An > > > >>>>> additional non-blocking "T poll()" method here is indeed optional > > and > > > >>>> could > > > >>>>> be used in cases like Flink does not want the thread to block > > > forever. > > > >>>> They > > > >>>>> can also be combined to have a "T poll(Timeout)", which is > exactly > > > what > > > >>>>> KafkaConsumer did. > > > >>>>> > > > >>>>> It sounds that you are proposing pattern 2 with something similar > > to > > > >> NIO2 > > > >>>>> AsynchronousByteChannel[1]. That API would work, except that the > > > >>>> signature > > > >>>>> returning future seems not necessary. If that is the case, a > minor > > > >> change > > > >>>>> on the current FLIP proposal to have "void advance(callback)" > > should > > > >>>> work. > > > >>>>> And this means the connectors MUST have their internal threads. > > > >>>>> > > > >>>>> BTW, one thing I am also trying to avoid is pushing users to > > perform > > > IO > > > >>>> in > > > >>>>> a method like "isBlocked()". If the method is expected to fetch > > > records > > > >>>>> (even if not returning them), naming it something more explicit > > would > > > >>>> help > > > >>>>> avoid confusion. > > > >>>>> > > > >>>>> Thanks, > > > >>>>> > > > >>>>> Jiangjie (Becket) Qin > > > >>>>> > > > >>>>> [1] > > > >>>>> > > > >>>> > > > >> > > > > > > https://docs.oracle.com/javase/8/docs/api/java/nio/channels/AsynchronousByteChannel.html > > > >>>>> > > > >>>>> On Fri, Nov 9, 2018 at 11:20 PM Piotr Nowojski < > > > >> pi...@data-artisans.com> > > > >>>>> wrote: > > > >>>>> > > > >>>>>> Hi > > > >>>>>> > > > >>>>>> Good point with select/epoll, however I do not see how they > > couldn’t > > > >> be > > > >>>>>> with Flink if we would like single task in Flink to be > > > single-threaded > > > >>>> (and > > > >>>>>> I believe we should pursue this goal). If your connector blocks > on > > > >>>>>> `select`, then it can not process/handle control messages from > > > Flink, > > > >>>> like > > > >>>>>> checkpoints, releasing resources and potentially output flushes. > > > This > > > >>>> would > > > >>>>>> require tight integration between connector and Flink’s main > event > > > >>>>>> loop/selects/etc. > > > >>>>>> > > > >>>>>> Looking at it from other perspective. Let’s assume that we have > a > > > >>>>>> connector implemented on top of `select`/`epoll`. In order to > > > >> integrate > > > >>>> it > > > >>>>>> with Flink’s checkpointing/flushes/resource releasing it will > have > > > to > > > >> be > > > >>>>>> executed in separate thread one way or another. At least if our > > API > > > >> will > > > >>>>>> enforce/encourage non blocking implementations with some kind of > > > >>>>>> notifications (`isBlocked()` or `notify()` callback), some > > > connectors > > > >>>> might > > > >>>>>> skip one layer of wapping threads. > > > >>>>>> > > > >>>>>> Btw, I don’t know if you understand my point. Having only > `poll()` > > > and > > > >>>>>> `take()` is not enough for single threaded task. If our source > > > >> interface > > > >>>>>> doesn’t provide `notify()` callback nor `CompletableFuture<?> > > > >>>>>> isBlocked(),`, there is no way to implement single threaded task > > > that > > > >>>> both > > > >>>>>> reads the data from the source connector and can also react to > > > system > > > >>>>>> events. Ok, non blocking `poll()` would allow that, but with > busy > > > >>>> looping. > > > >>>>>> > > > >>>>>> Piotrek > > > >>>>>> > > > >>>>>>> On 8 Nov 2018, at 06:56, Becket Qin <becket....@gmail.com> > > wrote: > > > >>>>>>> > > > >>>>>>> Hi Piotrek, > > > >>>>>>> > > > >>>>>>>> But I don’t see a reason why we should expose both blocking > > > `take()` > > > >>>> and > > > >>>>>>> non-blocking `poll()` methods to the Flink engine. Someone > (Flink > > > >>>> engine > > > >>>>>> or > > > >>>>>>> connector) would have to do the same busy > > > >>>>>>>> looping anyway and I think it would be better to have a > simpler > > > >>>>>> connector > > > >>>>>>> API (that would solve our problems) and force connectors to > > comply > > > >> one > > > >>>>>> way > > > >>>>>>> or another. > > > >>>>>>> > > > >>>>>>> If we let the block happen inside the connector, the blocking > > does > > > >> not > > > >>>>>> have > > > >>>>>>> to be a busy loop. For example, to do the block waiting > > > efficiently, > > > >>>> the > > > >>>>>>> connector can use java NIO selector().select which relies on OS > > > >> syscall > > > >>>>>>> like epoll[1] instead of busy looping. But if Flink engine > blocks > > > >>>> outside > > > >>>>>>> the connector, it pretty much has to do the busy loop. So if > > there > > > is > > > >>>>>> only > > > >>>>>>> one API to get the element, a blocking getNextElement() makes > > more > > > >>>> sense. > > > >>>>>>> In any case, we should avoid ambiguity. It has to be crystal > > clear > > > >>>> about > > > >>>>>>> whether a method is expected to be blocking or non-blocking. > > > >> Otherwise > > > >>>> it > > > >>>>>>> would be very difficult for Flink engine to do the right thing > > with > > > >> the > > > >>>>>>> connectors. At the first glance at getCurrent(), the expected > > > >> behavior > > > >>>> is > > > >>>>>>> not quite clear. > > > >>>>>>> > > > >>>>>>> That said, I do agree that functionality wise, poll() and > take() > > > kind > > > >>>> of > > > >>>>>>> overlap. But they are actually not quite different from > > > >>>>>>> isBlocked()/getNextElement(). Compared with isBlocked(), the > only > > > >>>>>>> difference is that poll() also returns the next record if it is > > > >>>>>> available. > > > >>>>>>> But I agree that the isBlocked() + getNextElement() is more > > > flexible > > > >> as > > > >>>>>>> users can just check the record availability, but not fetch the > > > next > > > >>>>>>> element. > > > >>>>>>> > > > >>>>>>>> In case of thread-less readers with only non-blocking > > > `queue.poll()` > > > >>>> (is > > > >>>>>>> that really a thing? I can not think about a real > implementation > > > that > > > >>>>>>> enforces such constraints) > > > >>>>>>> Right, it is pretty much a syntax sugar to allow user combine > the > > > >>>>>>> check-and-take into one method. It could be achieved with > > > >> isBlocked() + > > > >>>>>>> getNextElement(). > > > >>>>>>> > > > >>>>>>> [1] http://man7.org/linux/man-pages/man7/epoll.7.html > > > >>>>>>> > > > >>>>>>> Thanks, > > > >>>>>>> > > > >>>>>>> Jiangjie (Becket) Qin > > > >>>>>>> > > > >>>>>>> On Wed, Nov 7, 2018 at 11:58 PM Piotr Nowojski < > > > >>>> pi...@data-artisans.com> > > > >>>>>>> wrote: > > > >>>>>>> > > > >>>>>>>> Hi Becket, > > > >>>>>>>> > > > >>>>>>>> With my proposal, both of your examples would have to be > solved > > by > > > >> the > > > >>>>>>>> connector and solution to both problems would be the same: > > > >>>>>>>> > > > >>>>>>>> Pretend that connector is never blocked (`isBlocked() { return > > > >>>>>>>> NOT_BLOCKED; }`) and implement `getNextElement()` in blocking > > > >> fashion > > > >>>>>> (or > > > >>>>>>>> semi blocking with return of control from time to time to > allow > > > for > > > >>>>>>>> checkpointing, network flushing and other resource management > > > things > > > >>>> to > > > >>>>>>>> happen in the same main thread). In other words, exactly how > you > > > >> would > > > >>>>>>>> implement `take()` method or how the same source connector > would > > > be > > > >>>>>>>> implemented NOW with current source interface. The difference > > with > > > >>>>>> current > > > >>>>>>>> interface would be only that main loop would be outside of the > > > >>>>>> connector, > > > >>>>>>>> and instead of periodically releasing checkpointing lock, > > > >> periodically > > > >>>>>>>> `return null;` or `return Optional.empty();` from > > > >> `getNextElement()`. > > > >>>>>>>> > > > >>>>>>>> In case of thread-less readers with only non-blocking > > > `queue.poll()` > > > >>>> (is > > > >>>>>>>> that really a thing? I can not think about a real > implementation > > > >> that > > > >>>>>>>> enforces such constraints), we could provide a wrapper that > > hides > > > >> the > > > >>>>>> busy > > > >>>>>>>> looping. The same applies how to solve forever blocking > readers > > - > > > we > > > >>>>>> could > > > >>>>>>>> provider another wrapper running the connector in separate > > thread. > > > >>>>>>>> > > > >>>>>>>> But I don’t see a reason why we should expose both blocking > > > `take()` > > > >>>> and > > > >>>>>>>> non-blocking `poll()` methods to the Flink engine. Someone > > (Flink > > > >>>>>> engine or > > > >>>>>>>> connector) would have to do the same busy looping anyway and I > > > think > > > >>>> it > > > >>>>>>>> would be better to have a simpler connector API (that would > > solve > > > >> our > > > >>>>>>>> problems) and force connectors to comply one way or another. > > > >>>>>>>> > > > >>>>>>>> Piotrek > > > >>>>>>>> > > > >>>>>>>>> On 7 Nov 2018, at 10:55, Becket Qin <becket....@gmail.com> > > > wrote: > > > >>>>>>>>> > > > >>>>>>>>> Hi Piotr, > > > >>>>>>>>> > > > >>>>>>>>> I might have misunderstood you proposal. But let me try to > > > explain > > > >> my > > > >>>>>>>>> concern. I am thinking about the following case: > > > >>>>>>>>> 1. a reader has the following two interfaces, > > > >>>>>>>>> boolean isBlocked() > > > >>>>>>>>> T getNextElement() > > > >>>>>>>>> 2. the implementation of getNextElement() is non-blocking. > > > >>>>>>>>> 3. The reader is thread-less, i.e. it does not have any > > internal > > > >>>>>> thread. > > > >>>>>>>>> For example, it might just delegate the getNextElement() to a > > > >>>>>>>> queue.poll(), > > > >>>>>>>>> and isBlocked() is just queue.isEmpty(). > > > >>>>>>>>> > > > >>>>>>>>> How can Flink efficiently implement a blocking reading > behavior > > > >> with > > > >>>>>> this > > > >>>>>>>>> reader? Either a tight loop or a backoff interval is needed. > > > >> Neither > > > >>>> of > > > >>>>>>>>> them is ideal. > > > >>>>>>>>> > > > >>>>>>>>> Now let's say in the reader mentioned above implements a > > blocking > > > >>>>>>>>> getNextElement() method. Because there is no internal thread > in > > > the > > > >>>>>>>> reader, > > > >>>>>>>>> after isBlocked() returns false. Flink will still have to > loop > > on > > > >>>>>>>>> isBlocked() to check whether the next record is available. If > > the > > > >>>> next > > > >>>>>>>>> record reaches after 10 min, it is a tight loop for 10 min. > You > > > >> have > > > >>>>>>>>> probably noticed that in this case, even isBlocked() returns > a > > > >>>> future, > > > >>>>>>>> that > > > >>>>>>>>> future() will not be completed if Flink does not call some > > method > > > >>>> from > > > >>>>>>>> the > > > >>>>>>>>> reader, because the reader has no internal thread to complete > > > that > > > >>>>>> future > > > >>>>>>>>> by itself. > > > >>>>>>>>> > > > >>>>>>>>> Due to the above reasons, a blocking take() API would allow > > Flink > > > >> to > > > >>>>>> have > > > >>>>>>>>> an efficient way to read from a reader. There are many ways > to > > > wake > > > >>>> up > > > >>>>>>>> the > > > >>>>>>>>> blocking thread when checkpointing is needed depending on the > > > >>>>>>>>> implementation. But I think the poll()/take() API would also > > work > > > >> in > > > >>>>>> that > > > >>>>>>>>> case. > > > >>>>>>>>> > > > >>>>>>>>> Thanks, > > > >>>>>>>>> > > > >>>>>>>>> Jiangjie (Becket) Qin > > > >>>>>>>>> > > > >>>>>>>>> On Wed, Nov 7, 2018 at 4:31 PM Piotr Nowojski < > > > >>>> pi...@data-artisans.com > > > >>>>>>> > > > >>>>>>>>> wrote: > > > >>>>>>>>> > > > >>>>>>>>>> Hi, > > > >>>>>>>>>> > > > >>>>>>>>>> a) > > > >>>>>>>>>> > > > >>>>>>>>>>> BTW, regarding the isBlock() method, I have a few more > > > questions. > > > >>>> 21, > > > >>>>>>>> Is > > > >>>>>>>>>> a method isReady() with boolean as a return value > > > >>>>>>>>>>> equivalent? Personally I found it is a little bit confusing > > in > > > >> what > > > >>>>>> is > > > >>>>>>>>>> supposed to be returned when the future is completed. 22. if > > > >>>>>>>>>>> the implementation of isBlocked() is optional, how do the > > > callers > > > >>>>>> know > > > >>>>>>>>>> whether the method is properly implemented or not? > > > >>>>>>>>>>> Does not implemented mean it always return a completed > > future? > > > >>>>>>>>>> > > > >>>>>>>>>> `CompletableFuture<?> isBlocked()` is more or less an > > equivalent > > > >> to > > > >>>>>>>>>> `boolean hasNext()` which in case of “false” provides some > > kind > > > >> of a > > > >>>>>>>>>> listener/callback that notifies about presence of next > > element. > > > >>>> There > > > >>>>>>>> are > > > >>>>>>>>>> some minor details, like `CompletableFuture<?>` has a > minimal > > > two > > > >>>>>> state > > > >>>>>>>>>> logic: > > > >>>>>>>>>> > > > >>>>>>>>>> 1. Future is completed - we have more data > > > >>>>>>>>>> 2. Future not yet completed - we don’t have data now, but we > > > >>>> might/we > > > >>>>>>>> will > > > >>>>>>>>>> have in the future > > > >>>>>>>>>> > > > >>>>>>>>>> While `boolean hasNext()` and `notify()` callback are a bit > > more > > > >>>>>>>>>> complicated/dispersed and can lead/encourage `notify()` > spam. > > > >>>>>>>>>> > > > >>>>>>>>>> b) > > > >>>>>>>>>> > > > >>>>>>>>>>> 3. If merge the `advance` and `getCurrent` to one method > > like > > > >>>>>>>> `getNext` > > > >>>>>>>>>> the `getNext` would need return a > > > >>>>>>>>>>> `ElementWithTimestamp` because some sources want to add > > > timestamp > > > >>>> to > > > >>>>>>>>>> every element. IMO, this is not so memory friendly > > > >>>>>>>>>>> so I prefer this design. > > > >>>>>>>>>> > > > >>>>>>>>>> Guowei I don’t quite understand this. Could you elaborate > why > > > >>>> having a > > > >>>>>>>>>> separate `advance()` help? > > > >>>>>>>>>> > > > >>>>>>>>>> c) > > > >>>>>>>>>> > > > >>>>>>>>>> Regarding advance/poll/take. What’s the value of having two > > > >> separate > > > >>>>>>>>>> methods: poll and take? Which one of them should be called > and > > > >> which > > > >>>>>>>>>> implemented? What’s the benefit of having those methods > > compared > > > >> to > > > >>>>>>>> having > > > >>>>>>>>>> a one single method `getNextElement()` (or `pollElement() or > > > >>>> whatever > > > >>>>>> we > > > >>>>>>>>>> name it) with following contract: > > > >>>>>>>>>> > > > >>>>>>>>>> CompletableFuture<?> isBlocked(); > > > >>>>>>>>>> > > > >>>>>>>>>> /** > > > >>>>>>>>>> Return next element - will be called only if `isBlocked()` > is > > > >>>>>> completed. > > > >>>>>>>>>> Try to implement it in non blocking fashion, but if that’s > > > >>>> impossible > > > >>>>>> or > > > >>>>>>>>>> you just don’t need the effort, you can block in this > method. > > > >>>>>>>>>> */ > > > >>>>>>>>>> T getNextElement(); > > > >>>>>>>>>> > > > >>>>>>>>>> I mean, if the connector is implemented non-blockingly, > Flink > > > >> should > > > >>>>>> use > > > >>>>>>>>>> it that way. If it’s not, then `poll()` will `throw new > > > >>>>>>>>>> NotImplementedException()`. Implementing both of them and > > > >> providing > > > >>>>>>>> both of > > > >>>>>>>>>> them to Flink wouldn’t make a sense, thus why not merge them > > > into > > > >> a > > > >>>>>>>> single > > > >>>>>>>>>> method call that should preferably (but not necessarily need > > to) > > > >> be > > > >>>>>>>>>> non-blocking? It’s not like we are implementing general > > purpose > > > >>>>>> `Queue`, > > > >>>>>>>>>> which users might want to call either of `poll` or `take`. > We > > > >> would > > > >>>>>>>> always > > > >>>>>>>>>> prefer to call `poll`, but if it’s blocking, then still we > > have > > > no > > > >>>>>>>> choice, > > > >>>>>>>>>> but to call it and block on it. > > > >>>>>>>>>> > > > >>>>>>>>>> d) > > > >>>>>>>>>> > > > >>>>>>>>>>> 1. I agree with Piotr and Becket that the non-blocking > source > > > is > > > >>>> very > > > >>>>>>>>>>> important. But in addition to `Future/poll`, there may be > > > another > > > >>>> way > > > >>>>>>>> to > > > >>>>>>>>>>> achieve this. I think it may be not very memory friendly if > > > every > > > >>>>>>>> advance > > > >>>>>>>>>>> call return a Future. > > > >>>>>>>>>> > > > >>>>>>>>>> I didn’t want to mention this, to not clog my initial > > proposal, > > > >> but > > > >>>>>>>> there > > > >>>>>>>>>> is a simple solution for the problem: > > > >>>>>>>>>> > > > >>>>>>>>>> public interface SplitReader { > > > >>>>>>>>>> > > > >>>>>>>>>> (…) > > > >>>>>>>>>> > > > >>>>>>>>>> CompletableFuture<?> NOT_BLOCKED = > > > >>>>>>>>>> CompletableFuture.completedFuture(null); > > > >>>>>>>>>> > > > >>>>>>>>>> /** > > > >>>>>>>>>> * Returns a future that will be completed when the page > source > > > >>>>>>>> becomes > > > >>>>>>>>>> * unblocked. If the page source is not blocked, this method > > > >>>> should > > > >>>>>>>>>> return > > > >>>>>>>>>> * {@code NOT_BLOCKED}. > > > >>>>>>>>>> */ > > > >>>>>>>>>> default CompletableFuture<?> isBlocked() > > > >>>>>>>>>> { > > > >>>>>>>>>> return NOT_BLOCKED; > > > >>>>>>>>>> } > > > >>>>>>>>>> > > > >>>>>>>>>> If we are blocked and we are waiting for the IO, then > > creating a > > > >> new > > > >>>>>>>>>> Future is non-issue. Under full throttle/throughput and not > > > >> blocked > > > >>>>>>>> sources > > > >>>>>>>>>> returning a static `NOT_BLOCKED` constant should also solve > > the > > > >>>>>>>> problem. > > > >>>>>>>>>> > > > >>>>>>>>>> One more remark, non-blocking sources might be a necessity > in > > a > > > >>>> single > > > >>>>>>>>>> threaded model without a checkpointing lock. (Currently when > > > >> sources > > > >>>>>> are > > > >>>>>>>>>> blocked, they can release checkpointing lock and re-acquire > it > > > >> again > > > >>>>>>>>>> later). Non-blocking `poll`/`getNext()` would allow for > > > >> checkpoints > > > >>>> to > > > >>>>>>>>>> happen when source is idling. In that case either `notify()` > > or > > > my > > > >>>>>>>> proposed > > > >>>>>>>>>> `isBlocked()` would allow to avoid busy-looping. > > > >>>>>>>>>> > > > >>>>>>>>>> Piotrek > > > >>>>>>>>>> > > > >>>>>>>>>>> On 5 Nov 2018, at 03:59, Becket Qin <becket....@gmail.com> > > > >> wrote: > > > >>>>>>>>>>> > > > >>>>>>>>>>> Hi Thomas, > > > >>>>>>>>>>> > > > >>>>>>>>>>> The iterator-like API was also the first thing that came to > > me. > > > >> But > > > >>>>>> it > > > >>>>>>>>>>> seems a little confusing that hasNext() does not mean "the > > > stream > > > >>>> has > > > >>>>>>>> not > > > >>>>>>>>>>> ended", but means "the next record is ready", which is > > > >> repurposing > > > >>>>>> the > > > >>>>>>>>>> well > > > >>>>>>>>>>> known meaning of hasNext(). If we follow the > hasNext()/next() > > > >>>>>> pattern, > > > >>>>>>>> an > > > >>>>>>>>>>> additional isNextReady() method to indicate whether the > next > > > >> record > > > >>>>>> is > > > >>>>>>>>>>> ready seems more intuitive to me. > > > >>>>>>>>>>> > > > >>>>>>>>>>> Similarly, in poll()/take() pattern, another method of > > isDone() > > > >> is > > > >>>>>>>> needed > > > >>>>>>>>>>> to indicate whether the stream has ended or not. > > > >>>>>>>>>>> > > > >>>>>>>>>>> Compared with hasNext()/next()/isNextReady() pattern, > > > >>>>>>>>>>> isDone()/poll()/take() seems more flexible for the reader > > > >>>>>>>> implementation. > > > >>>>>>>>>>> When I am implementing a reader, I could have a couple of > > > >> choices: > > > >>>>>>>>>>> > > > >>>>>>>>>>> - A thread-less reader that does not have any internal > > thread. > > > >>>>>>>>>>> - When poll() is called, the same calling thread will > > perform a > > > >>>> bunch > > > >>>>>>>>>> of > > > >>>>>>>>>>> IO asynchronously. > > > >>>>>>>>>>> - When take() is called, the same calling thread will > > perform a > > > >>>>>>>>>> bunch > > > >>>>>>>>>>> of IO and wait until the record is ready. > > > >>>>>>>>>>> - A reader with internal threads performing network IO and > > put > > > >>>>>> records > > > >>>>>>>>>>> into a buffer. > > > >>>>>>>>>>> - When poll() is called, the calling thread simply reads > from > > > >> the > > > >>>>>>>>>>> buffer and return empty result immediately if there is no > > > >> record. > > > >>>>>>>>>>> - When take() is called, the calling thread reads from the > > > >> buffer > > > >>>>>>>>>> and > > > >>>>>>>>>>> block waiting if the buffer is empty. > > > >>>>>>>>>>> > > > >>>>>>>>>>> On the other hand, with the hasNext()/next()/isNextReady() > > API, > > > >> it > > > >>>> is > > > >>>>>>>>>> less > > > >>>>>>>>>>> intuitive for the reader developers to write the > thread-less > > > >>>> pattern. > > > >>>>>>>>>>> Although technically speaking one can still do the > > asynchronous > > > >> IO > > > >>>> to > > > >>>>>>>>>>> prepare the record in isNextReady(). But it is inexplicit > and > > > >> seems > > > >>>>>>>>>>> somewhat hacky. > > > >>>>>>>>>>> > > > >>>>>>>>>>> Thanks, > > > >>>>>>>>>>> > > > >>>>>>>>>>> Jiangjie (Becket) Qin > > > >>>>>>>>>>> > > > >>>>>>>>>>> > > > >>>>>>>>>>> > > > >>>>>>>>>>> On Mon, Nov 5, 2018 at 6:55 AM Thomas Weise < > t...@apache.org> > > > >>>> wrote: > > > >>>>>>>>>>> > > > >>>>>>>>>>>> Couple more points regarding discovery: > > > >>>>>>>>>>>> > > > >>>>>>>>>>>> The proposal mentions that discovery could be outside the > > > >>>> execution > > > >>>>>>>>>> graph. > > > >>>>>>>>>>>> Today, discovered partitions/shards are checkpointed. I > > > believe > > > >>>> that > > > >>>>>>>>>> will > > > >>>>>>>>>>>> also need to be the case in the future, even when > discovery > > > and > > > >>>>>>>> reading > > > >>>>>>>>>> are > > > >>>>>>>>>>>> split between different tasks. > > > >>>>>>>>>>>> > > > >>>>>>>>>>>> For cases such as resharding of a Kinesis stream, the > > > >> relationship > > > >>>>>>>>>> between > > > >>>>>>>>>>>> splits needs to be considered. Splits cannot be randomly > > > >>>> distributed > > > >>>>>>>>>> over > > > >>>>>>>>>>>> readers in certain situations. An example was mentioned > > here: > > > >>>>>>>>>>>> > > > >> https://github.com/apache/flink/pull/6980#issuecomment-435202809 > > > >>>>>>>>>>>> > > > >>>>>>>>>>>> Thomas > > > >>>>>>>>>>>> > > > >>>>>>>>>>>> > > > >>>>>>>>>>>> On Sun, Nov 4, 2018 at 1:43 PM Thomas Weise < > t...@apache.org > > > > > > >>>> wrote: > > > >>>>>>>>>>>> > > > >>>>>>>>>>>>> Thanks for getting the ball rolling on this! > > > >>>>>>>>>>>>> > > > >>>>>>>>>>>>> Can the number of splits decrease? Yes, splits can be > > closed > > > >> and > > > >>>> go > > > >>>>>>>>>> away. > > > >>>>>>>>>>>>> An example would be a shard merge in Kinesis (2 existing > > > shards > > > >>>>>> will > > > >>>>>>>> be > > > >>>>>>>>>>>>> closed and replaced with a new shard). > > > >>>>>>>>>>>>> > > > >>>>>>>>>>>>> Regarding advance/poll/take: IMO the least restrictive > > > approach > > > >>>>>> would > > > >>>>>>>>>> be > > > >>>>>>>>>>>>> the thread-less IO model (pull based, non-blocking, > caller > > > >>>>>> retrieves > > > >>>>>>>>>> new > > > >>>>>>>>>>>>> records when available). The current Kinesis API requires > > the > > > >> use > > > >>>>>> of > > > >>>>>>>>>>>>> threads. But that can be internal to the split reader and > > > does > > > >>>> not > > > >>>>>>>> need > > > >>>>>>>>>>>> to > > > >>>>>>>>>>>>> be a source API concern. In fact, that's what we are > > working > > > on > > > >>>>>> right > > > >>>>>>>>>> now > > > >>>>>>>>>>>>> as improvement to the existing consumer: Each shard > > consumer > > > >>>> thread > > > >>>>>>>>>> will > > > >>>>>>>>>>>>> push to a queue, the consumer main thread will poll the > > > >> queue(s). > > > >>>>>> It > > > >>>>>>>> is > > > >>>>>>>>>>>>> essentially a mapping from threaded IO to non-blocking. > > > >>>>>>>>>>>>> > > > >>>>>>>>>>>>> The proposed SplitReader interface would fit the > > thread-less > > > IO > > > >>>>>>>> model. > > > >>>>>>>>>>>>> Similar to an iterator, we find out if there is a new > > element > > > >>>>>>>> (hasNext) > > > >>>>>>>>>>>> and > > > >>>>>>>>>>>>> if so, move to it (next()). Separate calls deliver the > meta > > > >>>>>>>> information > > > >>>>>>>>>>>>> (timestamp, watermark). Perhaps advance call could offer > a > > > >>>> timeout > > > >>>>>>>>>>>> option, > > > >>>>>>>>>>>>> so that the caller does not end up in a busy wait. On the > > > other > > > >>>>>>>> hand, a > > > >>>>>>>>>>>>> caller processing multiple splits may want to cycle > through > > > >> fast, > > > >>>>>> to > > > >>>>>>>>>>>>> process elements of other splits as soon as they become > > > >>>> available. > > > >>>>>>>> The > > > >>>>>>>>>>>> nice > > > >>>>>>>>>>>>> thing is that this "split merge" logic can now live in > > Flink > > > >> and > > > >>>> be > > > >>>>>>>>>>>>> optimized and shared between different sources. > > > >>>>>>>>>>>>> > > > >>>>>>>>>>>>> Thanks, > > > >>>>>>>>>>>>> Thomas > > > >>>>>>>>>>>>> > > > >>>>>>>>>>>>> > > > >>>>>>>>>>>>> On Sun, Nov 4, 2018 at 6:34 AM Guowei Ma < > > > guowei....@gmail.com > > > >>> > > > >>>>>>>> wrote: > > > >>>>>>>>>>>>> > > > >>>>>>>>>>>>>> Hi, > > > >>>>>>>>>>>>>> Thanks Aljoscha for this FLIP. > > > >>>>>>>>>>>>>> > > > >>>>>>>>>>>>>> 1. I agree with Piotr and Becket that the non-blocking > > > source > > > >> is > > > >>>>>>>> very > > > >>>>>>>>>>>>>> important. But in addition to `Future/poll`, there may > be > > > >>>> another > > > >>>>>>>> way > > > >>>>>>>>>> to > > > >>>>>>>>>>>>>> achieve this. I think it may be not very memory friendly > > if > > > >>>> every > > > >>>>>>>>>>>> advance > > > >>>>>>>>>>>>>> call return a Future. > > > >>>>>>>>>>>>>> > > > >>>>>>>>>>>>>> public interface Listener { > > > >>>>>>>>>>>>>> public void notify(); > > > >>>>>>>>>>>>>> } > > > >>>>>>>>>>>>>> > > > >>>>>>>>>>>>>> public interface SplitReader() { > > > >>>>>>>>>>>>>> /** > > > >>>>>>>>>>>>>> * When there is no element temporarily, this will return > > > >>>> false. > > > >>>>>>>>>>>>>> * When elements is available again splitReader can call > > > >>>>>>>>>>>>>> listener.notify() > > > >>>>>>>>>>>>>> * In addition the frame would check `advance` > > periodically . > > > >>>>>>>>>>>>>> * Of course advance can always return true and ignore > the > > > >>>>>>>>>> listener > > > >>>>>>>>>>>>>> argument for simplicity. > > > >>>>>>>>>>>>>> */ > > > >>>>>>>>>>>>>> public boolean advance(Listener listener); > > > >>>>>>>>>>>>>> } > > > >>>>>>>>>>>>>> > > > >>>>>>>>>>>>>> 2. The FLIP tells us very clearly that how to create > all > > > >> Splits > > > >>>>>> and > > > >>>>>>>>>> how > > > >>>>>>>>>>>>>> to create a SplitReader from a Split. But there is no > > > strategy > > > >>>> for > > > >>>>>>>> the > > > >>>>>>>>>>>> user > > > >>>>>>>>>>>>>> to choose how to assign the splits to the tasks. I think > > we > > > >>>> could > > > >>>>>>>> add > > > >>>>>>>>>> a > > > >>>>>>>>>>>>>> Enum to let user to choose. > > > >>>>>>>>>>>>>> /** > > > >>>>>>>>>>>>>> public Enum SplitsAssignmentPolicy { > > > >>>>>>>>>>>>>> Location, > > > >>>>>>>>>>>>>> Workload, > > > >>>>>>>>>>>>>> Random, > > > >>>>>>>>>>>>>> Average > > > >>>>>>>>>>>>>> } > > > >>>>>>>>>>>>>> */ > > > >>>>>>>>>>>>>> > > > >>>>>>>>>>>>>> 3. If merge the `advance` and `getCurrent` to one > method > > > like > > > >>>>>>>>>> `getNext` > > > >>>>>>>>>>>>>> the `getNext` would need return a `ElementWithTimestamp` > > > >> because > > > >>>>>>>> some > > > >>>>>>>>>>>>>> sources want to add timestamp to every element. IMO, > this > > is > > > >> not > > > >>>>>> so > > > >>>>>>>>>>>> memory > > > >>>>>>>>>>>>>> friendly so I prefer this design. > > > >>>>>>>>>>>>>> > > > >>>>>>>>>>>>>> > > > >>>>>>>>>>>>>> > > > >>>>>>>>>>>>>> Thanks > > > >>>>>>>>>>>>>> > > > >>>>>>>>>>>>>> Piotr Nowojski <pi...@data-artisans.com> 于2018年11月1日周四 > > > >>>> 下午6:08写道： > > > >>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>> Hi, > > > >>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>> Thanks Aljoscha for starting this, it’s blocking quite > a > > > lot > > > >> of > > > >>>>>>>> other > > > >>>>>>>>>>>>>>> possible improvements. I have one proposal. Instead of > > > >> having a > > > >>>>>>>>>> method: > > > >>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>> boolean advance() throws IOException; > > > >>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>> I would replace it with > > > >>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>> /* > > > >>>>>>>>>>>>>>> * Return a future, which when completed means that > source > > > has > > > >>>>>> more > > > >>>>>>>>>>>> data > > > >>>>>>>>>>>>>>> and getNext() will not block. > > > >>>>>>>>>>>>>>> * If you wish to use benefits of non blocking > connectors, > > > >>>> please > > > >>>>>>>>>>>>>>> implement this method appropriately. > > > >>>>>>>>>>>>>>> */ > > > >>>>>>>>>>>>>>> default CompletableFuture<?> isBlocked() { > > > >>>>>>>>>>>>>>> return CompletableFuture.completedFuture(null); > > > >>>>>>>>>>>>>>> } > > > >>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>> And rename `getCurrent()` to `getNext()`. > > > >>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>> Couple of arguments: > > > >>>>>>>>>>>>>>> 1. I don’t understand the division of work between > > > >> `advance()` > > > >>>>>> and > > > >>>>>>>>>>>>>>> `getCurrent()`. What should be done in which, > especially > > > for > > > >>>>>>>>>> connectors > > > >>>>>>>>>>>>>>> that handle records in batches (like Kafka) and when > > should > > > >> you > > > >>>>>>>> call > > > >>>>>>>>>>>>>>> `advance` and when `getCurrent()`. > > > >>>>>>>>>>>>>>> 2. Replacing `boolean` with `CompletableFuture<?>` will > > > allow > > > >>>> us > > > >>>>>> in > > > >>>>>>>>>> the > > > >>>>>>>>>>>>>>> future to have asynchronous/non blocking connectors and > > > more > > > >>>>>>>>>>>> efficiently > > > >>>>>>>>>>>>>>> handle large number of blocked threads, without busy > > > waiting. > > > >>>>>> While > > > >>>>>>>>>> at > > > >>>>>>>>>>>> the > > > >>>>>>>>>>>>>>> same time it doesn’t add much complexity, since naive > > > >> connector > > > >>>>>>>>>>>>>>> implementations can be always blocking. > > > >>>>>>>>>>>>>>> 3. This also would allow us to use a fixed size thread > > pool > > > >> of > > > >>>>>> task > > > >>>>>>>>>>>>>>> executors, instead of one thread per task. > > > >>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>> Piotrek > > > >>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>>> On 31 Oct 2018, at 17:22, Aljoscha Krettek < > > > >>>> aljos...@apache.org > > > >>>>>>> > > > >>>>>>>>>>>>>>> wrote: > > > >>>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>>> Hi All, > > > >>>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>>> In order to finally get the ball rolling on the new > > source > > > >>>>>>>> interface > > > >>>>>>>>>>>>>>> that we have discussed for so long I finally created a > > > FLIP: > > > >>>>>>>>>>>>>>> > > > >>>>>>>>>>>> > > > >>>>>>>>>> > > > >>>>>>>> > > > >>>>>> > > > >>>> > > > >> > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-27%3A+Refactor+Source+Interface > > > >>>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>>> I cc'ed Thomas and Jamie because of the ongoing > > > >>>> work/discussion > > > >>>>>>>>>> about > > > >>>>>>>>>>>>>>> adding per-partition watermark support to the Kinesis > > > source > > > >>>> and > > > >>>>>>>>>>>> because > > > >>>>>>>>>>>>>>> this would enable generic implementation of event-time > > > >>>> alignment > > > >>>>>>>> for > > > >>>>>>>>>>>> all > > > >>>>>>>>>>>>>>> sources. Maybe we need another FLIP for the event-time > > > >>>> alignment > > > >>>>>>>>>> part, > > > >>>>>>>>>>>>>>> especially the part about information sharing between > > > >>>> operations > > > >>>>>>>> (I'm > > > >>>>>>>>>>>> not > > > >>>>>>>>>>>>>>> calling it state sharing because state has a special > > > meaning > > > >> in > > > >>>>>>>>>> Flink). > > > >>>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>>> Please discuss away! > > > >>>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>>> Aljoscha > > > >>>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>> > > > >>>>>>>>>>>>>>> > > > >>>>>>>>>>>> > > > >>>>>>>>>> > > > >>>>>>>>>> > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>> > > > >>>>>> > > > >>>> > > > >>>> > > > >> > > > >> > > > > > > > > > > > >
