rhauch commented on a change in pull request #10822: URL: https://github.com/apache/kafka/pull/10822#discussion_r655660680
########## File path: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/RestartRequest.java ########## @@ -0,0 +1,144 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.connect.runtime; + +import java.util.Objects; + +import org.apache.kafka.connect.connector.Connector; +import org.apache.kafka.connect.connector.Task; + +/** + * A request to restart a connector and/or task instances. Review comment: Let's mention the natural sort order. ########## File path: connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectCluster.java ########## @@ -559,6 +560,31 @@ public String endpointForResource(String resource) { return url + resource; } + /** + * Get the full URL of the endpoint that corresponds to the given REST resource using a worker + * that is not running any tasks or connector instance for the connectorName provided in the arguments + * + * @param resource the resource under the worker's admin endpoint + * @param connectorName the name of the connector + * @return the admin endpoint URL + * @throws ConnectException if no REST endpoint is available + */ + public String endpointForWorkerRunningNoResourceForConnector(String resource, String connectorName) { Review comment: What do you think about returning `Optional<String>` here (if every worker has at least one instance for the given connector) rather than throwing an exception? ########## File path: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java ########## @@ -1592,6 +1731,28 @@ public void onSessionKeyUpdate(SessionKey sessionKey) { } } } + + @Override + public void onRestartRequest(RestartRequest request) { + log.info("Received and enqueuing {}", request); + + synchronized (DistributedHerder.this) { + String connectorName = request.connectorName(); + //preserve the highest impact request + if (pendingRestartRequests.containsKey(connectorName)) { + RestartRequest existingRequest = pendingRestartRequests.get(connectorName); + if (request.compareTo(existingRequest) > 0) { + log.debug("Overwriting existing {} and enqueuing the higher impact {}", existingRequest, request); + pendingRestartRequests.put(connectorName, request); + } else { + log.debug("Preserving existing higher impact {} and ignoring incoming {}", existingRequest, request); + } + } else { + pendingRestartRequests.put(connectorName, request); + } Review comment: If we were to change `pendingRestartRequests` to a `ConcurrentMap`, then we could use the `compute(...)` functionality that actually would work even if we removed the synchronization. I wonder if this is a bit more readable anyway. WDYT? ``` pendingRestartRequests.compute(connectorName, (k, existing) -> { if (existing == null || request.compareTo(existing) > 0) { log.debug("Overwriting existing {} and enqueuing the higher impact {}", existing, request); return request; } log.debug("Preserving existing higher impact {} and ignoring incoming {}", existing, request); return existing; }); ``` ########## File path: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java ########## @@ -1063,6 +1076,132 @@ 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("Only the leader can process restart requests.", leaderUrl()), null); + } + + return null; + }, + forwardErrorCallback(callback) + ); + } + + /** + * Process all pending restart requests. There can be at most one request per connector. + * + * <p>This method is called from within the {@link #tick()} method. + */ + void processRestartRequests() { + List<RestartRequest> restartRequests; + synchronized (this) { + if (pendingRestartRequests.isEmpty()) { + return; + } + //dequeue into a local list to minimize the work being done within the synchronized block + restartRequests = new ArrayList<>(pendingRestartRequests.values()); + pendingRestartRequests.clear(); + } + for (RestartRequest restartRequest : restartRequests) { + try { + doRestartConnectorAndTasks(restartRequest); + } catch (Exception e) { + log.warn("Unexpected error while trying to process " + restartRequest + ", the restart request will be skipped.", e); + } + } + } + + /** + * Builds and and executes a restart plan for the connector and its tasks from <code>request</code>. + * Execution of a plan involves triggering the stop of eligible connector/tasks and then queuing the start for eligible connector/tasks. + * + * @param request the request to restart connector and tasks + */ + protected synchronized void 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; + } + 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) { + try { + startConnector(connectorName, (error, targetState) -> { + if (error == null) { + log.info("Connector '{}' restart successful", connectorName); + } else { + log.error("Connector '{}' restart failed", connectorName, error); + } + }); + } catch (Throwable t) { + log.error("Connector '{}' restart failed", connectorName, t); + } Review comment: The `startConnector(...)` method _should_ handle most of the errors by calling the callback, but there are still a few potential errors that could happen before the worker actually tries starting the connector. Do you think this try-catch is needed, though, since the try-catch where this `doRestartConnectorAndTasks(...)` is called will catch and log any unexpected change? -- 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
