Exactly-once delivery is more accurately called effectively-once delivery: it combines at-least-once processing with deduplication, so retries produce one externally visible effect. section 14.3
Retrying a batch without duplicating effects
A process moves 10 items from A through B to C, then fails after five. On restart, all 10 may enter B again. If B records which five it already sent to C, duplicates are detected and C receives one effective copy of each item. section 14.3
Apply effectively-once delivery by combining processing guarantees with the appropriate Akka mechanism:
- Use Akka Transactional with Transactional.source, TransactionalMessage, and Transactional.sink for Kafka processing. section 14.3
- Assign every running application instance a unique transactional ID; reuse causes ProducerFencedException. section 14.3
- Set isolation.level to read-committed and recognize the performance tradeoff against at-least-once delivery. section 14.3
Listing 14.4 Transactional source and sink
object ConsumerProducerEffectivelyOnce {
val transactionalId = args(0)
...
val drainingControl: Consumer.DrainingControl[_] =
Transactional #1
.source(
consumerSettings,
Subscriptions.topics("test5")
)
.map { msg: TransactionalMessage[String, String] => #2
ProducerMessage.single(
new ProducerRecord[String, String](
"test6",
msg.record.key,
msg.record.value),
msg.partitionOffset) #3
}
.toMat(Transactional.sink(producerSettings, transactionalId))( #4
Consumer.DrainingControl.apply)
.run()
}
Shows a transactional Kafka source-to-sink pipeline that preserves the input partition offset while producing the output record. section 14.3
Listing 10.5 Akka projection ingredients
import akka.projection.eventsourced.scaladsl.EventSourcedProvider
import akka.persistence.jdbc.query.scaladsl.JdbcReadJournal
import example.repository.scalike.ScalikeJdbcSession
import example.persistence.SPContainer
object CargosPerContainerProjection {
...
val sourceProvider =
EventSourcedProvider.eventsByTag[SPContainer.Event]( #1
system = system,
readJournalPluginId = JdbcReadJournal.Identifier,
tag = tag)
JdbcProjection.exactlyOnce(
projectionId =
ProjectionId("CargosPerContainerProjection", tag),
sourceProvider = sourceProvider,
handler =
() => new CPCProjectionHandler(tag, system, repository), #2
sessionFactory = () => new ScalikeJdbcSession()) #3
(system)
}
Shows an Akka projection reading events by tag and writing them through an exactly-once JDBC projection with a handler and database session. section 10.3
