flink-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Gyula Fóra <gyula.f...@gmail.com>
Subject Re: Change Streaming Source Function Interface
Date Thu, 30 Apr 2015 20:22:10 GMT
Okay, sounds very reasonable :)

On Thu, Apr 30, 2015 at 10:15 PM, Stephan Ewen <sewen@apache.org> wrote:

> For the variant with the "run()" method, this requires strong assumptions
> about the internal behavior of the source.
>
> Unless I am overlooking something, the source needs to guarantee this:
>
>  - It needs to lock internally and perform the state update and record emit
> call inside the locked scope
>
>  - It needs to use the same state object all the time, otherwise the driver
> and the source may lock different objects
>
>  - The second point makes it very hard to support sources that return
> copies (or shadow copies) of the state, to support asynchronous
> snapshotting.
>
>  - A per-element lock is an overhead that we could avoid with the "next()"
> approach.
>
>
> On Thu, Apr 30, 2015 at 10:04 PM, Gyula Fóra <gyula.fora@gmail.com> wrote:
>
> > Hi,
> >
> > The only thing we need is to guarantee that the source will not output
> any
> > records or update the state while we take the snapshot and send the
> > barrier. There are multiple ways of doing this I guess. We could simply
> > lock on these objects for instance or add the methods you wrote. If we
> > lock, we can assure that no user thread will find a way around the next()
> > and hasNext() (which would otherwise cause problems), and we can also
> keep
> > the current interface.
> >
> > I think we just need to figure out what is the preferable user interface
> > for sources, having a simple run and cancel methods or going with the
> > next(), hasNext etc. Or we could just support both.
> >
> > Gyula
> >
> > On Thu, Apr 30, 2015 at 8:30 PM, Stephan Ewen <sewen@apache.org> wrote:
> >
> > > Hi all!
> > >
> > > I think we need to change the interface of the streaming source
> function.
> > >
> > > The function currently has simply a run() method where it does its
> work,
> > > until canceled.
> > >
> > > With this, it is hard to write sources, where the state and the
> snapshot
> > > barriers are exactly aligned.
> > > When performing the checkpoint, the vertex will grab the state from the
> > > source and inject a checkpoint barrier. It is not clear that the
> injected
> > > barrier aligns with the state, because the source may have emitted more
> > > records since grabbing the state, or not emitted the record that is
> > > reflected in the state (offset).
> > >
> > > If we change the interface to a more iterator-like interface (hasNext()
> > and
> > > next()), then the vertex calls these methods and can checkpoint
> > in-between
> > > calling the methods.
> > > After hasNext() is a well defined point, where the state can be grabbed
> > and
> > > the barrier be emitted.
> > >
> > >
> > > Any opinions on that?
> > >
> > >
> > > Stephan
> > >
> >
>

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message