ZkJobCoordinator implements JobCoordinator { @VisibleForTesting void loadMetadataResources(JobModel jobModel) { try { MetadataResourceUtil metadataResourceUtil = createMetadataResourceUtil(jobModel, config); metadataResourceUtil.createResources(); if (coordinatorStreamStore != null) { CoordinatorStreamValueSerde jsonSerde = new CoordinatorStreamValueSerde(SetConfig.TYPE); NamespaceAwareCoordinatorStreamStore configStore = new NamespaceAwareCoordinatorStreamStore(coordinatorStreamStore, SetConfig.TYPE); for (Map.Entry<String, String> entry : config.entrySet()) { byte[] serializedValue = jsonSerde.toBytes(entry.getValue()); configStore.put(entry.getKey(), serializedValue); } configStore.flush(); if (new JobConfig(config).getStartpointEnabled()) { StartpointManager startpointManager = createStartpointManager(); startpointManager.start(); try { startpointManager.fanOut(JobModelUtil.getTaskToSystemStreamPartitions(jobModel)); } finally { startpointManager.stop(); } } } else { LOG.warn("No metadata store registered to this job coordinator. Config not written to the metadata store and no Startpoints fan out."); } } catch (IOException ex) { throw new SamzaException(String.format("IO exception while loading metadata resources."), ex); } } 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 testLoadMetadataResources() throws IOException { 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(); MetadataResourceUtil mockMetadataResourceUtil = mock(MetadataResourceUtil.class); doReturn(mockMetadataResourceUtil).when(zkJobCoordinator).createMetadataResourceUtil(any(), any(Config.class)); verifyZeroInteractions(mockStartpointManager); zkJobCoordinator.loadMetadataResources(jobModel); verify(mockMetadataResourceUtil).createResources(); verify(mockStartpointManager).start(); verify(mockStartpointManager).fanOut(any()); verify(mockStartpointManager).stop(); }