@Test public void onExecutorAdded() { TriConsumer<Long, Header, Object> handlerMock = mock(TriConsumer.class); SparkListener listener = new GarmadonSparkListener(handlerMock, DUMMY_HEADER.toSerializeHeader()); listener.onExecutorAdded(new SparkListenerExecutorAdded(1, "invokationExecutorId", new ExecutorInfo("infoHost", 12, new HashMap<>()))); checkExecutorEventHeader(handlerMock, DUMMY_HEADER.cloneAndOverride(Header.newBuilder() .withExecutorId("invokationExecutorId").build())); }
@Override public void onExecutorAdded(SparkListenerExecutorAdded executorAdded) { try { executorHostId.put(executorAdded.executorId(), executorAdded.executorInfo().executorHost()); sendExecutorStateEvent(executorAdded.time(), State.ADDED, executorAdded.executorId(), executorAdded.executorInfo().executorHost(), null, 0); } catch (Throwable t) { LOGGER.warn("Failed to send event for onExecutorAdded", t); } }
GarmadonSparkListener extends SparkListener { @Override public void onExecutorAdded(SparkListenerExecutorAdded executorAdded) { try { executorHostId.put(executorAdded.executorId(), executorAdded.executorInfo().executorHost()); sendExecutorStateEvent(executorAdded.time(), State.ADDED, executorAdded.executorId(), executorAdded.executorInfo().executorHost(), null, 0); } catch (Throwable t) { LOGGER.warn("Failed to send event for onExecutorAdded", t); } } }
GarmadonSparkListener extends SparkListener { @Override public void onExecutorAdded(SparkListenerExecutorAdded executorAdded) { try { executorHostId.put(executorAdded.executorId(), executorAdded.executorInfo().executorHost()); sendExecutorStateEvent(executorAdded.time(), State.ADDED, executorAdded.executorId(), executorAdded.executorInfo().executorHost(), null, 0); } catch (Throwable t) { LOGGER.warn("Failed to send event for onExecutorAdded", t); } } GarmadonSparkListener();  GarmadonSparkListener(TriConsumer<Long, Header, Object> eventHandler, Header.SerializedHeader header); }
GarmadonSparkListener extends SparkListener { @Override public void onExecutorAdded(SparkListenerExecutorAdded executorAdded) { try { executorHostId.put(executorAdded.executorId(), executorAdded.executorInfo().executorHost()); sendExecutorStateEvent(executorAdded.time(), State.ADDED, executorAdded.executorId(), executorAdded.executorInfo().executorHost(), null, 0); } catch (Throwable t) { LOGGER.warn("Failed to send event for onExecutorAdded", t); } } GarmadonSparkListener();  GarmadonSparkListener(TriConsumer<Long, Header, Object> eventHandler, Header.SerializedHeader header); @Override void onApplicationStart(SparkListenerApplicationStart applicationStart); @Override void onStageSubmitted(SparkListenerStageSubmitted stageSubmitted); @Override void onStageCompleted(SparkListenerStageCompleted stageCompleted); @Override void onTaskEnd(SparkListenerTaskEnd taskEnd); @Override void onExecutorAdded(SparkListenerExecutorAdded executorAdded); @Override void onExecutorRemoved(SparkListenerExecutorRemoved executorRemoved); @Override void onExecutorBlacklisted(SparkListenerExecutorBlacklisted executorBlacklisted); @Override void onExecutorUnblacklisted(SparkListenerExecutorUnblacklisted executorUnblacklisted); }
GarmadonSparkListener extends SparkListener { @Override public void onExecutorAdded(SparkListenerExecutorAdded executorAdded) { try { executorHostId.put(executorAdded.executorId(), executorAdded.executorInfo().executorHost()); sendExecutorStateEvent(executorAdded.time(), State.ADDED, executorAdded.executorId(), executorAdded.executorInfo().executorHost(), null, 0); } catch (Throwable t) { LOGGER.warn("Failed to send event for onExecutorAdded", t); } } GarmadonSparkListener();  GarmadonSparkListener(TriConsumer<Long, Header, Object> eventHandler, Header.SerializedHeader header); @Override void onApplicationStart(SparkListenerApplicationStart applicationStart); @Override void onStageSubmitted(SparkListenerStageSubmitted stageSubmitted); @Override void onStageCompleted(SparkListenerStageCompleted stageCompleted); @Override void onTaskEnd(SparkListenerTaskEnd taskEnd); @Override void onExecutorAdded(SparkListenerExecutorAdded executorAdded); @Override void onExecutorRemoved(SparkListenerExecutorRemoved executorRemoved); @Override void onExecutorBlacklisted(SparkListenerExecutorBlacklisted executorBlacklisted); @Override void onExecutorUnblacklisted(SparkListenerExecutorUnblacklisted executorUnblacklisted);  }