@Test public void testUpdatePartitionAssignmentsWithPausedInstances() throws Exception { String datastreamGroupName = "dg1"; Datastream ds = generateDatastream(0); ds.setMetadata(new StringMap()); ds.getMetadata().put(DatastreamMetadataConstants.TASK_PREFIX, datastreamGroupName); String clusterName = "testcluster"; HostTargetAssignment targetAssignment = new HostTargetAssignment(ImmutableList.of("p-0", "p-1"), "instance1"); _zkClient.ensurePath(KeyBuilder.instance(clusterName, "instance1-0000")); _zkClient.ensurePath(KeyBuilder.instance(clusterName, "PAUSED_INSTANCE")); _store.updatePartitionAssignments(ds.getName(), ds, targetAssignment, true); }
@Override public void updatePartitionAssignments(String key, Datastream datastream, HostTargetAssignment targetAssignment, boolean notifyLeader) throws DatastreamException { Validate.notNull(datastream, "null datastream"); Validate.notNull(key, "null key for datastream" + datastream); verifyHostname(targetAssignment.getTargetHost()); long currentTime = System.currentTimeMillis(); String datastreamGroupName = DatastreamUtils.getTaskPrefix(datastream); String path = KeyBuilder.getTargetAssignmentPath(_cluster, datastream.getConnectorName(), datastreamGroupName); _zkClient.ensurePath(path); if (_zkClient.exists(path)) { String json = targetAssignment.toJson(); _zkClient.ensurePath(path + '/' + currentTime); _zkClient.writeData(path + '/' + currentTime, json); } if (notifyLeader) { try { _zkClient.writeData(KeyBuilder.getTargetAssignmentBase(_cluster, datastream.getConnectorName()), String.valueOf(System.currentTimeMillis())); } catch (Exception e) { LOG.warn("Failed to touch the assignment update", e); throw new DatastreamException(e); } } }
ZookeeperBackedDatastreamStore implements DatastreamStore { @Override public void updatePartitionAssignments(String key, Datastream datastream, HostTargetAssignment targetAssignment, boolean notifyLeader) throws DatastreamException { Validate.notNull(datastream, "null datastream"); Validate.notNull(key, "null key for datastream" + datastream); verifyHostname(targetAssignment.getTargetHost()); long currentTime = System.currentTimeMillis(); String datastreamGroupName = DatastreamUtils.getTaskPrefix(datastream); String path = KeyBuilder.getTargetAssignmentPath(_cluster, datastream.getConnectorName(), datastreamGroupName); _zkClient.ensurePath(path); if (_zkClient.exists(path)) { String json = targetAssignment.toJson(); _zkClient.ensurePath(path + '/' + currentTime); _zkClient.writeData(path + '/' + currentTime, json); } if (notifyLeader) { try { _zkClient.writeData(KeyBuilder.getTargetAssignmentBase(_cluster, datastream.getConnectorName()), String.valueOf(System.currentTimeMillis())); } catch (Exception e) { LOG.warn("Failed to touch the assignment update", e); throw new DatastreamException(e); } } } }
ZookeeperBackedDatastreamStore implements DatastreamStore { @Override public void updatePartitionAssignments(String key, Datastream datastream, HostTargetAssignment targetAssignment, boolean notifyLeader) throws DatastreamException { Validate.notNull(datastream, "null datastream"); Validate.notNull(key, "null key for datastream" + datastream); verifyHostname(targetAssignment.getTargetHost()); long currentTime = System.currentTimeMillis(); String datastreamGroupName = DatastreamUtils.getTaskPrefix(datastream); String path = KeyBuilder.getTargetAssignmentPath(_cluster, datastream.getConnectorName(), datastreamGroupName); _zkClient.ensurePath(path); if (_zkClient.exists(path)) { String json = targetAssignment.toJson(); _zkClient.ensurePath(path + '/' + currentTime); _zkClient.writeData(path + '/' + currentTime, json); } if (notifyLeader) { try { _zkClient.writeData(KeyBuilder.getTargetAssignmentBase(_cluster, datastream.getConnectorName()), String.valueOf(System.currentTimeMillis())); } catch (Exception e) { LOG.warn("Failed to touch the assignment update", e); throw new DatastreamException(e); } } } ZookeeperBackedDatastreamStore(CachedDatastreamReader datastreamCache, ZkClient zkClient, String cluster); }
ZookeeperBackedDatastreamStore implements DatastreamStore { @Override public void updatePartitionAssignments(String key, Datastream datastream, HostTargetAssignment targetAssignment, boolean notifyLeader) throws DatastreamException { Validate.notNull(datastream, "null datastream"); Validate.notNull(key, "null key for datastream" + datastream); verifyHostname(targetAssignment.getTargetHost()); long currentTime = System.currentTimeMillis(); String datastreamGroupName = DatastreamUtils.getTaskPrefix(datastream); String path = KeyBuilder.getTargetAssignmentPath(_cluster, datastream.getConnectorName(), datastreamGroupName); _zkClient.ensurePath(path); if (_zkClient.exists(path)) { String json = targetAssignment.toJson(); _zkClient.ensurePath(path + '/' + currentTime); _zkClient.writeData(path + '/' + currentTime, json); } if (notifyLeader) { try { _zkClient.writeData(KeyBuilder.getTargetAssignmentBase(_cluster, datastream.getConnectorName()), String.valueOf(System.currentTimeMillis())); } catch (Exception e) { LOG.warn("Failed to touch the assignment update", e); throw new DatastreamException(e); } } } 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 updatePartitionAssignments(String key, Datastream datastream, HostTargetAssignment targetAssignment, boolean notifyLeader) throws DatastreamException { Validate.notNull(datastream, "null datastream"); Validate.notNull(key, "null key for datastream" + datastream); verifyHostname(targetAssignment.getTargetHost()); long currentTime = System.currentTimeMillis(); String datastreamGroupName = DatastreamUtils.getTaskPrefix(datastream); String path = KeyBuilder.getTargetAssignmentPath(_cluster, datastream.getConnectorName(), datastreamGroupName); _zkClient.ensurePath(path); if (_zkClient.exists(path)) { String json = targetAssignment.toJson(); _zkClient.ensurePath(path + '/' + currentTime); _zkClient.writeData(path + '/' + currentTime, json); } if (notifyLeader) { try { _zkClient.writeData(KeyBuilder.getTargetAssignmentBase(_cluster, datastream.getConnectorName()), String.valueOf(System.currentTimeMillis())); } catch (Exception e) { LOG.warn("Failed to touch the assignment update", e); throw new DatastreamException(e); } } } 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);  }