class 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 public  CassandraJobStore(CassandraStoreConfiguration configuration,
                             Session session,
                             TitusRuntime titusRuntime);
      CassandraJobStore(CassandraStoreConfiguration configuration,
                      Session session,
                      TitusRuntime titusRuntime,
                      ObjectMapper mapper,
                      int initialBucketCount,
                      int maxBucketSize);

}