Workers restarting is not a bug, it's standard often expected.
On Mon, Feb 19, 2018, 12:03 PM Romain Manni-Bucau <rmannibu...@gmail.com>
wrote:
> Nothing, as mentionned it is a bug so recovery is a bug recovery
> (procedure)
>
> Le 19 févr. 2018 19:42, "Eugene Kirpichov" <kirpic...@google.com> a
> écrit :
>
>> So what would you like to happen if there is a crash? The DoFn instance
>> no longer exists because the JVM it ran on no longer exists. What should
>> Teardown be called on?
>>
>> On Mon, Feb 19, 2018, 10:20 AM Romain Manni-Bucau <rmannibu...@gmail.com>
>> wrote:
>>
>>> This is what i want and not 999999 teardowns for 1000000 setups until
>>> there is an unexpected crash (= a bug).
>>>
>>> Le 19 févr. 2018 18:57, "Reuven Lax" <re...@google.com> a écrit :
>>>
>>>>
>>>>
>>>> On Mon, Feb 19, 2018 at 7:11 AM, Romain Manni-Bucau <
>>>> rmannibu...@gmail.com> wrote:
>>>>
>>>>>
>>>>>
>>>>> 2018-02-19 15:57 GMT+01:00 Reuven Lax <re...@google.com>:
>>>>>
>>>>>>
>>>>>>
>>>>>> On Mon, Feb 19, 2018 at 12:35 AM, Romain Manni-Bucau <
>>>>>> rmannibu...@gmail.com> wrote:
>>>>>>
>>>>>>> @Reuven: in practise it is created by pool of 256 but leads to the
>>>>>>> same pattern, the teardown is just a "if (iCreatedThem) releaseThem();"
>>>>>>>
>>>>>>
>>>>>> How do you control "256?" Even if you have a pool of 256 workers,
>>>>>> nothing in Beam guarantees how many threads and DoFns are created per
>>>>>> worker. In theory the runner might decide to create 1000 threads on each
>>>>>> worker.
>>>>>>
>>>>>
>>>>> Nop was the other way around, in this case on AWS you can get 256
>>>>> instances at once but not 512 (which will be 2x256). So when you compute
>>>>> the distribution you allocate to some fn the role to own the instance
>>>>> lookup and releasing.
>>>>>
>>>>
>>>> I still don't understand. Let's be more precise. If you write the
>>>> following code:
>>>>
>>>> pCollection.apply(ParDo.of(new MyDoFn()));
>>>>
>>>> There is no way to control how many instances of MyDoFn are created.
>>>> The runner might decided to create a million instances of this class across
>>>> your worker pool, which means that you will get a million Setup and
>>>> Teardown calls.
>>>>
>>>>
>>>>> Anyway this was just an example of an external resource you must
>>>>> release. Real topic is that beam should define asap a guaranteed generic
>>>>> lifecycle to let user embrace its programming model.
>>>>>
>>>>>
>>>>>>
>>>>>>
>>>>>>
>>>>>>> @Eugene:
>>>>>>> 1. wait logic is about passing the value which is not always
>>>>>>> possible (like 15% of cases from my raw estimate)
>>>>>>> 2. sdf: i'll try to detail why i mention SDF more here
>>>>>>>
>>>>>>>
>>>>>>> Concretely beam exposes a portable API (included in the SDK core).
>>>>>>> This API defines a *container* API and therefore implies bean
>>>>>>> lifecycles.
>>>>>>> I'll not detail them all but just use the sources and dofn (not sdf) to
>>>>>>> illustrate the idea I'm trying to develop.
>>>>>>>
>>>>>>> A. Source
>>>>>>>
>>>>>>> A source computes a partition plan with 2 primitives: estimateSize
>>>>>>> and split. As an user you can expect both to be called on the same bean
>>>>>>> instance to avoid to pay the same connection cost(s) twice. Concretely:
>>>>>>>
>>>>>>> connect()
>>>>>>> try {
>>>>>>> estimateSize()
>>>>>>> split()
>>>>>>> } finally {
>>>>>>> disconnect()
>>>>>>> }
>>>>>>>
>>>>>>> this is not guaranteed by the API so you must do:
>>>>>>>
>>>>>>> connect()
>>>>>>> try {
>>>>>>> estimateSize()
>>>>>>> } finally {
>>>>>>> disconnect()
>>>>>>> }
>>>>>>> connect()
>>>>>>> try {
>>>>>>> split()
>>>>>>> } finally {
>>>>>>> disconnect()
>>>>>>> }
>>>>>>>
>>>>>>> + a workaround with an internal estimate size since this primitive
>>>>>>> is often called in split but you dont want to connect twice in the
>>>>>>> second
>>>>>>> phase.
>>>>>>>
>>>>>>> Why do you need that? Simply cause you want to define an API to
>>>>>>> implement sources which initializes the source bean and destroys it.
>>>>>>> I insists it is a very very basic concern for such API. However beam
>>>>>>> doesn't embraces it and doesn't assume it so building any API on top of
>>>>>>> beam is very hurtful today and for direct beam users you hit the exact
>>>>>>> same
>>>>>>> issues - check how IO are implemented, the static utilities which create
>>>>>>> volatile connections preventing to reuse existing connection in a single
>>>>>>> method (
>>>>>>> https://github.com/apache/beam/blob/master/sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java#L862
>>>>>>> ).
>>>>>>>
>>>>>>> Same logic applies to the reader which is then created.
>>>>>>>
>>>>>>> B. DoFn & SDF
>>>>>>>
>>>>>>> As a fn dev you expect the same from the beam runtime: init(); try {
>>>>>>> while (...) process(); } finally { destroy(); } and that it is executed
>>>>>>> on
>>>>>>> the exact same instance to be able to be stateful at that level for
>>>>>>> expensive connections/operations/flow state handling.
>>>>>>>
>>>>>>> As you mentionned with the million example, this sequence should
>>>>>>> happen for each single instance so 1M times for your example.
>>>>>>>
>>>>>>> Now why did I mention SDF several times? Because SDF is a
>>>>>>> generalisation of both cases (source and dofn). Therefore it creates way
>>>>>>> more instances and requires to have a way more strict/explicit
>>>>>>> definition
>>>>>>> of the exact lifecycle and which instance does what. Since beam handles
>>>>>>> the
>>>>>>> full lifecycle of the bean instances it must provide init/destroy hooks
>>>>>>> (setup/teardown) which can be stateful.
>>>>>>>
>>>>>>> If you take the JDBC example which was mentionned earlier. Today,
>>>>>>> because of the teardown issue it uses bundles. Since bundles size is not
>>>>>>> defined - and will not with SDF, it must use a pool to be able to reuse
>>>>>>> a
>>>>>>> connection instance to not correct performances. Now with the SDF and
>>>>>>> the
>>>>>>> split increase, how do you handle the pool size? Generally in batch you
>>>>>>> use
>>>>>>> a single connection per thread to avoid to consume all database
>>>>>>> connections. With a pool you have 2 choices: 1. use a pool of 1, 2. use
>>>>>>> a
>>>>>>> pool a bit higher but multiplied by the number of beans you will likely
>>>>>>> x2
>>>>>>> or 3 the connection count and make the execution fail with "no more
>>>>>>> connection available". I you picked 1 (pool of #1), then you still have
>>>>>>> to
>>>>>>> have a reliable teardown by pool instance (close() generally) to ensure
>>>>>>> you
>>>>>>> release the pool and don't leak the connection information in the JVM.
>>>>>>> In
>>>>>>> all case you come back to the init()/destroy() lifecycle even if you
>>>>>>> fake
>>>>>>> to get connections with bundles.
>>>>>>>
>>>>>>> Just to make it obvious: SDF mentions are just cause SDF imply all
>>>>>>> the current issues with the loose definition of the bean lifecycles at
>>>>>>> an
>>>>>>> exponential level, nothing else.
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>> Romain Manni-Bucau
>>>>>>> @rmannibucau <https://twitter.com/rmannibucau> | Blog
>>>>>>> <https://rmannibucau.metawerx.net/> | Old Blog
>>>>>>> <http://rmannibucau.wordpress.com> | Github
>>>>>>> <https://github.com/rmannibucau> | LinkedIn
>>>>>>> <https://www.linkedin.com/in/rmannibucau> | Book
>>>>>>> <https://www.packtpub.com/application-development/java-ee-8-high-performance>
>>>>>>>
>>>>>>> 2018-02-18 22:32 GMT+01:00 Eugene Kirpichov <kirpic...@google.com>:
>>>>>>>
>>>>>>>> The kind of whole-transform lifecycle you're mentioning can be
>>>>>>>> accomplished using the Wait transform as I suggested in the thread
>>>>>>>> above,
>>>>>>>> and I believe it should become the canonical way to do that.
>>>>>>>>
>>>>>>>> (Would like to reiterate one more time, as the main author of most
>>>>>>>> design documents related to SDF and of its implementation in the Java
>>>>>>>> direct and dataflow runner that SDF is fully unrelated to the topic of
>>>>>>>> cleanup - I'm very confused as to why it keeps coming up)
>>>>>>>>
>>>>>>>> On Sun, Feb 18, 2018, 1:15 PM Romain Manni-Bucau <
>>>>>>>> rmannibu...@gmail.com> wrote:
>>>>>>>>
>>>>>>>>> I kind of agree except transforms lack a lifecycle too. My
>>>>>>>>> understanding is that sdf could be a way to unify it and clean the
>>>>>>>>> api.
>>>>>>>>>
>>>>>>>>> Otherwise how to normalize - single api - lifecycle of transforms?
>>>>>>>>>
>>>>>>>>> Le 18 févr. 2018 21:32, "Ben Chambers" <bchamb...@apache.org> a
>>>>>>>>> écrit :
>>>>>>>>>
>>>>>>>>>> Are you sure that focusing on the cleanup of specific DoFn's is
>>>>>>>>>> appropriate? Many cases where cleanup is necessary, it is around an
>>>>>>>>>> entire
>>>>>>>>>> composite PTransform. I think there have been discussions/proposals
>>>>>>>>>> around
>>>>>>>>>> a more methodical "cleanup" option, but those haven't been
>>>>>>>>>> implemented, to
>>>>>>>>>> the best of my knowledge.
>>>>>>>>>>
>>>>>>>>>> For instance, consider the steps of a FileIO:
>>>>>>>>>> 1. Write to a bunch (N shards) of temporary files
>>>>>>>>>> 2. When all temporary files are complete, attempt to do a bulk
>>>>>>>>>> copy to put them in the final destination.
>>>>>>>>>> 3. Cleanup all the temporary files.
>>>>>>>>>>
>>>>>>>>>> (This is often desirable because it minimizes the chance of
>>>>>>>>>> seeing partial/incomplete results in the final destination).
>>>>>>>>>>
>>>>>>>>>> In the above, you'd want step 1 to execute on many workers,
>>>>>>>>>> likely using a ParDo (say N different workers).
>>>>>>>>>> The move step should only happen once, so on one worker. This
>>>>>>>>>> means it will be a different DoFn, likely with some stuff done to
>>>>>>>>>> ensure it
>>>>>>>>>> runs on one worker.
>>>>>>>>>>
>>>>>>>>>> In such a case, cleanup / @TearDown of the DoFn is not enough. We
>>>>>>>>>> need an API for a PTransform to schedule some cleanup work for when
>>>>>>>>>> the
>>>>>>>>>> transform is "done". In batch this is relatively straightforward, but
>>>>>>>>>> doesn't exist. This is the source of some problems, such as BigQuery
>>>>>>>>>> sink
>>>>>>>>>> leaving files around that have failed to import into BigQuery.
>>>>>>>>>>
>>>>>>>>>> In streaming this is less straightforward -- do you want to wait
>>>>>>>>>> until the end of the pipeline? Or do you want to wait until the end
>>>>>>>>>> of the
>>>>>>>>>> window? In practice, you just want to wait until you know nobody
>>>>>>>>>> will need
>>>>>>>>>> the resource anymore.
>>>>>>>>>>
>>>>>>>>>> This led to some discussions around a "cleanup" API, where you
>>>>>>>>>> could have a transform that output resource objects. Each resource
>>>>>>>>>> object
>>>>>>>>>> would have logic for cleaning it up. And there would be something
>>>>>>>>>> that
>>>>>>>>>> indicated what parts of the pipeline needed that resource, and what
>>>>>>>>>> kind of
>>>>>>>>>> temporal lifetime those objects had. As soon as that part of the
>>>>>>>>>> pipeline
>>>>>>>>>> had advanced far enough that it would no longer need the resources,
>>>>>>>>>> they
>>>>>>>>>> would get cleaned up. This can be done at pipeline shutdown, or
>>>>>>>>>> incrementally during a streaming pipeline, etc.
>>>>>>>>>>
>>>>>>>>>> Would something like this be a better fit for your use case? If
>>>>>>>>>> not, why is handling teardown within a single DoFn sufficient?
>>>>>>>>>>
>>>>>>>>>> On Sun, Feb 18, 2018 at 11:53 AM Romain Manni-Bucau <
>>>>>>>>>> rmannibu...@gmail.com> wrote:
>>>>>>>>>>
>>>>>>>>>>> Yes 1M. Lets try to explain you simplifying the overall
>>>>>>>>>>> execution. Each instance - one fn so likely in a thread of a worker
>>>>>>>>>>> - has
>>>>>>>>>>> its lifecycle. Caricaturally: "new" and garbage collection.
>>>>>>>>>>>
>>>>>>>>>>> In practise, new is often an unsafe allocate (deserialization)
>>>>>>>>>>> but it doesnt matter here.
>>>>>>>>>>>
>>>>>>>>>>> What i want is any "new" to have a following setup before any
>>>>>>>>>>> process or stattbundle and the last time beam has the instance
>>>>>>>>>>> before it is
>>>>>>>>>>> gc-ed and after last finishbundle it calls teardown.
>>>>>>>>>>>
>>>>>>>>>>> It is as simple as it.
>>>>>>>>>>> This way no need to comibe fn in a way making a fn not self
>>>>>>>>>>> contained to implement basic transforms.
>>>>>>>>>>>
>>>>>>>>>>> Le 18 févr. 2018 20:07, "Reuven Lax" <re...@google.com> a
>>>>>>>>>>> écrit :
>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>> On Sun, Feb 18, 2018 at 10:50 AM, Romain Manni-Bucau <
>>>>>>>>>>>> rmannibu...@gmail.com> wrote:
>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>> Le 18 févr. 2018 19:28, "Ben Chambers" <bchamb...@apache.org>
>>>>>>>>>>>>> a écrit :
>>>>>>>>>>>>>
>>>>>>>>>>>>> It feels like his thread may be a bit off-track. Rather than
>>>>>>>>>>>>> focusing on the semantics of the existing methods -- which have
>>>>>>>>>>>>> been noted
>>>>>>>>>>>>> to be meet many existing use cases -- it would be helpful to
>>>>>>>>>>>>> focus on more
>>>>>>>>>>>>> on the reason you are looking for something with different
>>>>>>>>>>>>> semantics.
>>>>>>>>>>>>>
>>>>>>>>>>>>> Some possibilities (I'm not sure which one you are trying to
>>>>>>>>>>>>> do):
>>>>>>>>>>>>>
>>>>>>>>>>>>> 1. Clean-up some external, global resource, that was
>>>>>>>>>>>>> initialized once during the startup of the pipeline. If this is
>>>>>>>>>>>>> the case,
>>>>>>>>>>>>> how are you ensuring it was really only initialized once (and not
>>>>>>>>>>>>> once per
>>>>>>>>>>>>> worker, per thread, per instance, etc.)? How do you know when the
>>>>>>>>>>>>> pipeline
>>>>>>>>>>>>> should release it? If the answer is "when it reaches step X",
>>>>>>>>>>>>> then what
>>>>>>>>>>>>> about a streaming pipeline?
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>> When the dofn is no more needed logically ie when the batch is
>>>>>>>>>>>>> done or stream is stopped (manually or by a jvm shutdown)
>>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>> I'm really not following what this means.
>>>>>>>>>>>>
>>>>>>>>>>>> Let's say that a pipeline is running 1000 workers, and each
>>>>>>>>>>>> worker is running 1000 threads (each running a copy of the same
>>>>>>>>>>>> DoFn). How
>>>>>>>>>>>> many cleanups do you want (do you want 1000 * 1000 = 1M cleanups)
>>>>>>>>>>>> and when
>>>>>>>>>>>> do you want it called? When the entire pipeline is shut down? When
>>>>>>>>>>>> an
>>>>>>>>>>>> individual worker is about to shut down (which may be temporary -
>>>>>>>>>>>> may be
>>>>>>>>>>>> about to start back up)? Something else?
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>> 2. Finalize some resources that are used within some region of
>>>>>>>>>>>>> the pipeline. While, the DoFn lifecycle methods are not a good
>>>>>>>>>>>>> fit for this
>>>>>>>>>>>>> (they are focused on managing resources within the DoFn), you
>>>>>>>>>>>>> could model
>>>>>>>>>>>>> this on how FileIO finalizes the files that it produced. For
>>>>>>>>>>>>> instance:
>>>>>>>>>>>>> a) ParDo generates "resource IDs" (or some token that
>>>>>>>>>>>>> stores information about resources)
>>>>>>>>>>>>> b) "Require Deterministic Input" (to prevent retries from
>>>>>>>>>>>>> changing resource IDs)
>>>>>>>>>>>>> c) ParDo that initializes the resources
>>>>>>>>>>>>> d) Pipeline segments that use the resources, and eventually
>>>>>>>>>>>>> output the fact they're done
>>>>>>>>>>>>> e) "Require Deterministic Input"
>>>>>>>>>>>>> f) ParDo that frees the resources
>>>>>>>>>>>>>
>>>>>>>>>>>>> By making the use of the resource part of the data it is
>>>>>>>>>>>>> possible to "checkpoint" which resources may be in use or have
>>>>>>>>>>>>> been
>>>>>>>>>>>>> finished by using the require deterministic input. This is
>>>>>>>>>>>>> important to
>>>>>>>>>>>>> ensuring everything is actually cleaned up.
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>> I nees that but generic and not case by case to industrialize
>>>>>>>>>>>>> some api on top of beam.
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>> 3. Some other use case that I may be missing? If it is this
>>>>>>>>>>>>> case, could you elaborate on what you are trying to accomplish?
>>>>>>>>>>>>> That would
>>>>>>>>>>>>> help me understand both the problems with existing options and
>>>>>>>>>>>>> possibly
>>>>>>>>>>>>> what could be done to help.
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>> I understand there are sorkaround for almost all cases but
>>>>>>>>>>>>> means each transform is different in its lifecycle handling
>>>>>>>>>>>>> except i
>>>>>>>>>>>>> dislike it a lot at a scale and as a user since you cant put any
>>>>>>>>>>>>> unified
>>>>>>>>>>>>> practise on top of beam, it also makes beam very hard to
>>>>>>>>>>>>> integrate or to
>>>>>>>>>>>>> use to build higher level libraries or softwares.
>>>>>>>>>>>>>
>>>>>>>>>>>>> This is why i tried to not start the workaround discussions
>>>>>>>>>>>>> and just stay at API level.
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>> -- Ben
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>> On Sun, Feb 18, 2018 at 9:56 AM Romain Manni-Bucau <
>>>>>>>>>>>>> rmannibu...@gmail.com> wrote:
>>>>>>>>>>>>>
>>>>>>>>>>>>>> 2018-02-18 18:36 GMT+01:00 Eugene Kirpichov <
>>>>>>>>>>>>>> kirpic...@google.com>:
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> "Machine state" is overly low-level because many of the
>>>>>>>>>>>>>>> possible reasons can happen on a perfectly fine machine.
>>>>>>>>>>>>>>> If you'd like to rephrase it to "it will be called except in
>>>>>>>>>>>>>>> various situations where it's logically impossible or
>>>>>>>>>>>>>>> impractical to
>>>>>>>>>>>>>>> guarantee that it's called", that's fine. Or you can list some
>>>>>>>>>>>>>>> of the
>>>>>>>>>>>>>>> examples above.
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> Sounds ok to me
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> The main point for the user is, you *will* see
>>>>>>>>>>>>>>> non-preventable situations where it couldn't be called - it's
>>>>>>>>>>>>>>> not just
>>>>>>>>>>>>>>> intergalactic crashes - so if the logic is very important (e.g.
>>>>>>>>>>>>>>> cleaning up
>>>>>>>>>>>>>>> a large amount of temporary files, shutting down a large number
>>>>>>>>>>>>>>> of VMs you
>>>>>>>>>>>>>>> started etc), you have to express it using one of the other
>>>>>>>>>>>>>>> methods that
>>>>>>>>>>>>>>> have stricter guarantees (which obviously come at a cost, e.g.
>>>>>>>>>>>>>>> no
>>>>>>>>>>>>>>> pass-by-reference).
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> FinishBundle has the exact same guarantee sadly so not which
>>>>>>>>>>>>>> which other method you speak about. Concretely if you make it
>>>>>>>>>>>>>> really
>>>>>>>>>>>>>> unreliable - this is what best effort sounds to me - then users
>>>>>>>>>>>>>> can use it
>>>>>>>>>>>>>> to clean anything but if you make it "can happen but it is
>>>>>>>>>>>>>> unexpected and
>>>>>>>>>>>>>> means something happent" then it is fine to have a manual - or
>>>>>>>>>>>>>> auto if
>>>>>>>>>>>>>> fancy - recovery procedure. This is where it makes all the
>>>>>>>>>>>>>> difference and
>>>>>>>>>>>>>> impacts the developpers, ops (all users basically).
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> On Sun, Feb 18, 2018 at 9:16 AM Romain Manni-Bucau <
>>>>>>>>>>>>>>> rmannibu...@gmail.com> wrote:
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Agree Eugene except that "best effort" means that. It is
>>>>>>>>>>>>>>>> also often used to say "at will" and this is what triggered
>>>>>>>>>>>>>>>> this thread.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> I'm fine using "except if the machine state prevents it"
>>>>>>>>>>>>>>>> but "best effort" is too open and can be very badly and
>>>>>>>>>>>>>>>> wrongly perceived
>>>>>>>>>>>>>>>> by users (like I did).
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Romain Manni-Bucau
>>>>>>>>>>>>>>>> @rmannibucau <https://twitter.com/rmannibucau> | Blog
>>>>>>>>>>>>>>>> <https://rmannibucau.metawerx.net/> | Old Blog
>>>>>>>>>>>>>>>> <http://rmannibucau.wordpress.com> | Github
>>>>>>>>>>>>>>>> <https://github.com/rmannibucau> | LinkedIn
>>>>>>>>>>>>>>>> <https://www.linkedin.com/in/rmannibucau> | Book
>>>>>>>>>>>>>>>> <https://www.packtpub.com/application-development/java-ee-8-high-performance>
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> 2018-02-18 18:13 GMT+01:00 Eugene Kirpichov <
>>>>>>>>>>>>>>>> kirpic...@google.com>:
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> It will not be called if it's impossible to call it: in
>>>>>>>>>>>>>>>>> the example situation you have (intergalactic crash), and in
>>>>>>>>>>>>>>>>> a number of
>>>>>>>>>>>>>>>>> more common cases: eg in case the worker container has
>>>>>>>>>>>>>>>>> crashed (eg user
>>>>>>>>>>>>>>>>> code in a different thread called a C library over JNI and it
>>>>>>>>>>>>>>>>> segfaulted),
>>>>>>>>>>>>>>>>> JVM bug, crash due to user code OOM, in case the worker has
>>>>>>>>>>>>>>>>> lost network
>>>>>>>>>>>>>>>>> connectivity (then it may be called but it won't be able to
>>>>>>>>>>>>>>>>> do anything
>>>>>>>>>>>>>>>>> useful), in case this is running on a preemptible VM and it
>>>>>>>>>>>>>>>>> was preempted
>>>>>>>>>>>>>>>>> by the underlying cluster manager without notice or if the
>>>>>>>>>>>>>>>>> worker was too
>>>>>>>>>>>>>>>>> busy with other stuff (eg calling other Teardown functions)
>>>>>>>>>>>>>>>>> until the
>>>>>>>>>>>>>>>>> preemption timeout elapsed, in case the underlying hardware
>>>>>>>>>>>>>>>>> simply failed
>>>>>>>>>>>>>>>>> (which happens quite often at scale), and in many other
>>>>>>>>>>>>>>>>> conditions.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> "Best effort" is the commonly used term to describe such
>>>>>>>>>>>>>>>>> behavior. Please feel free to file bugs for cases where you
>>>>>>>>>>>>>>>>> observed a
>>>>>>>>>>>>>>>>> runner not call Teardown in a situation where it was possible
>>>>>>>>>>>>>>>>> to call it
>>>>>>>>>>>>>>>>> but the runner made insufficient effort.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> On Sun, Feb 18, 2018, 9:02 AM Romain Manni-Bucau <
>>>>>>>>>>>>>>>>> rmannibu...@gmail.com> wrote:
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> 2018-02-18 18:00 GMT+01:00 Eugene Kirpichov <
>>>>>>>>>>>>>>>>>> kirpic...@google.com>:
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> On Sun, Feb 18, 2018, 2:06 AM Romain Manni-Bucau <
>>>>>>>>>>>>>>>>>>> rmannibu...@gmail.com> wrote:
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>> Le 18 févr. 2018 00:23, "Kenneth Knowles" <
>>>>>>>>>>>>>>>>>>>> k...@google.com> a écrit :
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>> On Sat, Feb 17, 2018 at 3:09 PM, Romain Manni-Bucau <
>>>>>>>>>>>>>>>>>>>> rmannibu...@gmail.com> wrote:
>>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>>> If you give an example of a high-level need (e.g. "I'm
>>>>>>>>>>>>>>>>>>>>> trying to write an IO for system $x and it requires the
>>>>>>>>>>>>>>>>>>>>> following
>>>>>>>>>>>>>>>>>>>>> initialization and the following cleanup logic and the
>>>>>>>>>>>>>>>>>>>>> following processing
>>>>>>>>>>>>>>>>>>>>> in between") I'll be better able to help you.
>>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>>> Take a simple example of a transform requiring a
>>>>>>>>>>>>>>>>>>>>> connection. Using bundles is a perf killer since size is
>>>>>>>>>>>>>>>>>>>>> not controlled.
>>>>>>>>>>>>>>>>>>>>> Using teardown doesnt allow you to release the connection
>>>>>>>>>>>>>>>>>>>>> since it is a
>>>>>>>>>>>>>>>>>>>>> best effort thing. Not releasing the connection makes you
>>>>>>>>>>>>>>>>>>>>> pay a lot - aws
>>>>>>>>>>>>>>>>>>>>> ;) - or prevents you to launch other processings -
>>>>>>>>>>>>>>>>>>>>> concurrent limit.
>>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>> For this example @Teardown is an exact fit. If things
>>>>>>>>>>>>>>>>>>>> die so badly that @Teardown is not called then nothing
>>>>>>>>>>>>>>>>>>>> else can be called
>>>>>>>>>>>>>>>>>>>> to close the connection either. What AWS service are you
>>>>>>>>>>>>>>>>>>>> thinking of that
>>>>>>>>>>>>>>>>>>>> stays open for a long time when everything at the other
>>>>>>>>>>>>>>>>>>>> end has died?
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>> You assume connections are kind of stateless but some
>>>>>>>>>>>>>>>>>>>> (proprietary) protocols requires some closing exchanges
>>>>>>>>>>>>>>>>>>>> which are not only
>>>>>>>>>>>>>>>>>>>> "im leaving".
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>> For aws i was thinking about starting some services -
>>>>>>>>>>>>>>>>>>>> machines - on the fly in a pipeline startup and closing
>>>>>>>>>>>>>>>>>>>> them at the end. If
>>>>>>>>>>>>>>>>>>>> teardown is not called you leak machines and money. You
>>>>>>>>>>>>>>>>>>>> can say it can be
>>>>>>>>>>>>>>>>>>>> done another way...as the full pipeline ;).
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>> I dont want to be picky but if beam cant handle its
>>>>>>>>>>>>>>>>>>>> components lifecycle it can be used at scale for generic
>>>>>>>>>>>>>>>>>>>> pipelines and if
>>>>>>>>>>>>>>>>>>>> bound to some particular IO.
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>> What does prevent to enforce teardown - ignoring the
>>>>>>>>>>>>>>>>>>>> interstellar crash case which cant be handled by any human
>>>>>>>>>>>>>>>>>>>> system? Nothing
>>>>>>>>>>>>>>>>>>>> technically. Why do you push to not handle it? Is it due
>>>>>>>>>>>>>>>>>>>> to some legacy
>>>>>>>>>>>>>>>>>>>> code on dataflow or something else?
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> Teardown *is* already documented and implemented this
>>>>>>>>>>>>>>>>>>> way (best-effort). So I'm not sure what kind of change
>>>>>>>>>>>>>>>>>>> you're asking for.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> Remove "best effort" from the javadoc. If it is not call
>>>>>>>>>>>>>>>>>> then it is a bug and we are done :).
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>> Also what does it mean for the users? Direct runner
>>>>>>>>>>>>>>>>>>>> does it so if a user udes the RI in test, he will get a
>>>>>>>>>>>>>>>>>>>> different behavior
>>>>>>>>>>>>>>>>>>>> in prod? Also dont forget the user doesnt know what the
>>>>>>>>>>>>>>>>>>>> IOs he composes use
>>>>>>>>>>>>>>>>>>>> so this is so impacting for the whole product than he must
>>>>>>>>>>>>>>>>>>>> be handled IMHO.
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>> I understand the portability culture is new in big data
>>>>>>>>>>>>>>>>>>>> world but it is not a reason to ignore what people did for
>>>>>>>>>>>>>>>>>>>> years and do it
>>>>>>>>>>>>>>>>>>>> wrong before doing right ;).
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>> My proposal is to list what can prevent to guarantee -
>>>>>>>>>>>>>>>>>>>> in the normal IT conditions - the execution of teardown.
>>>>>>>>>>>>>>>>>>>> Then we see if we
>>>>>>>>>>>>>>>>>>>> can handle it and only if there is a technical reason we
>>>>>>>>>>>>>>>>>>>> cant we make it
>>>>>>>>>>>>>>>>>>>> experimental/unsupported in the api. I know spark and
>>>>>>>>>>>>>>>>>>>> flink can, any
>>>>>>>>>>>>>>>>>>>> unknown blocker for other runners?
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>> Technical note: even a kill should go through java
>>>>>>>>>>>>>>>>>>>> shutdown hooks otherwise your environment (beam enclosing
>>>>>>>>>>>>>>>>>>>> software) is
>>>>>>>>>>>>>>>>>>>> fully unhandled and your overall system is uncontrolled.
>>>>>>>>>>>>>>>>>>>> Only case where it
>>>>>>>>>>>>>>>>>>>> is not true is when the software is always owned by a
>>>>>>>>>>>>>>>>>>>> vendor and never
>>>>>>>>>>>>>>>>>>>> installed on customer environment. In this case it belongd
>>>>>>>>>>>>>>>>>>>> to the vendor to
>>>>>>>>>>>>>>>>>>>> handle beam API and not to beam to adjust its API for a
>>>>>>>>>>>>>>>>>>>> vendor - otherwise
>>>>>>>>>>>>>>>>>>>> all unsupported features by one runner should be made
>>>>>>>>>>>>>>>>>>>> optional right?
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>> All state is not about network, even in distributed
>>>>>>>>>>>>>>>>>>>> systems so this is key to have an explicit and defined
>>>>>>>>>>>>>>>>>>>> lifecycle.
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>> Kenn
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>
>>>>>>
>>>>>
>>>>
