NIFI-10975 Add Kubernetes Leader Election and State Provider (#6779)

* NIFI-10975 Added Kubernetes Leader Election and State Provider
- Added Kubernetes Leader Election Manager based on Kubernetes Leases
- Added Kubernetes State Provider based on Kubernetes ConfigMaps
- Added nifi-kubernetes-client for generalized access to Fabric8 Kubernetes Client
- Added nifi.cluster.leader.election.implementation Property defaulting to CuratorLeaderElectionManager
- Refactored LeaderElectionManager to nifi-framework-api for Extension Discovering Manager
- Refactored shared ZooKeeper configuration to nifi-framework-cluster-zookeeper

* NIFI-10975 Updated Kubernetes Client and StateMap
- Upgraded Kubernetes Client from 6.2.0 to 6.3.0
- Added getStateVersion to StateMap and deprecated getVersion
- Updated Docker start.sh with additional properties

* NIFI-10975 Corrected MockStateManager.assertStateSet()
* NIFI-10975 Upgraded Kubernetes Client from 6.3.0 to 6.3.1
* NIFI-10975 Corrected unregister leader and disabled release on cancel

* NIFI-10975 Corrected findLeader handling of Lease expiration
- Changed LeaderElectionManager.getLeader() return to Optional String

* NIFI-10975 Corrected StandardNiFiServiceFacade handling of Optional Leader
* NIFI-10975 Changed getLeader() to call findLeader() to avoid stale cached values
* NIFI-10975 Updated LeaderElectionCommand to run LeaderElector in loop
* NIFI-10975 Rebased on project version 2.0.0-SNAPSHOT

* NIFI-10975 Corrected Gson and AspectJ versions
- Updated versions to match current main branch and avoid reverting

nifi-api/src/main/java/org/apache/nifi/components/state/StateMap.java

@@ -18,6 +18,7 @@
 package org.apache.nifi.components.state;
 
 import java.util.Map;
+import java.util.Optional;
 
 /**
  * Provides a representation of a component's state at some point in time.
@@ -29,13 +30,27 @@ public interface StateMap {
      * Though this number is monotonically increasing, it should not be expected to increment always
      * from X to X+1. I.e., version numbers may be skipped.
      *
+     * @deprecated This method should be replaced with getStateVersion()
+     *
      * @return the version associated with the state
      */
+    @Deprecated
     long getVersion();
 
     /**
-     * Returns the value associated with the given key
+     * Get state version is not guaranteed to be numeric, but can be used to compare against an expected version.
+     * The default implementation uses the available version number and considers -1 as indicating an empty version
      *
+     * @return State version or empty when not known
+     */
+    default Optional<String> getStateVersion() {
+        final long version = getVersion();
+        return version == -1 ? Optional.empty() : Optional.of(String.valueOf(version));
+    }
+
+    /**
+     * Returns the value associated with the given key
+     *n
      * @param key the key whose value should be retrieved
      * @return the value associated with the given key, or <code>null</code> if no value is associated
      *         with this key.

nifi-assembly/pom.xml

@@ -919,6 +919,12 @@ language governing permissions and limitations under the License. -->
             <version>2.0.0-SNAPSHOT</version>
             <type>nar</type>
         </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-framework-kubernetes-nar</artifactId>
+            <version>2.0.0-SNAPSHOT</version>
+            <type>nar</type>
+        </dependency>
         <!-- AspectJ library needed by the Java Agent used for native library loading (see bootstrap.conf) -->
         <dependency>
             <groupId>org.aspectj</groupId>

nifi-commons/nifi-kubernetes-client/pom.xml

@@ -0,0 +1,36 @@
+<?xml version="1.0"?>
+<!--
+  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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.nifi</groupId>
+        <artifactId>nifi-commons</artifactId>
+        <version>2.0.0-SNAPSHOT</version>
+    </parent>
+    <artifactId>nifi-kubernetes-client</artifactId>
+    <description>Minimal abstraction for access Kubernetes REST API Resources</description>
+    <dependencies>
+        <dependency>
+            <groupId>io.fabric8</groupId>
+            <artifactId>kubernetes-client-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.fabric8</groupId>
+            <artifactId>kubernetes-client</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

nifi-commons/nifi-kubernetes-client/src/main/java/org/apache/nifi/kubernetes/client/KubernetesClientProvider.java

@@ -0,0 +1,31 @@
+/*
+ * 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.nifi.kubernetes.client;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+
+/**
+ * Abstraction for providing a configured Kubernetes Client
+ */
+public interface KubernetesClientProvider {
+    /**
+     * Get configured Kubernetes Client
+     *
+     * @return Kubernetes Client
+     */
+    KubernetesClient getKubernetesClient();
+}

nifi-commons/nifi-kubernetes-client/src/main/java/org/apache/nifi/kubernetes/client/NamespaceProvider.java

@@ -0,0 +1,29 @@
+/*
+ * 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.nifi.kubernetes.client;
+
+/**
+ * Kubernetes Namespace Provider
+ */
+public interface NamespaceProvider {
+    /**
+     * Get Namespace
+     *
+     * @return Kubernetes Namespace
+     */
+    String getNamespace();
+}

nifi-commons/nifi-kubernetes-client/src/main/java/org/apache/nifi/kubernetes/client/ServiceAccountNamespaceProvider.java

@@ -0,0 +1,53 @@
+/*
+ * 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.nifi.kubernetes.client;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+/**
+ * Service Account Namespace Provider based on standard file location
+ */
+public class ServiceAccountNamespaceProvider implements NamespaceProvider {
+    protected static final String NAMESPACE_PATH = "/var/run/secrets/kubernetes.io/serviceaccount/namespace";
+
+    protected static final String DEFAULT_NAMESPACE = "default";
+
+    /**
+     * Get Namespace from Service Account location or return default namespace when not found
+     *
+     * @return Kubernetes Namespace
+     */
+    @Override
+    public String getNamespace() {
+        final Path namespacePath = Paths.get(NAMESPACE_PATH);
+        return Files.isReadable(namespacePath) ? getNamespace(namespacePath) : DEFAULT_NAMESPACE;
+    }
+
+    private String getNamespace(final Path namespacePath) {
+        try {
+            final byte[] bytes = Files.readAllBytes(namespacePath);
+            return new String(bytes, StandardCharsets.UTF_8).trim();
+        } catch (final IOException e) {
+            throw new UncheckedIOException("Read Service Account namespace failed", e);
+        }
+    }
+}

nifi-commons/nifi-kubernetes-client/src/main/java/org/apache/nifi/kubernetes/client/StandardKubernetesClientProvider.java

@@ -0,0 +1,44 @@
+/*
+ * 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.nifi.kubernetes.client;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientBuilder;
+
+/**
+ * Standard implementation of Kubernetes Client Provider with default properties
+ */
+public class StandardKubernetesClientProvider implements KubernetesClientProvider {
+    private volatile KubernetesClient kubernetesClient;
+
+    /**
+     * Get Kubernetes Client with default configuration discovery
+     *
+     * @return Kubernetes Client
+     */
+    @Override
+    public KubernetesClient getKubernetesClient() {
+        if (kubernetesClient == null) {
+            kubernetesClient = buildKubernetesClient();
+        }
+        return kubernetesClient;
+    }
+
+    private KubernetesClient buildKubernetesClient() {
+        return new KubernetesClientBuilder().build();
+    }
+}

nifi-commons/nifi-kubernetes-client/src/test/java/org/apache/nifi/kubernetes/client/ServiceAccountNamespaceProviderTest.java

@@ -0,0 +1,48 @@
+/*
+ * 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.nifi.kubernetes.client;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+class ServiceAccountNamespaceProviderTest {
+    ServiceAccountNamespaceProvider provider;
+
+    @BeforeEach
+    void setProvider() {
+        provider = new ServiceAccountNamespaceProvider();
+    }
+
+    @Test
+    void testGetNamespace() {
+        final String namespace = provider.getNamespace();
+
+        final Path namespacePath = Paths.get(ServiceAccountNamespaceProvider.NAMESPACE_PATH);
+        if (Files.isReadable(namespacePath)) {
+            assertNotNull(namespace);
+        } else {
+            assertEquals(ServiceAccountNamespaceProvider.DEFAULT_NAMESPACE, namespace);
+        }
+    }
+}

nifi-commons/nifi-kubernetes-client/src/test/java/org/apache/nifi/kubernetes/client/StandardKubernetesClientProviderTest.java

@@ -0,0 +1,41 @@
+/*
+ * 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.nifi.kubernetes.client;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+class StandardKubernetesClientProviderTest {
+    StandardKubernetesClientProvider provider;
+
+    @BeforeEach
+    void setProvider() {
+        provider = new StandardKubernetesClientProvider();
+    }
+
+    @Timeout(5)
+    @Test
+    void testGetKubernetesClient() {
+        final KubernetesClient kubernetesClient = provider.getKubernetesClient();
+
+        assertNotNull(kubernetesClient);
+    }
+}

nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java

@@ -252,6 +252,7 @@ public class NiFiProperties extends ApplicationProperties {
     public static final String CLUSTER_PROTOCOL_HEARTBEAT_INTERVAL = "nifi.cluster.protocol.heartbeat.interval";
     public static final String CLUSTER_PROTOCOL_HEARTBEAT_MISSABLE_MAX = "nifi.cluster.protocol.heartbeat.missable.max";
     public static final String CLUSTER_PROTOCOL_IS_SECURE = "nifi.cluster.protocol.is.secure";
+    public static final String CLUSTER_LEADER_ELECTION_IMPLEMENTATION = "nifi.cluster.leader.election.implementation";
 
     // cluster node properties
     public static final String CLUSTER_IS_NODE = "nifi.cluster.is.node";
@@ -405,6 +406,7 @@ public class NiFiProperties extends ApplicationProperties {
     public static final String DEFAULT_CLUSTER_NODE_READ_TIMEOUT = "5 sec";
     public static final String DEFAULT_CLUSTER_NODE_CONNECTION_TIMEOUT = "5 sec";
     public static final int DEFAULT_CLUSTER_NODE_MAX_CONCURRENT_REQUESTS = 100;
+    public static final String DEFAULT_CLUSTER_LEADER_ELECTION_IMPLEMENTATION = "CuratorLeaderElectionManager";
 
     // cluster node defaults
     public static final int DEFAULT_CLUSTER_NODE_PROTOCOL_THREADS = 10;

nifi-commons/pom.xml

@@ -33,6 +33,7 @@
         <module>nifi-hl7-query-language</module>
         <module>nifi-json-utils</module>
         <module>nifi-jetty-configuration</module>
+        <module>nifi-kubernetes-client</module>
         <module>nifi-logging-utils</module>
         <module>nifi-metrics</module>
         <module>nifi-parameter</module>

nifi-docker/dockerhub/sh/start.sh

@@ -89,6 +89,10 @@ prop_replace 'nifi.cluster.flow.election.max.wait.time'     "${NIFI_ELECTION_MAX
 prop_replace 'nifi.cluster.flow.election.max.candidates'    "${NIFI_ELECTION_MAX_CANDIDATES:-}"
 prop_replace 'nifi.web.proxy.context.path'                  "${NIFI_WEB_PROXY_CONTEXT_PATH:-}"
 
+# Set leader election and state management properties
+prop_replace 'nifi.cluster.leader.election.implementation'      "${NIFI_LEADER_ELECTION_IMPLEMENTATION:-CuratorLeaderElectionManager}"
+prop_replace 'nifi.state.management.provider.cluster'           "${NIFI_STATE_MANAGEMENT_CLUSTER_PROVIDER:-zk-provider}"
+
 # Set analytics properties
 prop_replace 'nifi.analytics.predict.enabled'                   "${NIFI_ANALYTICS_PREDICT_ENABLED:-false}"
 prop_replace 'nifi.analytics.predict.interval'                  "${NIFI_ANALYTICS_PREDICT_INTERVAL:-3 mins}"

nifi-docs/src/main/asciidoc/administration-guide.adoc

@@ -4193,6 +4193,16 @@ Configure these properties for cluster nodes.
 |====
 |*Property*|*Description*
 |`nifi.cluster.is.node`|Set this to `true` if the instance is a node in a cluster. The default value is `false`.
+|`nifi.cluster.leader.election.implementation`|The Cluster Leader Election implementation class name or simple class
+name.
+
+The default value is `CuratorLeaderElectionManager` for ZooKeeper Leader Election using `nifi.zookeeper` settings.
+
+The implementation can be set to `KubernetesLeaderElectionManager` for Leader Election using
+link:https://kubernetes.io/docs/concepts/architecture/leases/[Kubernetes Leases]. The Kubernetes namespace for Leases
+will be read from the
+link:https://kubernetes.io/docs/reference/access-authn-authz/service-accounts-admin/[Service Account] namespace secret.
+The Kubernetes namespace will be set to `default` if the Service Account secret is not found.
 |`nifi.cluster.node.address`|The fully qualified address of the node. It is blank by default.
 |`nifi.cluster.node.protocol.port`|The node's protocol port. It is blank by default.
 |`nifi.cluster.node.protocol.max.threads`|The maximum number of threads that should be used to communicate with other nodes in the cluster. This property defaults to `50`. When a request is made to one node, it must be forwarded to the coordinator. The coordinator then replicates it to all nodes. There could be up to `n+2` threads for a given request, where `n` = number of nodes in your cluster. As an example, if 4 requests are made, a 5 node cluster will use `4 * 7 = 28` threads.