Asynchronous execution inside poll method.

[radoslawtwardy]

Hi, I have some code:

    ParallelStreamProcessor<String, String> eosStreamProcessor = ParallelStreamProcessor.createEosStreamProcessor(options);

    eosStreamProcessor.subscribe(java.util.regex.Pattern.compile("test\\..*"));

    eosStreamProcessor.poll(record -> process(record));`

Method process do some IO operations. So i don't want block consumer threads for this IO operations. It is possibility to execute process method in own thread pool (asynchronous), and then notify in some way that record was processed.

It is real problem for me, because I must be able to handle thousands simultaneous process invocations - but setting maxConcurrency to for. example. 50k seems vary bad (it means that i will have thread pool with 50k threads).

[radoslawtwardy]

I try to analyze your code in order to implement own ParallelStreamProcessor (following the example of vertx implementation) - but i don't understand one thing: why in vertx implementation thread poll size is also same as max queue size ? If I understand do doc ('maxConcurrency') doesn't suite to code:

Controls the maximum degree of concurrency to occur. Used to limit concurrent calls to external systems to a maximum to prevent overloading them or to a degree, using up quotas. [...] In the core module, this sets the number of threads to use in the core's thread pool.
So I assume that in vertx (and my) implementation thread pool will be created in another way - but not, worse, as I wrote in code is assumption that max queue size is always thread pool size.

Any tips for me ? Your parallel client is great - I'd rather not write own client.

[astubbs]

Is it possible for you to write a vert.x Vertical to do your io? That way it's very simple - no need to reimplement anything. See https://vertx.io/docs/vertx-core/java/#_verticles.

I just realised I never implemented a generic version, only an http version. Let me take a quick stab, and maybe you'd like to improve on it.. one sec.

The max concurrency setting in the code module went through some changes - I may need to look at how' it's referenced in the vert.x module. The vert.x module hasn't been used much so it may have lapsed a bit.

[astubbs]

Ah - sorry for the long delay. Have you looked at the vert.x module? That does exactly what you're after, and is already implemented. If you don't want to do http, you can implement a vert.x Vertical to do your IO. Let me know once you've had a look at that module. If you can provide some sample code of your async io function, I can help write an example that uses it.

Ok sorry yea I see you've already looked at that. Let me see...

[radoslawtwardy]

Hi, I wrote own ParallelEoSStreamProcessor implementation. But I also tried a vert.x implementation. Both has some problem - maxConcurrency = thread pool size, and possible throughput is maxConcurrency * factor - so I if want handle 10k messages i must have huge number of threads.
I have forked project for test. If i separate maxConcurrency and thread pool size settings - then it is ok for me, but it is big work around, because then consumer loses all back pressure (which is not important in my case, but generally it is).

[astubbs]

Let's move this here #94 (comment)

[astubbs]

Is your fork published? doesn't seem to be on this list: https://github.com/confluentinc/parallel-consumer/network/members

[astubbs]

Use the generic existing http support in the vert.x module for this, for generic CompletableFuture or Vert.x Future support, this is being added in this PR #94.