Scheduler { public CompletableFuture<Void> createJob(@NonNull DyneinJobSpec jobSpec) { CompletableFuture<Void> ret = new CompletableFuture<>(); log.info("Create new job {}", jobSpec); String serializedJobSpec = jobSpecTransformer.serializeJobSpec(jobSpec); switch (jobSpec.getSchedulePolicy().getType()) { case IMMEDIATE: { ret = asyncClient .add(serializedJobSpec, jobSpec.getQueueName()) .whenComplete( (it, ex) -> { if (ex == null) { log.info( "DyneinJob {} triggered, enqueue to queue {}, type: IMMEDIATE", jobSpec.getJobToken(), jobSpec.getQueueName()); metrics.dispatchJob(jobSpec.getQueueName()); } else { log.error( "Failed to trigger DyneinJob {} to queue {}, type: IMMEDIATE", jobSpec.getJobToken(), jobSpec.getQueueName(), ex); } }); break; } case SCHEDULED: { ret = asyncClient .add(serializedJobSpec, inboundQueueName) .whenComplete( (it, ex) -> { if (ex == null) { log.info( "Added job spec {} to inbound job queue {}, with token {}", jobSpec.getName(), inboundQueueName, jobSpec.getJobToken()); metrics.enqueueToInboundQueue(jobSpec); } else { log.error( "Exception when adding job spec {} to inbound job queue {}, with token {}", jobSpec.getName(), inboundQueueName, jobSpec.getJobToken(), ex); } }); break; } case SQS_DELAYED: { long delayMs = TimeUtils.getDelayMillis(jobSpec.getSchedulePolicy(), clock); if (delayMs > TimeUnit.MINUTES.toMillis(15)) { ret.completeExceptionally( new IllegalArgumentException( "Delay " + delayMs + "ms is longer than 15 minutes, cannot be scheduled with SQS_DELAYED")); } else { ret = asyncClient .add( serializedJobSpec, jobSpec.getQueueName(), (int) TimeUnit.MILLISECONDS.toSeconds(delayMs)) .whenComplete( (it, ex) -> { if (ex == null) { log.info( "DyneinJob {} triggered, enqueue to queue {}, type: SQS_DELAYED", jobSpec.getJobToken(), jobSpec.getQueueName()); metrics.dispatchJob(jobSpec.getQueueName()); } else { log.error( "Failed to trigger DyneinJob {} to queue {}, type: SQS_DELAYED", jobSpec.getJobToken(), jobSpec.getQueueName(), ex); } }); } break; } default: ret.completeExceptionally( new UnsupportedOperationException("Unsupported Job Schedule Policy Type")); } return ret; }  String prepareJob(@NonNull PrepareJobRequest request); CompletableFuture<Void> createJob(@NonNull DyneinJobSpec jobSpec); CompletableFuture<Void> deleteJob(String token); CompletableFuture<DyneinJobSpec> getJob(String token);  }
@Test public void testScheduledJob() { DyneinJobSpec jobSpec = DyneinJobSpec.builder() .name("AddJob") .queueName("test-queue") .schedulePolicy( JobSchedulePolicy.builder() .type(JobScheduleType.SCHEDULED) .delayMillis(1000L) .build()) .build(); when(asyncClient.add(jobSpecTransformer.serializeJobSpec(jobSpec), "inbound-test")) .thenReturn(CompletableFuture.completedFuture(null)); CompletableFuture<Void> ret = scheduler.createJob(jobSpec); verify(asyncClient).add(jobSpecTransformer.serializeJobSpec(jobSpec), "inbound-test"); assertNull(ret.join()); }
@Test public void testImmediateJob() { when(asyncClient.add(any(String.class), eq("test-queue"))) .thenReturn(CompletableFuture.completedFuture(null)); DyneinJobSpec jobSpec = DyneinJobSpec.builder() .name("AddJob") .queueName("test-queue") .schedulePolicy( JobSchedulePolicy.builder().type(JobScheduleType.IMMEDIATE).delayMillis(0L).build()) .build(); CompletableFuture<Void> ret = scheduler.createJob(jobSpec); verify(asyncClient).add(jobSpecTransformer.serializeJobSpec(jobSpec), "test-queue"); assertNull(ret.join()); }
@Test public void testSQSDelayedJob_withInstant() { when(asyncClient.add(any(String.class), eq("test-queue"), any(Integer.class))) .thenReturn(CompletableFuture.completedFuture(null)); DyneinJobSpec jobSpec = DyneinJobSpec.builder() .name("AddJob") .queueName("test-queue") .schedulePolicy( JobSchedulePolicy.builder() .type(JobScheduleType.SQS_DELAYED) .delayMillis(5000L) .build()) .build(); CompletableFuture<Void> ret = scheduler.createJob(jobSpec); verify(asyncClient).add(jobSpecTransformer.serializeJobSpec(jobSpec), "test-queue", 5); assertNull(ret.join()); }
@Test public void testSQSDelayedJob_withDelay() { when(asyncClient.add(any(String.class), eq("test-queue"), any(Integer.class))) .thenReturn(CompletableFuture.completedFuture(null)); DyneinJobSpec jobSpec = DyneinJobSpec.builder() .name("AddJob") .queueName("test-queue") .schedulePolicy( JobSchedulePolicy.builder() .type(JobScheduleType.SQS_DELAYED) .epochMillis(Instant.now().plusMillis(5000L).toEpochMilli()) .build()) .build(); CompletableFuture<Void> ret = scheduler.createJob(jobSpec); verify(asyncClient).add(jobSpecTransformer.serializeJobSpec(jobSpec), "test-queue", 5); assertNull(ret.join()); }
@Test(expected = IllegalArgumentException.class) public void testSQSDelayedJob_DelayTooLong() throws Throwable { DyneinJobSpec jobSpec = DyneinJobSpec.builder() .name("AddJob") .queueName("error-queue") .schedulePolicy( JobSchedulePolicy.builder() .type(JobScheduleType.SQS_DELAYED) .epochMillis(Instant.now().plusMillis(1000000L).toEpochMilli()) .build()) .build(); try { scheduler.createJob(jobSpec).join(); } catch (Exception e) { Throwable t = e; while (t.getCause() != null) { t = t.getCause(); } throw t; } }
@Test public void testScheduledJobFailure() { DyneinJobSpec jobSpec = DyneinJobSpec.builder() .name("AddJob") .queueName("error-queue") .schedulePolicy( JobSchedulePolicy.builder() .type(JobScheduleType.SCHEDULED) .delayMillis(1000L) .build()) .build(); CompletableFuture<Void> error = new CompletableFuture<>(); error.completeExceptionally(new Exception()); when(asyncClient.add(jobSpecTransformer.serializeJobSpec(jobSpec), "inbound-test")) .thenReturn(error); CompletableFuture<Void> ret = scheduler.createJob(jobSpec); verify(asyncClient).add(jobSpecTransformer.serializeJobSpec(jobSpec), "inbound-test"); try { ret.get(1000, TimeUnit.MILLISECONDS); } catch (TimeoutException timeout) { fail("Future does not seem to complete when failure in adding to inbound queue (SCHEDULED)."); } catch (Exception ex) { } }
@Test public void testImmediateJobFailure() { DyneinJobSpec jobSpec = DyneinJobSpec.builder() .name("AddJob") .queueName("test-error") .schedulePolicy( JobSchedulePolicy.builder().type(JobScheduleType.IMMEDIATE).delayMillis(0L).build()) .build(); CompletableFuture<Void> error = new CompletableFuture<>(); error.completeExceptionally(new Exception()); when(asyncClient.add(jobSpecTransformer.serializeJobSpec(jobSpec), "test-error")) .thenReturn(error); CompletableFuture<Void> ret = scheduler.createJob(jobSpec); verify(asyncClient).add(jobSpecTransformer.serializeJobSpec(jobSpec), "test-error"); try { ret.get(1000, TimeUnit.MILLISECONDS); } catch (TimeoutException timeout) { fail( "Future does not seem to complete when failure in adding to destination queue (IMMEDIATE)."); } catch (Exception ex) { } }
@Test public void testSQSDelayedJobFailure() { DyneinJobSpec jobSpec = DyneinJobSpec.builder() .name("AddJob") .queueName("test-error") .schedulePolicy( JobSchedulePolicy.builder() .type(JobScheduleType.SQS_DELAYED) .epochMillis(Instant.now().plusMillis(5000L).toEpochMilli()) .build()) .build(); CompletableFuture<Void> error = new CompletableFuture<>(); error.completeExceptionally(new Exception()); when(asyncClient.add(jobSpecTransformer.serializeJobSpec(jobSpec), "test-error", 5)) .thenReturn(error); CompletableFuture<Void> ret = scheduler.createJob(jobSpec); verify(asyncClient).add(jobSpecTransformer.serializeJobSpec(jobSpec), "test-error", 5); try { ret.get(1000, TimeUnit.MILLISECONDS); } catch (TimeoutException timeout) { fail( "Future does not seem to complete when failure in adding to destination queue (SQS_DELAYED)."); } catch (Exception ex) { } }