Hi, its not a issue, i followed your Git repository and trying to implement same for my project, (my stack is reactive mongo and playframework) ,in your code you created consumers using (Consumer.committableSource), i also did the same, but the problem is i need to re-process around 40 lakh to 50 lakh records, but it takes more time( nearly 1 day). and the problem is when i see the cpu percentages , my analytics is using very less cpu percentage.

my code is:
testConsumer.create("testconsumer")(context.system)
.groupedWithin(batchSize, batchAwaitTime milliseconds)
.mapAsync(incidentParallelism)({ records =>
//logger.warn("batch size for Incidents is " + records.size)
var offsetBatch = CommittableOffsetBatch.empty
val incidentList = records map { eachRecord =>
offsetBatch = offsetBatch.updated(eachRecord.committableOffset)
Json.parse(eachRecord.record.value()).as[IncidentModel]
}
Future.sequence(processRecords(incidentList.toList, isIncidentsNeedsProcessing)).map { _ =>
offsetBatch
}
})
.mapAsync(2)(_.commitScaladsl())
.runWith(Sink.ignore) recover {
case e: Exception => logger.warn(" Exception occurred in actor " + e)
}

my consumer configuration is :

akka.kafka.consumer {

poll-interval = 50ms
poll-timeout = 2s
stop-timeout = 30s

close-timeout = 100s
commit-timeout = 100s

wakeup-timeout = 3s

max-wakeups = 10

use-dispatcher = "akka.kafka.default-dispatcher"

kafka-clients {
enable.auto.commit = false

heartbeat.interval.ms = 1500

session.timeout.ms = 10000

max.poll.records = 500

}
}
iam trying hard to know the mistake i did. help me !!!

https://github.com/softwaremill/activator-reactive-kafka-scala/blob/master/app/reactivekafka/LoggingConsumer.scala#L23-L32

Seems it's not up to date anymore, and people keep referring to it since it's a top result in google search