partitioned_kafka_akka.java

Consumer
  .plainPartitionedSource(consumerSettings, Subscriptions.topics(topicName))
  .mapAsync(maxPartitions, pair -> {
      Source<ConsumerRecord<String, String>, NotUsed> source = pair.second();
      return source
              .map(record -> {
                  // processing logic comes here
                  return record;
              })
              .runWith(Sink.ignore(), materializer);
  })
  .runWith(Sink.ignore(), materializer)
  .toCompletableFuture()
  .handle(AppSupport.doneHandler())
  .join();