FederatedZKLogMetadataStore extends NamespaceWatcher implements LogMetadataStore, Watcher, Runnable,
        FutureEventListener<Set<URI>> { @Override public Future<URI> createLog(final String logName) { if (duplicatedLogFound.get()) { return duplicatedLogException(duplicatedLogName.get()); } Promise<URI> createPromise = new Promise<URI>(); doCreateLog(logName, createPromise); return postStateCheck(createPromise); } FederatedZKLogMetadataStore(
            DistributedLogConfiguration conf,
            URI namespace,
            ZooKeeperClient zkc,
            OrderedScheduler scheduler); static void createFederatedNamespace(URI namespace, ZooKeeperClient zkc); @Override void run(); @Override void onSuccess(Set<URI> uris); @Override void onFailure(Throwable cause); @Override void process(WatchedEvent watchedEvent); @Override Future<URI> createLog(final String logName); @Override Future<Optional<URI>> getLogLocation(final String logName); @Override Future<Iterator<String>> getLogs(); @Override void registerNamespaceListener(NamespaceListener listener);  }
@Test(timeout = 60000) public void testCreateLog() throws Exception { DistributedLogConfiguration conf = new DistributedLogConfiguration(); conf.addConfiguration(baseConf); ZooKeeperClient anotherZkc = TestZooKeeperClientBuilder.newBuilder() .uri(uri) .sessionTimeoutMs(zkSessionTimeoutMs) .build(); FederatedZKLogMetadataStore anotherMetadataStore = new FederatedZKLogMetadataStore(conf, uri, anotherZkc, scheduler); for (int i = 0; i < 2 * maxLogsPerSubnamespace; i++) { LogMetadataStore createStore, checkStore; if (i % 2 == 0) { createStore = metadataStore; checkStore = anotherMetadataStore; } else { createStore = anotherMetadataStore; checkStore = metadataStore; } String logName = "test-create-log-" + i; URI logUri = FutureUtils.result(createStore.createLog(logName)); Optional<URI> logLocation = FutureUtils.result(checkStore.getLogLocation(logName)); assertTrue("Log " + logName + " doesn't exist", logLocation.isPresent()); assertEquals("Different log location " + logLocation.get() + " is found", logUri, logLocation.get()); } assertEquals(2, metadataStore.getSubnamespaces().size()); assertEquals(2, anotherMetadataStore.getSubnamespaces().size()); }
@Test(timeout = 60000, expected = LogExistsException.class) public void testCreateLogWhenLogExists() throws Exception { String logName = "test-create-log-when-log-exists"; URI logUri = FutureUtils.result(metadataStore.createLog(logName)); assertEquals(uri, logUri); FutureUtils.result(metadataStore.createLog(logName)); }