### Question:
ClusterControllerLeaderImpl implements ClusterControllerLeader, RegistrationListener { @Override public void processAsLeader() throws Exception { log.info("Become controller leader to monitor servers for assigning storage containers."); performServerChangesPermits.release(); try { this.regClient.watchWritableBookies(this).get(); } catch (Exception e) { log.warn("Controller leader fails to watch servers : {}, giving up leadership", e.getMessage()); throw e; } while (true) { try { checkSuspension(); processServerChange(); } catch (InterruptedException ie) { log.warn("Controller leader is interrupted, giving up leadership"); this.regClient.unwatchWritableBookies(this); throw ie; } catch (Exception e) { if (!suspended) { log.warn("Controller leader encountered exceptions on processing server changes," + " giving up leadership"); this.regClient.unwatchWritableBookies(this); throw e; } } } } ClusterControllerLeaderImpl(ClusterMetadataStore clusterMetadataStore,
                                StorageContainerController scController,
                                RegistrationClient regClient,
                                Duration scheduleDuration); @Override void suspend(); @Override void resume(); @Override void processAsLeader(); @Override void onBookiesChanged(Versioned<Set<BookieId>> bookies);  }### Answer:
@Test public void testProcessAsLeader() throws Exception { clusterController.suspend(); assertTrue(clusterController.isSuspended()); leaderExecutor.submit(() -> { try { clusterController.processAsLeader(); } catch (Exception e) { log.info("Encountered exception when cluster controller processes as a leader", e); } }); clusterController.resume(); assertFalse(clusterController.isSuspended()); FutureUtils.complete(watchFuture, null); assertNotNull(regListenerRef); assertFalse(coordSem.tryAcquire(1, TimeUnit.SECONDS)); assertTrue(clusterController.getLastSuccessfulAssigmentAt() < 0); Set<BookieId> cluster = Sets.newSet(BookieId.parse("127.0.0.1:4181")); Version version = new LongVersion(0L); regListenerRef.get().onBookiesChanged(new Versioned<>(cluster, version)); coordSem.acquire(); assertTrue(clusterController.getLastSuccessfulAssigmentAt() > 0); long lastSuccessfulAssignmentAt = clusterController.getLastSuccessfulAssigmentAt(); regListenerRef.get().onBookiesChanged(new Versioned<>(cluster, version)); assertFalse(coordSem.tryAcquire(200, TimeUnit.MILLISECONDS)); assertEquals(lastSuccessfulAssignmentAt, clusterController.getLastSuccessfulAssigmentAt()); cluster.add(BookieId.parse("127.0.0.1:4182")); cluster.add(BookieId.parse("127.0.0.1:4183")); cluster.add(BookieId.parse("127.0.0.1:4184")); cluster.add(BookieId.parse("127.0.0.1:4185")); version = new LongVersion(1L); regListenerRef.get().onBookiesChanged(new Versioned<>(cluster, version)); coordSem.acquire(); assertTrue(clusterController.getLastSuccessfulAssigmentAt() > lastSuccessfulAssignmentAt); lastSuccessfulAssignmentAt = clusterController.getLastSuccessfulAssigmentAt(); regListenerRef.get().onBookiesChanged(new Versioned<>(Collections.emptySet(), new LongVersion(2L))); assertFalse(coordSem.tryAcquire(1, TimeUnit.SECONDS)); assertEquals(lastSuccessfulAssignmentAt, clusterController.getLastSuccessfulAssigmentAt()); }