This is an automated email from the ASF dual-hosted git repository.

rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git

commit 918eeefea29f428d486580cf91229a3a05243025
Author: Benoit Tellier <[email protected]>
AuthorDate: Mon Mar 2 10:45:03 2020 +0700

    JAMES-2897 Use SERIAL consistency level for event sourcing
    
    Conditional inserts and deletes are used. Not using SERIAL
    consistency level breaks linearizability
    
    CF 
https://docs.datastax.com/en/ddac/doc/datastax_enterprise/dbInternals/dbIntLtwtTransactions.html
---
 .../apache/james/eventsourcing/eventstore/cassandra/EventStoreDao.scala | 2 ++
 1 file changed, 2 insertions(+)

diff --git 
a/event-sourcing/event-store-cassandra/src/main/scala/org/apache/james/eventsourcing/eventstore/cassandra/EventStoreDao.scala
 
b/event-sourcing/event-store-cassandra/src/main/scala/org/apache/james/eventsourcing/eventstore/cassandra/EventStoreDao.scala
index acac08b..95a9239 100644
--- 
a/event-sourcing/event-store-cassandra/src/main/scala/org/apache/james/eventsourcing/eventstore/cassandra/EventStoreDao.scala
+++ 
b/event-sourcing/event-store-cassandra/src/main/scala/org/apache/james/eventsourcing/eventstore/cassandra/EventStoreDao.scala
@@ -18,6 +18,7 @@
  ****************************************************************/
 package org.apache.james.eventsourcing.eventstore.cassandra
 
+import com.datastax.driver.core.ConsistencyLevel.SERIAL
 import com.datastax.driver.core._
 import com.datastax.driver.core.querybuilder.QueryBuilder
 import com.datastax.driver.core.querybuilder.QueryBuilder.{bindMarker, 
insertInto}
@@ -66,6 +67,7 @@ class EventStoreDao @Inject() (val session: Session, val 
jsonEventSerializer: Js
 
   private[cassandra] def getEventsOfAggregate(aggregateId: AggregateId): 
SMono[History] = {
     val preparedStatement = select.bind.setString(AGGREGATE_ID, 
aggregateId.asAggregateKey)
+      .setConsistencyLevel(SERIAL)
     val rows: SFlux[Row] = 
SFlux[Row](cassandraAsyncExecutor.executeRows(preparedStatement))
 
     val events: SFlux[Event] = rows.map(toEvent)


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to