gobblin-temporal/src/main/java/org/apache/gobblin/temporal/yarn/YarnService.java [647:779]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - amrmClientAsync.removeContainerRequest(firstMatchingContainerRequest); } } containerLaunchExecutor.submit(new Runnable() { @Override public void run() { try { LOGGER.info("Starting container " + containerId); nmClientAsync.startContainerAsync(container, newContainerLaunchContext(containerInfo)); } catch (IOException ioe) { LOGGER.error("Failed to start container " + containerId, ioe); } } }); } } @Override public void onShutdownRequest() { if (eventSubmitter.isPresent()) { eventSubmitter.get().submit(GobblinYarnEventConstants.EventNames.SHUTDOWN_REQUEST); } LOGGER.info("Received shutdown request from the ResourceManager"); this.done = true; eventBus.post(new ClusterManagerShutdownRequest()); } @Override public void onNodesUpdated(List updatedNodes) { for (NodeReport nodeReport : updatedNodes) { LOGGER.info("Received node update report: " + nodeReport); } } @Override public float getProgress() { return this.done ? 1.0f : 0.0f; } @Override public void onError(Throwable t) { if (eventSubmitter.isPresent()) { eventSubmitter.get().submit(GobblinYarnEventConstants.EventNames.ERROR, GobblinYarnEventConstants.EventMetadata.ERROR_EXCEPTION, Throwables.getStackTraceAsString(t)); } LOGGER.error("Received error: " + t, t); this.done = true; eventBus.post(new ClusterManagerShutdownRequest()); } } /** * A custom implementation of {@link NMClientAsync.CallbackHandler}. */ class NMClientCallbackHandler implements NMClientAsync.CallbackHandler { @Override public void onContainerStarted(ContainerId containerId, Map allServiceResponse) { if (eventSubmitter.isPresent()) { eventSubmitter.get().submit(GobblinYarnEventConstants.EventNames.CONTAINER_STARTED, GobblinYarnMetricTagNames.CONTAINER_ID, containerId.toString()); } LOGGER.info(String.format("Container %s has been started", containerId)); } @Override public void onContainerStatusReceived(ContainerId containerId, ContainerStatus containerStatus) { if (eventSubmitter.isPresent()) { eventSubmitter.get().submit(GobblinYarnEventConstants.EventNames.CONTAINER_STATUS_RECEIVED, buildContainerStatusEventMetadata(containerStatus).build()); } LOGGER.info(String.format("Received container status for container %s: %s", containerId, containerStatus)); } @Override public void onContainerStopped(ContainerId containerId) { if (eventSubmitter.isPresent()) { eventSubmitter.get().submit(GobblinYarnEventConstants.EventNames.CONTAINER_STOPPED, GobblinYarnMetricTagNames.CONTAINER_ID, containerId.toString()); } LOGGER.info(String.format("Container %s has been stopped", containerId)); if (containerMap.isEmpty()) { synchronized (allContainersStopped) { allContainersStopped.notify(); } } } @Override public void onStartContainerError(ContainerId containerId, Throwable t) { if (eventSubmitter.isPresent()) { eventSubmitter.get().submit(GobblinYarnEventConstants.EventNames.CONTAINER_START_ERROR, GobblinYarnMetricTagNames.CONTAINER_ID, containerId.toString(), GobblinYarnEventConstants.EventMetadata.ERROR_EXCEPTION, Throwables.getStackTraceAsString(t)); } LOGGER.error(String.format("Failed to start container %s due to error %s", containerId, t)); } @Override public void onGetContainerStatusError(ContainerId containerId, Throwable t) { if (eventSubmitter.isPresent()) { eventSubmitter.get().submit(GobblinYarnEventConstants.EventNames.CONTAINER_GET_STATUS_ERROR, GobblinYarnMetricTagNames.CONTAINER_ID, containerId.toString(), GobblinYarnEventConstants.EventMetadata.ERROR_EXCEPTION, Throwables.getStackTraceAsString(t)); } LOGGER.error(String.format("Failed to get status for container %s due to error %s", containerId, t)); } @Override public void onStopContainerError(ContainerId containerId, Throwable t) { if (eventSubmitter.isPresent()) { eventSubmitter.get().submit(GobblinYarnEventConstants.EventNames.CONTAINER_STOP_ERROR, GobblinYarnMetricTagNames.CONTAINER_ID, containerId.toString(), GobblinYarnEventConstants.EventMetadata.ERROR_EXCEPTION, Throwables.getStackTraceAsString(t)); } LOGGER.error(String.format("Failed to stop container %s due to error %s", containerId, t)); } } // Class encapsulates Container instance, WorkerProfile name to print, WorkerProfile, and // initial startup command @Getter class ContainerInfo { private final Container container; - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - gobblin-yarn/src/main/java/org/apache/gobblin/yarn/YarnService.java [1041:1173]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - amrmClientAsync.removeContainerRequest(firstMatchingContainerRequest); } containerLaunchExecutor.submit(new Runnable() { @Override public void run() { try { LOGGER.info("Starting container " + containerId); nmClientAsync.startContainerAsync(container, newContainerLaunchContext(containerInfo)); } catch (IOException ioe) { LOGGER.error("Failed to start container " + containerId, ioe); } } }); } } @Override public void onShutdownRequest() { if (eventSubmitter.isPresent()) { eventSubmitter.get().submit(GobblinYarnEventConstants.EventNames.SHUTDOWN_REQUEST); } LOGGER.info("Received shutdown request from the ResourceManager"); this.done = true; eventBus.post(new ClusterManagerShutdownRequest()); } @Override public void onNodesUpdated(List updatedNodes) { for (NodeReport nodeReport : updatedNodes) { LOGGER.info("Received node update report: " + nodeReport); } } @Override public float getProgress() { return this.done ? 1.0f : 0.0f; } @Override public void onError(Throwable t) { if (eventSubmitter.isPresent()) { eventSubmitter.get().submit(GobblinYarnEventConstants.EventNames.ERROR, GobblinYarnEventConstants.EventMetadata.ERROR_EXCEPTION, Throwables.getStackTraceAsString(t)); } LOGGER.error("Received error: " + t, t); this.done = true; eventBus.post(new ClusterManagerShutdownRequest()); } } /** * A custom implementation of {@link NMClientAsync.CallbackHandler}. */ class NMClientCallbackHandler implements NMClientAsync.CallbackHandler { @Override public void onContainerStarted(ContainerId containerId, Map allServiceResponse) { if (eventSubmitter.isPresent()) { eventSubmitter.get().submit(GobblinYarnEventConstants.EventNames.CONTAINER_STARTED, GobblinYarnMetricTagNames.CONTAINER_ID, containerId.toString()); } LOGGER.info(String.format("Container %s has been started", containerId)); } @Override public void onContainerStatusReceived(ContainerId containerId, ContainerStatus containerStatus) { if (eventSubmitter.isPresent()) { eventSubmitter.get().submit(GobblinYarnEventConstants.EventNames.CONTAINER_STATUS_RECEIVED, buildContainerStatusEventMetadata(containerStatus).build()); } LOGGER.info(String.format("Received container status for container %s: %s", containerId, containerStatus)); } @Override public void onContainerStopped(ContainerId containerId) { if (eventSubmitter.isPresent()) { eventSubmitter.get().submit(GobblinYarnEventConstants.EventNames.CONTAINER_STOPPED, GobblinYarnMetricTagNames.CONTAINER_ID, containerId.toString()); } LOGGER.info(String.format("Container %s has been stopped", containerId)); if (containerMap.isEmpty()) { synchronized (allContainersStopped) { allContainersStopped.notify(); } } } @Override public void onStartContainerError(ContainerId containerId, Throwable t) { if (eventSubmitter.isPresent()) { eventSubmitter.get().submit(GobblinYarnEventConstants.EventNames.CONTAINER_START_ERROR, GobblinYarnMetricTagNames.CONTAINER_ID, containerId.toString(), GobblinYarnEventConstants.EventMetadata.ERROR_EXCEPTION, Throwables.getStackTraceAsString(t)); } LOGGER.error(String.format("Failed to start container %s due to error %s", containerId, t)); } @Override public void onGetContainerStatusError(ContainerId containerId, Throwable t) { if (eventSubmitter.isPresent()) { eventSubmitter.get().submit(GobblinYarnEventConstants.EventNames.CONTAINER_GET_STATUS_ERROR, GobblinYarnMetricTagNames.CONTAINER_ID, containerId.toString(), GobblinYarnEventConstants.EventMetadata.ERROR_EXCEPTION, Throwables.getStackTraceAsString(t)); } LOGGER.error(String.format("Failed to get status for container %s due to error %s", containerId, t)); } @Override public void onStopContainerError(ContainerId containerId, Throwable t) { if (eventSubmitter.isPresent()) { eventSubmitter.get().submit(GobblinYarnEventConstants.EventNames.CONTAINER_STOP_ERROR, GobblinYarnMetricTagNames.CONTAINER_ID, containerId.toString(), GobblinYarnEventConstants.EventMetadata.ERROR_EXCEPTION, Throwables.getStackTraceAsString(t)); } LOGGER.error(String.format("Failed to stop container %s due to error %s", containerId, t)); } } // Class encapsulates Container instances, Helix participant IDs of the containers, Helix Tag, and // initial startup command @Getter class ContainerInfo { private final Container container; - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -