kpatelatwork commented on a change in pull request #10822:
URL: https://github.com/apache/kafka/pull/10822#discussion_r656356999
##########
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();
Review comment:
fixed
##########
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>.
Review comment:
Thanks for catching this
--
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
