ZkJobCoordinator implements JobCoordinator { @Override public JobModel getJobModel() { return newJobModel; } 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 testShouldRemoveBufferedEventsInDebounceQueueOnSessionExpiration() { 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(new JobModel(new MapConfig(), new HashMap<>())); ScheduleAfterDebounceTime mockDebounceTimer = Mockito.mock(ScheduleAfterDebounceTime.class); ZkJobCoordinator zkJobCoordinator = Mockito.spy(new ZkJobCoordinator("TEST_PROCESSOR_ID", new MapConfig(), new NoOpMetricsRegistry(), zkUtils, zkMetadataStore, coordinatorStreamStore)); zkJobCoordinator.debounceTimer = mockDebounceTimer; zkJobCoordinator.zkSessionMetrics = new ZkSessionMetrics(new MetricsRegistryMap()); final ZkSessionStateChangedListener zkSessionStateChangedListener = zkJobCoordinator.new ZkSessionStateChangedListener(); zkSessionStateChangedListener.handleStateChanged(Watcher.Event.KeeperState.Expired); verify(zkUtils).incGeneration(); verify(mockDebounceTimer).cancelAllScheduledActions(); verify(mockDebounceTimer).scheduleAfterDebounceTime(eq("ZK_SESSION_EXPIRED"), eq(0L), Mockito.any(Runnable.class)); Assert.assertEquals(1, zkJobCoordinator.zkSessionMetrics.zkSessionExpirations.getCount()); }
@Test public void testZookeeperSessionMetricsAreUpdatedCoorrectly() { 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(new JobModel(new MapConfig(), new HashMap<>())); ScheduleAfterDebounceTime mockDebounceTimer = Mockito.mock(ScheduleAfterDebounceTime.class); ZkJobCoordinator zkJobCoordinator = Mockito.spy(new ZkJobCoordinator("TEST_PROCESSOR_ID", new MapConfig(), new NoOpMetricsRegistry(), zkUtils, zkMetadataStore, coordinatorStreamStore)); zkJobCoordinator.debounceTimer = mockDebounceTimer; zkJobCoordinator.zkSessionMetrics = new ZkSessionMetrics(new MetricsRegistryMap()); final ZkSessionStateChangedListener zkSessionStateChangedListener = zkJobCoordinator.new ZkSessionStateChangedListener(); zkSessionStateChangedListener.handleStateChanged(Watcher.Event.KeeperState.Disconnected); zkSessionStateChangedListener.handleStateChanged(Watcher.Event.KeeperState.SyncConnected); zkSessionStateChangedListener.handleStateChanged(Watcher.Event.KeeperState.AuthFailed); Assert.assertEquals(1, zkJobCoordinator.zkSessionMetrics.zkSessionErrors.getCount()); zkSessionStateChangedListener.handleSessionEstablishmentError(new SamzaException("Test exception")); Assert.assertEquals(1, zkJobCoordinator.zkSessionMetrics.zkSessionDisconnects.getCount()); Assert.assertEquals(1, zkJobCoordinator.zkSessionMetrics.zkSyncConnected.getCount()); Assert.assertEquals(2, zkJobCoordinator.zkSessionMetrics.zkSessionErrors.getCount()); }