@Test(timeout = 10000L) public void testLogNotFoundHandling() throws Throwable { final int dataPort = NetUtils.getAvailablePort(); Configuration config = new Configuration(); config.setInteger(NettyShuffleEnvironmentOptions.DATA_PORT, dataPort); config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100); config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX, 200); config.setString(ConfigConstants.TASK_MANAGER_LOG_PATH_KEY, "/i/dont/exist"); try (TaskSubmissionTestEnvironment env = new TaskSubmissionTestEnvironment.Builder(jobId) .setConfiguration(config) .setLocalCommunication(false) .build()) { TaskExecutorGateway tmGateway = env.getTaskExecutorGateway(); try { CompletableFuture<TransientBlobKey> logFuture = tmGateway.requestFileUpload(FileType.LOG, timeout); logFuture.get(); } catch (Exception e) { assertThat(e.getMessage(), containsString("The file LOG does not exist on the TaskExecutor.")); } } }
@Override public CompletableFuture<TransientBlobKey> requestFileUpload(FileType fileType, Time timeout) { log.debug("Request file {} upload.", fileType); final String filePath; switch (fileType) { case LOG: filePath = taskManagerConfiguration.getTaskManagerLogPath(); break; case STDOUT: filePath = taskManagerConfiguration.getTaskManagerStdoutPath(); break; default: filePath = null; } if (filePath != null && !filePath.isEmpty()) { final File file = new File(filePath); if (file.exists()) { final TransientBlobCache transientBlobService = blobCacheService.getTransientBlobService(); final TransientBlobKey transientBlobKey; try (FileInputStream fileInputStream = new FileInputStream(file)) { transientBlobKey = transientBlobService.putTransient(fileInputStream); } catch (IOException e) { log.debug("Could not upload file {}.", fileType, e); return FutureUtils.completedExceptionally(new FlinkException("Could not upload file " + fileType + '.', e)); } return CompletableFuture.completedFuture(transientBlobKey); } else { log.debug("The file {} does not exist on the TaskExecutor {}.", fileType, getResourceID()); return FutureUtils.completedExceptionally(new FlinkException("The file " + fileType + " does not exist on the TaskExecutor.")); } } else { log.debug("The file {} is unavailable on the TaskExecutor {}.", fileType, getResourceID()); return FutureUtils.completedExceptionally(new FlinkException("The file " + fileType + " is not available on the TaskExecutor.")); } }
TaskExecutor extends RpcEndpoint implements TaskExecutorGateway { @Override public CompletableFuture<TransientBlobKey> requestFileUpload(FileType fileType, Time timeout) { log.debug("Request file {} upload.", fileType); final String filePath; switch (fileType) { case LOG: filePath = taskManagerConfiguration.getTaskManagerLogPath(); break; case STDOUT: filePath = taskManagerConfiguration.getTaskManagerStdoutPath(); break; default: filePath = null; } if (filePath != null && !filePath.isEmpty()) { final File file = new File(filePath); if (file.exists()) { final TransientBlobCache transientBlobService = blobCacheService.getTransientBlobService(); final TransientBlobKey transientBlobKey; try (FileInputStream fileInputStream = new FileInputStream(file)) { transientBlobKey = transientBlobService.putTransient(fileInputStream); } catch (IOException e) { log.debug("Could not upload file {}.", fileType, e); return FutureUtils.completedExceptionally(new FlinkException("Could not upload file " + fileType + '.', e)); } return CompletableFuture.completedFuture(transientBlobKey); } else { log.debug("The file {} does not exist on the TaskExecutor {}.", fileType, getResourceID()); return FutureUtils.completedExceptionally(new FlinkException("The file " + fileType + " does not exist on the TaskExecutor.")); } } else { log.debug("The file {} is unavailable on the TaskExecutor {}.", fileType, getResourceID()); return FutureUtils.completedExceptionally(new FlinkException("The file " + fileType + " is not available on the TaskExecutor.")); } } }
TaskExecutor extends RpcEndpoint implements TaskExecutorGateway { @Override public CompletableFuture<TransientBlobKey> requestFileUpload(FileType fileType, Time timeout) { log.debug("Request file {} upload.", fileType); final String filePath; switch (fileType) { case LOG: filePath = taskManagerConfiguration.getTaskManagerLogPath(); break; case STDOUT: filePath = taskManagerConfiguration.getTaskManagerStdoutPath(); break; default: filePath = null; } if (filePath != null && !filePath.isEmpty()) { final File file = new File(filePath); if (file.exists()) { final TransientBlobCache transientBlobService = blobCacheService.getTransientBlobService(); final TransientBlobKey transientBlobKey; try (FileInputStream fileInputStream = new FileInputStream(file)) { transientBlobKey = transientBlobService.putTransient(fileInputStream); } catch (IOException e) { log.debug("Could not upload file {}.", fileType, e); return FutureUtils.completedExceptionally(new FlinkException("Could not upload file " + fileType + '.', e)); } return CompletableFuture.completedFuture(transientBlobKey); } else { log.debug("The file {} does not exist on the TaskExecutor {}.", fileType, getResourceID()); return FutureUtils.completedExceptionally(new FlinkException("The file " + fileType + " does not exist on the TaskExecutor.")); } } else { log.debug("The file {} is unavailable on the TaskExecutor {}.", fileType, getResourceID()); return FutureUtils.completedExceptionally(new FlinkException("The file " + fileType + " is not available on the TaskExecutor.")); } } TaskExecutor(
			RpcService rpcService,
			TaskManagerConfiguration taskManagerConfiguration,
			HighAvailabilityServices haServices,
			TaskManagerServices taskExecutorServices,
			HeartbeatServices heartbeatServices,
			TaskManagerMetricGroup taskManagerMetricGroup,
			@Nullable String metricQueryServiceAddress,
			BlobCacheService blobCacheService,
			FatalErrorHandler fatalErrorHandler,
			PartitionTable<JobID> partitionTable); }
TaskExecutor extends RpcEndpoint implements TaskExecutorGateway { @Override public CompletableFuture<TransientBlobKey> requestFileUpload(FileType fileType, Time timeout) { log.debug("Request file {} upload.", fileType); final String filePath; switch (fileType) { case LOG: filePath = taskManagerConfiguration.getTaskManagerLogPath(); break; case STDOUT: filePath = taskManagerConfiguration.getTaskManagerStdoutPath(); break; default: filePath = null; } if (filePath != null && !filePath.isEmpty()) { final File file = new File(filePath); if (file.exists()) { final TransientBlobCache transientBlobService = blobCacheService.getTransientBlobService(); final TransientBlobKey transientBlobKey; try (FileInputStream fileInputStream = new FileInputStream(file)) { transientBlobKey = transientBlobService.putTransient(fileInputStream); } catch (IOException e) { log.debug("Could not upload file {}.", fileType, e); return FutureUtils.completedExceptionally(new FlinkException("Could not upload file " + fileType + '.', e)); } return CompletableFuture.completedFuture(transientBlobKey); } else { log.debug("The file {} does not exist on the TaskExecutor {}.", fileType, getResourceID()); return FutureUtils.completedExceptionally(new FlinkException("The file " + fileType + " does not exist on the TaskExecutor.")); } } else { log.debug("The file {} is unavailable on the TaskExecutor {}.", fileType, getResourceID()); return FutureUtils.completedExceptionally(new FlinkException("The file " + fileType + " is not available on the TaskExecutor.")); } } TaskExecutor(
			RpcService rpcService,
			TaskManagerConfiguration taskManagerConfiguration,
			HighAvailabilityServices haServices,
			TaskManagerServices taskExecutorServices,
			HeartbeatServices heartbeatServices,
			TaskManagerMetricGroup taskManagerMetricGroup,
			@Nullable String metricQueryServiceAddress,
			BlobCacheService blobCacheService,
			FatalErrorHandler fatalErrorHandler,
			PartitionTable<JobID> partitionTable); @Override CompletableFuture<Boolean> canBeReleased(); @Override void onStart(); @Override CompletableFuture<Void> onStop(); @Override CompletableFuture<StackTraceSampleResponse> requestStackTraceSample(
			final ExecutionAttemptID executionAttemptId,
			final int sampleId,
			final int numSamples,
			final Time delayBetweenSamples,
			final int maxStackTraceDepth,
			final Time timeout); @Override CompletableFuture<Acknowledge> submitTask(
			TaskDeploymentDescriptor tdd,
			JobMasterId jobMasterId,
			Time timeout); @Override CompletableFuture<Acknowledge> cancelTask(ExecutionAttemptID executionAttemptID, Time timeout); @Override CompletableFuture<Acknowledge> updatePartitions(
			final ExecutionAttemptID executionAttemptID,
			Iterable<PartitionInfo> partitionInfos,
			Time timeout); @Override void releasePartitions(JobID jobId, Collection<ResultPartitionID> partitionIds); @Override void heartbeatFromJobManager(ResourceID resourceID, AllocatedSlotReport allocatedSlotReport); @Override void heartbeatFromResourceManager(ResourceID resourceID); @Override CompletableFuture<Acknowledge> triggerCheckpoint(
			ExecutionAttemptID executionAttemptID,
			long checkpointId,
			long checkpointTimestamp,
			CheckpointOptions checkpointOptions,
			boolean advanceToEndOfEventTime); @Override CompletableFuture<Acknowledge> confirmCheckpoint(
			ExecutionAttemptID executionAttemptID,
			long checkpointId,
			long checkpointTimestamp); @Override CompletableFuture<Acknowledge> requestSlot(
		final SlotID slotId,
		final JobID jobId,
		final AllocationID allocationId,
		final String targetAddress,
		final ResourceManagerId resourceManagerId,
		final Time timeout); @Override CompletableFuture<Acknowledge> freeSlot(AllocationID allocationId, Throwable cause, Time timeout); @Override CompletableFuture<TransientBlobKey> requestFileUpload(FileType fileType, Time timeout); @Override CompletableFuture<SerializableOptional<String>> requestMetricQueryServiceAddress(Time timeout); @Override void disconnectJobManager(JobID jobId, Exception cause); @Override void disconnectResourceManager(Exception cause); ResourceID getResourceID(); }
TaskExecutor extends RpcEndpoint implements TaskExecutorGateway { @Override public CompletableFuture<TransientBlobKey> requestFileUpload(FileType fileType, Time timeout) { log.debug("Request file {} upload.", fileType); final String filePath; switch (fileType) { case LOG: filePath = taskManagerConfiguration.getTaskManagerLogPath(); break; case STDOUT: filePath = taskManagerConfiguration.getTaskManagerStdoutPath(); break; default: filePath = null; } if (filePath != null && !filePath.isEmpty()) { final File file = new File(filePath); if (file.exists()) { final TransientBlobCache transientBlobService = blobCacheService.getTransientBlobService(); final TransientBlobKey transientBlobKey; try (FileInputStream fileInputStream = new FileInputStream(file)) { transientBlobKey = transientBlobService.putTransient(fileInputStream); } catch (IOException e) { log.debug("Could not upload file {}.", fileType, e); return FutureUtils.completedExceptionally(new FlinkException("Could not upload file " + fileType + '.', e)); } return CompletableFuture.completedFuture(transientBlobKey); } else { log.debug("The file {} does not exist on the TaskExecutor {}.", fileType, getResourceID()); return FutureUtils.completedExceptionally(new FlinkException("The file " + fileType + " does not exist on the TaskExecutor.")); } } else { log.debug("The file {} is unavailable on the TaskExecutor {}.", fileType, getResourceID()); return FutureUtils.completedExceptionally(new FlinkException("The file " + fileType + " is not available on the TaskExecutor.")); } } TaskExecutor(
			RpcService rpcService,
			TaskManagerConfiguration taskManagerConfiguration,
			HighAvailabilityServices haServices,
			TaskManagerServices taskExecutorServices,
			HeartbeatServices heartbeatServices,
			TaskManagerMetricGroup taskManagerMetricGroup,
			@Nullable String metricQueryServiceAddress,
			BlobCacheService blobCacheService,
			FatalErrorHandler fatalErrorHandler,
			PartitionTable<JobID> partitionTable); @Override CompletableFuture<Boolean> canBeReleased(); @Override void onStart(); @Override CompletableFuture<Void> onStop(); @Override CompletableFuture<StackTraceSampleResponse> requestStackTraceSample(
			final ExecutionAttemptID executionAttemptId,
			final int sampleId,
			final int numSamples,
			final Time delayBetweenSamples,
			final int maxStackTraceDepth,
			final Time timeout); @Override CompletableFuture<Acknowledge> submitTask(
			TaskDeploymentDescriptor tdd,
			JobMasterId jobMasterId,
			Time timeout); @Override CompletableFuture<Acknowledge> cancelTask(ExecutionAttemptID executionAttemptID, Time timeout); @Override CompletableFuture<Acknowledge> updatePartitions(
			final ExecutionAttemptID executionAttemptID,
			Iterable<PartitionInfo> partitionInfos,
			Time timeout); @Override void releasePartitions(JobID jobId, Collection<ResultPartitionID> partitionIds); @Override void heartbeatFromJobManager(ResourceID resourceID, AllocatedSlotReport allocatedSlotReport); @Override void heartbeatFromResourceManager(ResourceID resourceID); @Override CompletableFuture<Acknowledge> triggerCheckpoint(
			ExecutionAttemptID executionAttemptID,
			long checkpointId,
			long checkpointTimestamp,
			CheckpointOptions checkpointOptions,
			boolean advanceToEndOfEventTime); @Override CompletableFuture<Acknowledge> confirmCheckpoint(
			ExecutionAttemptID executionAttemptID,
			long checkpointId,
			long checkpointTimestamp); @Override CompletableFuture<Acknowledge> requestSlot(
		final SlotID slotId,
		final JobID jobId,
		final AllocationID allocationId,
		final String targetAddress,
		final ResourceManagerId resourceManagerId,
		final Time timeout); @Override CompletableFuture<Acknowledge> freeSlot(AllocationID allocationId, Throwable cause, Time timeout); @Override CompletableFuture<TransientBlobKey> requestFileUpload(FileType fileType, Time timeout); @Override CompletableFuture<SerializableOptional<String>> requestMetricQueryServiceAddress(Time timeout); @Override void disconnectJobManager(JobID jobId, Exception cause); @Override void disconnectResourceManager(Exception cause); ResourceID getResourceID(); static final String TASK_MANAGER_NAME; }