class CassandraJobStore implements JobStore {

    @Override
    public Completable moveTask(Job jobFrom, Job jobTo, Task taskAfter) {
        return Observable.fromCallable((Callable<Statement>) () -> {
            checkIfJobIsActive(jobFrom.getId());
            checkIfJobIsActive(jobTo.getId());

            String taskJsonString = ObjectMappers.writeValueAsString(mapper, taskAfter);
            transactionLogger().logBeforeUpdate(insertActiveTaskStatement, "moveTask", taskJsonString);

            BatchStatement batchStatement = new BatchStatement();
            batchStatement.add(insertActiveJobStatement.bind(jobFrom.getId(), ObjectMappers.writeValueAsString(mapper, jobFrom)));
            batchStatement.add(insertActiveJobStatement.bind(jobTo.getId(), ObjectMappers.writeValueAsString(mapper, jobTo)));
            batchStatement.add(insertActiveTaskStatement.bind(taskAfter.getId(), taskJsonString));
            batchStatement.add(deleteActiveTaskIdStatement.bind(jobFrom.getId(), taskAfter.getId()));
            batchStatement.add(insertActiveTaskIdStatement.bind(jobTo.getId(), taskAfter.getId()));

            return batchStatement;
        }).flatMap(this::execute).toCompletable().doOnCompleted(() -> transactionLogger().logAfterUpdate(insertActiveTaskStatement, "moveTask", taskAfter));
    }

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

}