ZkJobCoordinator implements JobCoordinator { void doOnProcessorChange() { List<ProcessorNode> processorNodes = zkUtils.getAllProcessorNodes(); List<String> currentProcessorIds = new ArrayList<>(); for (ProcessorNode processorNode : processorNodes) { currentProcessorIds.add(processorNode.getProcessorData().getProcessorId()); } Set<String> uniqueProcessorIds = new HashSet<>(currentProcessorIds); if (currentProcessorIds.size() != uniqueProcessorIds.size()) { LOG.info("Processors: {} has duplicates. Not generating JobModel.", currentProcessorIds); return; } LOG.info("Generating new JobModel with processors: {}.", currentProcessorIds); JobModel jobModel = generateNewJobModel(processorNodes); if (!hasLoadedMetadataResources) { loadMetadataResources(jobModel); hasLoadedMetadataResources = true; } String currentJMVersion = zkUtils.getJobModelVersion(); String nextJMVersion = zkUtils.getNextJobModelVersion(currentJMVersion); LOG.info("pid=" + processorId + "Generated new JobModel with version: " + nextJMVersion + " and processors: " + currentProcessorIds); publishJobModelToMetadataStore(jobModel, nextJMVersion); barrier.create(nextJMVersion, currentProcessorIds); zkUtils.publishJobModelVersion(currentJMVersion, nextJMVersion); LOG.info("pid=" + processorId + "Published new Job Model. Version = " + nextJMVersion); debounceTimer.scheduleAfterDebounceTime(ON_ZK_CLEANUP, 0, () -> zkUtils.cleanupZK(NUM_VERSIONS_TO_LEAVE)); } ZkJobCoordinator(String processorId, Config config, MetricsRegistry metricsRegistry, ZkUtils zkUtils, MetadataStore jobModelMetadataStore, MetadataStore coordinatorStreamStore); @Override void start(); @Override void stop(); @Override void setListener(JobCoordinatorListener listener); @Override JobModel getJobModel(); @Override String getProcessorId(); void onProcessorChange(List<String> processors); @VisibleForTesting ZkUtils getZkUtils();  }
@Test public void testDoOnProcessorChange() { ZkKeyBuilder keyBuilder = Mockito.mock(ZkKeyBuilder.class); ZkClient mockZkClient = Mockito.mock(ZkClient.class); when(keyBuilder.getJobModelVersionBarrierPrefix()).thenReturn(TEST_BARRIER_ROOT); ZkUtils zkUtils = Mockito.mock(ZkUtils.class); when(zkUtils.getKeyBuilder()).thenReturn(keyBuilder); when(zkUtils.getZkClient()).thenReturn(mockZkClient); when(zkUtils.getJobModel(TEST_JOB_MODEL_VERSION)).thenReturn(jobModel); StartpointManager mockStartpointManager = Mockito.mock(StartpointManager.class); ZkJobCoordinator zkJobCoordinator = Mockito.spy(new ZkJobCoordinator("TEST_PROCESSOR_ID", config, new NoOpMetricsRegistry(), zkUtils, zkMetadataStore, coordinatorStreamStore)); doReturn(mockStartpointManager).when(zkJobCoordinator).createStartpointManager(); doReturn(jobModel).when(zkJobCoordinator).generateNewJobModel(any()); doNothing().when(zkJobCoordinator).loadMetadataResources(jobModel); zkJobCoordinator.doOnProcessorChange(); verify(zkUtils).publishJobModelVersion(anyString(), anyString()); verify(zkJobCoordinator).loadMetadataResources(eq(jobModel)); }