Allow configuring additional options when restarting connectors

CHANGELOG.md

@@ -16,6 +16,8 @@
 * Extend the EntityOperator, Cruise Control and KafkaExporter deployment to support PDB via the template section in the CR spec.
 * Added support for [KIP-1073](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1073:+Return+fenced+brokers+in+DescribeCluster+response)
   to get the list of the registered brokers by using the Kafka Admin API. It replaces the usage of the `.status.registeredNodeIds` field in Kafka.
+* Added support for [KIP-745](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=181308623) in Kafka Connector and Mirror Maker 2, allowing the usage of 
+  `includeTasks` and `onlyFailed` arguments in kafka connectors restart.
 
 ### Major changes, deprecations and removals
 

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/AbstractConnectOperator.java

@@ -7,6 +7,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.fabric8.kubernetes.api.model.ConfigMap;
 import io.fabric8.kubernetes.api.model.DefaultKubernetesResourceList;
+import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.LocalObjectReference;
 import io.fabric8.kubernetes.api.model.ObjectMeta;
 import io.fabric8.kubernetes.api.model.ServiceAccount;
@@ -131,6 +132,17 @@ public abstract class AbstractConnectOperator<C extends KubernetesClient, T exte
     protected final SharedEnvironmentProvider sharedEnvironmentProvider;
     protected final int port;
 
+    /**
+     * This optional argument can be used to include tasks in the restart connector operation.
+     * */
+    protected static final String STRIMZI_IO_RESTART_INCLUDE_TASKS_ARG = "includeTasks";
+
+    /**
+     * This optional argument can be used to restart connector only failed tasks.
+     * */
+    protected static final String STRIMZI_IO_RESTART_ONLY_FAILED_ARG = "onlyFailed";
+
+
     /**
      * Constructor
      *
@@ -746,8 +758,18 @@ private static int nextAutoRestartBackOffIntervalInMinutes(int restartCount)
     @SuppressWarnings({ "rawtypes" })
     private Future<List<Condition>> maybeRestartConnector(Reconciliation reconciliation, String host, KafkaConnectApi apiClient, String connectorName, CustomResource resource, List<Condition> conditions) {
         if (hasRestartAnnotation(resource, connectorName)) {
-            LOGGER.debugCr(reconciliation, "Restarting connector {}", connectorName);
-            return VertxUtil.completableFutureToVertxFuture(apiClient.restart(host, port, connectorName, false, false))
+
+            if (!restartAnnotationIsValid(resource)) {
+                LOGGER.warnCr(reconciliation, "Invalid annotation format");
+                conditions.add(StatusUtils.buildWarningCondition("RestartConnector", "Invalid annotation format"));
+                return Future.succeededFuture(conditions);
+            }
+
+            boolean restartIncludeTasks = restartAnnotationHasIncludeTasksArg(resource);
+            boolean restartOnlyFailedTasks = restartAnnotationHasOnlyFailedTasksArg(resource);
+            LOGGER.infoCr(reconciliation, "Restarting connector {}, IncludeTasks {}, OnlyFailedTasks {}", connectorName, restartIncludeTasks, restartOnlyFailedTasks);
+
+            return VertxUtil.completableFutureToVertxFuture(apiClient.restart(host, port, connectorName, restartIncludeTasks, restartOnlyFailedTasks))
                     .compose(ignored -> removeRestartAnnotation(reconciliation, resource)
                         .compose(v -> Future.succeededFuture(conditions)),
                         throwable -> {
@@ -1059,6 +1081,36 @@ private Future<ConnectorStatusAndConditions> updateConnectorTopics(Reconciliatio
     @SuppressWarnings({ "rawtypes" })
     abstract boolean hasRestartAnnotation(CustomResource resource, String connectorName);
 
+    /**
+     * Checks if restart annotation value is valid
+     *
+     * @param resource          Resource instance to check
+     *
+     * @return True if the provided resource has valid restart annotation. False otherwise.
+     * */
+    @SuppressWarnings({ "rawtypes" })
+    abstract boolean restartAnnotationIsValid(CustomResource resource);
+
+    /**
+     * Checks whether the provided resource instance (a KafkaConnector or KafkaMirrorMaker2) has argument includeTasks in restart annotation.
+     *
+     * @param resource          Resource instance to check
+     *
+     * @return True if the provided resource has argument includeTasks in restart annotation. False otherwise.
+     */
+    @SuppressWarnings({ "rawtypes" })
+    abstract boolean restartAnnotationHasIncludeTasksArg(CustomResource resource);
+
+    /**
+     * Checks whether the provided resource instance (a KafkaConnector or KafkaMirrorMaker2) has argument onlyFailedTasks in restart annotation.
+     *
+     * @param resource          Resource instance to check
+     *
+     * @return True if the provided resource has argument onlyFailedTasks in restart annotation. False otherwise.
+     */
+    @SuppressWarnings({ "rawtypes" })
+    abstract boolean restartAnnotationHasOnlyFailedTasksArg(HasMetadata resource);
+
     /**
      * Returns the ID of the connector task to be restarted from the (a KafkaConnector or KafkaMirrorMaker2) custom resource.
      *

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaConnectAssemblyOperator.java

@@ -4,6 +4,7 @@
  */
 package io.strimzi.operator.cluster.operator.assembly;
 
+import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.LabelSelector;
 import io.fabric8.kubernetes.api.model.LabelSelectorBuilder;
 import io.fabric8.kubernetes.api.model.LabelSelectorRequirement;
@@ -62,6 +63,7 @@
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
 import static io.strimzi.api.ResourceAnnotations.ANNO_STRIMZI_IO_CONNECTOR_OFFSETS;
@@ -80,6 +82,13 @@ public class KafkaConnectAssemblyOperator extends AbstractConnectOperator<Kubern
     private final CrdOperator<KubernetesClient, KafkaConnector, KafkaConnectorList> connectorOperator;
     private final ConnectBuildOperator connectBuildOperator;
 
+    /**
+     * Pattern for validation of restart connector annotation value.
+     * */
+    private static final Pattern STRIMZI_IO_RESTART_ARGS_PATTERN = Pattern.compile("^includeTasks," +
+            "onlyFailed$|^onlyFailed,includeTasks$|^includeTasks$|^onlyFailed$|^true$");
+
+
     /**
      * Constructor
      *
@@ -613,7 +622,56 @@ private Future<Void> maybeUpdateConnectorStatus(Reconciliation reconciliation, K
     @SuppressWarnings({ "rawtypes" })
     @Override
     protected boolean hasRestartAnnotation(CustomResource resource, String connectorName) {
-        return Annotations.booleanAnnotation(resource, ANNO_STRIMZI_IO_RESTART, false);
+        return Annotations.stringAnnotation(resource, ANNO_STRIMZI_IO_RESTART, null) != null;
+    }
+
+    /**
+     * Checks if restart annotation value has valid combination of values. These are valid and not valid combinations:
+     * strimzi.io/restart=includeTasks,onlyFailed          # restart with args: includeTasks=true and onlyFailed=true
+     * strimzi.io/restart=includeTasks                     # restart with args: includeTasks=true and onlyFailed=false
+     * strimzi.io/restart=onlyFailed                       # restart with args: includeTasks=false and onlyFailed=true
+     * strimzi.io/restart=true                             # restart with args: includeTasks=false and onlyFailed=false
+     * strimzi.io/restart=false,includeTasks,onlyFailed    # do not restart, fail and log error because you can't set args and boolean value together
+     * strimzi.io/restart=true,includeTasks,onlyFailed     # do not restart, fail and log error because you can't set args and boolean value together
+     * strimzi.io/restart=includeTasks,wrongArg            # do not restart, fail and log error because wrongArg is not supported
+     *
+     * @param resource          Resource instance to check
+     *
+     * @return True if the provided resource has valid restart annotation. False otherwise.
+     * */
+    @SuppressWarnings({ "rawtypes" })
+    protected boolean restartAnnotationIsValid(CustomResource resource) {
+        String restartValue = Annotations.stringAnnotation(resource, ANNO_STRIMZI_IO_RESTART, "");
+        return STRIMZI_IO_RESTART_ARGS_PATTERN.matcher(restartValue).matches();
+    }
+
+    /**
+     * Checks whether the provided resource instance (a KafkaConnector or KafkaMirrorMaker2) has argument includeTasks in restart annotation.
+     *
+     * @param resource          Resource instance to check
+     *
+     * @return True if the provided resource has argument includeTasks in restart annotation. False otherwise.
+     */
+    @SuppressWarnings({ "rawtypes" })
+    @Override
+    protected boolean restartAnnotationHasIncludeTasksArg(CustomResource resource) {
+        return Annotations.stringAnnotation(resource, ANNO_STRIMZI_IO_RESTART, "")
+                .contains(STRIMZI_IO_RESTART_INCLUDE_TASKS_ARG);
+    }
+
+
+    /**
+     * Checks whether the provided resource instance (a KafkaConnector or KafkaMirrorMaker2) has argument onlyFailedTasks in restart annotation.
+     *
+     * @param resource          Resource instance to check
+     *
+     * @return True if the provided resource has argument onlyFailedTasks in restart annotation. False otherwise.
+     */
+    @SuppressWarnings({ "rawtypes" })
+    @Override
+    protected boolean restartAnnotationHasOnlyFailedTasksArg(HasMetadata resource) {
+        return Annotations.stringAnnotation(resource, ANNO_STRIMZI_IO_RESTART, "")
+                .contains(STRIMZI_IO_RESTART_ONLY_FAILED_ARG);
     }
 
     /**

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaMirrorMaker2AssemblyOperator.java

@@ -4,6 +4,7 @@
  */
 package io.strimzi.operator.cluster.operator.assembly;
 
+import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.rbac.ClusterRoleBinding;
 import io.fabric8.kubernetes.client.CustomResource;
 import io.fabric8.kubernetes.client.KubernetesClient;
@@ -42,6 +43,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
 import static io.strimzi.api.ResourceAnnotations.ANNO_STRIMZI_IO_CONNECTOR_OFFSETS;
@@ -63,6 +65,14 @@
 public class KafkaMirrorMaker2AssemblyOperator extends AbstractConnectOperator<KubernetesClient, KafkaMirrorMaker2, KafkaMirrorMaker2List, KafkaMirrorMaker2Spec, KafkaMirrorMaker2Status> {
     private static final ReconciliationLogger LOGGER = ReconciliationLogger.create(KafkaMirrorMaker2AssemblyOperator.class.getName());
 
+    /**
+     * Pattern for validation of MirrorMaker2 restart connector annotation value.
+     * */
+    private static final Pattern STRIMZI_IO_RESTART_CONNECTOR_MM2_ARGS_PATTERN = Pattern
+            .compile("^([a-zA-Z0-9-_]+):includeTasks,onlyFailed$|^([a-zA-Z0-9-_]+):onlyFailed,includeTasks$" +
+                    "|^([a-zA-Z0-9-_]+):includeTasks$|^([a-zA-Z0-9-_]+):onlyFailed$|^([a-zA-Z0-9-_]+)$");
+
+
     /**
      * Constructor
      *
@@ -311,7 +321,54 @@ private Future<Void> maybeUpdateMirrorMaker2Status(Reconciliation reconciliation
     @Override
     protected boolean hasRestartAnnotation(CustomResource resource, String connectorName) {
         String restartAnnotationConnectorName = Annotations.stringAnnotation(resource, ANNO_STRIMZI_IO_RESTART_CONNECTOR, null);
-        return connectorName.equals(restartAnnotationConnectorName);
+        return restartAnnotationConnectorName != null && restartAnnotationConnectorName.contains(connectorName);
+    }
+
+    /**
+     * Checks if restart annotation value has valid combination of values. These are valid and not valid combinations:
+     * strimzi.io/restart-connector=mirrormaker_connector_name:includeTasks,onlyFailed # restart with args: includeTasks=true and onlyFailed=true
+     * strimzi.io/restart-connector=mirrormaker_connector_name:includeTasks # restart with args: includeTasks=true and onlyFailed=false
+     * strimzi.io/restart-connector=mirrormaker_connector_name:onlyFailed # restart with args: includeTasks=false and onlyFailed=true
+     * strimzi.io/restart-connector=mirrormaker_connector_name # restart with args: includeTasks=false and onlyFailed=false
+     * strimzi.io/restart-connector=includeTasks,onlyFailed # do not restart, fail and log error because connector name is required
+     *
+     * @param resource          Resource instance to check
+     *
+     * @return True if the provided resource has valid restart annotation. False otherwise.
+     * */
+    @SuppressWarnings({ "rawtypes" })
+    protected boolean restartAnnotationIsValid(CustomResource resource) {
+        String restartValue = Annotations.stringAnnotation(resource, ANNO_STRIMZI_IO_RESTART_CONNECTOR, "");
+        return STRIMZI_IO_RESTART_CONNECTOR_MM2_ARGS_PATTERN.matcher(restartValue).matches();
+    }
+
+    /**
+     * Checks whether the provided resource instance (a KafkaConnector or KafkaMirrorMaker2) has argument includeTasks in restart annotation.
+     *
+     * @param resource          Resource instance to check
+     *
+     * @return True if the provided resource has argument includeTasks in restart annotation. False otherwise.
+     */
+    @SuppressWarnings({ "rawtypes" })
+    @Override
+    protected boolean restartAnnotationHasIncludeTasksArg(CustomResource resource) {
+        return Annotations.stringAnnotation(resource, ANNO_STRIMZI_IO_RESTART_CONNECTOR, "")
+                .contains(STRIMZI_IO_RESTART_INCLUDE_TASKS_ARG);
+    }
+
+
+    /**
+     * Checks whether the provided resource instance (a KafkaConnector or KafkaMirrorMaker2) has argument onlyFailedTasks in restart annotation.
+     *
+     * @param resource          Resource instance to check
+     *
+     * @return True if the provided resource has argument onlyFailedTasks in restart annotation. False otherwise.
+     */
+    @SuppressWarnings({ "rawtypes" })
+    @Override
+    protected boolean restartAnnotationHasOnlyFailedTasksArg(HasMetadata resource) {
+        return Annotations.stringAnnotation(resource, ANNO_STRIMZI_IO_RESTART_CONNECTOR, "")
+                .contains(STRIMZI_IO_RESTART_ONLY_FAILED_ARG);
     }
 
     /**