Can we do parallel Kafka Streams processing with CompletableFutures

Is it possible to do parallel work within a Kafka stream application using Java CompletableFutures?

I want to read from 1 Kafka topic, create two windowed counts, 1 for minute and another for hour but do them in parallel.

I wrote some sample code. I am able to get this to work but looking at the Kafka stream documentation, since KafkaStreams assigns 1 task per partition and it can't go beyond one-thread I'm not sure if this code will have desired effect.

CompletableFuture completableFutureOfMinute = CompletableFuture.runAsync(() -> {
        inputStream.groupByKey()
                .windowedBy(TimeWindows.of(Duration.ofMinutes(1)).grace(Duration.ofMinutes(1)))
                .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as(
                        "minute-store")
                        .withRetention(Duration.ofMinutes(1)))
                .toStream()
                .to("result-topic");
    });

    CompletableFuture completableFutureOfHour = CompletableFuture.runAsync(() -> {
        inputStream.groupByKey()
                .windowedBy(TimeWindows.of(Duration.ofHours(1)).grace(Duration.ofHours(1)))
                .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as(
                        "hour-store")
                        .withRetention(Duration.ofHours(1)))
                .toStream()
                .to("result-topic-2", produced);
    });

    final CompletableFuture<Void> combinedFutures = CompletableFuture.allOf(completableFutureOfMinute,
            completableFutureOfHour);

    try {
        combinedFutures.get();
    } catch (final Exception ex) {

    }

Your program does not seem to be right.

Note, that using the DSL, you basically assemble a dataflow program and data processing only starts when you call KafkaStreams#start(). Thus, using Futures while specifying your processing logic does not help, as no data is processed yet.

Kafka Streams parallelizes based on tasks. Thus, if you want to process both windows in parallel, you would need to "replicate" the input topic to split your program (called Topology) into independent parts (call SubTopology):

<!-- language java -->

KStream input = builder.stream(...);
input.groupByKey().windowBy(/* 1 min */).count(...);
input.repartition().groupByKey().windowBy(/* 1 hour */).count();

Using repartition() your program will be split into two sub-topologies and you will get tasks for each sub-topology that can be processed by different threads in parallel.

However, I actually doubt if this program will increase your throughput. If you really want to increase your throughput, you should increase the number of input topic partitions to get more parallel tasks.