At-least-once delivery ensures that a record is processed before its Kafka offset is committed. If processing or committing is interrupted, the record can be read and processed again after restart rather than being lost. The guarantee favors retaining work over preventing duplicate work. section 14.1.4 section 14.5
Use a committable consumer source to receive a CommittableMessage, process its ConsumerRecord, and pass its CommittableOffset to a commit stage. Commit after each record for simplicity, or batch commits to avoid slowing a stream with many messages. section 14.1.4 section 14.1.4
Commit batching is controlled by settings that balance how frequently offsets are persisted:
- maxBatch limits the number of messages in one commit; its default is 1,000. section 14.1.4
- maxInterval commits after a time limit when the batch has not filled; its default is 10 seconds. section 14.1.4
- maxParallelism controls asynchronous commit parallelism; its default is 100. section 14.1.4
- Drain the stream during shutdown so in-flight records can finish processing and commit, reducing avoidable reprocessing. section 14.1.4
Consume, transform, produce, then commit
A Kafka pipeline can read from test3 with Consumer.committableSource, wrap each record for test4 with ProducerMessage.single, carry the original committable offset as pass-through data, and write through Producer.committableSink. The offset is committed as part of this at-least-once flow. section 14.2.2 section 14.2.2
Duplicates are the expected failure consequence. If a batch moving items from A to B fails after five of ten items, restarting from the first item can send the initial five again. Effectively-once delivery retains at-least-once behavior but deduplicates repeated items. section 14.3 section 14.3
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 stream shows the core sequence: read committable records, transform each record, pass its offset into Committer.flow, and attach draining control for shutdown. section 14.1.4
