How to correctly close a StreamsApp with a parallel-consumer ?

[bmaggi]

I got my application stuck once because the parallel-consumer auto close himself without throwing an exception, so other threads where still running while the parallel-consumer was stopped. Which is something I want to avoid ;)

My current solution is to launch a thread to check regularly if the ParallelEoSStreamProcessor isClosedOrFailed.
I feel that it's probably not the best option, is there any mechanism that I missed ?

=> How to correctly close a StreamsApp with a parallel-consumer ?

For example : replacing the ParallelStreamProcessor used in StreamsApp by a this failing implementation :

public class MyParallelEoSStreamProcessor<K, V>   extends ParallelEoSStreamProcessor<K, V>  {
    public MyParallelEoSStreamProcessor(ParallelConsumerOptions<K, V> newOptions) {
        super(newOptions);
    }

    @Override
    protected void controlLoop(Function userFunction, Consumer callback) throws TimeoutException, ExecutionException, InterruptedException {
        throw new ExecutionException(new TimeoutException("example error"));
    }
}

Will lead to these logs and a never closing app

2023-07-24T13:48:03.458  INFO  [io.confluent.parallelconsumer.examples.streams.StreamsAppBMA] - start stream
2023-07-24T13:48:04.232  INFO  [io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor] - Confluent Parallel Consumer initialise... groupId: esp-bmaggi-testacf78362-a35c-445d-a29f-c2ff77459938, Options: ParallelConsumerOptions(consumer=org.apache.kafka.clients.consumer.KafkaConsumer@55342f40, producer=org.apache.kafka.clients.producer.KafkaProducer@a4ca3f6, managedExecutorService=java:comp/DefaultManagedExecutorService, managedThreadFactory=java:comp/DefaultManagedThreadFactory, allowEagerProcessingDuringTransactionCommit=false, commitLockAcquisitionTimeout=PT5M, produceLockAcquisitionTimeout=PT1M, commitInterval=PT5S, ordering=KEY, commitMode=PERIODIC_CONSUMER_ASYNCHRONOUS, maxConcurrency=16, defaultMessageRetryDelay=PT1S, retryDelayProvider=null, sendTimeout=PT10S, offsetCommitTimeout=PT10S, batchSize=1, thresholdForTimeSpendInQueueWarning=PT10S, maxFailureHistory=10)
2023-07-24T13:48:04.381  INFO  [pl.tlinkowski.unij.api.UniJLoader] - pl.tlinkowski.unij.service.api.collect.UnmodifiableListFactory service: selected pl.tlinkowski.unij.service.collect.jdk8.Jdk8UnmodifiableListFactory (priority=40)
2023-07-24T13:48:04.391  INFO  [io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor] - Control loop starting up...
2023-07-24T13:48:04.392  ERROR [io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor] - Error from poll control thread, will attempt controlled shutdown, then rethrow. Error: java.util.concurrent.TimeoutException: example error
java.util.concurrent.ExecutionException: java.util.concurrent.TimeoutException: example error
	at io.confluent.parallelconsumer.MyParallelEoSStreamProcessor.controlLoop(MyParallelEoSStreamProcessor.java:19) ~[classes/:?]
	at io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor.lambda$supervisorLoop$9(AbstractParallelEoSStreamProcessor.java:667) ~[classes/:?]
	at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
	at java.lang.Thread.run(Thread.java:833) [?:?]
Caused by: java.util.concurrent.TimeoutException: example error
	... 6 more
2023-07-24T13:48:04.441  INFO  [pl.tlinkowski.unij.api.UniJLoader] - pl.tlinkowski.unij.service.api.collect.UnmodifiableMapFactory service: selected pl.tlinkowski.unij.service.collect.jdk8.Jdk8UnmodifiableMapFactory (priority=40)
2023-07-24T13:48:04.621  ERROR [io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor] - PC closed due to error: java.lang.RuntimeException: Error from poll control thread: java.util.concurrent.TimeoutException: example error

PS launching an throw new RuntimeException("example"); instead of logginf the processing will have a similar effect.

[rkolesnev]

Hi,
Excuse me for delay in looking into this.

I have tried to reproduce and yes - i noticed that if controlLoop dies - the exception is not thrown to main thread until close() is called - as state of controlTaskFutureResult is not checked until then - this is due to the fact that PC runs asynchronously - it is not possible to throw out the exception from background thread to main thread without blocking (or periodic checking etc).

Given that execution is asynchronous - i am not sure of the top of my head what the best approach would be to monitor it - either as you mentioned - create a thread that checks status of the parallel consumer periodically add and expose an listener hook that is triggered on close / exception...

Afaik the behaviour is similar to Kafka Streams actually - as it runs in async thread as well - so it cannot really throw out an exception to main thread on error.

Feel free to raise an issue if you have a suggestion on how it could be improved.