CassandraJobStore implements JobStore { @Override public Observable<Pair<List<Task>, Integer>> retrieveTasksForJob(String jobId) { return Observable.fromCallable(() -> { checkIfJobIsActive(jobId); return retrieveActiveTaskIdsForJobStatement.bind(jobId).setFetchSize(Integer.MAX_VALUE); }).flatMap(retrieveActiveTaskIdsForJob -> execute(retrieveActiveTaskIdsForJob).flatMap(taskIdsResultSet -> { List<String> taskIds = taskIdsResultSet.all().stream() .map(row -> row.getString(0)) .flatMap(taskId -> { if (fitBadDataInjection.isPresent()) { List<String> effectiveTaskIds = new ArrayList<>(); String effectiveTaskId = fitBadDataInjection.get().afterImmediate(JobStoreFitAction.ErrorKind.LostTaskIds.name(), taskId); if (effectiveTaskId != null) { effectiveTaskIds.add(effectiveTaskId); } String phantomId = fitBadDataInjection.get().afterImmediate(JobStoreFitAction.ErrorKind.PhantomTaskIds.name(), taskId); if (phantomId != null && !phantomId.equals(taskId)) { effectiveTaskIds.add(phantomId); } return effectiveTaskIds.stream(); } return Stream.of(taskId); }) .collect(Collectors.toList()); List<Observable<ResultSet>> observables = taskIds.stream().map(retrieveActiveTaskStatement::bind).map(this::execute).collect(Collectors.toList()); return Observable.merge(observables, getConcurrencyLimit()).flatMapIterable(tasksResultSet -> { List<Either<Task, Throwable>> tasks = new ArrayList<>(); for (Row row : tasksResultSet.all()) { String value = row.getString(0); String effectiveValue; if (fitBadDataInjection.isPresent()) { effectiveValue = fitBadDataInjection.get().afterImmediate(JobStoreFitAction.ErrorKind.CorruptedRawTaskRecords.name(), value); } else { effectiveValue = value; } Task task; try { task = deserializeTask(effectiveValue); if (!fitBadDataInjection.isPresent()) { tasks.add(Either.ofValue(task)); } else { Task effectiveTask = fitBadDataInjection.get().afterImmediate(JobStoreFitAction.ErrorKind.CorruptedTaskRecords.name(), task); effectiveTask = fitBadDataInjection.get().afterImmediate(JobStoreFitAction.ErrorKind.DuplicatedEni.name(), effectiveTask); effectiveTask = fitBadDataInjection.get().afterImmediate(JobStoreFitAction.ErrorKind.CorruptedTaskPlacementData.name(), effectiveTask); tasks.add(Either.ofValue(effectiveTask)); } transactionLogger().logAfterRead(retrieveActiveTaskStatement, "retrieveTasksForJob", task); } catch (Exception e) { logger.error("Cannot map serialized task data to Task class: {}", effectiveValue, e); tasks.add(Either.ofError(e)); } } return tasks; }); })).toList().map(taskErrorPairs -> { List<Task> tasks = taskErrorPairs.stream().filter(Either::hasValue).map(Either::getValue).collect(Collectors.toList()); int errors = (int) taskErrorPairs.stream().filter(Either::hasError).count(); return Pair.of(tasks, errors); }); } @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 testRetrieveTasksForJob() { 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); Task task = createTaskObject(job); store.storeTask(task).await(); Pair<List<Task>, Integer> tasks = store.retrieveTasksForJob(job.getId()).toBlocking().first(); assertThat(tasks.getLeft().get(0)).isEqualTo(task); Task archivedTask = store.retrieveArchivedTasksForJob(job.getId()).toBlocking().firstOrDefault(null); assertThat(archivedTask).isNull(); }