DatabaseSessionStoreManager extends BasicDatabaseStoreManager<DatabaseSessionStoreManager.Dao> implements SessionStoreManager { @SuppressWarnings("unchecked") @VisibleForTesting List<ArchivedTask> loadTaskArchive(String data) { try { return (List<ArchivedTask>) taskArchiveMapper.readValue(data, taskArchiveMapper.getTypeFactory().constructParametrizedType(List.class, List.class, ArchivedTask.class)); } catch (IOException ex) { throw new RuntimeException("Failed to load task archive", ex); } } @Inject  DatabaseSessionStoreManager(ConfigFactory cf, TransactionManager transactionManager, ConfigMapper cfm, ObjectMapper mapper, DatabaseConfig config); @Override SessionStore getSessionStore(int siteId); @Override Instant getStoreTime(); @Override int getSiteIdOfTask(long taskId); @Override StoredSessionAttemptWithSession getAttemptWithSessionById(long attemptId); @Override AttemptStateFlags getAttemptStateFlags(long attemptId); @Override boolean isAnyNotDoneAttempts(); @Override List<Long> findAllReadyTaskIds(int maxEntries); @Override List<StoredSessionAttempt> findActiveAttemptsCreatedBefore(Instant createdBefore, long lastId, int limit); @Override List<TaskAttemptSummary> findTasksStartedBeforeWithState(TaskStateCode[] states, Instant startedBefore, long lastId, int limit); @Override Optional<T> lockAttemptIfExists(long attemptId, AttemptLockAction<T> func); @Override List<TaskStateSummary> findRecentlyChangedTasks(Instant updatedSince, long lastId); @Override List<Long> findTasksByState(TaskStateCode state, long lastId); @Override List<TaskAttemptSummary> findRootTasksByStates(TaskStateCode[] states, long lastId); @Override List<Long> findDirectParentsOfBlockedTasks(long lastId); @Override boolean requestCancelAttempt(long attemptId); @Override int trySetRetryWaitingToReady(); @Override Optional<T> lockTaskIfExists(long taskId, TaskLockAction<T> func); @Override Optional<T> lockTaskIfNotLocked(long taskId, TaskLockAction<T> func); @Override Optional<T> lockTaskIfExists(long taskId, TaskLockActionWithDetails<T> func); @Override Optional<T> lockTaskIfNotLocked(long taskId, TaskLockActionWithDetails<T> func); @Override void lockReadySessionMonitors(Instant currentTime, SessionMonitorAction func); @Override List<TaskRelation> getTaskRelations(long attemptId); @Override List<Config> getExportParams(List<Long> idList); @Override List<ParameterUpdate> getStoreParams(List<Long> idList); @Override List<Config> getErrors(List<Long> idList); @Override void lockReadyDelayedAttempts(Instant currentTime, DelayedAttemptAction func);  }
@Test public void verifyStoredArchiveTaskBackwardCompatibility() { Config oldTask = newConfig() .set("parentId", null) .set("fullName", "+test") .set("config", newConfig() .set("local", newConfig().set("do", "that")) .set("export", newConfig().set("database", "mydb"))) .set("taskType", 0) .set("state", "success") .set("stateFlags", 8) .set("id", 132) .set("attemptId", 793) .set("upstreams", ImmutableList.of(8)) .set("updatedAt", "2017-01-24T12:06:01.993761631Z") .set("retryAt", null) .set("stateParams", newConfig()) .set("subtaskConfig", newConfig() .set("+sub", newConfig() .set("echo>", "a"))) .set("exportParams", newConfig() .set("database", "mydb")) .set("storeParams", newConfig() .set("field.id", 32)) .set("report", newConfig() .set("inputs", ImmutableList.of()) .set("outputs", ImmutableList.of())) .set("error", newConfig()) .set("ThisColumnIsRemovedAndShouldBeIgnored", 10) ; String data = "[" + oldTask.toString() + "]"; List<ArchivedTask> tasks = ((DatabaseSessionStoreManager) manager).loadTaskArchive(data); assertThat(tasks.size(), is(1)); assertThat(tasks.get(0), is( ImmutableArchivedTask.builder() .id(132L) .parentId(Optional.absent()) .attemptId(793L) .fullName("+test") .config(TaskConfig.validate( newConfig() .set("do", "that") .set("_export", newConfig() .set("database", "mydb")) )) .taskType(TaskType.of(0)) .state(TaskStateCode.SUCCESS) .stateFlags(TaskStateFlags.empty().withInitialTask()) .upstreams(ImmutableList.of(8L)) .updatedAt(Instant.parse("2017-01-24T12:06:01.993761631Z")) .retryAt(Optional.absent()) .startedAt(Optional.absent()) .stateParams(newConfig()) .retryCount(0) .subtaskConfig(newConfig().set("+sub", newConfig().set("echo>", "a"))) .resetStoreParams(ImmutableList.of()) .exportParams(newConfig().set("database", "mydb")) .storeParams(newConfig().set("field.id", 32)) .report(Optional.of(TaskReport.empty())) .error(newConfig()) .resumingTaskId(Optional.absent()) .build() )); }