@Test public void testDeleteTasksWithPrefix() throws InterruptedException { String testCluster = "testDeleteTaskWithPrefix"; String connectorType = "connectorType"; ZkClientInterceptingAdapter adapter = createInterceptingZkAdapter(testCluster); adapter.connect(); adapter._zkClient.ensurePath(KeyBuilder.connector(testCluster, connectorType)); Assert.assertEquals(adapter.cleanUpOrphanConnectorTaskLocks(false), 0); DatastreamTaskImpl lockTask = new DatastreamTaskImpl(); lockTask.setId("task" + "lock"); lockTask.setTaskPrefix("taskPrefix" + "lock"); lockTask.setConnectorType(connectorType); lockTask.setZkAdapter(adapter); List<DatastreamTask> tasks = new ArrayList<>(); for (int i = 0; i < 10; i++) { DatastreamTaskImpl dsTask = new DatastreamTaskImpl(); dsTask.setId("task" + i); String taskPrefix = "taskPrefix" + i; dsTask.setTaskPrefix(taskPrefix); dsTask.setConnectorType(connectorType); dsTask.setZkAdapter(adapter); tasks.add(dsTask); adapter.acquireTask(dsTask, Duration.ofSeconds(2)); } updateInstanceAssignment(adapter, adapter.getInstanceName(), tasks); adapter.acquireTask(lockTask, Duration.ofSeconds(2)); ZkClient zkClient = Mockito.spy(adapter.getZkClient()); for (int j = 0; j < 8; j++) { adapter.deleteTasksWithPrefix(connectorType, "taskPrefix" + j); } Mockito.verify(zkClient, Mockito.never()).getChildren(any()); Mockito.verify(zkClient, Mockito.never()).getChildren(any(), anyBoolean()); List<String> leftOverTasks = zkClient.getChildren(KeyBuilder.connector(testCluster, connectorType)); Assert.assertEquals(leftOverTasks.size(), 3); adapter.cleanUpOrphanConnectorTasks(false); leftOverTasks = zkClient.getChildren(KeyBuilder.connector(testCluster, connectorType)); Assert.assertEquals(leftOverTasks.size(), 3); Assert.assertEquals(adapter.cleanUpOrphanConnectorTaskLocks(false), 1 * 2); Map<String, Set<String>> taskLocks = adapter.getAllConnectorTaskLocks(connectorType); Assert.assertEquals(taskLocks.size(), 11); Assert.assertEquals(taskLocks.get("taskPrefix0").size(), 1); Assert.assertEquals(taskLocks.get("taskPrefixlock").size(), 1); leftOverTasks = zkClient.getChildren(KeyBuilder.connector(testCluster, connectorType)); Assert.assertEquals(leftOverTasks.size(), 3); updateInstanceAssignment(adapter, adapter.getInstanceName(), Collections.emptyList()); leftOverTasks = zkClient.getChildren(KeyBuilder.connector(testCluster, connectorType)); Assert.assertEquals(leftOverTasks.size(), 3); adapter.cleanUpOrphanConnectorTasks(false); leftOverTasks = zkClient.getChildren(KeyBuilder.connector(testCluster, connectorType)); Assert.assertEquals(leftOverTasks.size(), 3); adapter.cleanUpOrphanConnectorTasks(true); Assert.assertEquals(adapter.cleanUpOrphanConnectorTaskLocks(true), 11 * 2); taskLocks = adapter.getAllConnectorTaskLocks(connectorType); Assert.assertEquals(taskLocks.size(), 11); Assert.assertEquals(taskLocks.get("taskPrefix0").size(), 1); Assert.assertEquals(taskLocks.get("taskPrefixlock").size(), 1); Thread.sleep(5000); taskLocks = adapter.getAllConnectorTaskLocks(connectorType); Assert.assertEquals(taskLocks.size(), 0); leftOverTasks = zkClient.getChildren(KeyBuilder.connector(testCluster, connectorType)); Assert.assertEquals(leftOverTasks.size(), 1); Assert.assertEquals(leftOverTasks.get(0), KeyBuilder.DATASTREAM_TASK_LOCK_ROOT_NAME); adapter.releaseTask(lockTask); leftOverTasks = zkClient.getChildren(KeyBuilder.connector(testCluster, connectorType)); Assert.assertEquals(leftOverTasks.size(), 1); adapter.disconnect(); }
public void deleteTasksWithPrefix(String connector, String taskPrefix) { Set<String> tasksToDelete = _liveTaskMap.values() .stream() .flatMap(Collection::stream) .filter(x -> x.getTaskPrefix().equals(taskPrefix) && x.getConnectorType().equals(connector)) .map(DatastreamTask::getDatastreamTaskName) .collect(Collectors.toSet()); LOG.info("Cleaning up stale connector tasks: {} for connector: {}", tasksToDelete, connector); for (String taskToDelete : tasksToDelete) { deleteConnectorTask(connector, taskToDelete); } }
ZkAdapter { public void deleteTasksWithPrefix(String connector, String taskPrefix) { Set<String> tasksToDelete = _liveTaskMap.values() .stream() .flatMap(Collection::stream) .filter(x -> x.getTaskPrefix().equals(taskPrefix) && x.getConnectorType().equals(connector)) .map(DatastreamTask::getDatastreamTaskName) .collect(Collectors.toSet()); LOG.info("Cleaning up stale connector tasks: {} for connector: {}", tasksToDelete, connector); for (String taskToDelete : tasksToDelete) { deleteConnectorTask(connector, taskToDelete); } } }
ZkAdapter { public void deleteTasksWithPrefix(String connector, String taskPrefix) { Set<String> tasksToDelete = _liveTaskMap.values() .stream() .flatMap(Collection::stream) .filter(x -> x.getTaskPrefix().equals(taskPrefix) && x.getConnectorType().equals(connector)) .map(DatastreamTask::getDatastreamTaskName) .collect(Collectors.toSet()); LOG.info("Cleaning up stale connector tasks: {} for connector: {}", tasksToDelete, connector); for (String taskToDelete : tasksToDelete) { deleteConnectorTask(connector, taskToDelete); } } ZkAdapter(String zkServers, String cluster, String defaultTransportProviderName, int sessionTimeoutMs,
      int connectionTimeoutMs, int operationRetryTimeoutMs, long debounceTimerMs, ZkAdapterListener listener); @VisibleForTesting  ZkAdapter(String zkServers, String cluster, String defaultTransportProviderName, int sessionTimeoutMs,
      int connectionTimeoutMs, long debounceTimerMs, ZkAdapterListener listener); }
ZkAdapter { public void deleteTasksWithPrefix(String connector, String taskPrefix) { Set<String> tasksToDelete = _liveTaskMap.values() .stream() .flatMap(Collection::stream) .filter(x -> x.getTaskPrefix().equals(taskPrefix) && x.getConnectorType().equals(connector)) .map(DatastreamTask::getDatastreamTaskName) .collect(Collectors.toSet()); LOG.info("Cleaning up stale connector tasks: {} for connector: {}", tasksToDelete, connector); for (String taskToDelete : tasksToDelete) { deleteConnectorTask(connector, taskToDelete); } } ZkAdapter(String zkServers, String cluster, String defaultTransportProviderName, int sessionTimeoutMs,
      int connectionTimeoutMs, int operationRetryTimeoutMs, long debounceTimerMs, ZkAdapterListener listener); @VisibleForTesting  ZkAdapter(String zkServers, String cluster, String defaultTransportProviderName, int sessionTimeoutMs,
      int connectionTimeoutMs, long debounceTimerMs, ZkAdapterListener listener); synchronized boolean isLeader(); String getInstanceName(); void disconnect(); void connect(); boolean updateDatastream(Datastream datastream); void deleteTasksWithPrefix(String connector, String taskPrefix); void deleteDatastream(String datastreamName); List<String> getAllInstances(); void touchAllInstanceAssignments(); List<String> getLiveInstances(); List<String> getInstanceAssignment(String instance); Map<String, Set<DatastreamTask>> getAllAssignedDatastreamTasks(); DatastreamTaskImpl getAssignedDatastreamTask(String instance, String taskName); void updateAllAssignments(Map<String, List<DatastreamTask>> assignmentsByInstance); void addConnectorType(String connectorType); void zkSaveInstanceError(String message); String getDatastreamTaskStateForKey(DatastreamTask datastreamTask, String key); void setDatastreamTaskStateForKey(DatastreamTask datastreamTask, String key, String value); void cleanUpDeadInstanceDataAndOtherUnusedTasks(
      Map<String, Set<DatastreamTask>> previousAssignmentByInstance,
      Map<String, List<DatastreamTask>> newAssignmentsByInstance,
      List<String> liveInstances); void cleanUpOldUnusedTasksFromConnector(Map<String, Set<DatastreamTask>> previousAssignmentByInstance,
      Map<String, List<DatastreamTask>> newAssignmentsByInstance); int cleanUpOrphanConnectorTasks(boolean cleanUpOrphanTasksInConnector); int cleanUpOrphanConnectorTaskLocks(boolean cleanUpOrphanTaskLocksInConnector); void acquireTask(DatastreamTaskImpl task, Duration timeout); boolean checkIsTaskLocked(String connectorType, String taskPrefix, String taskName); void waitForDependencies(DatastreamTaskImpl task, Duration timeout); void cleanUpPartitionMovement(String connectorType, String datastreamGroupName, long timestamp); List<String> getDatastreamsNeedPartitionMovement(String connectorType); Map<String, Set<String>> getPartitionMovement(String connectorType, String datastreamGroupName,
      long timestamp); void releaseTask(DatastreamTaskImpl task); static String parseHostnameFromZkInstance(String instance); }
ZkAdapter { public void deleteTasksWithPrefix(String connector, String taskPrefix) { Set<String> tasksToDelete = _liveTaskMap.values() .stream() .flatMap(Collection::stream) .filter(x -> x.getTaskPrefix().equals(taskPrefix) && x.getConnectorType().equals(connector)) .map(DatastreamTask::getDatastreamTaskName) .collect(Collectors.toSet()); LOG.info("Cleaning up stale connector tasks: {} for connector: {}", tasksToDelete, connector); for (String taskToDelete : tasksToDelete) { deleteConnectorTask(connector, taskToDelete); } } ZkAdapter(String zkServers, String cluster, String defaultTransportProviderName, int sessionTimeoutMs,
      int connectionTimeoutMs, int operationRetryTimeoutMs, long debounceTimerMs, ZkAdapterListener listener); @VisibleForTesting  ZkAdapter(String zkServers, String cluster, String defaultTransportProviderName, int sessionTimeoutMs,
      int connectionTimeoutMs, long debounceTimerMs, ZkAdapterListener listener); synchronized boolean isLeader(); String getInstanceName(); void disconnect(); void connect(); boolean updateDatastream(Datastream datastream); void deleteTasksWithPrefix(String connector, String taskPrefix); void deleteDatastream(String datastreamName); List<String> getAllInstances(); void touchAllInstanceAssignments(); List<String> getLiveInstances(); List<String> getInstanceAssignment(String instance); Map<String, Set<DatastreamTask>> getAllAssignedDatastreamTasks(); DatastreamTaskImpl getAssignedDatastreamTask(String instance, String taskName); void updateAllAssignments(Map<String, List<DatastreamTask>> assignmentsByInstance); void addConnectorType(String connectorType); void zkSaveInstanceError(String message); String getDatastreamTaskStateForKey(DatastreamTask datastreamTask, String key); void setDatastreamTaskStateForKey(DatastreamTask datastreamTask, String key, String value); void cleanUpDeadInstanceDataAndOtherUnusedTasks(
      Map<String, Set<DatastreamTask>> previousAssignmentByInstance,
      Map<String, List<DatastreamTask>> newAssignmentsByInstance,
      List<String> liveInstances); void cleanUpOldUnusedTasksFromConnector(Map<String, Set<DatastreamTask>> previousAssignmentByInstance,
      Map<String, List<DatastreamTask>> newAssignmentsByInstance); int cleanUpOrphanConnectorTasks(boolean cleanUpOrphanTasksInConnector); int cleanUpOrphanConnectorTaskLocks(boolean cleanUpOrphanTaskLocksInConnector); void acquireTask(DatastreamTaskImpl task, Duration timeout); boolean checkIsTaskLocked(String connectorType, String taskPrefix, String taskName); void waitForDependencies(DatastreamTaskImpl task, Duration timeout); void cleanUpPartitionMovement(String connectorType, String datastreamGroupName, long timestamp); List<String> getDatastreamsNeedPartitionMovement(String connectorType); Map<String, Set<String>> getPartitionMovement(String connectorType, String datastreamGroupName,
      long timestamp); void releaseTask(DatastreamTaskImpl task); static String parseHostnameFromZkInstance(String instance);  }