Re: DirectRunner in test - await completion of workers threads?

[Ismaël Mejía]

It seems there is still an issue with teardown not being called in
failed tasks, just created BEAM-4040 to track it.

On Thu, Apr 5, 2018 at 4:45 PM, Tim Robertson  wrote:
> Will do - I'll report the result on https://github.com/apache/beam/pull/4905
>
> On Thu, Apr 5, 2018 at 11:45 AM, Ismaël Mejía  wrote:
>>
>> For info, Romain's PR was merged today, can you confirm if this fixes
>> the issue Tim.
>>
>> On Sun, Apr 1, 2018 at 9:21 PM, Tim Robertson 
>> wrote:
>> > Thanks all.
>> >
>> > I went with what I outlined above, which you can see in this test.
>> >
>> > https://github.com/timrobertson100/beam/blob/BEAM-3848/sdks/java/io/solr/src/test/java/org/apache/beam/sdk/io/solr/SolrIOTest.java#L285
>> >
>> > That forms part of this PR https://github.com/apache/beam/pull/4956
>> >
>> > I'll monitor Romain's PR and back it out when appropriate.
>> >
>> >
>> >
>> >
>> >
>> > On Sun, Apr 1, 2018 at 8:20 PM, Jean-Baptiste Onofré 
>> > wrote:
>> >>
>> >> Indeed. It's exactly what Romain's PR is about.
>> >>
>> >> Regards
>> >> JB
>> >> Le 1 avr. 2018, à 19:33, Reuven Lax  a écrit:
>> >>>
>> >>> Correct - teardown is currently run in the direct runner, but
>> >>> asynchronously. I believe Romain's pending PRs should solve this for
>> >>> your
>> >>> use case.
>> >>>
>> >>> On Sun, Apr 1, 2018 at 3:13 AM Tim Robertson <
>> >>> timrobertson...@gmail.com>
>> >>> wrote:
>> 
>>  Thanks for confirming Romain - also for the very fast reply!
>> 
>>  I'll continue with the workaround and reference BEAM-3409 inline as
>>  justification.
>>  I'm trying to wrap this up before travel next week, but if I get a
>>  chance I'll try and run this scenario (BEAM-3848) with your patch.
>> 
>> 
>> 
>>  On Sun, Apr 1, 2018 at 12:05 PM, Romain Manni-Bucau
>>   wrote:
>> >
>> > Hi
>> >
>> > I have the same blocker and created
>> >
>> > https://github.com/apache/beam/pull/4790 and
>> > https://github.com/apache/beam/pull/4965 to solve part of it
>> >
>> >
>> >
>> > Le 1 avr. 2018 11:35, "Tim Robertson" < timrobertson...@gmail.com> a
>> > écrit :
>> >
>> > Hi devs
>> >
>> > I'm working on SolrIO tests for failure scenarios (i.e. an exception
>> > will come out of the pipeline execution).  I see that the exception
>> > is
>> > surfaced to the driver while " direct-runner-worker" threads are
>> > still
>> > running.  This causes issue because:
>> >
>> >   1. The Solr tests do thread leak detection, and a
>> > solrClient.close()
>> > is what removes the object
>> >   2. @Teardown is not necessarily called which is what would close
>> > the
>> > solrClient
>> >
>> > I can unregister all the solrClients that have been spawned.
>> > However I
>> > have seen race conditions where there are still threads running
>> > creating and
>> > registering clients. I need to someone ensure that all workers
>> > related to
>> > the pipeline execution are indeed finished so no new ones are
>> > created after
>> > the first exception is passed up.
>> >
>> > Currently I have this (psuedo code) which works, but I suspect
>> > someone
>> > can suggest a better approach:
>> >
>> > // store the state of clients registered for object leak check
>> > Set existingClients = registeredSolrClients();
>> > try {
>> >   pipeline.run();
>> >
>> > } catch (Pipeline.PipelineExecutionException e) {
>> >
>> >
>> >   // Hack: await all bundle workers completing
>> >   while (namedThreadStillExists("direct-runner-worker")) {
>> > Thread.sleep(100);
>> >   }
>> >
>> >   // remove all solrClients created in this execution only
>> >   // since the teardown may not have done so
>> >   for (Object o : ObjectReleaseTracker.OBJECTS.keySet()) {
>> > if (o instanceof SolrClient && !existingClients.contains(o)) {
>> >   ObjectReleaseTracker.release(o);
>> > }
>> >   }
>> >
>> >   // now we can do our assertions
>> >
>> >
>> > expectedLogs.verifyWarn(String.format(SolrIO.Write.WriteFn.RETRY_ATTEMPT_LOG,
>> > 1));
>> >
>> >
>> > Please do point out the obvious if I am missing it - I am a newbie
>> > here...
>> >
>> > Thank you all very much,
>> > Tim
>> > ( timrobertson...@gmail.com on the slack apache/beam channel)
>> >
>> >
>> >
>> 
>> >
>
>

Re: DirectRunner in test - await completion of workers threads?

[Tim Robertson]

Will do - I'll report the result on https://github.com/apache/beam/pull/4905


On Thu, Apr 5, 2018 at 11:45 AM, Ismaël Mejía  wrote:

> For info, Romain's PR was merged today, can you confirm if this fixes
> the issue Tim.
>
> On Sun, Apr 1, 2018 at 9:21 PM, Tim Robertson 
> wrote:
> > Thanks all.
> >
> > I went with what I outlined above, which you can see in this test.
> > https://github.com/timrobertson100/beam/blob/
> BEAM-3848/sdks/java/io/solr/src/test/java/org/apache/beam/
> sdk/io/solr/SolrIOTest.java#L285
> >
> > That forms part of this PR https://github.com/apache/beam/pull/4956
> >
> > I'll monitor Romain's PR and back it out when appropriate.
> >
> >
> >
> >
> >
> > On Sun, Apr 1, 2018 at 8:20 PM, Jean-Baptiste Onofré 
> > wrote:
> >>
> >> Indeed. It's exactly what Romain's PR is about.
> >>
> >> Regards
> >> JB
> >> Le 1 avr. 2018, à 19:33, Reuven Lax  a écrit:
> >>>
> >>> Correct - teardown is currently run in the direct runner, but
> >>> asynchronously. I believe Romain's pending PRs should solve this for
> your
> >>> use case.
> >>>
> >>> On Sun, Apr 1, 2018 at 3:13 AM Tim Robertson <
> timrobertson...@gmail.com>
> >>> wrote:
> 
>  Thanks for confirming Romain - also for the very fast reply!
> 
>  I'll continue with the workaround and reference BEAM-3409 inline as
>  justification.
>  I'm trying to wrap this up before travel next week, but if I get a
>  chance I'll try and run this scenario (BEAM-3848) with your patch.
> 
> 
> 
>  On Sun, Apr 1, 2018 at 12:05 PM, Romain Manni-Bucau
>   wrote:
> >
> > Hi
> >
> > I have the same blocker and created
> >
> > https://github.com/apache/beam/pull/4790 and
> > https://github.com/apache/beam/pull/4965 to solve part of it
> >
> >
> >
> > Le 1 avr. 2018 11:35, "Tim Robertson" < timrobertson...@gmail.com> a
> > écrit :
> >
> > Hi devs
> >
> > I'm working on SolrIO tests for failure scenarios (i.e. an exception
> > will come out of the pipeline execution).  I see that the exception
> is
> > surfaced to the driver while " direct-runner-worker" threads are
> still
> > running.  This causes issue because:
> >
> >   1. The Solr tests do thread leak detection, and a
> solrClient.close()
> > is what removes the object
> >   2. @Teardown is not necessarily called which is what would close
> the
> > solrClient
> >
> > I can unregister all the solrClients that have been spawned.
> However I
> > have seen race conditions where there are still threads running
> creating and
> > registering clients. I need to someone ensure that all workers
> related to
> > the pipeline execution are indeed finished so no new ones are
> created after
> > the first exception is passed up.
> >
> > Currently I have this (psuedo code) which works, but I suspect
> someone
> > can suggest a better approach:
> >
> > // store the state of clients registered for object leak check
> > Set existingClients = registeredSolrClients();
> > try {
> >   pipeline.run();
> >
> > } catch (Pipeline.PipelineExecutionException e) {
> >
> >
> >   // Hack: await all bundle workers completing
> >   while (namedThreadStillExists("direct-runner-worker")) {
> > Thread.sleep(100);
> >   }
> >
> >   // remove all solrClients created in this execution only
> >   // since the teardown may not have done so
> >   for (Object o : ObjectReleaseTracker.OBJECTS.keySet()) {
> > if (o instanceof SolrClient && !existingClients.contains(o)) {
> >   ObjectReleaseTracker.release(o);
> > }
> >   }
> >
> >   // now we can do our assertions
> >
> > expectedLogs.verifyWarn(String.format(SolrIO.Write.
> WriteFn.RETRY_ATTEMPT_LOG,
> > 1));
> >
> >
> > Please do point out the obvious if I am missing it - I am a newbie
> > here...
> >
> > Thank you all very much,
> > Tim
> > ( timrobertson...@gmail.com on the slack apache/beam channel)
> >
> >
> >
> 
> >
>

Re: DirectRunner in test - await completion of workers threads?

[Tim Robertson]

Thanks all.

I went with what I outlined above, which you can see in this test.
https://github.com/timrobertson100/beam/blob/BEAM-3848/sdks/java/io/solr/src/test/java/org/apache/beam/sdk/io/solr/SolrIOTest.java#L285

That forms part of this PR https://github.com/apache/beam/pull/4956

I'll monitor Romain's PR and back it out when appropriate.





On Sun, Apr 1, 2018 at 8:20 PM, Jean-Baptiste Onofré 
wrote:

> Indeed. It's exactly what Romain's PR is about.
>
> Regards
> JB
> Le 1 avr. 2018, à 19:33, Reuven Lax  a écrit:
>
>> Correct - teardown is currently run in the direct runner, but
>> asynchronously. I believe Romain's pending PRs should solve this for your
>> use case.
>>
>> On Sun, Apr 1, 2018 at 3:13 AM Tim Robertson < timrobertson...@gmail.com>
>> wrote:
>>
>>> Thanks for confirming Romain - also for the very fast reply!
>>>
>>> I'll continue with the workaround and reference BEAM-3409 inline as
>>> justification.
>>> I'm trying to wrap this up before travel next week, but if I get a
>>> chance I'll try and run this scenario (BEAM-3848) with your patch.
>>>
>>>
>>>
>>> On Sun, Apr 1, 2018 at 12:05 PM, Romain Manni-Bucau <
>>> rmannibu...@gmail.com> wrote:
>>>
 Hi

 I have the same blocker and created

 https://github.com/apache/beam/pull/4790 and
 https://github.com/apache/beam/pull/4965 to solve part of it



 Le 1 avr. 2018 11:35, "Tim Robertson" < timrobertson...@gmail.com> a
 écrit :

 Hi devs

 I'm working on SolrIO tests for failure scenarios (i.e. an exception
 will come out of the pipeline execution).  I see that the exception is
 surfaced to the driver while " direct-runner-worker" threads are still
 running.  This causes issue because:

   1. The Solr tests do thread leak detection, and a solrClient.close()
 is what removes the object
   2. @Teardown is not necessarily called which is what would close the
 solrClient

 I can unregister all the solrClients that have been spawned.  However I
 have seen race conditions where there are still threads running creating
 and registering clients. I need to someone ensure that all workers related
 to the pipeline execution are indeed finished so no new ones are created
 after the first exception is passed up.

 Currently I have this (psuedo code) which works, but I suspect someone
 can suggest a better approach:

 // store the state of clients registered for object leak check
 Set existingClients = registeredSolrClients();
 try {
   pipeline.run();

 } catch (Pipeline.PipelineExecutionException e) {


 // Hack: await all bundle workers completing
 while (namedThreadStillExists("direct-runner-worker")) {
 Thread.sleep(100);
 }

 // remove all solrClients created in this execution only
 // since the teardown may not have done so
 for (Object o : ObjectReleaseTracker.OBJECTS.keySet()) {
 if (o instanceof SolrClient && !existingClients.contains(o)) {
 ObjectReleaseTracker.release(o);
 }
 }

 // now we can do our assertions
 expectedLogs.verifyWarn(String.format(SolrIO.Write.WriteFn.
 RETRY_ATTEMPT_LOG, 1));


 Please do point out the obvious if I am missing it - I am a newbie
 here...

 Thank you all very much,
 Tim
 ( timrobertson...@gmail.com on the slack apache/beam channel)




>>>

Re: DirectRunner in test - await completion of workers threads?

[Jean-Baptiste Onofré]

Indeed. It's exactly what Romain's PR is about.

Regards
JB

Le 1 avr. 2018 à 19:33, à 19:33, Reuven Lax  a écrit:
>Correct - teardown is currently run in the direct runner, but
>asynchronously. I believe Romain's pending PRs should solve this for
>your
>use case.
>
>On Sun, Apr 1, 2018 at 3:13 AM Tim Robertson
>
>wrote:
>
>> Thanks for confirming Romain - also for the very fast reply!
>>
>> I'll continue with the workaround and reference BEAM-3409 inline as
>> justification.
>> I'm trying to wrap this up before travel next week, but if I get a
>chance
>> I'll try and run this scenario (BEAM-3848) with your patch.
>>
>>
>>
>> On Sun, Apr 1, 2018 at 12:05 PM, Romain Manni-Bucau
>> > wrote:
>>
>>> Hi
>>>
>>> I have the same blocker and created
>>>
>>> https://github.com/apache/beam/pull/4790 and
>>> https://github.com/apache/beam/pull/4965 to solve part of it
>>>
>>>
>>>
>>> Le 1 avr. 2018 11:35, "Tim Robertson"  a
>>> écrit :
>>>
>>> Hi devs
>>>
>>> I'm working on SolrIO tests for failure scenarios (i.e. an exception
>will
>>> come out of the pipeline execution).  I see that the exception is
>surfaced
>>> to the driver while "direct-runner-worker" threads are still
>running.
>>> This causes issue because:
>>>
>>>   1. The Solr tests do thread leak detection, and a
>solrClient.close()
>>> is what removes the object
>>>   2. @Teardown is not necessarily called which is what would close
>the
>>> solrClient
>>>
>>> I can unregister all the solrClients that have been spawned.
>However I
>>> have seen race conditions where there are still threads running
>creating
>>> and registering clients. I need to someone ensure that all workers
>related
>>> to the pipeline execution are indeed finished so no new ones are
>created
>>> after the first exception is passed up.
>>>
>>> Currently I have this (psuedo code) which works, but I suspect
>someone
>>> can suggest a better approach:
>>>
>>> // store the state of clients registered for object leak check
>>> Set existingClients = registeredSolrClients();
>>> try {
>>>   pipeline.run();
>>>
>>> } catch (Pipeline.PipelineExecutionException e) {
>>>
>>>
>>> // Hack: await all bundle workers completing
>>> while (namedThreadStillExists("direct-runner-worker")) {
>>> Thread.sleep(100);
>>> }
>>>
>>> // remove all solrClients created in this execution only
>>> // since the teardown may not have done so
>>> for (Object o : ObjectReleaseTracker.OBJECTS.keySet()) {
>>> if (o instanceof SolrClient && !existingClients.contains(o)) {
>>> ObjectReleaseTracker.release(o);
>>> }
>>> }
>>>
>>> // now we can do our assertions
>>> expectedLogs.verifyWarn(String.format(SolrIO.Write.WriteFn.
>>> RETRY_ATTEMPT_LOG, 1));
>>>
>>>
>>> Please do point out the obvious if I am missing it - I am a newbie
>here...
>>>
>>> Thank you all very much,
>>> Tim
>>> (timrobertson...@gmail.com on the slack apache/beam channel)
>>>
>>>
>>>
>>>
>>

Re: DirectRunner in test - await completion of workers threads?

[Reuven Lax]

Correct - teardown is currently run in the direct runner, but
asynchronously. I believe Romain's pending PRs should solve this for your
use case.

On Sun, Apr 1, 2018 at 3:13 AM Tim Robertson 
wrote:

> Thanks for confirming Romain - also for the very fast reply!
>
> I'll continue with the workaround and reference BEAM-3409 inline as
> justification.
> I'm trying to wrap this up before travel next week, but if I get a chance
> I'll try and run this scenario (BEAM-3848) with your patch.
>
>
>
> On Sun, Apr 1, 2018 at 12:05 PM, Romain Manni-Bucau  > wrote:
>
>> Hi
>>
>> I have the same blocker and created
>>
>> https://github.com/apache/beam/pull/4790 and
>> https://github.com/apache/beam/pull/4965 to solve part of it
>>
>>
>>
>> Le 1 avr. 2018 11:35, "Tim Robertson"  a
>> écrit :
>>
>> Hi devs
>>
>> I'm working on SolrIO tests for failure scenarios (i.e. an exception will
>> come out of the pipeline execution).  I see that the exception is surfaced
>> to the driver while "direct-runner-worker" threads are still running.
>> This causes issue because:
>>
>>   1. The Solr tests do thread leak detection, and a solrClient.close()
>> is what removes the object
>>   2. @Teardown is not necessarily called which is what would close the
>> solrClient
>>
>> I can unregister all the solrClients that have been spawned.  However I
>> have seen race conditions where there are still threads running creating
>> and registering clients. I need to someone ensure that all workers related
>> to the pipeline execution are indeed finished so no new ones are created
>> after the first exception is passed up.
>>
>> Currently I have this (psuedo code) which works, but I suspect someone
>> can suggest a better approach:
>>
>> // store the state of clients registered for object leak check
>> Set existingClients = registeredSolrClients();
>> try {
>>   pipeline.run();
>>
>> } catch (Pipeline.PipelineExecutionException e) {
>>
>>
>> // Hack: await all bundle workers completing
>> while (namedThreadStillExists("direct-runner-worker")) {
>> Thread.sleep(100);
>> }
>>
>> // remove all solrClients created in this execution only
>> // since the teardown may not have done so
>> for (Object o : ObjectReleaseTracker.OBJECTS.keySet()) {
>> if (o instanceof SolrClient && !existingClients.contains(o)) {
>> ObjectReleaseTracker.release(o);
>> }
>> }
>>
>> // now we can do our assertions
>> expectedLogs.verifyWarn(String.format(SolrIO.Write.WriteFn.
>> RETRY_ATTEMPT_LOG, 1));
>>
>>
>> Please do point out the obvious if I am missing it - I am a newbie here...
>>
>> Thank you all very much,
>> Tim
>> (timrobertson...@gmail.com on the slack apache/beam channel)
>>
>>
>>
>>
>