### Question:
ZkStorageContainerManager extends AbstractLifecycleComponent<StorageConfiguration> implements StorageContainerManager, Consumer<Void> { private CompletableFuture<StorageContainer> startStorageContainer(long scId) { log.info("Starting storage container ({})", scId); StorageContainer sc = liveContainers.get(scId); if (null != sc) { log.warn("Storage container ({}) is already started", scId); return FutureUtils.value(sc); } else { pendingStartStopContainers.add(scId); return registry .startStorageContainer(scId) .whenComplete((container, cause) -> { try { if (null != cause) { log.warn("Failed to start storage container ({})", scId, cause); } else { log.info("Successfully started storage container ({})", scId); addStorageContainer(scId, container); } } finally { pendingStartStopContainers.remove(scId); } }); } } ZkStorageContainerManager(Endpoint myEndpoint,
                                     StorageConfiguration conf,
                                     ClusterMetadataStore clusterMetadataStore,
                                     StorageContainerRegistry registry,
                                     StatsLogger statsLogger); @Override Endpoint getStorageContainer(long scId); @Override void accept(Void aVoid);  }### Answer:
@Test public void testStartContainerOnFailures() throws Exception { scManager.close(); long containerId = 11L; AtomicBoolean returnGoodContainer = new AtomicBoolean(false); CompletableFuture<StorageContainer> startFuture = new CompletableFuture<>(); StorageContainer goodSc = createStorageContainer(containerId, startFuture, FutureUtils.Void()); mockScFactory = (scId) -> { if (returnGoodContainer.get()) { return goodSc; } else { return createStorageContainer( scId, FutureUtils.exception(new Exception("Failed to start")), FutureUtils.Void() ); } }; scRegistry = spy(new StorageContainerRegistryImpl(mockScFactory)); scManager = new ZkStorageContainerManager( myEndpoint, new StorageConfiguration(new CompositeConfiguration()) .setClusterControllerScheduleInterval(1, TimeUnit.SECONDS), clusterMetadataStore, scRegistry, NullStatsLogger.INSTANCE); scManager.start(); ClusterAssignmentData cad = ClusterAssignmentData.newBuilder() .putServers( NetUtils.endpointToString(myEndpoint), ServerAssignmentData.newBuilder() .addContainers(containerId) .build()) .build(); clusterMetadataStore.updateClusterAssignmentData(cad); verify(scRegistry, timeout(10000).atLeastOnce()).startStorageContainer(eq(containerId)); assertEquals(0, scManager.getLiveContainers().size()); returnGoodContainer.set(true); FutureUtils.complete(startFuture, goodSc); MoreAsserts.assertUtil( ignored -> scManager.getLiveContainers().size() >= 1, () -> null); assertEquals(1, scManager.getLiveContainers().size()); assertTrue(scManager.getLiveContainers().containsKey(containerId)); }