Push query disconnection/reconnect

[blatda]

I have a push query running through:
_subscription = _kContext.CreateQuery<T>(queryStreamParameters).ToObservable().Subscribe(monitor);
If the connection to the kSqlDB is lost, the OnError method is invoked and the push query keeps running on the server and cannot be reconnected.
Is there any strategy to:

1. terminate the orphaned push query?
2. reconnect query or create a new one, which will start where the previous ended?

What I'm actually supposed to do inside the OnError method?

Thanks for the Client library, that's an extraordinary contribution to the kSqlDb community! :-)

[tomasfabian]

Hi @blatda David!

1. Could you please try out the bellow example, please? Does it do what you would expect regarding the termination of running queries?

Subscription? _subscription = null;

_subscription = await _kContext.CreateQuery<T>(queryStreamParameters)
  .SubscribeAsync(onNext: value =>
    {
    }, onError: async error =>
    {
      await restApiProvider.TerminatePushQueryAsync(subscription!.QueryId, cts.Token);
    },
    onCompleted: () => { }, cts.Token);

Is the monitor variable an observer in your example? You probably doesn't have to call ToObservable() extension method in your pipeline. Let me know if I missed something. If the above example satisfies the termination of orphaned queries we could maybe find a better way how to achieve it even internally called for the Subscribe method, but I'm afraid that it would be a breaking change.

2. I'm not aware about such an option (reconnections). I even raised a question in StackOverflow once. If someone will find something useful in the ksqldb's doc let me know, please.
   As a workaround you could try to create a stream/table which is backed by a topic (eg CREATE STREAM AS SELECT) and then consume it with a lower lever Kafka consumer. In that way you would have the commiting of offsets under your control. The downside is that you have to provide Kafka brokers (BootstrapServers as an additional config) etc.

Thank you very much for your appreciation!