ZkMetadataResolver implements MetadataResolver { @Override public DLMetadata resolve(URI uri) throws IOException { String dlPath = uri.getPath(); PathUtils.validatePath(dlPath); String[] parts = StringUtils.split(dlPath, '/'); if (null == parts || 0 == parts.length) { throw new IOException("Invalid dlPath to resolve dl metadata : " + dlPath); } for (int i = parts.length; i >= 0; i--) { String pathToResolve = String.format("/%s", StringUtils.join(parts, '/', 0, i)); byte[] data; try { data = zkc.get().getData(pathToResolve, false, new Stat()); } catch (KeeperException.NoNodeException nne) { continue; } catch (KeeperException ke) { throw new IOException("Fail to resolve dl path : " + pathToResolve); } catch (InterruptedException ie) { throw new IOException("Interrupted when resolving dl path : " + pathToResolve); } if (null == data || data.length == 0) { continue; } try { return DLMetadata.deserialize(uri, data); } catch (IOException ie) { } } throw new IOException("No bkdl config bound under dl path : " + dlPath); } ZkMetadataResolver(ZooKeeperClient zkc); @Override DLMetadata resolve(URI uri);  }
@Test(timeout = 60000) public void testResolveFailures() throws Exception { try { resolver.resolve(createURI("/unexisted/path")); fail("Should fail if no metadata resolved."); } catch (IOException e) { } Utils.zkCreateFullPathOptimistic(zkc, "/existed/path", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); try { resolver.resolve(createURI("/existed/path")); fail("Should fail if no metadata resolved."); } catch (IOException e) { } }
@Test(timeout = 60000) public void testResolve() throws Exception { DLMetadata dlMetadata = DLMetadata.create(bkdlConfig); dlMetadata.create(createURI("/messaging/distributedlog-testresolve")); DLMetadata dlMetadata2 = DLMetadata.create(bkdlConfig2); dlMetadata2.create(createURI("/messaging/distributedlog-testresolve/child")); assertEquals(dlMetadata, resolver.resolve(createURI("/messaging/distributedlog-testresolve"))); assertEquals(dlMetadata2, resolver.resolve(createURI("/messaging/distributedlog-testresolve/child"))); assertEquals(dlMetadata2, resolver.resolve(createURI("/messaging/distributedlog-testresolve/child/unknown"))); Utils.zkCreateFullPathOptimistic(zkc, "/messaging/distributedlog-testresolve/child/child2", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); assertEquals(dlMetadata2, resolver.resolve(createURI("/messaging/distributedlog-testresolve/child/child2"))); }