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  CassandraJobStore(CassandraStoreConfiguration configuration,
                             Session session,
                             TitusRuntime titusRuntime);   CassandraJobStore(CassandraStoreConfiguration configuration,
                      Session session,
                      TitusRuntime titusRuntime,
                      ObjectMapper mapper,
                      int initialBucketCount,
                      int maxBucketSize); @Override Completable init(); @Override Observable<Pair<List<Job<?>>, Integer>> retrieveJobs(); @Override Observable<Job<?>> retrieveJob(String jobId); @Override Completable storeJob(Job job); @Override Completable updateJob(Job job); @Override Completable deleteJob(Job job); @Override Observable<Pair<List<Task>, Integer>> retrieveTasksForJob(String jobId); @Override Observable<Task> retrieveTask(String taskId); @Override Completable storeTask(Task task); @Override Completable updateTask(Task task); @Override Completable replaceTask(Task oldTask, Task newTask); @Override Completable moveTask(Job jobFrom, Job jobTo, Task taskAfter); @Override Completable deleteTask(Task task); @Override Observable<Job<?>> retrieveArchivedJob(String jobId); @Override Observable<Task> retrieveArchivedTasksForJob(String jobId); @Override Observable<Task> retrieveArchivedTask(String taskId);  }
@Test public void testMoveTask() { JobStore store = getJobStore(); store.init().await(); Job<ServiceJobExt> jobFrom = createServiceJobObject(); store.storeJob(jobFrom).await(); Job<ServiceJobExt> jobTo = createServiceJobObject(); store.storeJob(jobTo).await(); Task task = createServiceTaskObject(jobFrom); store.storeTask(task).await(); Job<ServiceJobExt> updatedFromJob = JobFunctions.incrementJobSize(jobFrom, -1); Job<ServiceJobExt> updatedToJob = JobFunctions.incrementJobSize(jobTo, 1); Task updatedTask = task.toBuilder().withJobId(updatedToJob.getId()).build(); store.moveTask(updatedFromJob, updatedToJob, updatedTask).await(); Job<?> jobFromLoaded = store.retrieveJob(jobFrom.getId()).toBlocking().first(); assertThat(JobFunctions.getJobDesiredSize(jobFromLoaded)).isEqualTo(0); Pair<List<Task>, Integer> jobFromTasksLoaded = store.retrieveTasksForJob(jobFrom.getId()).toBlocking().first(); assertThat(jobFromTasksLoaded.getLeft()).hasSize(0); Job<?> jobToLoaded = store.retrieveJob(jobTo.getId()).toBlocking().first(); assertThat(JobFunctions.getJobDesiredSize(jobToLoaded)).isEqualTo(2); Pair<List<Task>, Integer> jobToTasksLoaded = store.retrieveTasksForJob(jobTo.getId()).toBlocking().first(); assertThat(jobToTasksLoaded.getLeft()).hasSize(1); jobToTasksLoaded.getLeft().forEach(t -> assertThat(t.getJobId()).isEqualTo(jobTo.getId())); }