ZookeeperBackedDatastreamStore implements DatastreamStore { @Override public Datastream getDatastream(String key) { if (key == null) { return null; } String path = getZnodePath(key); String json = _zkClient.readData(path, true ); if (json == null) { return null; } return DatastreamUtils.fromJSON(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);  }
@Test public void testGetCorruptedDatastream() { _zkClient.ensurePath("/testcluster/dms/datastream1"); String data = "Corrupted data"; _zkClient.writeData("/testcluster/dms/datastream1", data); Assert.assertNull(_store.getDatastream("datastream1")); }