RMContextImplEventRunnable implements Runnable { public void sendAppEvent(ApplicationId applicationId, RMApp rmApp) { if (cacheFinishedApp.getIfPresent(applicationId.toString()) == null) { Header.Builder headerBuilder = Header.newBuilder() .withId(applicationId.toString()) .withApplicationID(applicationId.toString()) .withUser(rmApp.getUser()) .withApplicationName(rmApp.getName()) .withFramework(rmApp.getApplicationType().toUpperCase()); ApplicationEvent.Builder eventBuilder = ApplicationEvent.newBuilder() .setState(rmApp.getState().name()) .setQueue(rmApp.getQueue()); rmApp.getApplicationTags().stream() .filter(tag -> YARN_TAGS_TO_EXTRACT.stream().noneMatch(tag::startsWith) && !tag.contains(":")) .forEach(eventBuilder::addYarnTags); rmApp.getApplicationTags().stream() .filter(tag -> tag.contains(":") && YARN_TAGS_TO_EXTRACT.stream().anyMatch(tag::startsWith)) .map(tag -> { int idx = tag.indexOf(':'); String key = tag.substring(0, idx); String value = tag.substring(idx + 1); return new String[] {key, value}; }) .forEach(splitTag -> BUILDERS.get(splitTag[0]).accept(splitTag[1], eventBuilder)); eventBuilder.setFinalStatus(rmApp.getFinalApplicationStatus().name()); eventBuilder.setStartTime(rmApp.getStartTime()); eventBuilder.setFinishTime(rmApp.getFinishTime()); RMAppMetrics rmAppMetrics = rmApp.getRMAppMetrics(); if (rmAppMetrics != null) { eventBuilder.setMemorySeconds(rmAppMetrics.getMemorySeconds()); eventBuilder.setVcoreSeconds(rmAppMetrics.getVcoreSeconds()); } RMAppAttempt rmAppAttempt = rmApp.getCurrentAppAttempt(); if (rmAppAttempt != null) { headerBuilder.withAttemptID(rmAppAttempt.getAppAttemptId().toString()); Container container = rmAppAttempt.getMasterContainer(); if (container != null) { eventBuilder.setAmContainerId(container.getId().toString()); } } if (rmApp.getTrackingUrl() != null) { eventBuilder.setTrackingUrl(normalizeTrackingUrl(rmApp.getTrackingUrl())); } if (rmApp.getOriginalTrackingUrl() != null && !"N/A".equals(rmApp.getOriginalTrackingUrl())) { eventBuilder.setOriginalTrackingUrl(normalizeTrackingUrl(rmApp.getOriginalTrackingUrl())); } eventHandler.accept(System.currentTimeMillis(), headerBuilder.build(), eventBuilder.build()); if (rmApp.getState() == RMAppState.FINISHED || rmApp.getState() == RMAppState.KILLED || rmApp.getState() == RMAppState.FAILED) { cacheFinishedApp.put(applicationId.toString(), rmApp.getState().name()); } } } RMContextImplEventRunnable(RMContextImpl rmContext, TriConsumer<Long, Header, Object> eventHandler); String normalizeTrackingUrl(String trackingUrl); void sendAppEvent(ApplicationId applicationId, RMApp rmApp); @Override void run();  }
@Test public void RMContextImplEventRunnable_sendAppEvent_should_not_failed_for_null_component_in_rmapp() { RMApp rmApp = mockRMApp("container_id_001", null, "project", "workflow", FinalApplicationStatus.SUCCEEDED); rmContextImplEventRunnable.sendAppEvent(applicationId, rmApp); verify(eventHandler).accept(any(Long.class), any(Header.class), argument.capture()); ResourceManagerEventProtos.ApplicationEvent appEvent = argument.getValue(); assertThat(appEvent.getAmContainerId()).isEqualTo("container_id_001"); assertThat(appEvent.getTrackingUrl()).isEqualTo(""); assertThat(appEvent.getProjectName()).isEqualTo("project"); assertThat(appEvent.getWorkflowName()).isEqualTo("workflow"); assertThat(appEvent.getFinalStatus()).isEqualTo("SUCCEEDED"); }