ZKAssistedDiscovery implements Discovery<byte[]>,
    Component<com.datatorrent.api.Context>, Configurable, Serializable { @Override public Collection<Service<byte[]>> discover() { try { new EnsurePath(basePath).ensure(curatorFramework.getZookeeperClient()); Collection<ServiceInstance<byte[]>> services = discovery.queryForInstances(serviceName); ArrayList<Service<byte[]>> returnable = new ArrayList<Service<byte[]>>(services.size()); for (final ServiceInstance<byte[]> service : services) { returnable.add(new Service<byte[]>() { @Override public String getHost() { return service.getAddress(); } @Override public int getPort() { return service.getPort(); } @Override public byte[] getPayload() { return service.getPayload(); } @Override public String getId() { return service.getId(); } @Override public String toString() { return "{" + getId() + " => " + getHost() + ':' + getPort() + '}'; } }); } return returnable; } catch (Exception e) { throw Throwables.propagate(e); } } ZKAssistedDiscovery(); @Override void unadvertise(Service<byte[]> service); @Override void advertise(Service<byte[]> service); void doAdvertise(Service<byte[]> service, boolean flag); @Override Collection<Service<byte[]>> discover(); @Override String toString(); @Override int hashCode(); @Override boolean equals(Object obj); String getConnectionString(); void setConnectionString(String connectionString); String getBasePath(); void setBasePath(String basePath); int getConnectionTimeoutMillis(); void setConnectionTimeoutMillis(int connectionTimeoutMillis); int getConnectionRetryCount(); void setConnectionRetryCount(int connectionRetryCount); int getConntectionRetrySleepMillis(); void setConntectionRetrySleepMillis(int conntectionRetrySleepMillis); String getServiceName(); void setServiceName(String serviceName); @Override void configure(org.apache.flume.Context context); @Override void setup(com.datatorrent.api.Context context); @Override void teardown();  }
@Test public void testDiscover() { ZKAssistedDiscovery discovery = new ZKAssistedDiscovery(); discovery.setServiceName("ApexFlumeTest"); discovery.setConnectionString("localhost:2181"); discovery.setBasePath("/HelloApex"); discovery.setup(null); assertNotNull("Discovered Sinks", discovery.discover()); discovery.teardown(); }