in genie-client/src/integTest/java/com/netflix/genie/client/JobClientIntegrationTest.java [64:356]
void canSubmitJob() throws Exception {
final String dummyClusterId = this.createDummyCluster();
final String sleepCommandId = this.createSleepCommand();
final String dateCommandId = this.createDateCommand();
final String echoCommandId = this.createEchoCommand();
final List<ClusterCriteria> clusterCriteriaList = Lists.newArrayList(
new ClusterCriteria(Sets.newHashSet(DUMMY_TAG))
);
final JobRequest sleepJob = new JobRequest.Builder(
UUID.randomUUID().toString(),
UUID.randomUUID().toString(),
UUID.randomUUID().toString(),
clusterCriteriaList,
Sets.newHashSet(SLEEP_TAG)
)
.withCommandArgs(Lists.newArrayList("1"))
.withDisableLogArchival(true)
.build();
final JobRequest killJob = new JobRequest.Builder(
UUID.randomUUID().toString(),
UUID.randomUUID().toString(),
UUID.randomUUID().toString(),
clusterCriteriaList,
Sets.newHashSet(SLEEP_TAG)
)
.withCommandArgs(Lists.newArrayList("60"))
.withDisableLogArchival(true)
.build();
final JobRequest timeoutJob = new JobRequest.Builder(
UUID.randomUUID().toString(),
UUID.randomUUID().toString(),
UUID.randomUUID().toString(),
clusterCriteriaList,
Sets.newHashSet(SLEEP_TAG)
)
.withCommandArgs(Lists.newArrayList("60"))
.withTimeout(1)
.withDisableLogArchival(true)
.build();
final JobRequest dateJob = new JobRequest.Builder(
UUID.randomUUID().toString(),
UUID.randomUUID().toString(),
UUID.randomUUID().toString(),
clusterCriteriaList,
Sets.newHashSet(DATE_TAG)
)
.withDisableLogArchival(true)
.build();
final JobRequest echoJob = new JobRequest.Builder(
UUID.randomUUID().toString(),
UUID.randomUUID().toString(),
UUID.randomUUID().toString(),
clusterCriteriaList,
Sets.newHashSet(ECHO_TAG)
)
.withCommandArgs(Lists.newArrayList("hello"))
.withDisableLogArchival(true)
.build();
final String sleepJobId;
final byte[] attachmentBytes = UUID.randomUUID().toString().getBytes(StandardCharsets.UTF_8);
try (ByteArrayInputStream bis = new ByteArrayInputStream(attachmentBytes)) {
final Map<String, InputStream> attachments = ImmutableMap.of("attachment.txt", bis);
sleepJobId = this.jobClient.submitJobWithAttachments(sleepJob, attachments);
}
final String killJobId = this.jobClient.submitJob(killJob);
final Thread killThread = new Thread(
() -> {
try {
while (this.jobClient.getJobStatus(killJobId) != JobStatus.RUNNING) {
Thread.sleep(10);
}
this.jobClient.killJob(killJobId);
} catch (final Exception e) {
Assertions.fail(e.getMessage(), e);
}
}
);
killThread.start();
final String timeoutJobId = this.jobClient.submitJob(timeoutJob);
final String dateJobId = this.jobClient.submitJob(dateJob);
final String echoJobId = this.jobClient.submitJob(echoJob);
final Map<String, JobStatus> expectedStatuses = ImmutableMap.<String, JobStatus>builder()
.put(sleepJobId, JobStatus.SUCCEEDED)
.put(killJobId, JobStatus.KILLED)
.put(timeoutJobId, JobStatus.KILLED)
.put(dateJobId, JobStatus.SUCCEEDED)
.put(echoJobId, JobStatus.SUCCEEDED)
.build();
final long waitStart = System.currentTimeMillis();
final long maxTotalWait = 120000;
for (final Map.Entry<String, JobStatus> entry : expectedStatuses.entrySet()) {
final String jobId = entry.getKey();
final JobStatus status = entry.getValue();
log.info("Waiting for job: {} (expected final status: {})", jobId, status.name());
final long timeElapsed = System.currentTimeMillis() - waitStart;
final long timeLeft = maxTotalWait - timeElapsed;
if (timeLeft <= 0) {
throw new TimeoutException("Timed out waiting for jobs to complete");
}
Assertions
.assertThat(this.jobClient.waitForCompletion(jobId, timeLeft, 100))
.isEqualByComparingTo(entry.getValue());
}
// Some basic checking of fields
Assertions.assertThat(this.jobClient.getJob(sleepJobId).getName()).isEqualTo(sleepJob.getName());
Assertions.assertThat(this.jobClient.getJobRequest(sleepJobId).getUser()).isEqualTo(sleepJob.getUser());
Assertions.assertThat(this.jobClient.getJobExecution(sleepJobId)).isNotNull();
Assertions.assertThat(this.jobClient.getJobMetadata(sleepJobId)).isNotNull();
Assertions.assertThat(this.jobClient.getJobCluster(sleepJobId).getId()).isPresent().contains(dummyClusterId);
Assertions.assertThat(this.jobClient.getJobCommand(sleepJobId).getId()).isPresent().contains(sleepCommandId);
Assertions.assertThat(this.jobClient.getJobCommand(dateJobId).getId()).isPresent().contains(dateCommandId);
Assertions.assertThat(this.jobClient.getJobCommand(echoJobId).getId()).isPresent().contains(echoCommandId);
Assertions.assertThat(this.jobClient.getJobApplications(sleepJobId)).isEmpty();
Assertions
.assertThat(IOUtils.toString(this.jobClient.getJobStdout(echoJobId), StandardCharsets.UTF_8))
.isEqualTo("hello\n");
Assertions
.assertThat(IOUtils.toString(this.jobClient.getJobStdout(echoJobId, null, null), StandardCharsets.UTF_8))
.isEqualTo("hello\n");
Assertions
.assertThat(IOUtils.toString(this.jobClient.getJobStdout(echoJobId, 4L, null), StandardCharsets.UTF_8))
.isEqualTo("o\n");
Assertions
.assertThat(IOUtils.toString(this.jobClient.getJobStdout(echoJobId, 0L, 3L), StandardCharsets.UTF_8))
.isEqualTo("hell");
Assertions
.assertThat(IOUtils.toString(this.jobClient.getJobStdout(echoJobId, null, 2L), StandardCharsets.UTF_8))
.isEqualTo("o\n");
Assertions
.assertThat(IOUtils.toString(this.jobClient.getJobStderr(echoJobId), StandardCharsets.UTF_8))
.isBlank();
Assertions
.assertThat(IOUtils.toString(
this.jobClient.getJobOutputFile(echoJobId, "stdout"), StandardCharsets.UTF_8))
.isEqualTo("hello\n");
Assertions
.assertThat(IOUtils.toString(
this.jobClient.getJobOutputFile(echoJobId, "run"), StandardCharsets.UTF_8))
.isNotBlank();
Assertions
.assertThat(IOUtils.toString(
this.jobClient.getJobOutputFile(echoJobId, ""), StandardCharsets.UTF_8))
.isNotBlank();
Assertions
.assertThat(IOUtils.toString(
this.jobClient.getJobOutputFile(echoJobId, null), StandardCharsets.UTF_8))
.isNotBlank();
// Some quick find jobs calls
Assertions
.assertThat(this.jobClient.getJobs())
.extracting(JobSearchResult::getId)
.containsExactlyInAnyOrder(sleepJobId, killJobId, timeoutJobId, dateJobId, echoJobId);
Assertions
.assertThat(
this.jobClient.getJobs(
null,
null,
null,
null,
null,
null,
null,
null,
echoCommandId,
null,
null,
null,
null,
null,
null
)
)
.extracting(JobSearchResult::getId)
.containsExactlyInAnyOrder(echoJobId);
Assertions
.assertThat(
this.jobClient.getJobs(
null,
null,
null,
Sets.newHashSet(JobStatus.KILLED.name()),
null,
null,
null,
null,
null,
null,
null,
null,
null
)
)
.extracting(JobSearchResult::getId)
.containsExactlyInAnyOrder(killJobId, timeoutJobId);
final List<String> ids = Lists.newArrayList(sleepJobId, killJobId, timeoutJobId, dateJobId, echoJobId);
// Paginate, 1 result per page
for (int i = 0; i < ids.size(); i++) {
final List<JobSearchResult> page = this.jobClient.getJobs(
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
1,
SortAttribute.CREATED,
SortDirection.ASC,
i
);
Assertions.assertThat(page.size()).isEqualTo(1);
Assertions.assertThat(page.get(0).getId()).isEqualTo(ids.get(i));
}
// Paginate, 1 result per page, reverse order
Collections.reverse(ids);
for (int i = 0; i < ids.size(); i++) {
final List<JobSearchResult> page = this.jobClient.getJobs(
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
1,
SortAttribute.CREATED,
SortDirection.DESC,
i
);
Assertions.assertThat(page.size()).isEqualTo(1);
Assertions.assertThat(page.get(0).getId()).isEqualTo(ids.get(i));
}
// Ask for page beyond end of results
Assertions.assertThat(
this.jobClient.getJobs(
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
10,
SortAttribute.CREATED,
SortDirection.DESC,
1
)
).isEmpty();
}