@Test(expectedExceptions = DatastreamAlreadyExistsException.class) public void testCreateDuplicateDatastreams() { try { Datastream ds = generateDatastream(0); _store.createDatastream(ds.getName(), ds); } catch (DatastreamAlreadyExistsException e) { Assert.fail(); } Datastream ds = generateDatastream(0); _store.createDatastream(ds.getName(), ds); }
@Override public void createDatastream(String key, Datastream datastream) { Validate.notNull(datastream, "null datastream"); Validate.notNull(key, "null key for datastream" + datastream); String path = getZnodePath(key); if (_zkClient.exists(path)) { String content = _zkClient.ensureReadData(path); String errorMessage = String.format("Datastream already exists: path=%s, content=%s", key, content); LOG.warn(errorMessage); throw new DatastreamAlreadyExistsException(errorMessage); } _zkClient.ensurePath(path); String json = DatastreamUtils.toJSON(datastream); _zkClient.writeData(path, json); }
ZookeeperBackedDatastreamStore implements DatastreamStore { @Override public void createDatastream(String key, Datastream datastream) { Validate.notNull(datastream, "null datastream"); Validate.notNull(key, "null key for datastream" + datastream); String path = getZnodePath(key); if (_zkClient.exists(path)) { String content = _zkClient.ensureReadData(path); String errorMessage = String.format("Datastream already exists: path=%s, content=%s", key, content); LOG.warn(errorMessage); throw new DatastreamAlreadyExistsException(errorMessage); } _zkClient.ensurePath(path); String json = DatastreamUtils.toJSON(datastream); _zkClient.writeData(path, json); } }
ZookeeperBackedDatastreamStore implements DatastreamStore { @Override public void createDatastream(String key, Datastream datastream) { Validate.notNull(datastream, "null datastream"); Validate.notNull(key, "null key for datastream" + datastream); String path = getZnodePath(key); if (_zkClient.exists(path)) { String content = _zkClient.ensureReadData(path); String errorMessage = String.format("Datastream already exists: path=%s, content=%s", key, content); LOG.warn(errorMessage); throw new DatastreamAlreadyExistsException(errorMessage); } _zkClient.ensurePath(path); String json = DatastreamUtils.toJSON(datastream); _zkClient.writeData(path, json); } ZookeeperBackedDatastreamStore(CachedDatastreamReader datastreamCache, ZkClient zkClient, String cluster); }
ZookeeperBackedDatastreamStore implements DatastreamStore { @Override public void createDatastream(String key, Datastream datastream) { Validate.notNull(datastream, "null datastream"); Validate.notNull(key, "null key for datastream" + datastream); String path = getZnodePath(key); if (_zkClient.exists(path)) { String content = _zkClient.ensureReadData(path); String errorMessage = String.format("Datastream already exists: path=%s, content=%s", key, content); LOG.warn(errorMessage); throw new DatastreamAlreadyExistsException(errorMessage); } _zkClient.ensurePath(path); String json = DatastreamUtils.toJSON(datastream); _zkClient.writeData(path, json); } ZookeeperBackedDatastreamStore(CachedDatastreamReader datastreamCache, ZkClient zkClient, String cluster); @Override Datastream getDatastream(String key); @Override Stream<String> getAllDatastreams(); @Override void updateDatastream(String key, Datastream datastream, boolean notifyLeader); @Override void createDatastream(String key, Datastream datastream); @Override void updatePartitionAssignments(String key, Datastream datastream, HostTargetAssignment targetAssignment,
      boolean notifyLeader); @Override void deleteDatastream(String key); }
ZookeeperBackedDatastreamStore implements DatastreamStore { @Override public void createDatastream(String key, Datastream datastream) { Validate.notNull(datastream, "null datastream"); Validate.notNull(key, "null key for datastream" + datastream); String path = getZnodePath(key); if (_zkClient.exists(path)) { String content = _zkClient.ensureReadData(path); String errorMessage = String.format("Datastream already exists: path=%s, content=%s", key, content); LOG.warn(errorMessage); throw new DatastreamAlreadyExistsException(errorMessage); } _zkClient.ensurePath(path); String json = DatastreamUtils.toJSON(datastream); _zkClient.writeData(path, json); } ZookeeperBackedDatastreamStore(CachedDatastreamReader datastreamCache, ZkClient zkClient, String cluster); @Override Datastream getDatastream(String key); @Override Stream<String> getAllDatastreams(); @Override void updateDatastream(String key, Datastream datastream, boolean notifyLeader); @Override void createDatastream(String key, Datastream datastream); @Override void updatePartitionAssignments(String key, Datastream datastream, HostTargetAssignment targetAssignment,
      boolean notifyLeader); @Override void deleteDatastream(String key);  }