ZookeeperBackedDatastreamStore implements DatastreamStore { @Override public void updateDatastream(String key, Datastream datastream, boolean notifyLeader) throws DatastreamException { Datastream oldDatastream = getDatastream(key); if (oldDatastream == null) { throw new DatastreamException("Datastream does not exists, can not be updated: " + key); } String json = DatastreamUtils.toJSON(datastream); _zkClient.writeData(getZnodePath(key), json); if (notifyLeader) { notifyLeaderOfDataChange(); } } 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);  }
@Test public void testUpdateDatastream() throws Exception { Datastream ds = generateDatastream(0); Assert.assertNull(_store.getDatastream(ds.getName())); _store.createDatastream(ds.getName(), ds); Datastream ds2 = _store.getDatastream(ds.getName()); Assert.assertNotNull(ds2); Assert.assertEquals(ds2, ds); ds.getMetadata().put("key", "value"); ds.getSource().setConnectionString("updated"); _store.updateDatastream(ds.getName(), ds, false); ds2 = _store.getDatastream(ds.getName()); Assert.assertEquals(ds, ds2); }