NIFI-16086 - Per Flow Registry Client sync frequency

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

@@ -2886,7 +2886,7 @@ This cleanup mechanism takes into account only automatically created archived _f
 |`nifi.flow.configuration.archive.max.count`*|The number of archive files allowed. NiFi will delete the oldest archive files so that only N latest archives can be kept, if this property is specified.
 |`nifi.flowcontroller.autoResumeState`|Indicates whether -upon restart- the components on the NiFi graph should return to their last state. When running in cluster, all nodes should have the same value. The default value is `true`.
 |`nifi.flowcontroller.graceful.shutdown.period`|Indicates the shutdown period. The default value is `10 secs`.
-|`nifi.flowcontroller.registry.sync.interval`|Specifies the recurring interval at which NiFi synchronizes the flow configuration with Flow Registry Clients. The default value is `30 min`.
+|`nifi.flowcontroller.registry.sync.interval`|Specifies the default recurring interval at which NiFi synchronizes the flow configuration with Flow Registry Clients. The default value is `30 min`. This value is used for any Flow Registry Client that does not configure its own `Synchronization Interval` property; a Flow Registry Client that sets that property is synchronized at its own interval instead.
 |`nifi.flowservice.writedelay.interval`|When many changes are made to the _flow.json_, this property specifies how long to wait before writing out the changes, so as to batch the changes into a single write. The default value is `500 ms`.
 |`nifi.administrative.yield.duration`|If a component allows an unexpected exception to escape, it is considered a bug. As a result, the framework will pause (or administratively yield) the component for this amount of time. This is done so that the component does not use up massive amounts of system resources, since it is known to have problems in the existing state. The default value is `30 secs`.
 |`nifi.bored.yield.duration`|When a component has no work to do (i.e., is "bored"), this is the amount of time it will wait before checking to see if it has new data to work on. This way, it does not use up CPU resources by checking for new work too often. When setting this property, be aware that it could add extra latency for components that do not constantly have work to do, as once they go into this "bored" state, they will wait this amount of time before checking for more work. The default value is `10 ms`.

nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/src/main/java/org/apache/nifi/registry/flow/git/AbstractGitFlowRegistryClient.java

@@ -162,6 +162,7 @@ public void initialize(final FlowRegistryClientInitializationContext context) {
         combinedPropertyDescriptors.add(PARAMETER_CONTEXT_VALUES);
         combinedPropertyDescriptors.add(COMMIT_AUTHOR_SOURCE);
         combinedPropertyDescriptors.add(SSL_CONTEXT_SERVICE);
+        combinedPropertyDescriptors.add(SYNCHRONIZATION_INTERVAL);
         propertyDescriptors = Collections.unmodifiableList(combinedPropertyDescriptors);
 
         flowSnapshotSerializer = createFlowSnapshotSerializer();

nifi-extension-bundles/nifi-flow-registry-client-bundle/nifi-flow-registry-client-services/src/main/java/org/apache/nifi/registry/flow/NifiRegistryFlowRegistryClient.java

@@ -114,7 +114,8 @@ private String getProposedUri(final FlowRegistryClientConfigurationContext conte
     protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
         return Arrays.asList(
                 PROPERTY_URL,
-                SSL_CONTEXT_SERVICE
+                SSL_CONTEXT_SERVICE,
+                SYNCHRONIZATION_INTERVAL
         );
     }
 

nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java

@@ -1390,24 +1390,18 @@ public void initializeFlow(final QueueProvider queueProvider) throws IOException
             }, 0L, 30L, TimeUnit.SECONDS);
 
             final String registrySyncInterval = nifiProperties.getProperty("nifi.flowcontroller.registry.sync.interval", "30 min");
-            final long registrySyncIntervalSeconds = FormatUtils.getTimeDuration(registrySyncInterval, TimeUnit.SECONDS);
+            final long defaultRegistrySyncIntervalSeconds = FormatUtils.getTimeDuration(registrySyncInterval, TimeUnit.SECONDS);
 
-            LOG.info("Scheduled Flow Registry synchronization every {}", registrySyncInterval);
+            // The synchronization task runs on a fixed tick but synchronizes each Flow Registry Client's Process Groups only when
+            // that client's configured interval has elapsed. The tick is bounded so that short per-client intervals are honored
+            // reasonably closely while avoiding needlessly frequent iterations for the typical (minutes) interval.
+            final long registrySyncTickSeconds = Math.max(1, Math.min(defaultRegistrySyncIntervalSeconds, 30));
 
-            // Schedule the flow registry synchronization task
-            timerDrivenEngineRef.get().scheduleWithFixedDelay(() -> {
-                final ProcessGroup rootGroup = flowManager.getRootGroup();
-                final List<ProcessGroup> allGroups = rootGroup.findAllProcessGroups();
-                allGroups.add(rootGroup);
+            LOG.info("Scheduled Flow Registry synchronization with a default interval of {} and a check interval of {} seconds; "
+                    + "individual Flow Registry Clients may override the interval via the Synchronization Interval property", registrySyncInterval, registrySyncTickSeconds);
 
-                for (final ProcessGroup group : allGroups) {
-                    try {
-                        group.synchronizeWithFlowRegistry(flowManager);
-                    } catch (final Exception e) {
-                        LOG.error("Failed to synchronize {} with Flow Registry", group, e);
-                    }
-                }
-            }, 300, registrySyncIntervalSeconds, TimeUnit.SECONDS);
+            final RegistryFlowSynchronizationTask registrySynchronizationTask = new RegistryFlowSynchronizationTask(flowManager, defaultRegistrySyncIntervalSeconds);
+            timerDrivenEngineRef.get().scheduleWithFixedDelay(registrySynchronizationTask, 300, registrySyncTickSeconds, TimeUnit.SECONDS);
 
             initialized.set(true);
         } finally {

nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/RegistryFlowSynchronizationTask.java

@@ -0,0 +1,119 @@
+/*
+ * 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.controller;
+
+import org.apache.nifi.controller.flow.FlowManager;
+import org.apache.nifi.groups.ProcessGroup;
+import org.apache.nifi.registry.flow.AbstractFlowRegistryClient;
+import org.apache.nifi.registry.flow.FlowRegistryClientNode;
+import org.apache.nifi.registry.flow.VersionControlInformation;
+import org.apache.nifi.util.FormatUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Periodic task that synchronizes version-controlled Process Groups with their external Flow Registries. The task is
+ * scheduled on a fixed tick, but each Process Group is only synchronized at the interval configured on its Flow
+ * Registry Client via {@link AbstractFlowRegistryClient#SYNCHRONIZATION_INTERVAL}. When a Flow Registry Client does not
+ * specify an interval, the configurable default (the {@code nifi.flowcontroller.registry.sync.interval} property) is
+ * used. Process Groups are grouped by their Flow Registry Client so that all groups belonging to a given client are
+ * synchronized together on that client's cadence.
+ */
+final class RegistryFlowSynchronizationTask implements Runnable {
+    private static final Logger logger = LoggerFactory.getLogger(RegistryFlowSynchronizationTask.class);
+
+    private final FlowManager flowManager;
+    private final long defaultIntervalSeconds;
+    private final Map<String, Long> lastSynchronizationTimestamps = new ConcurrentHashMap<>();
+
+    RegistryFlowSynchronizationTask(final FlowManager flowManager, final long defaultIntervalSeconds) {
+        this.flowManager = flowManager;
+        this.defaultIntervalSeconds = defaultIntervalSeconds;
+    }
+
+    @Override
+    public void run() {
+        final ProcessGroup rootGroup = flowManager.getRootGroup();
+        final List<ProcessGroup> allGroups = rootGroup.findAllProcessGroups();
+        allGroups.add(rootGroup);
+
+        final Map<String, List<ProcessGroup>> groupsByRegistryClientId = new HashMap<>();
+        for (final ProcessGroup group : allGroups) {
+            final VersionControlInformation versionControlInformation = group.getVersionControlInformation();
+            if (versionControlInformation == null) {
+                continue;
+            }
+
+            groupsByRegistryClientId.computeIfAbsent(versionControlInformation.getRegistryIdentifier(), key -> new ArrayList<>()).add(group);
+        }
+
+        final long now = System.currentTimeMillis();
+        for (final Map.Entry<String, List<ProcessGroup>> entry : groupsByRegistryClientId.entrySet()) {
+            final String registryClientId = entry.getKey();
+            final long intervalSeconds = getEffectiveIntervalSeconds(registryClientId);
+            final Long lastSynchronization = lastSynchronizationTimestamps.get(registryClientId);
+
+            if (lastSynchronization != null && (now - lastSynchronization) < TimeUnit.SECONDS.toMillis(intervalSeconds)) {
+                continue;
+            }
+
+            for (final ProcessGroup group : entry.getValue()) {
+                try {
+                    group.synchronizeWithFlowRegistry(flowManager);
+                } catch (final Exception e) {
+                    logger.error("Failed to synchronize {} with Flow Registry", group, e);
+                }
+            }
+
+            lastSynchronizationTimestamps.put(registryClientId, now);
+        }
+
+        // Stop tracking Flow Registry Clients that no longer have any version-controlled Process Groups so that a client
+        // is synchronized immediately if it is removed and later re-added.
+        lastSynchronizationTimestamps.keySet().retainAll(groupsByRegistryClientId.keySet());
+    }
+
+    long getEffectiveIntervalSeconds(final String registryClientId) {
+        final FlowRegistryClientNode clientNode = flowManager.getFlowRegistryClient(registryClientId);
+        if (clientNode == null) {
+            return defaultIntervalSeconds;
+        }
+
+        final String configuredInterval = clientNode.getEffectivePropertyValue(AbstractFlowRegistryClient.SYNCHRONIZATION_INTERVAL);
+        return parseIntervalSeconds(configuredInterval, defaultIntervalSeconds);
+    }
+
+    static long parseIntervalSeconds(final String configuredInterval, final long defaultIntervalSeconds) {
+        if (configuredInterval == null || configuredInterval.isBlank()) {
+            return defaultIntervalSeconds;
+        }
+
+        try {
+            return FormatUtils.getTimeDuration(configuredInterval.trim(), TimeUnit.SECONDS);
+        } catch (final IllegalArgumentException e) {
+            logger.warn("Configured Flow Registry Synchronization Interval [{}] is not valid; using default of {} seconds", configuredInterval, defaultIntervalSeconds);
+            return defaultIntervalSeconds;
+        }
+    }
+}

nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/RegistryFlowSynchronizationTaskTest.java

@@ -0,0 +1,68 @@
+/*
+ * 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.controller;
+
+import org.apache.nifi.controller.flow.FlowManager;
+import org.apache.nifi.registry.flow.AbstractFlowRegistryClient;
+import org.apache.nifi.registry.flow.FlowRegistryClientNode;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class RegistryFlowSynchronizationTaskTest {
+
+    private static final long DEFAULT_INTERVAL_SECONDS = 1800L;
+
+    @Test
+    void testParseIntervalSecondsFallsBackToDefaultWhenNotConfigured() {
+        assertEquals(DEFAULT_INTERVAL_SECONDS, RegistryFlowSynchronizationTask.parseIntervalSeconds(null, DEFAULT_INTERVAL_SECONDS));
+        assertEquals(DEFAULT_INTERVAL_SECONDS, RegistryFlowSynchronizationTask.parseIntervalSeconds("", DEFAULT_INTERVAL_SECONDS));
+        assertEquals(DEFAULT_INTERVAL_SECONDS, RegistryFlowSynchronizationTask.parseIntervalSeconds("   ", DEFAULT_INTERVAL_SECONDS));
+    }
+
+    @Test
+    void testParseIntervalSecondsParsesConfiguredDuration() {
+        assertEquals(300L, RegistryFlowSynchronizationTask.parseIntervalSeconds("5 min", DEFAULT_INTERVAL_SECONDS));
+        assertEquals(45L, RegistryFlowSynchronizationTask.parseIntervalSeconds("45 secs", DEFAULT_INTERVAL_SECONDS));
+        assertEquals(45L, RegistryFlowSynchronizationTask.parseIntervalSeconds("  45 secs  ", DEFAULT_INTERVAL_SECONDS));
+    }
+
+    @Test
+    void testParseIntervalSecondsFallsBackToDefaultWhenInvalid() {
+        assertEquals(DEFAULT_INTERVAL_SECONDS, RegistryFlowSynchronizationTask.parseIntervalSeconds("not-a-duration", DEFAULT_INTERVAL_SECONDS));
+    }
+
+    @Test
+    void testGetEffectiveIntervalSeconds() {
+        final FlowManager flowManager = Mockito.mock(FlowManager.class);
+        final RegistryFlowSynchronizationTask task = new RegistryFlowSynchronizationTask(flowManager, DEFAULT_INTERVAL_SECONDS);
+
+        final FlowRegistryClientNode configuredClient = Mockito.mock(FlowRegistryClientNode.class);
+        Mockito.when(configuredClient.getEffectivePropertyValue(AbstractFlowRegistryClient.SYNCHRONIZATION_INTERVAL)).thenReturn("10 min");
+        Mockito.when(flowManager.getFlowRegistryClient("configured")).thenReturn(configuredClient);
+        assertEquals(600L, task.getEffectiveIntervalSeconds("configured"));
+
+        final FlowRegistryClientNode unconfiguredClient = Mockito.mock(FlowRegistryClientNode.class);
+        Mockito.when(unconfiguredClient.getEffectivePropertyValue(AbstractFlowRegistryClient.SYNCHRONIZATION_INTERVAL)).thenReturn(null);
+        Mockito.when(flowManager.getFlowRegistryClient("unconfigured")).thenReturn(unconfiguredClient);
+        assertEquals(DEFAULT_INTERVAL_SECONDS, task.getEffectiveIntervalSeconds("unconfigured"));
+
+        Mockito.when(flowManager.getFlowRegistryClient("missing")).thenReturn(null);
+        assertEquals(DEFAULT_INTERVAL_SECONDS, task.getEffectiveIntervalSeconds("missing"));
+    }
+}

nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/flow/registry/FileSystemFlowRegistryClient.java

@@ -47,7 +47,6 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
@@ -87,7 +86,7 @@ public class FileSystemFlowRegistryClient extends AbstractFlowRegistryClient {
 
     @Override
     protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
-        return Collections.singletonList(DIRECTORY);
+        return List.of(DIRECTORY, SYNCHRONIZATION_INTERVAL);
     }
 
     @Override

pom.xml

@@ -118,7 +118,7 @@
         <node.version>v24.14.1</node.version>
 
         <!-- NiFi build -->
-        <nifi-api.version>2.9.0</nifi-api.version>
+        <nifi-api.version>2.10.0-SNAPSHOT</nifi-api.version>
         <nifi.nar.maven.plugin.version>2.3.0</nifi.nar.maven.plugin.version>
 
         <!-- CSPs SDK -->