JdbcTaskRepository implements TaskRepository { @Override @SuppressWarnings({"unchecked"}) public boolean createIfNotExists(Execution execution) { try { Optional<Execution> existingExecution = getExecution(execution.taskInstance); if (existingExecution.isPresent()) { LOG.debug("Execution not created, it already exists. Due: {}", existingExecution.get().executionTime); return false; } jdbcRunner.execute( "insert into " + tableName + "(task_name, task_instance, task_data, execution_time, picked, version) values(?, ?, ?, ?, ?, ?)", (PreparedStatement p) -> { p.setString(1, execution.taskInstance.getTaskName()); p.setString(2, execution.taskInstance.getId()); p.setObject(3, serializer.serialize(execution.taskInstance.getData())); jdbcCustomization.setInstant(p, 4, execution.executionTime); p.setBoolean(5, false); p.setLong(6, 1L); }); return true; } catch (SQLRuntimeException e) { LOG.debug("Exception when inserting execution. Assuming it to be a constraint violation.", e); Optional<Execution> existingExecution = getExecution(execution.taskInstance); if (!existingExecution.isPresent()) { throw new RuntimeException("Failed to add new execution.", e); } LOG.debug("Execution not created, another thread created it."); return false; } } 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; }
@Test public void test_createIfNotExists() { Instant now = Instant.now(); TaskInstance<Void> instance1 = oneTimeTask.instance("id1"); TaskInstance<Void> instance2 = oneTimeTask.instance("id2"); assertTrue(taskRepository.createIfNotExists(new Execution(now, instance1))); assertFalse(taskRepository.createIfNotExists(new Execution(now, instance1))); assertTrue(taskRepository.createIfNotExists(new Execution(now, instance2))); }