@Test public void get_dead_executions_should_not_include_previously_unresolved() { Instant now = Instant.now(); final Instant timeDied = now.minus(Duration.ofDays(5)); createDeadExecution(oneTimeTask.instance("id1"), timeDied); TaskResolver taskResolverMissingTask = new TaskResolver(testableRegistry); JdbcTaskRepository repoMissingTask = new JdbcTaskRepository(DB.getDataSource(), DEFAULT_TABLE_NAME, taskResolverMissingTask, new SchedulerName.Fixed(SCHEDULER_NAME)); assertThat(taskResolverMissingTask.getUnresolved(), hasSize(0)); assertEquals(0, testableRegistry.getCount(SchedulerStatsEvent.UNRESOLVED_TASK)); assertThat(repoMissingTask.getDeadExecutions(timeDied), hasSize(0)); assertThat(taskResolverMissingTask.getUnresolved(), hasSize(1)); assertEquals(1, testableRegistry.getCount(SchedulerStatsEvent.UNRESOLVED_TASK)); assertThat(repoMissingTask.getDeadExecutions(timeDied), hasSize(0)); assertThat(taskResolverMissingTask.getUnresolved(), hasSize(1)); assertEquals(1, testableRegistry.getCount(SchedulerStatsEvent.UNRESOLVED_TASK)); }
@Override public List<Execution> getDeadExecutions(Instant olderThan) { final UnresolvedFilter unresolvedFilter = new UnresolvedFilter(taskResolver.getUnresolved()); return jdbcRunner.query( "select * from " + tableName + " where picked = ? and last_heartbeat <= ? " + unresolvedFilter.andCondition() + " order by last_heartbeat asc", (PreparedStatement p) -> { int index = 1; p.setBoolean(index++, true); jdbcCustomization.setInstant(p, index++, olderThan); unresolvedFilter.setParameters(p, index); }, new ExecutionResultSetMapper() ); }
JdbcTaskRepository implements TaskRepository { @Override public List<Execution> getDeadExecutions(Instant olderThan) { final UnresolvedFilter unresolvedFilter = new UnresolvedFilter(taskResolver.getUnresolved()); return jdbcRunner.query( "select * from " + tableName + " where picked = ? and last_heartbeat <= ? " + unresolvedFilter.andCondition() + " order by last_heartbeat asc", (PreparedStatement p) -> { int index = 1; p.setBoolean(index++, true); jdbcCustomization.setInstant(p, index++, olderThan); unresolvedFilter.setParameters(p, index); }, new ExecutionResultSetMapper() ); } }
JdbcTaskRepository implements TaskRepository { @Override public List<Execution> getDeadExecutions(Instant olderThan) { final UnresolvedFilter unresolvedFilter = new UnresolvedFilter(taskResolver.getUnresolved()); return jdbcRunner.query( "select * from " + tableName + " where picked = ? and last_heartbeat <= ? " + unresolvedFilter.andCondition() + " order by last_heartbeat asc", (PreparedStatement p) -> { int index = 1; p.setBoolean(index++, true); jdbcCustomization.setInstant(p, index++, olderThan); unresolvedFilter.setParameters(p, index); }, new ExecutionResultSetMapper() ); } JdbcTaskRepository(DataSource dataSource, String tableName, TaskResolver taskResolver, SchedulerName schedulerSchedulerName);  JdbcTaskRepository(DataSource dataSource, JdbcCustomization jdbcCustomization, String tableName, TaskResolver taskResolver, SchedulerName schedulerSchedulerName);  JdbcTaskRepository(DataSource dataSource, JdbcCustomization jdbcCustomization, String tableName, TaskResolver taskResolver, SchedulerName schedulerSchedulerName, Serializer serializer); }
JdbcTaskRepository implements TaskRepository { @Override public List<Execution> getDeadExecutions(Instant olderThan) { final UnresolvedFilter unresolvedFilter = new UnresolvedFilter(taskResolver.getUnresolved()); return jdbcRunner.query( "select * from " + tableName + " where picked = ? and last_heartbeat <= ? " + unresolvedFilter.andCondition() + " order by last_heartbeat asc", (PreparedStatement p) -> { int index = 1; p.setBoolean(index++, true); jdbcCustomization.setInstant(p, index++, olderThan); unresolvedFilter.setParameters(p, index); }, new ExecutionResultSetMapper() ); } JdbcTaskRepository(DataSource dataSource, String tableName, TaskResolver taskResolver, SchedulerName schedulerSchedulerName);  JdbcTaskRepository(DataSource dataSource, JdbcCustomization jdbcCustomization, String tableName, TaskResolver taskResolver, SchedulerName schedulerSchedulerName);  JdbcTaskRepository(DataSource dataSource, JdbcCustomization jdbcCustomization, String tableName, TaskResolver taskResolver, SchedulerName schedulerSchedulerName, Serializer serializer); @Override @SuppressWarnings({"unchecked"}) boolean createIfNotExists(Execution execution); @Override void getScheduledExecutions(Consumer<Execution> consumer); @Override void getScheduledExecutions(String taskName, Consumer<Execution> consumer); @Override List<Execution> getDue(Instant now, int limit); @Override void remove(Execution execution); @Override boolean reschedule(Execution execution, Instant nextExecutionTime, Instant lastSuccess, Instant lastFailure, int consecutiveFailures); @Override boolean reschedule(Execution execution, Instant nextExecutionTime, Object newData, Instant lastSuccess, Instant lastFailure, int consecutiveFailures); @Override @SuppressWarnings({"unchecked"}) Optional<Execution> pick(Execution e, Instant timePicked); @Override List<Execution> getDeadExecutions(Instant olderThan); @Override void updateHeartbeat(Execution e, Instant newHeartbeat); @Override List<Execution> getExecutionsFailingLongerThan(Duration interval); Optional<Execution> getExecution(TaskInstance taskInstance); Optional<Execution> getExecution(String taskName, String taskInstanceId); @Override int removeExecutions(String taskName); }
JdbcTaskRepository implements TaskRepository { @Override public List<Execution> getDeadExecutions(Instant olderThan) { final UnresolvedFilter unresolvedFilter = new UnresolvedFilter(taskResolver.getUnresolved()); return jdbcRunner.query( "select * from " + tableName + " where picked = ? and last_heartbeat <= ? " + unresolvedFilter.andCondition() + " order by last_heartbeat asc", (PreparedStatement p) -> { int index = 1; p.setBoolean(index++, true); jdbcCustomization.setInstant(p, index++, olderThan); unresolvedFilter.setParameters(p, index); }, new ExecutionResultSetMapper() ); } JdbcTaskRepository(DataSource dataSource, String tableName, TaskResolver taskResolver, SchedulerName schedulerSchedulerName);  JdbcTaskRepository(DataSource dataSource, JdbcCustomization jdbcCustomization, String tableName, TaskResolver taskResolver, SchedulerName schedulerSchedulerName);  JdbcTaskRepository(DataSource dataSource, JdbcCustomization jdbcCustomization, String tableName, TaskResolver taskResolver, SchedulerName schedulerSchedulerName, Serializer serializer); @Override @SuppressWarnings({"unchecked"}) boolean createIfNotExists(Execution execution); @Override void getScheduledExecutions(Consumer<Execution> consumer); @Override void getScheduledExecutions(String taskName, Consumer<Execution> consumer); @Override List<Execution> getDue(Instant now, int limit); @Override void remove(Execution execution); @Override boolean reschedule(Execution execution, Instant nextExecutionTime, Instant lastSuccess, Instant lastFailure, int consecutiveFailures); @Override boolean reschedule(Execution execution, Instant nextExecutionTime, Object newData, Instant lastSuccess, Instant lastFailure, int consecutiveFailures); @Override @SuppressWarnings({"unchecked"}) Optional<Execution> pick(Execution e, Instant timePicked); @Override List<Execution> getDeadExecutions(Instant olderThan); @Override void updateHeartbeat(Execution e, Instant newHeartbeat); @Override List<Execution> getExecutionsFailingLongerThan(Duration interval); Optional<Execution> getExecution(TaskInstance taskInstance); Optional<Execution> getExecution(String taskName, String taskInstanceId); @Override int removeExecutions(String taskName); static final String DEFAULT_TABLE_NAME; }