CassandraModSeqProvider implements ModSeqProvider { @Override public ModSeq nextModSeq(Mailbox mailbox) throws MailboxException { CassandraId mailboxId = (CassandraId) mailbox.getMailboxId(); return nextModSeq(mailboxId) .blockOptional() .orElseThrow(() -> new MailboxException("Can not retrieve modseq for " + mailboxId)); } @Inject  CassandraModSeqProvider(Session session, CassandraConfiguration cassandraConfiguration,
                                   CassandraConsistenciesConfiguration consistenciesConfiguration); @Override ModSeq nextModSeq(Mailbox mailbox); @Override ModSeq nextModSeq(MailboxId mailboxId); @Override ModSeq highestModSeq(Mailbox mailbox); @Override ModSeq highestModSeq(MailboxId mailboxId); Mono<ModSeq> nextModSeq(CassandraId mailboxId); static final String MOD_SEQ_CONDITION; }
@Test void failedInsertsShouldBeRetried(CassandraCluster cassandra) throws Exception { Barrier insertBarrier = new Barrier(2); Barrier retryBarrier = new Barrier(1); cassandra.getConf() .registerScenario( executeNormally() .times(2) .whenQueryStartsWith("SELECT nextModseq FROM modseq WHERE mailboxId=:mailboxId;"), awaitOn(insertBarrier) .thenExecuteNormally() .times(2) .whenQueryStartsWith("INSERT INTO modseq (nextModseq,mailboxId) VALUES (:nextModseq,:mailboxId) IF NOT EXISTS;"), awaitOn(retryBarrier) .thenExecuteNormally() .times(1) .whenQueryStartsWith("SELECT nextModseq FROM modseq WHERE mailboxId=:mailboxId;")); CompletableFuture<ModSeq> operation1 = modSeqProvider.nextModSeq(CASSANDRA_ID) .subscribeOn(Schedulers.elastic()) .toFuture(); CompletableFuture<ModSeq> operation2 = modSeqProvider.nextModSeq(CASSANDRA_ID) .subscribeOn(Schedulers.elastic()) .toFuture(); insertBarrier.awaitCaller(); insertBarrier.releaseCaller(); retryBarrier.awaitCaller(); retryBarrier.releaseCaller(); cassandra.getConf() .execute(QueryBuilder.delete().from(TABLE_NAME) .where(QueryBuilder.eq(MAILBOX_ID, CASSANDRA_ID.asUuid()))); retryBarrier.releaseCaller(); assertThatCode(() -> operation1.get(1, TimeUnit.SECONDS)).doesNotThrowAnyException(); assertThatCode(() -> operation2.get(1, TimeUnit.SECONDS)).doesNotThrowAnyException(); }
@Test void nextModSeqShouldGenerateUniqueValuesWhenParallelCalls() throws ExecutionException, InterruptedException { int nbEntries = 10; ConcurrentSkipListSet<ModSeq> modSeqs = new ConcurrentSkipListSet<>(); ConcurrentTestRunner.builder() .operation( (threadNumber, step) -> modSeqs.add(modSeqProvider.nextModSeq(mailbox))) .threadCount(10) .operationCount(nbEntries) .runSuccessfullyWithin(Duration.ofMinutes(1)); assertThat(modSeqs).hasSize(100); }