@Test public void testTaskCancelBeforeRunFailure() throws Exception { InMemorySingleTaskRunner inMemorySingleTaskRunner = createInMemoryTaskRunner(); inMemorySingleTaskRunner.initClusterSingleTask(false); try { inMemorySingleTaskRunner.task.cancel(); } catch (Exception e) { Assert.assertTrue(e instanceof IllegalStateException); Assert.assertTrue(e.getMessage().contains("Failed to initialize")); } }
public void cancel() { int retryCount = 0 ; int maxRetry = ConfigUtils.getInt(_dynamicConfig, MAX_RETRY_WAITING_FOR_INIT_KEY, DEFAULT_MAX_RETRY_WAITING_FOR_INIT); try { _lock.lock(); try { while (_taskAttempt == null) { if (!_taskAttemptBuilt.await(5, TimeUnit.SECONDS) && ++retryCount > maxRetry) { throw new IllegalStateException("Failed to initialize taskAttempt object before cancel"); } } } finally { _lock.unlock(); } if (_taskAttempt != null) { _logger.info("Task cancelled: Shutdown starting for tasks with jobId: {}", _jobId); _taskAttempt.shutdownTasks(); _logger.info("Task cancelled: Shutdown complete for tasks with jobId: {}", _jobId); } else { throw new IllegalStateException("TaskAttempt not initialized while passing conditional barrier"); } } catch (InterruptedException e) { throw new RuntimeException("Interrupted while shutting down task with jobId: " + _jobId, e); } }
SingleTask { public void cancel() { int retryCount = 0 ; int maxRetry = ConfigUtils.getInt(_dynamicConfig, MAX_RETRY_WAITING_FOR_INIT_KEY, DEFAULT_MAX_RETRY_WAITING_FOR_INIT); try { _lock.lock(); try { while (_taskAttempt == null) { if (!_taskAttemptBuilt.await(5, TimeUnit.SECONDS) && ++retryCount > maxRetry) { throw new IllegalStateException("Failed to initialize taskAttempt object before cancel"); } } } finally { _lock.unlock(); } if (_taskAttempt != null) { _logger.info("Task cancelled: Shutdown starting for tasks with jobId: {}", _jobId); _taskAttempt.shutdownTasks(); _logger.info("Task cancelled: Shutdown complete for tasks with jobId: {}", _jobId); } else { throw new IllegalStateException("TaskAttempt not initialized while passing conditional barrier"); } } catch (InterruptedException e) { throw new RuntimeException("Interrupted while shutting down task with jobId: " + _jobId, e); } } }
SingleTask { public void cancel() { int retryCount = 0 ; int maxRetry = ConfigUtils.getInt(_dynamicConfig, MAX_RETRY_WAITING_FOR_INIT_KEY, DEFAULT_MAX_RETRY_WAITING_FOR_INIT); try { _lock.lock(); try { while (_taskAttempt == null) { if (!_taskAttemptBuilt.await(5, TimeUnit.SECONDS) && ++retryCount > maxRetry) { throw new IllegalStateException("Failed to initialize taskAttempt object before cancel"); } } } finally { _lock.unlock(); } if (_taskAttempt != null) { _logger.info("Task cancelled: Shutdown starting for tasks with jobId: {}", _jobId); _taskAttempt.shutdownTasks(); _logger.info("Task cancelled: Shutdown complete for tasks with jobId: {}", _jobId); } else { throw new IllegalStateException("TaskAttempt not initialized while passing conditional barrier"); } } catch (InterruptedException e) { throw new RuntimeException("Interrupted while shutting down task with jobId: " + _jobId, e); } } SingleTask(String jobId, Path workUnitFilePath, Path jobStateFilePath, FileSystem fs,
      TaskAttemptBuilder taskAttemptBuilder, StateStores stateStores, Config dynamicConfig);   SingleTask(String jobId, Path workUnitFilePath, Path jobStateFilePath, FileSystem fs,
      TaskAttemptBuilder taskAttemptBuilder, StateStores stateStores, Config dynamicConfig, boolean skipGetJobState); }
SingleTask { public void cancel() { int retryCount = 0 ; int maxRetry = ConfigUtils.getInt(_dynamicConfig, MAX_RETRY_WAITING_FOR_INIT_KEY, DEFAULT_MAX_RETRY_WAITING_FOR_INIT); try { _lock.lock(); try { while (_taskAttempt == null) { if (!_taskAttemptBuilt.await(5, TimeUnit.SECONDS) && ++retryCount > maxRetry) { throw new IllegalStateException("Failed to initialize taskAttempt object before cancel"); } } } finally { _lock.unlock(); } if (_taskAttempt != null) { _logger.info("Task cancelled: Shutdown starting for tasks with jobId: {}", _jobId); _taskAttempt.shutdownTasks(); _logger.info("Task cancelled: Shutdown complete for tasks with jobId: {}", _jobId); } else { throw new IllegalStateException("TaskAttempt not initialized while passing conditional barrier"); } } catch (InterruptedException e) { throw new RuntimeException("Interrupted while shutting down task with jobId: " + _jobId, e); } } SingleTask(String jobId, Path workUnitFilePath, Path jobStateFilePath, FileSystem fs,
      TaskAttemptBuilder taskAttemptBuilder, StateStores stateStores, Config dynamicConfig);   SingleTask(String jobId, Path workUnitFilePath, Path jobStateFilePath, FileSystem fs,
      TaskAttemptBuilder taskAttemptBuilder, StateStores stateStores, Config dynamicConfig, boolean skipGetJobState); void run(); void cancel(); }
SingleTask { public void cancel() { int retryCount = 0 ; int maxRetry = ConfigUtils.getInt(_dynamicConfig, MAX_RETRY_WAITING_FOR_INIT_KEY, DEFAULT_MAX_RETRY_WAITING_FOR_INIT); try { _lock.lock(); try { while (_taskAttempt == null) { if (!_taskAttemptBuilt.await(5, TimeUnit.SECONDS) && ++retryCount > maxRetry) { throw new IllegalStateException("Failed to initialize taskAttempt object before cancel"); } } } finally { _lock.unlock(); } if (_taskAttempt != null) { _logger.info("Task cancelled: Shutdown starting for tasks with jobId: {}", _jobId); _taskAttempt.shutdownTasks(); _logger.info("Task cancelled: Shutdown complete for tasks with jobId: {}", _jobId); } else { throw new IllegalStateException("TaskAttempt not initialized while passing conditional barrier"); } } catch (InterruptedException e) { throw new RuntimeException("Interrupted while shutting down task with jobId: " + _jobId, e); } } SingleTask(String jobId, Path workUnitFilePath, Path jobStateFilePath, FileSystem fs,
      TaskAttemptBuilder taskAttemptBuilder, StateStores stateStores, Config dynamicConfig);   SingleTask(String jobId, Path workUnitFilePath, Path jobStateFilePath, FileSystem fs,
      TaskAttemptBuilder taskAttemptBuilder, StateStores stateStores, Config dynamicConfig, boolean skipGetJobState); void run(); void cancel(); static final String MAX_RETRY_WAITING_FOR_INIT_KEY; static final int DEFAULT_MAX_RETRY_WAITING_FOR_INIT; }