kpatelatwork commented on a change in pull request #10822:
URL: https://github.com/apache/kafka/pull/10822#discussion_r655408049
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
##########
@@ -1063,6 +1076,112 @@ 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);
+ }
+ }
+ }
+
+ 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) {
+ 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);
+ }
Review comment:
@rhauch we can add a retryCount in the RestartRequest and start with a
default of 5 and decrement it every time we re-enqueue. If the tick() happens
every second then we can exhaust the retries pretty fast so we may need to add
backoff with a delay before processing the request. Adding retry logic to this
PR can make the PR complex, I would vote for adding it in a separate PR.
--
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
