CassandraJobStore implements JobStore { @Override public Completable deleteJob(Job job) { return Observable.fromCallable(() -> { String jobId = job.getId(); checkIfJobIsActive(jobId); return jobId; }).flatMap(jobId -> retrieveTasksForJob(jobId).flatMap(tasksAndErrors -> { List<Task> tasks = tasksAndErrors.getLeft(); int errors = tasksAndErrors.getRight(); if (errors > 0) { logger.warn("Some tasks records could not be loaded during the job delete operation. Ignoring them: {}", errors); } List<Task> fixedTasks = checkTaskConsistency(tasks); List<Completable> completables = fixedTasks.stream().map(this::deleteTask).collect(Collectors.toList()); return Completable.merge(Observable.from(completables), getConcurrencyLimit()).toObservable(); })).toList().flatMap(ignored -> { BatchStatement statement = getArchiveJobBatchStatement(job); transactionLogger().logBeforeDelete(deleteActiveJobStatement, "deleteJob", job); return execute(statement).doOnNext(rs -> transactionLogger().logAfterDelete(deleteActiveJobStatement, "deleteJob", job)); }).flatMap(ignored -> { activeJobIdsBucketManager.deleteItem(job.getId()); return Observable.empty(); }).toCompletable(); } @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 testDeleteJob() { JobStore store = getJobStore(); Job<BatchJobExt> job = createBatchJobObject(); store.init().await(); store.storeJob(job).await(); Pair<List<Job<?>>, Integer> jobsAndErrors = store.retrieveJobs().toBlocking().first(); assertThat(jobsAndErrors.getLeft().get(0)).isEqualTo(job); store.deleteJob(job).await(); jobsAndErrors = store.retrieveJobs().toBlocking().first(); assertThat(jobsAndErrors.getLeft()).isEmpty(); }