@Test public void verifyRetries() throws Exception { TDOperator operator = new TDOperator(client, "foobar"); String jobStateKey = "fooJob"; Config state = configFactory.create(); String domainKey; { TaskExecutionException e = runJobIteration(operator, state, jobStateKey, pollInterval, retryInterval, jobStarter); verifyZeroInteractions(jobStarter); state = e.getStateParams(configFactory).get(); assertThat(state.has(jobStateKey), is(true)); TDOperator.JobState jobState = state.get(jobStateKey, TDOperator.JobState.class); domainKey = jobState.domainKey().get(); } int errorPollIteration = 0; for (int i = 0; i < 7; i++) { { errorPollIteration++; reset(jobStarter); when(jobStarter.startJob(any(TDOperator.class), anyString())) .thenThrow(new TDClientException(TDClientException.ErrorType.EXECUTION_FAILURE, "Error")); TaskExecutionException e = runJobIteration(operator, state, jobStateKey, pollInterval, retryInterval, jobStarter); verify(jobStarter).startJob(operator, domainKey); state = e.getStateParams(configFactory).get(); TDOperator.JobState jobState = state.get(jobStateKey, TDOperator.JobState.class); assertThat(jobState.domainKey(), is(Optional.of(domainKey))); assertThat(jobState.pollIteration(), is(Optional.absent())); assertThat(jobState.errorPollIteration(), is(Optional.of(errorPollIteration))); } { errorPollIteration++; reset(jobStarter); when(jobStarter.startJob(any(TDOperator.class), anyString())) .thenThrow(new TDClientHttpException(TDClientException.ErrorType.SERVER_ERROR, "Service Unavailable", 503, null)); TaskExecutionException e = runJobIteration(operator, state, jobStateKey, pollInterval, retryInterval, jobStarter); verify(jobStarter).startJob(operator, domainKey); state = e.getStateParams(configFactory).get(); TDOperator.JobState jobState = state.get(jobStateKey, TDOperator.JobState.class); assertThat(jobState.domainKey(), is(Optional.of(domainKey))); assertThat(jobState.pollIteration(), is(Optional.absent())); assertThat(jobState.errorPollIteration(), is(Optional.of(errorPollIteration))); } } String jobId = "badf00d4711"; { errorPollIteration = 0; reset(jobStarter); when(jobStarter.startJob(any(TDOperator.class), anyString())).thenReturn(jobId); TaskExecutionException e = runJobIteration(operator, state, jobStateKey, pollInterval, retryInterval, jobStarter); verify(jobStarter).startJob(operator, domainKey); state = e.getStateParams(configFactory).get(); TDOperator.JobState jobState = state.get(jobStateKey, TDOperator.JobState.class); assertThat(jobState.pollIteration(), is(Optional.absent())); assertThat(jobState.errorPollIteration(), is(Optional.absent())); assertThat(jobState.jobId(), is(Optional.of(jobId))); } Optional<Integer> pollIteration = Optional.absent(); TDJob.Status[] statuses = {TDJob.Status.QUEUED, TDJob.Status.RUNNING}; for (TDJob.Status status : statuses) { for (int i = 0; i < 2; i++) { { errorPollIteration++; reset(client); when(client.jobStatus(jobId)) .thenThrow(new TDClientException(TDClientException.ErrorType.EXECUTION_FAILURE, "Error")); TaskExecutionException e = runJobIteration(operator, state, jobStateKey, pollInterval, retryInterval, jobStarter); verify(client, times(4)).jobStatus(jobId); state = e.getStateParams(configFactory).get(); TDOperator.JobState jobState = state.get(jobStateKey, TDOperator.JobState.class); assertThat(jobState.domainKey(), is(Optional.of(domainKey))); assertThat(jobState.pollIteration(), is(pollIteration)); assertThat(jobState.errorPollIteration(), is(Optional.of(errorPollIteration))); } { errorPollIteration++; reset(client); when(client.jobStatus(jobId)) .thenThrow(new TDClientHttpException(TDClientException.ErrorType.SERVER_ERROR, "Service Unavailable", 503, null)); TaskExecutionException e = runJobIteration(operator, state, jobStateKey, pollInterval, retryInterval, jobStarter); verify(client, times(4)).jobStatus(jobId); state = e.getStateParams(configFactory).get(); TDOperator.JobState jobState = state.get(jobStateKey, TDOperator.JobState.class); assertThat(jobState.domainKey(), is(Optional.of(domainKey))); assertThat(jobState.pollIteration(), is(pollIteration)); assertThat(jobState.errorPollIteration(), is(Optional.of(errorPollIteration))); } } { pollIteration = pollIteration.transform(it -> it + 1).or(Optional.of(1)); errorPollIteration = 0; reset(client); when(client.jobStatus(jobId)).thenReturn(summary(jobId, status)); TaskExecutionException e = runJobIteration(operator, state, jobStateKey, pollInterval, retryInterval, jobStarter); verify(client).jobStatus(jobId); state = e.getStateParams(configFactory).get(); TDOperator.JobState jobState = state.get(jobStateKey, TDOperator.JobState.class); assertThat(jobState.pollIteration(), is(pollIteration)); } } when(client.jobStatus(jobId)).thenReturn(summary(jobId, TDJob.Status.SUCCESS)); TDJobOperator jobOperator = operator.runJob(TaskState.of(state), 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();  }