This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 2fe3fa79b5f NIFI-16086 Added Flow Registry Client Synchronization
Interval property (#11404)
2fe3fa79b5f is described below
commit 2fe3fa79b5f6e30f831dd49a43343a3bbe12b485
Author: Pierre Villard <[email protected]>
AuthorDate: Mon Jul 13 22:05:49 2026 +0200
NIFI-16086 Added Flow Registry Client Synchronization Interval property
(#11404)
Signed-off-by: David Handermann <[email protected]>
---
.../src/main/asciidoc/administration-guide.adoc | 2 +-
.../flow/git/AbstractGitFlowRegistryClient.java | 1 +
.../flow/NifiRegistryFlowRegistryClient.java | 3 +-
.../org/apache/nifi/controller/FlowController.java | 23 ++--
.../RegistryFlowSynchronizationTask.java | 119 +++++++++++++++++++++
.../RegistryFlowSynchronizationTaskTest.java | 69 ++++++++++++
.../registry/FileSystemFlowRegistryClient.java | 3 +-
7 files changed, 201 insertions(+), 19 deletions(-)
diff --git a/nifi-docs/src/main/asciidoc/administration-guide.adoc
b/nifi-docs/src/main/asciidoc/administration-guide.adoc
index a2af47b3055..c3cec4e8024 100644
--- a/nifi-docs/src/main/asciidoc/administration-guide.adoc
+++ b/nifi-docs/src/main/asciidoc/administration-guide.adoc
@@ -2891,7 +2891,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 [...]
diff --git
a/nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/src/main/java/org/apache/nifi/registry/flow/git/AbstractGitFlowRegistryClient.java
b/nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/src/main/java/org/apache/nifi/registry/flow/git/AbstractGitFlowRegistryClient.java
index fc0f2bf2d79..2fa3fac9953 100644
---
a/nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/src/main/java/org/apache/nifi/registry/flow/git/AbstractGitFlowRegistryClient.java
+++
b/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 abstract class AbstractGitFlowRegistryClient extends
AbstractFlowRegistry
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();
diff --git
a/nifi-extension-bundles/nifi-flow-registry-client-bundle/nifi-flow-registry-client-services/src/main/java/org/apache/nifi/registry/flow/NifiRegistryFlowRegistryClient.java
b/nifi-extension-bundles/nifi-flow-registry-client-bundle/nifi-flow-registry-client-services/src/main/java/org/apache/nifi/registry/flow/NifiRegistryFlowRegistryClient.java
index 4c82e0eaf58..8f3e06db762 100644
---
a/nifi-extension-bundles/nifi-flow-registry-client-bundle/nifi-flow-registry-client-services/src/main/java/org/apache/nifi/registry/flow/NifiRegistryFlowRegistryClient.java
+++
b/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 @@ public class NifiRegistryFlowRegistryClient extends
AbstractFlowRegistryClient {
protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
return Arrays.asList(
PROPERTY_URL,
- SSL_CONTEXT_SERVICE
+ SSL_CONTEXT_SERVICE,
+ SYNCHRONIZATION_INTERVAL
);
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index ede57506249..674e2464307 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -1390,24 +1390,17 @@ public class FlowController implements
ReportingTaskProvider, FlowAnalysisRulePr
}, 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 with Sync Interval [{} s] Check
Interval [{} s]", 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 {
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/RegistryFlowSynchronizationTask.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/RegistryFlowSynchronizationTask.java
new file mode 100644
index 00000000000..a2c94bbd36f
--- /dev/null
+++
b/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;
+ }
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/RegistryFlowSynchronizationTaskTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/RegistryFlowSynchronizationTaskTest.java
new file mode 100644
index 00000000000..8683d73cad6
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/RegistryFlowSynchronizationTaskTest.java
@@ -0,0 +1,69 @@
+/*
+ * 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 static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+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 = mock(FlowManager.class);
+ final RegistryFlowSynchronizationTask task = new
RegistryFlowSynchronizationTask(flowManager, DEFAULT_INTERVAL_SECONDS);
+
+ final FlowRegistryClientNode configuredClient =
mock(FlowRegistryClientNode.class);
+
when(configuredClient.getEffectivePropertyValue(AbstractFlowRegistryClient.SYNCHRONIZATION_INTERVAL)).thenReturn("10
min");
+
when(flowManager.getFlowRegistryClient("configured")).thenReturn(configuredClient);
+ assertEquals(600L, task.getEffectiveIntervalSeconds("configured"));
+
+ final FlowRegistryClientNode unconfiguredClient =
mock(FlowRegistryClientNode.class);
+
when(unconfiguredClient.getEffectivePropertyValue(AbstractFlowRegistryClient.SYNCHRONIZATION_INTERVAL)).thenReturn(null);
+
when(flowManager.getFlowRegistryClient("unconfigured")).thenReturn(unconfiguredClient);
+ assertEquals(DEFAULT_INTERVAL_SECONDS,
task.getEffectiveIntervalSeconds("unconfigured"));
+
+ when(flowManager.getFlowRegistryClient("missing")).thenReturn(null);
+ assertEquals(DEFAULT_INTERVAL_SECONDS,
task.getEffectiveIntervalSeconds("missing"));
+ }
+}
diff --git
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/flow/registry/FileSystemFlowRegistryClient.java
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/flow/registry/FileSystemFlowRegistryClient.java
index 1d7eebf0eda..1910247ed2a 100644
---
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/flow/registry/FileSystemFlowRegistryClient.java
+++
b/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.Files;
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
