@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) { } }
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; }
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; } }
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; }  }
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); }
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);  }