CommittableMessage is the element emitted by an Akka Kafka committable consumer source. It carries the consumed record together with a CommittableOffset, so downstream processing can acknowledge the message by committing its offset. section 14.1.4
Process a message and expose its offset
A stream can inspect the record, then pass its committableOffset to the commit stage: Consumer.committableSource(consumerSettings, Subscriptions.topics("test2")) .map { msg: CommittableMessage[String, String] => println(s"${msg.record.key} => ${msg.record.value.toUpperCase}") msg.committableOffset } .via(Committer.flow(committerSettings.withMaxBatch(100))) Committing after processing provides at-least-once delivery semantics. Batch commits can use a maximum record count, maximum interval, and asynchronous parallelism. section 14.1.4
import akka.kafka.scaladsl.Committer
import akka.kafka.scaladsl.Consumer.DrainingControl
import akka.kafka.CommitterSettings
import akka.kafka.ConsumerMessage.CommittableMessage
object ConsumerCommittableSource {
...
val drainingControl: DrainingControl[_] =
Consumer
.committableSource( #1
consumerSettings,
Subscriptions.topics("test2"))
.map { msg: CommittableMessage[String, String] => #1
println(
s"${msg.record.key} => ${msg.record.value.toUpperCase}")
msg.committableOffset #2
}
.via(Committer.flow(committerSettings.withMaxBatch(100))) #3
.toMat(Sink.seq)(DrainingControl.apply) #4
.run()
...
}
Shows the complete source-to-committer pipeline, including extraction of CommittableOffset, batch configuration, and DrainingControl for orderly shutdown. section 14.1.4
The same acknowledgement information can travel through a producer pipeline. ProducerMessage.single stores the CommittableOffset as an envelope passThrough, allowing committableSink to write the output record and commit the consumed input. section 14.2.2
Operationally, CommittableMessage supports:
- Processing each record before committing its offset. section 14.1.4
- Reducing commit overhead by batching offsets, such as every 100 records or after 10 seconds. section 14.1.4
- Draining in-flight messages before shutdown to avoid reprocessing consumed but uncommitted work. section 14.1.4
