erdody commented on a change in pull request #10822:
URL: https://github.com/apache/kafka/pull/10822#discussion_r651509086
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
##########
@@ -193,8 +194,8 @@
private short backoffRetries;
// visible for testing
- // The pending restart requests for the connectors;
- final NavigableSet<RestartRequest> pendingRestartRequests = new
TreeSet<>();
+ // The latest pending restart requests for the connectors;
+ final Map<String, RestartRequest> pendingRestartRequests = new
ConcurrentHashMap<>();
Review comment:
AFAICS, since all accesses are synchronized, this doesn't need to be
concurrent.
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
##########
@@ -1063,6 +1076,104 @@ public int generation() {
return generation;
}
+ @Override
+ public void restartConnectorAndTasks(
+ RestartRequest request,
+ Callback<ConnectorStateInfo> callback
+ ) {
+ final String connectorName = request.connectorName();
+ addRequest(
+ () -> {
+ if (checkRebalanceNeeded(callback)) {
+ return null;
+ }
+ if
(!configState.connectors().contains(request.connectorName())) {
+ callback.onCompletion(new NotFoundException("Unknown
connector: " + connectorName), null);
+ return null;
+ }
+ if (isLeader()) {
+ // Write a restart request to the config backing
store, to be executed asynchronously in tick()
+ configBackingStore.putRestartRequest(request);
+ // Compute and send the response that this was accepted
+ Optional<RestartPlan> maybePlan =
buildRestartPlanFor(request);
+ if (!maybePlan.isPresent()) {
+ callback.onCompletion(new
NotFoundException("Status for connector " + connectorName + " not found",
null), null);
+ } else {
+ RestartPlan plan = maybePlan.get();
+ callback.onCompletion(null,
plan.restartConnectorStateInfo());
+ }
+ } else {
+ callback.onCompletion(new NotLeaderException("Cannot
process restart request since it is not assigned to this member", leaderUrl()),
null);
+ }
+
+ return null;
+ },
+ forwardErrorCallback(callback)
+ );
+ }
+
+ /**
+ * Process all pending restart requests. There can be at most one request
per connector, because of how
+ * {@link RestartRequest#equals(Object)} and {@link
RestartRequest#hashCode()} are based only on the connector name.
+ *
+ * <p>This method is called from within the {@link #tick()} method. It is
synchronized so that all pending restart requests
+ * are processed at once before any additional requests are added.
+ */
+ private synchronized void processRestartRequests() {
+ RestartRequest request;
+ while ((request = pendingRestartRequests.pollFirst()) != null) {
+ doRestartConnectorAndTasks(request);
+ }
+ }
+
+ protected synchronized boolean doRestartConnectorAndTasks(RestartRequest
request) {
+ final String connectorName = request.connectorName();
+ Optional<RestartPlan> maybePlan = buildRestartPlanFor(request);
+ if (!maybePlan.isPresent()) {
+ log.debug("Skipping restart of connector '{}' since no status is
available: {}", connectorName, request);
+ return false;
+ }
+ RestartPlan plan = maybePlan.get();
+ log.info("Executing {}", plan);
+
+
+ // If requested, stop the connector and any tasks, marking each as
restarting
+ final ExtendedAssignment currentAssignments = assignment;
+ final Collection<ConnectorTaskId> assignedIdsToRestart =
plan.taskIdsToRestart()
+ .stream()
+
.filter(taskId -> currentAssignments.tasks().contains(taskId))
+
.collect(Collectors.toList());
+ final boolean restartConnector = plan.restartConnector() &&
currentAssignments.connectors().contains(connectorName);
+ final boolean restartTasks = !assignedIdsToRestart.isEmpty();
+ if (restartConnector) {
+ worker.stopAndAwaitConnector(connectorName);
+ recordRestarting(connectorName);
+ }
+ if (restartTasks) {
+ // Stop the tasks and mark as restarting
+ worker.stopAndAwaitTasks(assignedIdsToRestart);
+ assignedIdsToRestart.forEach(this::recordRestarting);
+ }
+
+ // Now restart the connector and tasks
+ if (restartConnector) {
+ startConnector(connectorName, (error, targetState) -> {
+ if (error == null) {
+ log.info("Connector {} successfully restarted",
connectorName);
+ } else {
+ log.error("Failed to restart connector '" + connectorName
+ "'", error);
+ }
+ });
+ }
+ if (restartTasks) {
+ log.debug("Restarting {} of {} tasks for {}",
plan.restartTaskCount(), plan.totalTaskCount(), request);
+ plan.taskIdsToRestart().forEach(this::startTask);
+ log.debug("Restarted {} of {} tasks for {} as requested",
plan.restartTaskCount(), plan.totalTaskCount(), request);
+ }
+ log.info("Completed {}", plan);
+ return restartConnector || restartTasks;
Review comment:
The main problem is that you're testing the code you added for tests,
not that the actual actions are executed.
Unless you're also have coverage to verify that there's correspondence
between the different results and the actions that need to happen in each case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
