Can we shutdown a pipeline based on some condition

[Sachin Mittal <sj...@gmail.com>]

Hi,
I am kind of building a batch/streaming hybrid beam application.
Data is fed into a kinesis stream and the beam pipeline is run.

I want to stop the pipeline if no new data is fed into the stream for a
certain period of time, say 5 minutes.

Is there a way of achieving this ?

Right now I only see something like this:

pipeline.run().waitUntilFinish();

Instead of waiting until finish can we have some conditional finish in any
way ?

Thanks
Sachin

Re: Can we shutdown a pipeline based on some condition

[Sachin Mittal <sj...@gmail.com>]

Hi Pavel,
I have tried the setting you have suggested but this does not seem to work.
Looks like this setting bounds the unbounded stream till the specified time
and then closes the stream and it is only after that the records are
applied downstream.

In my case withMaxReadTime = 3 minutes

Logs are like this:
------------------------------------------------------------------------
13:06:38.267 [direct-runner-worker] INFO a.p.f.b.c.k.ShardReadersPool -
Starting to read test-batch stream from [shardId-000000000000] shards
13:09:39.002 [direct-runner-worker] INFO a.p.f.b.c.k.ShardReadersPool -
Closing shard iterators pool
13:09:42.632 [pool-16-thread-1] INFO a.p.f.b.c.k.ShardReadersPool - Kinesis
Shard read loop has finished


As you can see, it reads all the records and then closes the reader and
only when the read loop is finished, the downstream operators process these
records further.

So if downstream I again right to the same Kinesis stream, it is never read
as the loop is closed.

I guess I may still need the unbounded stream so it continues to listen to
any fresh records added to the stream, but I would like the beam
application to shutdown after a specified time.

Would some like this work:

pipeline.run().waitUntilFinish(Duration.standardMinutes(3));

This will run the pipeline for 3 minutes and then shuts it down and by
then it has computed all the derived data.


Thanks

Sachin



On Thu, May 4, 2023 at 11:26 PM Pavel Solomin <p....@gmail.com> wrote:

> Hello!
>
> In case of KinesisIO there is a param which you can set - withMaxReadTime
> I think many IOs implement it. It does not check if there is still some
> outstanding data, simply finishes the pipeline when this time is over.
>
> Best Regards,
> Pavel Solomin
>
> Tel: +351 962 950 692 | Skype: pavel_solomin | Linkedin
> <https://www.linkedin.com/in/pavelsolomin>
>
>
>
>
>
> On Thu, 4 May 2023 at 18:20, Sachin Mittal <sj...@gmail.com> wrote:
>
>> Hi,
>> I am kind of building a batch/streaming hybrid beam application.
>> Data is fed into a kinesis stream and the beam pipeline is run.
>>
>> I want to stop the pipeline if no new data is fed into the stream for a
>> certain period of time, say 5 minutes.
>>
>> Is there a way of achieving this ?
>>
>> Right now I only see something like this:
>>
>> pipeline.run().waitUntilFinish();
>>
>> Instead of waiting until finish can we have some conditional finish in
>> any way ?
>>
>> Thanks
>> Sachin
>>
>

Re: Can we shutdown a pipeline based on some condition

[Pavel Solomin <p....@gmail.com>]

Hello!

In case of KinesisIO there is a param which you can set - withMaxReadTime
I think many IOs implement it. It does not check if there is still some
outstanding data, simply finishes the pipeline when this time is over.

Best Regards,
Pavel Solomin

Tel: +351 962 950 692 | Skype: pavel_solomin | Linkedin
<https://www.linkedin.com/in/pavelsolomin>





On Thu, 4 May 2023 at 18:20, Sachin Mittal <sj...@gmail.com> wrote:

> Hi,
> I am kind of building a batch/streaming hybrid beam application.
> Data is fed into a kinesis stream and the beam pipeline is run.
>
> I want to stop the pipeline if no new data is fed into the stream for a
> certain period of time, say 5 minutes.
>
> Is there a way of achieving this ?
>
> Right now I only see something like this:
>
> pipeline.run().waitUntilFinish();
>
> Instead of waiting until finish can we have some conditional finish in any
> way ?
>
> Thanks
> Sachin
>