Implements #93 - Generic Vert.x future support

[astubbs]

fixes #93

[astubbs]

@radoslawtwardy can you give me an example of the async io code you're using? I'd like something more concrete to use in my tests for this.

[radoslawtwardy]

So as I said i tried to wrote own ParallelEoSStreamProcessor (which inside call async-htpp-client and some other thinks), but also i tried to use VertxParallelStreamProcessor, so maybe so example with it:

    ParallelConsumerOptions<String, String> options = ParallelConsumerOptions.<String, String>builder()
            .ordering(ParallelConsumerOptions.ProcessingOrder.UNORDERED)
            .maxConcurrency(10000) //I assume that processing mainly wait for IO
            .consumer(consumer)
            .build();

    VertxParallelStreamProcessor<String, String> vertxProcessor = VertxParallelStreamProcessor.createEosStreamProcessor(consumer, null, options);
    vertxProcessor.subscribe(of("test"));

    vertxProcessor.vertxHttpWebClient(
            (client, record) -> client.post(9000, "localhost", "/someEndpoint").sendJsonObject(new JsonObject("someBody")),
            resp -> resp.onSuccess(s -> logger.debug(s.statusMessage()))

This cause that i has 10000 threads !

I can show you own implementation of ParallelEoSStreamProcessor (which has same problem) - but i wrote it in scala (i'm scala developer)

[radoslawtwardy]

I've written some very simple code in Java - using ParallelEoSStreamProcessor to show you intention:

    //some simple and naive implementation of thread pool for IO operations - replace with some fork join pool
    ThreadPoolExecutor executor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), 32, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue());

    KafkaConsumer<String, String> consumer = new KafkaConsumer(kafkaConsumerProps);
    ParallelConsumerOptions<String, String> options = ParallelConsumerOptions.<String, String>builder()
            .ordering(ParallelConsumerOptions.ProcessingOrder.UNORDERED)
            .maxConcurrency(10000) //I assume that processing mainly wait for IO
            .consumer(consumer)
            .build();

    ParallelStreamProcessor<String, String> processor = ParallelStreamProcessor.createEosStreamProcessor(options);
    processor.subscribe(of("test"));

    processor.poll(record -> executor.execute(new Runnable() {
        @Override
        public void run() {
            //do some IO operations with record
        }
    }));

Of course using future is much better and elegant, but I think simplicity of this code show my case. I want delegate main work to custom thread pool which is designed for some kind of IO operations. It can be not only one thread pool (maybe few).
Your commit (if I understood) has good API, but not resolve my problem, that throughput are closely related with pool size of 'workerPool'

[astubbs]

I don't understand - this PR adds future support for generic async functions to the vert.x module, which uses a small number of threads.

Is your fork published so I can see it? I am also a "Scala developer"? doesn't seem to be on this list: https://github.com/confluentinc/parallel-consumer/network/members.

You don't need to reimplement thread pools or a custom ParallelEoSStreamProcessor in order to use generic async operations - the vert.x module already does that for you (with this PR). It uses a thread pool by default of the number of cores as you've shown.

ParallelEoSStreamProcessor (which inside call async-htpp-client

You also don't need a custom implementation just to have async http requests - the vert.x http functions are already available in the vert.x module.

This cause that i has 10000 threads !

Are you saying that when you tried the http client in the vert.x module, you got 10,000 threads? That shouldn't happen, and if it did, please open an issue as it's a bug. You should get roughly the same number of threads as you have cores, unless it's overridden.

do some IO operations with record

Also if you could show some examples of the complex IO operations you're doing and the libraries used and what they return in the function that would really help as well.

[radoslawtwardy]

As you suggested i create issue #97

[radoslawtwardy]

Together with #98 this seems nice for me. Thanks!

[birdayz]

hey! this is pretty interesting. sorry for being annoying, but are there any plans to finish this PR ?

[astubbs]

Thanks @birdayz - yes! Just finishing #54 first - paying down some decently sized tech debt first before adding more features...

[astubbs]

hey! this is pretty interesting. sorry for being annoying, but are there any plans to finish this PR ?

Yes. This is now supported in the Reactor module ( #116 ). The fixes for that, will enable this with Vert.x I think.