CommittableOffset is the Kafka position attached to a consumed record. Carry it through the stream and commit it only after processing succeeds to obtain at-least-once delivery. section 14.1.4
Commit a consumed record
A committable source emits CommittableMessage values. Process the record, then map the message to msg.committableOffset and pass those offsets to Committer.flow. The flow accepts Committable values and produces Done. 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()
...
}
This complete pipeline shows record processing, CommittableOffset extraction, batched committing, and draining during shutdown. section 14.1.4
Use the offset in several operational patterns:
- Commit offsets individually for a simple pipeline, or batch them with maximum batch and interval limits; defaults are 1,000 messages and 10 seconds. section 14.1.4
- Tune asynchronous commit parallelism with maxParallelism; the default is 100. section 14.1.4
- Preserve the offset as passThrough in ProducerMessage.single when consuming from one Kafka topic and producing to another through committableSink. section 14.2.2
- Drain the stream before shutdown so consumed messages still in flight can finish processing and committing. section 14.1.4
