class CassandraJobStore implements JobStore {

    @Override
    public Completable storeJob(Job job) {
        return Observable
                .fromCallable((Callable<Statement>) () -> {
                    String jobId = job.getId();
                    checkIfJobAlreadyExists(jobId);

                    String jobJsonString = writeJobToString(job);

                    int bucket = activeJobIdsBucketManager.getNextBucket();
                    activeJobIdsBucketManager.addItem(bucket, jobId);
                    Statement jobStatement = insertActiveJobStatement.bind(jobId, jobJsonString);
                    Statement jobIdStatement = insertActiveJobIdStatement.bind(bucket, jobId);

                    BatchStatement batchStatement = new BatchStatement();
                    batchStatement.add(jobStatement);
                    batchStatement.add(jobIdStatement);

                    transactionLogger().logBeforeCreate(insertActiveJobStatement, "storeJob", job);

                    return batchStatement;
                })
                .flatMap(statement -> execute(statement)
                        .doOnNext(rs -> transactionLogger().logAfterCreate(insertActiveJobStatement, "storeJob", job))
                        .doOnError(throwable -> activeJobIdsBucketManager.deleteItem(job.getId()))
                )
                .toCompletable();
    }

    @Inject public  CassandraJobStore(CassandraStoreConfiguration configuration,
                             Session session,
                             TitusRuntime titusRuntime);
      CassandraJobStore(CassandraStoreConfiguration configuration,
                      Session session,
                      TitusRuntime titusRuntime,
                      ObjectMapper mapper,
                      int initialBucketCount,
                      int maxBucketSize);

}