@Test public void testRunJob() throws Exception { TDOperator operator = new TDOperator(client, "foobar"); Config state0 = configFactory.create(); String jobStateKey = "fooJob"; TDOperator.JobState jobState1; Config state1; { TaskExecutionException e = runJobIteration(operator, state0, jobStateKey, pollInterval, retryInterval, jobStarter); verifyZeroInteractions(jobStarter); state1 = e.getStateParams(configFactory).get(); assertThat(state1.has(jobStateKey), is(true)); jobState1 = state1.get(jobStateKey, TDOperator.JobState.class); assertThat(jobState1.domainKey().isPresent(), is(true)); } String jobId = "badf00d4711"; Config state2; TDOperator.JobState jobState2; { when(jobStarter.startJob(any(TDOperator.class), anyString())).thenReturn(jobId); TaskExecutionException e = runJobIteration(operator, state1, jobStateKey, pollInterval, retryInterval, jobStarter); state2 = e.getStateParams(configFactory).get(); verify(jobStarter).startJob(operator, jobState1.domainKey().get()); jobState2 = state2.get(jobStateKey, TDOperator.JobState.class); assertThat(jobState2.jobId(), is(Optional.of(jobId))); } Config state3; TDOperator.JobState jobState3; { when(client.jobStatus(jobId)).thenReturn(summary(jobId, TDJob.Status.RUNNING)); TaskExecutionException e = runJobIteration(operator, state2, jobStateKey, pollInterval, retryInterval, jobStarter); state3 = e.getStateParams(configFactory).get(); jobState3 = state3.get(jobStateKey, TDOperator.JobState.class); assertThat(jobState3.pollIteration(), is(Optional.of(1))); } when(client.jobStatus(jobId)).thenReturn(summary(jobId, TDJob.Status.SUCCESS)); TDJobOperator jobOperator = operator.runJob(TaskState.of(state3), jobStateKey, pollInterval, retryInterval, jobStarter); assertThat(jobOperator.getJobId(), is(jobId)); verifyNoMoreInteractions(jobStarter); }
public TDJobOperator runJob(TaskState state, String key, DurationInterval pollInterval, DurationInterval retryInterval, JobStarter starter) { if (state.params().has("jobId")) { Config jobState = state.params().getNestedOrSetEmpty(key); if (!jobState.isEmpty()) { throw new AssertionError(); } jobState.setOptional("jobId", state.params().getOptional("jobId", String.class)); jobState.setOptional("domainKey", state.params().getOptional("domainKey", String.class)); jobState.setOptional("pollIteration", state.params().getOptional("pollIteration", Integer.class)); state.params().remove("jobId"); state.params().remove("domainKey"); state.params().remove("pollIteration"); } JobState jobState = state.params().get(key, JobState.class, JobState.empty()); Optional<String> domainKey = jobState.domainKey(); if (!domainKey.isPresent()) { state.params().set(key, jobState.withDomainKey(UUID.randomUUID().toString())); throw state.pollingTaskExecutionException(0); } Optional<String> jobId = jobState.jobId(); if (!jobId.isPresent()) { assert domainKey.isPresent(); String newJobId; try { newJobId = starter.startJob(this, domainKey.get()); } catch (TDClientException e) { logger.warn("failed to start job: domainKey={}", domainKey.get(), e); if (isDeterministicClientException(e)) { throw e; } throw errorPollingException(state, key, jobState, retryInterval); } jobState = jobState.withErrorPollIteration(Optional.absent()); state.params().set(key, jobState.withJobId(newJobId)); throw state.pollingTaskExecutionException((int) pollInterval.min().getSeconds()); } TDJobOperator job = newJobOperator(jobId.get()); TDJobSummary status; try { status = job.checkStatus(); } catch (TDClientException e) { logger.warn("failed to check job status: domainKey={}, jobId={}", domainKey.get(), jobId.get(), e); if (isDeterministicClientException(e)) { throw e; } throw errorPollingException(state, key, jobState, retryInterval); } jobState = jobState.withErrorPollIteration(Optional.absent()); if (!status.getStatus().isFinished()) { throw pollingException(state, key, jobState, pollInterval); } if (status.getStatus() != SUCCESS) { TDJob jobInfo; try { jobInfo = job.getJobInfo(); } catch (TDClientException e) { logger.warn("failed to get job failure info: domainKey={}, jobId={}, status={}", domainKey.get(), jobId.get(), status.getStatus(), e); if (isDeterministicClientException(e)) { throw e; } throw errorPollingException(state, key, jobState, retryInterval); } String message = jobInfo.getCmdOut() + "\n" + jobInfo.getStdErr(); throw new TaskExecutionException(message); } return job; }
TDOperator implements Closeable { public TDJobOperator runJob(TaskState state, String key, DurationInterval pollInterval, DurationInterval retryInterval, JobStarter starter) { if (state.params().has("jobId")) { Config jobState = state.params().getNestedOrSetEmpty(key); if (!jobState.isEmpty()) { throw new AssertionError(); } jobState.setOptional("jobId", state.params().getOptional("jobId", String.class)); jobState.setOptional("domainKey", state.params().getOptional("domainKey", String.class)); jobState.setOptional("pollIteration", state.params().getOptional("pollIteration", Integer.class)); state.params().remove("jobId"); state.params().remove("domainKey"); state.params().remove("pollIteration"); } JobState jobState = state.params().get(key, JobState.class, JobState.empty()); Optional<String> domainKey = jobState.domainKey(); if (!domainKey.isPresent()) { state.params().set(key, jobState.withDomainKey(UUID.randomUUID().toString())); throw state.pollingTaskExecutionException(0); } Optional<String> jobId = jobState.jobId(); if (!jobId.isPresent()) { assert domainKey.isPresent(); String newJobId; try { newJobId = starter.startJob(this, domainKey.get()); } catch (TDClientException e) { logger.warn("failed to start job: domainKey={}", domainKey.get(), e); if (isDeterministicClientException(e)) { throw e; } throw errorPollingException(state, key, jobState, retryInterval); } jobState = jobState.withErrorPollIteration(Optional.absent()); state.params().set(key, jobState.withJobId(newJobId)); throw state.pollingTaskExecutionException((int) pollInterval.min().getSeconds()); } TDJobOperator job = newJobOperator(jobId.get()); TDJobSummary status; try { status = job.checkStatus(); } catch (TDClientException e) { logger.warn("failed to check job status: domainKey={}, jobId={}", domainKey.get(), jobId.get(), e); if (isDeterministicClientException(e)) { throw e; } throw errorPollingException(state, key, jobState, retryInterval); } jobState = jobState.withErrorPollIteration(Optional.absent()); if (!status.getStatus().isFinished()) { throw pollingException(state, key, jobState, pollInterval); } if (status.getStatus() != SUCCESS) { TDJob jobInfo; try { jobInfo = job.getJobInfo(); } catch (TDClientException e) { logger.warn("failed to get job failure info: domainKey={}, jobId={}, status={}", domainKey.get(), jobId.get(), status.getStatus(), e); if (isDeterministicClientException(e)) { throw e; } throw errorPollingException(state, key, jobState, retryInterval); } String message = jobInfo.getCmdOut() + "\n" + jobInfo.getStdErr(); throw new TaskExecutionException(message); } return job; } }
TDOperator implements Closeable { public TDJobOperator runJob(TaskState state, String key, DurationInterval pollInterval, DurationInterval retryInterval, JobStarter starter) { if (state.params().has("jobId")) { Config jobState = state.params().getNestedOrSetEmpty(key); if (!jobState.isEmpty()) { throw new AssertionError(); } jobState.setOptional("jobId", state.params().getOptional("jobId", String.class)); jobState.setOptional("domainKey", state.params().getOptional("domainKey", String.class)); jobState.setOptional("pollIteration", state.params().getOptional("pollIteration", Integer.class)); state.params().remove("jobId"); state.params().remove("domainKey"); state.params().remove("pollIteration"); } JobState jobState = state.params().get(key, JobState.class, JobState.empty()); Optional<String> domainKey = jobState.domainKey(); if (!domainKey.isPresent()) { state.params().set(key, jobState.withDomainKey(UUID.randomUUID().toString())); throw state.pollingTaskExecutionException(0); } Optional<String> jobId = jobState.jobId(); if (!jobId.isPresent()) { assert domainKey.isPresent(); String newJobId; try { newJobId = starter.startJob(this, domainKey.get()); } catch (TDClientException e) { logger.warn("failed to start job: domainKey={}", domainKey.get(), e); if (isDeterministicClientException(e)) { throw e; } throw errorPollingException(state, key, jobState, retryInterval); } jobState = jobState.withErrorPollIteration(Optional.absent()); state.params().set(key, jobState.withJobId(newJobId)); throw state.pollingTaskExecutionException((int) pollInterval.min().getSeconds()); } TDJobOperator job = newJobOperator(jobId.get()); TDJobSummary status; try { status = job.checkStatus(); } catch (TDClientException e) { logger.warn("failed to check job status: domainKey={}, jobId={}", domainKey.get(), jobId.get(), e); if (isDeterministicClientException(e)) { throw e; } throw errorPollingException(state, key, jobState, retryInterval); } jobState = jobState.withErrorPollIteration(Optional.absent()); if (!status.getStatus().isFinished()) { throw pollingException(state, key, jobState, pollInterval); } if (status.getStatus() != SUCCESS) { TDJob jobInfo; try { jobInfo = job.getJobInfo(); } catch (TDClientException e) { logger.warn("failed to get job failure info: domainKey={}, jobId={}, status={}", domainKey.get(), jobId.get(), status.getStatus(), e); if (isDeterministicClientException(e)) { throw e; } throw errorPollingException(state, key, jobState, retryInterval); } String message = jobInfo.getCmdOut() + "\n" + jobInfo.getStdErr(); throw new TaskExecutionException(message); } return job; } TDOperator(TDClient client, String database); }
TDOperator implements Closeable { public TDJobOperator runJob(TaskState state, String key, DurationInterval pollInterval, DurationInterval retryInterval, JobStarter starter) { if (state.params().has("jobId")) { Config jobState = state.params().getNestedOrSetEmpty(key); if (!jobState.isEmpty()) { throw new AssertionError(); } jobState.setOptional("jobId", state.params().getOptional("jobId", String.class)); jobState.setOptional("domainKey", state.params().getOptional("domainKey", String.class)); jobState.setOptional("pollIteration", state.params().getOptional("pollIteration", Integer.class)); state.params().remove("jobId"); state.params().remove("domainKey"); state.params().remove("pollIteration"); } JobState jobState = state.params().get(key, JobState.class, JobState.empty()); Optional<String> domainKey = jobState.domainKey(); if (!domainKey.isPresent()) { state.params().set(key, jobState.withDomainKey(UUID.randomUUID().toString())); throw state.pollingTaskExecutionException(0); } Optional<String> jobId = jobState.jobId(); if (!jobId.isPresent()) { assert domainKey.isPresent(); String newJobId; try { newJobId = starter.startJob(this, domainKey.get()); } catch (TDClientException e) { logger.warn("failed to start job: domainKey={}", domainKey.get(), e); if (isDeterministicClientException(e)) { throw e; } throw errorPollingException(state, key, jobState, retryInterval); } jobState = jobState.withErrorPollIteration(Optional.absent()); state.params().set(key, jobState.withJobId(newJobId)); throw state.pollingTaskExecutionException((int) pollInterval.min().getSeconds()); } TDJobOperator job = newJobOperator(jobId.get()); TDJobSummary status; try { status = job.checkStatus(); } catch (TDClientException e) { logger.warn("failed to check job status: domainKey={}, jobId={}", domainKey.get(), jobId.get(), e); if (isDeterministicClientException(e)) { throw e; } throw errorPollingException(state, key, jobState, retryInterval); } jobState = jobState.withErrorPollIteration(Optional.absent()); if (!status.getStatus().isFinished()) { throw pollingException(state, key, jobState, pollInterval); } if (status.getStatus() != SUCCESS) { TDJob jobInfo; try { jobInfo = job.getJobInfo(); } catch (TDClientException e) { logger.warn("failed to get job failure info: domainKey={}, jobId={}, status={}", domainKey.get(), jobId.get(), status.getStatus(), e); if (isDeterministicClientException(e)) { throw e; } throw errorPollingException(state, key, jobState, retryInterval); } String message = jobInfo.getCmdOut() + "\n" + jobInfo.getStdErr(); throw new TaskExecutionException(message); } return job; } TDOperator(TDClient client, String database); static TDOperator fromConfig(SystemDefaultConfig systemDefaultConfig, Map<String, String> env, Config config, SecretProvider secrets); static String escapeHiveIdent(String ident); static String escapePrestoIdent(String ident); static String escapeHiveTableName(TableParam table); static String escapePrestoTableName(TableParam table); TDOperator withDatabase(String anotherDatabase); String getDatabase(); void ensureDatabaseCreated(String name); void ensureDatabaseDeleted(String name); void ensureTableCreated(String tableName); void ensureTableDeleted(String tableName); void ensureExistentTableRenamed(String existentTable, String toName); boolean tableExists(String table); long lookupConnection(String name); String submitNewJobWithRetry(TDJobRequest req); String submitNewJobWithRetry(Submitter submitter); TDJobOperator newJobOperator(String jobId); TDJobOperator runJob(TaskState state, String key, DurationInterval pollInterval, DurationInterval retryInterval, JobStarter starter); @Override void close(); }
TDOperator implements Closeable { public TDJobOperator runJob(TaskState state, String key, DurationInterval pollInterval, DurationInterval retryInterval, JobStarter starter) { if (state.params().has("jobId")) { Config jobState = state.params().getNestedOrSetEmpty(key); if (!jobState.isEmpty()) { throw new AssertionError(); } jobState.setOptional("jobId", state.params().getOptional("jobId", String.class)); jobState.setOptional("domainKey", state.params().getOptional("domainKey", String.class)); jobState.setOptional("pollIteration", state.params().getOptional("pollIteration", Integer.class)); state.params().remove("jobId"); state.params().remove("domainKey"); state.params().remove("pollIteration"); } JobState jobState = state.params().get(key, JobState.class, JobState.empty()); Optional<String> domainKey = jobState.domainKey(); if (!domainKey.isPresent()) { state.params().set(key, jobState.withDomainKey(UUID.randomUUID().toString())); throw state.pollingTaskExecutionException(0); } Optional<String> jobId = jobState.jobId(); if (!jobId.isPresent()) { assert domainKey.isPresent(); String newJobId; try { newJobId = starter.startJob(this, domainKey.get()); } catch (TDClientException e) { logger.warn("failed to start job: domainKey={}", domainKey.get(), e); if (isDeterministicClientException(e)) { throw e; } throw errorPollingException(state, key, jobState, retryInterval); } jobState = jobState.withErrorPollIteration(Optional.absent()); state.params().set(key, jobState.withJobId(newJobId)); throw state.pollingTaskExecutionException((int) pollInterval.min().getSeconds()); } TDJobOperator job = newJobOperator(jobId.get()); TDJobSummary status; try { status = job.checkStatus(); } catch (TDClientException e) { logger.warn("failed to check job status: domainKey={}, jobId={}", domainKey.get(), jobId.get(), e); if (isDeterministicClientException(e)) { throw e; } throw errorPollingException(state, key, jobState, retryInterval); } jobState = jobState.withErrorPollIteration(Optional.absent()); if (!status.getStatus().isFinished()) { throw pollingException(state, key, jobState, pollInterval); } if (status.getStatus() != SUCCESS) { TDJob jobInfo; try { jobInfo = job.getJobInfo(); } catch (TDClientException e) { logger.warn("failed to get job failure info: domainKey={}, jobId={}, status={}", domainKey.get(), jobId.get(), status.getStatus(), e); if (isDeterministicClientException(e)) { throw e; } throw errorPollingException(state, key, jobState, retryInterval); } String message = jobInfo.getCmdOut() + "\n" + jobInfo.getStdErr(); throw new TaskExecutionException(message); } return job; } TDOperator(TDClient client, String database); static TDOperator fromConfig(SystemDefaultConfig systemDefaultConfig, Map<String, String> env, Config config, SecretProvider secrets); static String escapeHiveIdent(String ident); static String escapePrestoIdent(String ident); static String escapeHiveTableName(TableParam table); static String escapePrestoTableName(TableParam table); TDOperator withDatabase(String anotherDatabase); String getDatabase(); void ensureDatabaseCreated(String name); void ensureDatabaseDeleted(String name); void ensureTableCreated(String tableName); void ensureTableDeleted(String tableName); void ensureExistentTableRenamed(String existentTable, String toName); boolean tableExists(String table); long lookupConnection(String name); String submitNewJobWithRetry(TDJobRequest req); String submitNewJobWithRetry(Submitter submitter); TDJobOperator newJobOperator(String jobId); TDJobOperator runJob(TaskState state, String key, DurationInterval pollInterval, DurationInterval retryInterval, JobStarter starter); @Override void close();  }