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 054f49a16cc NIFI-16507 Add rebase handlers for name, concurrent tasks,
scheduled state and scheduling changes (#11853)
054f49a16cc is described below
commit 054f49a16cc7521533bbca57c7f13614b5b73414
Author: Pierre Villard <[email protected]>
AuthorDate: Mon Oct 26 20:11:14 2026 +0100
NIFI-16507 Add rebase handlers for name, concurrent tasks, scheduled state
and scheduling changes (#11853)
Signed-off-by: David Handermann <[email protected]>
---
.../diff/ConcurrentTasksChangedRebaseHandler.java | 76 +++++++++++
.../flow/diff/NameChangedRebaseHandler.java | 56 ++++++++
.../registry/flow/diff/RebaseConflictCode.java | 20 +++
.../registry/flow/diff/RebaseHandlerUtils.java | 41 ++++++
.../diff/ScheduledStateChangedRebaseHandler.java | 77 +++++++++++
.../flow/diff/SchedulingChangedRebaseHandler.java | 99 ++++++++++++++
.../registry/flow/diff/StandardRebaseEngine.java | 5 +
.../ConcurrentTasksChangedRebaseHandlerTest.java | 147 +++++++++++++++++++++
.../flow/diff/NameChangedRebaseHandlerTest.java | 127 ++++++++++++++++++
.../nifi/registry/flow/diff/RebaseEngineTest.java | 19 ++-
.../ScheduledStateChangedRebaseHandlerTest.java | 145 ++++++++++++++++++++
.../diff/SchedulingChangedRebaseHandlerTest.java | 143 ++++++++++++++++++++
12 files changed, 954 insertions(+), 1 deletion(-)
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/ConcurrentTasksChangedRebaseHandler.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/ConcurrentTasksChangedRebaseHandler.java
new file mode 100644
index 00000000000..70034b38781
--- /dev/null
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/ConcurrentTasksChangedRebaseHandler.java
@@ -0,0 +1,76 @@
+/*
+ * 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.registry.flow.diff;
+
+import org.apache.nifi.flow.VersionedComponent;
+import org.apache.nifi.flow.VersionedPort;
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
+import org.apache.nifi.flow.VersionedRemoteGroupPort;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Consumer;
+
+public class ConcurrentTasksChangedRebaseHandler implements RebaseHandler {
+
+ private static final Logger logger =
LoggerFactory.getLogger(ConcurrentTasksChangedRebaseHandler.class);
+
+ @Override
+ public DifferenceType getSupportedType() {
+ return DifferenceType.CONCURRENT_TASKS_CHANGED;
+ }
+
+ @Override
+ public RebaseAnalysis.ClassifiedDifference classify(final FlowDifference
localDifference, final Set<FlowDifference> upstreamDifferences,
+ final
VersionedProcessGroup targetSnapshot) {
+ final VersionedComponent localComponent =
localDifference.getComponentB();
+ if (getSetter(localComponent).isEmpty()) {
+ return
RebaseAnalysis.ClassifiedDifference.unsupported(localDifference,
RebaseConflictCode.UNSUPPORTED_COMPONENT_TYPE,
+ "Concurrent tasks change is not supported for component %s
of type %s".formatted(localComponent.getIdentifier(),
localComponent.getComponentType()));
+ }
+ if
(RebaseHandlerUtils.findConflictingUpstreamDifference(localDifference,
upstreamDifferences).isPresent()) {
+ return
RebaseAnalysis.ClassifiedDifference.conflicting(localDifference,
RebaseConflictCode.SAME_CONCURRENT_TASKS,
+ "Both local and upstream flows modified the concurrent
tasks of component " + localComponent.getIdentifier());
+ }
+ return RebaseAnalysis.ClassifiedDifference.compatible(localDifference);
+ }
+
+ @Override
+ public void apply(final FlowDifference localDifference, final
VersionedProcessGroup mergedFlow) {
+ final String componentIdentifier =
localDifference.getComponentB().getIdentifier();
+ final VersionedComponent component =
RebaseHandlerUtils.findComponentById(mergedFlow, componentIdentifier);
+ if (component == null) {
+ logger.warn("Unable to apply concurrent tasks change: component
[{}] not found in merged flow", componentIdentifier);
+ return;
+ }
+ getSetter(component).ifPresent(setter -> setter.accept((Integer)
localDifference.getValueB()));
+ }
+
+ private Optional<Consumer<Integer>> getSetter(final VersionedComponent
component) {
+ return switch (component) {
+ case VersionedProcessor processor ->
Optional.of(processor::setConcurrentlySchedulableTaskCount);
+ case VersionedPort port ->
Optional.of(port::setConcurrentlySchedulableTaskCount);
+ case VersionedRemoteGroupPort remoteGroupPort ->
Optional.of(remoteGroupPort::setConcurrentlySchedulableTaskCount);
+ case VersionedProcessGroup processGroup ->
Optional.of(processGroup::setMaxConcurrentTasks);
+ default -> Optional.empty();
+ };
+ }
+}
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/NameChangedRebaseHandler.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/NameChangedRebaseHandler.java
new file mode 100644
index 00000000000..23c530086e3
--- /dev/null
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/NameChangedRebaseHandler.java
@@ -0,0 +1,56 @@
+/*
+ * 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.registry.flow.diff;
+
+import org.apache.nifi.flow.VersionedComponent;
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Set;
+
+public class NameChangedRebaseHandler implements RebaseHandler {
+
+ private static final Logger logger =
LoggerFactory.getLogger(NameChangedRebaseHandler.class);
+
+ @Override
+ public DifferenceType getSupportedType() {
+ return DifferenceType.NAME_CHANGED;
+ }
+
+ @Override
+ public RebaseAnalysis.ClassifiedDifference classify(final FlowDifference
localDifference, final Set<FlowDifference> upstreamDifferences,
+ final
VersionedProcessGroup targetSnapshot) {
+ if
(RebaseHandlerUtils.findConflictingUpstreamDifference(localDifference,
upstreamDifferences).isPresent()) {
+ return
RebaseAnalysis.ClassifiedDifference.conflicting(localDifference,
RebaseConflictCode.SAME_COMPONENT_NAME,
+ "Both local and upstream flows modified the name of
component " + localDifference.getComponentB().getIdentifier());
+ }
+ return RebaseAnalysis.ClassifiedDifference.compatible(localDifference);
+ }
+
+ @Override
+ public void apply(final FlowDifference localDifference, final
VersionedProcessGroup mergedFlow) {
+ final String componentIdentifier =
localDifference.getComponentB().getIdentifier();
+ final VersionedComponent component =
RebaseHandlerUtils.findComponentById(mergedFlow, componentIdentifier);
+ if (component == null) {
+ logger.warn("Unable to apply name change: component [{}] not found
in merged flow", componentIdentifier);
+ return;
+ }
+ component.setName((String) localDifference.getValueB());
+ }
+}
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseConflictCode.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseConflictCode.java
index ad651b6d017..60b165c433c 100644
---
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseConflictCode.java
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseConflictCode.java
@@ -47,6 +47,26 @@ public enum RebaseConflictCode {
*/
SAME_COMPONENT_COMMENTS,
+ /**
+ * Both the local and upstream flows modified the name of the same
component to different values.
+ */
+ SAME_COMPONENT_NAME,
+
+ /**
+ * Both the local and upstream flows modified the concurrent tasks of the
same component to different values.
+ */
+ SAME_CONCURRENT_TASKS,
+
+ /**
+ * Both the local and upstream flows modified the scheduled state of the
same component to different values.
+ */
+ SAME_SCHEDULED_STATE,
+
+ /**
+ * Both the local and upstream flows modified the scheduling strategy or
run schedule of the same processor to different values.
+ */
+ SAME_SCHEDULING,
+
/**
* The component targeted by the local change does not exist in the target
version.
*/
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseHandlerUtils.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseHandlerUtils.java
index c1d2702d5c6..da4c82efa0b 100644
---
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseHandlerUtils.java
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseHandlerUtils.java
@@ -26,8 +26,13 @@ import org.apache.nifi.flow.VersionedLabel;
import org.apache.nifi.flow.VersionedPort;
import org.apache.nifi.flow.VersionedProcessGroup;
import org.apache.nifi.flow.VersionedProcessor;
+import org.apache.nifi.flow.VersionedRemoteGroupPort;
import org.apache.nifi.flow.VersionedRemoteProcessGroup;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
class RebaseHandlerUtils {
private RebaseHandlerUtils() {
@@ -77,6 +82,14 @@ class RebaseHandlerUtils {
if (identifier.equals(remoteProcessGroup.getIdentifier())) {
return remoteProcessGroup;
}
+ final VersionedRemoteGroupPort remoteInputPort =
findRemoteGroupPortById(remoteProcessGroup.getInputPorts(), identifier);
+ if (remoteInputPort != null) {
+ return remoteInputPort;
+ }
+ final VersionedRemoteGroupPort remoteOutputPort =
findRemoteGroupPortById(remoteProcessGroup.getOutputPorts(), identifier);
+ if (remoteOutputPort != null) {
+ return remoteOutputPort;
+ }
}
for (final VersionedProcessGroup childGroup :
group.getProcessGroups()) {
@@ -89,6 +102,34 @@ class RebaseHandlerUtils {
return null;
}
+ private static VersionedRemoteGroupPort findRemoteGroupPortById(final
Set<VersionedRemoteGroupPort> remotePorts, final String identifier) {
+ if (remotePorts == null) {
+ return null;
+ }
+ for (final VersionedRemoteGroupPort remotePort : remotePorts) {
+ if (identifier.equals(remotePort.getIdentifier())) {
+ return remotePort;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Finds an upstream difference of the same type on the same component
whose new value differs from the local new value.
+ */
+ static Optional<FlowDifference> findConflictingUpstreamDifference(final
FlowDifference localDifference, final Set<FlowDifference> upstreamDifferences) {
+ final String componentIdentifier =
localDifference.getComponentB().getIdentifier();
+ for (final FlowDifference upstreamDifference : upstreamDifferences) {
+ if (upstreamDifference.getDifferenceType() ==
localDifference.getDifferenceType()
+ && upstreamDifference.getComponentA() != null
+ &&
componentIdentifier.equals(upstreamDifference.getComponentA().getIdentifier())
+ && !Objects.equals(localDifference.getValueB(),
upstreamDifference.getValueB())) {
+ return Optional.of(upstreamDifference);
+ }
+ }
+ return Optional.empty();
+ }
+
static VersionedProcessGroup findProcessGroupById(final
VersionedProcessGroup group, final String identifier) {
if (identifier.equals(group.getIdentifier()) ||
identifier.equals(group.getInstanceIdentifier())) {
return group;
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/ScheduledStateChangedRebaseHandler.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/ScheduledStateChangedRebaseHandler.java
new file mode 100644
index 00000000000..1a64ee56419
--- /dev/null
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/ScheduledStateChangedRebaseHandler.java
@@ -0,0 +1,77 @@
+/*
+ * 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.registry.flow.diff;
+
+import org.apache.nifi.flow.ScheduledState;
+import org.apache.nifi.flow.VersionedComponent;
+import org.apache.nifi.flow.VersionedPort;
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
+import org.apache.nifi.flow.VersionedRemoteGroupPort;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Consumer;
+
+public class ScheduledStateChangedRebaseHandler implements RebaseHandler {
+
+ private static final Logger logger =
LoggerFactory.getLogger(ScheduledStateChangedRebaseHandler.class);
+
+ @Override
+ public DifferenceType getSupportedType() {
+ return DifferenceType.SCHEDULED_STATE_CHANGED;
+ }
+
+ @Override
+ public RebaseAnalysis.ClassifiedDifference classify(final FlowDifference
localDifference, final Set<FlowDifference> upstreamDifferences,
+ final
VersionedProcessGroup targetSnapshot) {
+ final VersionedComponent localComponent =
localDifference.getComponentB();
+ if (getSetter(localComponent).isEmpty()) {
+ return
RebaseAnalysis.ClassifiedDifference.unsupported(localDifference,
RebaseConflictCode.UNSUPPORTED_COMPONENT_TYPE,
+ "Scheduled state change is not supported for component %s
of type %s".formatted(localComponent.getIdentifier(),
localComponent.getComponentType()));
+ }
+ if
(RebaseHandlerUtils.findConflictingUpstreamDifference(localDifference,
upstreamDifferences).isPresent()) {
+ return
RebaseAnalysis.ClassifiedDifference.conflicting(localDifference,
RebaseConflictCode.SAME_SCHEDULED_STATE,
+ "Both local and upstream flows modified the scheduled
state of component " + localComponent.getIdentifier());
+ }
+ return RebaseAnalysis.ClassifiedDifference.compatible(localDifference);
+ }
+
+ @Override
+ public void apply(final FlowDifference localDifference, final
VersionedProcessGroup mergedFlow) {
+ final String componentIdentifier =
localDifference.getComponentB().getIdentifier();
+ final VersionedComponent component =
RebaseHandlerUtils.findComponentById(mergedFlow, componentIdentifier);
+ if (component == null) {
+ logger.warn("Unable to apply scheduled state change: component
[{}] not found in merged flow", componentIdentifier);
+ return;
+ }
+ getSetter(component).ifPresent(setter ->
setter.accept((ScheduledState) localDifference.getValueB()));
+ }
+
+ private Optional<Consumer<ScheduledState>> getSetter(final
VersionedComponent component) {
+ return switch (component) {
+ case VersionedProcessor processor ->
Optional.of(processor::setScheduledState);
+ case VersionedPort port -> Optional.of(port::setScheduledState);
+ case VersionedRemoteGroupPort remoteGroupPort ->
Optional.of(remoteGroupPort::setScheduledState);
+ case VersionedProcessGroup processGroup ->
Optional.of(processGroup::setScheduledState);
+ default -> Optional.empty();
+ };
+ }
+}
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/SchedulingChangedRebaseHandler.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/SchedulingChangedRebaseHandler.java
new file mode 100644
index 00000000000..e41e56dc226
--- /dev/null
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/SchedulingChangedRebaseHandler.java
@@ -0,0 +1,99 @@
+/*
+ * 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.registry.flow.diff;
+
+import org.apache.nifi.flow.VersionedComponent;
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Handles scheduling strategy and run schedule changes on processors. The two
fields are evaluated together because a run
+ * schedule is only meaningful for its scheduling strategy: an upstream change
to either field conflicts with the local change
+ * unless it results in the same value as the local processor.
+ */
+public class SchedulingChangedRebaseHandler implements RebaseHandler {
+
+ private static final Logger logger =
LoggerFactory.getLogger(SchedulingChangedRebaseHandler.class);
+
+ private final DifferenceType supportedType;
+
+ public SchedulingChangedRebaseHandler(final DifferenceType supportedType) {
+ if (supportedType != DifferenceType.SCHEDULING_STRATEGY_CHANGED &&
supportedType != DifferenceType.RUN_SCHEDULE_CHANGED) {
+ throw new IllegalArgumentException("Unsupported difference type "
+ supportedType);
+ }
+ this.supportedType = supportedType;
+ }
+
+ @Override
+ public DifferenceType getSupportedType() {
+ return supportedType;
+ }
+
+ @Override
+ public RebaseAnalysis.ClassifiedDifference classify(final FlowDifference
localDifference, final Set<FlowDifference> upstreamDifferences,
+ final
VersionedProcessGroup targetSnapshot) {
+ final VersionedComponent localComponent =
localDifference.getComponentB();
+ final String componentIdentifier = localComponent.getIdentifier();
+ if (!(localComponent instanceof VersionedProcessor localProcessor)) {
+ return
RebaseAnalysis.ClassifiedDifference.unsupported(localDifference,
RebaseConflictCode.UNSUPPORTED_COMPONENT_TYPE,
+ "Scheduling change is not supported for component %s of
type %s".formatted(componentIdentifier, localComponent.getComponentType()));
+ }
+
+ for (final FlowDifference upstreamDifference : upstreamDifferences) {
+ if (upstreamDifference.getComponentA() == null ||
!componentIdentifier.equals(upstreamDifference.getComponentA().getIdentifier()))
{
+ continue;
+ }
+
+ final String localValue;
+ if (upstreamDifference.getDifferenceType() ==
DifferenceType.SCHEDULING_STRATEGY_CHANGED) {
+ localValue = localProcessor.getSchedulingStrategy();
+ } else if (upstreamDifference.getDifferenceType() ==
DifferenceType.RUN_SCHEDULE_CHANGED) {
+ localValue = localProcessor.getSchedulingPeriod();
+ } else {
+ continue;
+ }
+
+ if (!Objects.equals(localValue, upstreamDifference.getValueB())) {
+ return
RebaseAnalysis.ClassifiedDifference.conflicting(localDifference,
RebaseConflictCode.SAME_SCHEDULING,
+ "Both local and upstream flows modified the scheduling
of processor " + componentIdentifier);
+ }
+ }
+ return RebaseAnalysis.ClassifiedDifference.compatible(localDifference);
+ }
+
+ @Override
+ public void apply(final FlowDifference localDifference, final
VersionedProcessGroup mergedFlow) {
+ final String componentIdentifier =
localDifference.getComponentB().getIdentifier();
+ if (!(RebaseHandlerUtils.findComponentById(mergedFlow,
componentIdentifier) instanceof VersionedProcessor processor)) {
+ logger.warn("Unable to apply scheduling change: processor [{}] not
found in merged flow", componentIdentifier);
+ return;
+ }
+
+ final String value = (String) localDifference.getValueB();
+ if (supportedType == DifferenceType.SCHEDULING_STRATEGY_CHANGED) {
+ processor.setSchedulingStrategy(value);
+ } else {
+ processor.setSchedulingPeriod(value);
+ }
+ }
+}
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardRebaseEngine.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardRebaseEngine.java
index 57153781cdf..9ebbcb6ecb6 100644
---
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardRebaseEngine.java
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardRebaseEngine.java
@@ -50,6 +50,11 @@ public class StandardRebaseEngine implements RebaseEngine {
registerHandler(new PropertyChangedRebaseHandler());
registerHandler(new PropertyAddedRebaseHandler());
registerHandler(new CommentsChangedRebaseHandler());
+ registerHandler(new NameChangedRebaseHandler());
+ registerHandler(new ConcurrentTasksChangedRebaseHandler());
+ registerHandler(new ScheduledStateChangedRebaseHandler());
+ registerHandler(new
SchedulingChangedRebaseHandler(DifferenceType.SCHEDULING_STRATEGY_CHANGED));
+ registerHandler(new
SchedulingChangedRebaseHandler(DifferenceType.RUN_SCHEDULE_CHANGED));
}
public StandardRebaseEngine(final Map<DifferenceType, RebaseHandler>
handlerRegistry) {
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/ConcurrentTasksChangedRebaseHandlerTest.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/ConcurrentTasksChangedRebaseHandlerTest.java
new file mode 100644
index 00000000000..6969d4fd007
--- /dev/null
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/ConcurrentTasksChangedRebaseHandlerTest.java
@@ -0,0 +1,147 @@
+/*
+ * 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.registry.flow.diff;
+
+import org.apache.nifi.flow.VersionedComponent;
+import org.apache.nifi.flow.VersionedControllerService;
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
+import org.apache.nifi.flow.VersionedRemoteGroupPort;
+import org.apache.nifi.flow.VersionedRemoteProcessGroup;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class ConcurrentTasksChangedRebaseHandlerTest {
+
+ private static final String ROOT = "root";
+ private static final String PROC_A = "proc-a";
+ private static final String PROC_B = "proc-b";
+ private static final String SERVICE_A = "service-a";
+ private static final String REMOTE_GROUP = "remote-group";
+ private static final String REMOTE_INPUT_PORT = "remote-input-port";
+ private static final int ORIGINAL_TASKS = 1;
+ private static final int LOCAL_TASKS = 6;
+ private static final int UPSTREAM_TASKS = 3;
+
+ private ConcurrentTasksChangedRebaseHandler handler;
+
+ @BeforeEach
+ void setup() {
+ handler = new ConcurrentTasksChangedRebaseHandler();
+ }
+
+ @Test
+ void testNoUpstreamChange() {
+ final FlowDifference localDifference =
createDifference(createProcessor(PROC_A), LOCAL_TASKS);
+ final FlowDifference upstreamOtherComponent =
createDifference(createProcessor(PROC_B), UPSTREAM_TASKS);
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, Set.of(upstreamOtherComponent), createRoot());
+
+ assertEquals(RebaseClassification.COMPATIBLE,
result.getClassification());
+ }
+
+ @Test
+ void testUpstreamChangedToSameValue() {
+ final FlowDifference localDifference =
createDifference(createProcessor(PROC_A), LOCAL_TASKS);
+ final FlowDifference upstreamDifference =
createDifference(createProcessor(PROC_A), LOCAL_TASKS);
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, Set.of(upstreamDifference), createRoot());
+
+ assertEquals(RebaseClassification.COMPATIBLE,
result.getClassification());
+ }
+
+ @Test
+ void testUpstreamChangedToDifferentValue() {
+ final FlowDifference localDifference =
createDifference(createProcessor(PROC_A), LOCAL_TASKS);
+ final FlowDifference upstreamDifference =
createDifference(createProcessor(PROC_A), UPSTREAM_TASKS);
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, Set.of(upstreamDifference), createRoot());
+
+ assertEquals(RebaseClassification.CONFLICTING,
result.getClassification());
+ assertEquals(RebaseConflictCode.SAME_CONCURRENT_TASKS,
result.getConflictCode());
+ }
+
+ @Test
+ void testUnsupportedComponentType() {
+ final VersionedControllerService service = new
VersionedControllerService();
+ service.setIdentifier(SERVICE_A);
+ final FlowDifference localDifference = createDifference(service,
LOCAL_TASKS);
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, Collections.emptySet(), createRoot());
+
+ assertEquals(RebaseClassification.UNSUPPORTED,
result.getClassification());
+ assertEquals(RebaseConflictCode.UNSUPPORTED_COMPONENT_TYPE,
result.getConflictCode());
+ }
+
+ @Test
+ void testApplyProcessor() {
+ final VersionedProcessGroup mergedFlow = createRoot();
+ final VersionedProcessor processor = createProcessor(PROC_A);
+ processor.setConcurrentlySchedulableTaskCount(ORIGINAL_TASKS);
+ mergedFlow.getProcessors().add(processor);
+
+ handler.apply(createDifference(createProcessor(PROC_A), LOCAL_TASKS),
mergedFlow);
+
+ assertEquals(LOCAL_TASKS,
processor.getConcurrentlySchedulableTaskCount());
+ }
+
+ @Test
+ void testApplyRemoteGroupPort() {
+ final VersionedProcessGroup mergedFlow = createRoot();
+ final VersionedRemoteGroupPort remotePort = createRemoteGroupPort();
+ remotePort.setConcurrentlySchedulableTaskCount(ORIGINAL_TASKS);
+ final VersionedRemoteProcessGroup remoteProcessGroup = new
VersionedRemoteProcessGroup();
+ remoteProcessGroup.setIdentifier(REMOTE_GROUP);
+ remoteProcessGroup.setInputPorts(Set.of(remotePort));
+ mergedFlow.getRemoteProcessGroups().add(remoteProcessGroup);
+
+ final FlowDifference localDifference =
createDifference(createRemoteGroupPort(), LOCAL_TASKS);
+ assertEquals(RebaseClassification.COMPATIBLE,
handler.classify(localDifference, Collections.emptySet(),
mergedFlow).getClassification());
+ handler.apply(localDifference, mergedFlow);
+
+ assertEquals(LOCAL_TASKS,
remotePort.getConcurrentlySchedulableTaskCount());
+ }
+
+ private FlowDifference createDifference(final VersionedComponent
component, final int newValue) {
+ return new
StandardFlowDifference(DifferenceType.CONCURRENT_TASKS_CHANGED, component,
component,
+ ORIGINAL_TASKS, newValue, "Concurrent tasks changed");
+ }
+
+ private VersionedProcessor createProcessor(final String identifier) {
+ final VersionedProcessor processor = new VersionedProcessor();
+ processor.setIdentifier(identifier);
+ return processor;
+ }
+
+ private VersionedRemoteGroupPort createRemoteGroupPort() {
+ final VersionedRemoteGroupPort remotePort = new
VersionedRemoteGroupPort();
+ remotePort.setIdentifier(REMOTE_INPUT_PORT);
+ return remotePort;
+ }
+
+ private VersionedProcessGroup createRoot() {
+ final VersionedProcessGroup root = new VersionedProcessGroup();
+ root.setIdentifier(ROOT);
+ return root;
+ }
+}
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/NameChangedRebaseHandlerTest.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/NameChangedRebaseHandlerTest.java
new file mode 100644
index 00000000000..8beae1c5f1e
--- /dev/null
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/NameChangedRebaseHandlerTest.java
@@ -0,0 +1,127 @@
+/*
+ * 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.registry.flow.diff;
+
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class NameChangedRebaseHandlerTest {
+
+ private static final String ROOT = "root";
+ private static final String CHILD_GROUP = "child-group";
+ private static final String PROC_A = "proc-a";
+ private static final String PROC_B = "proc-b";
+ private static final String ORIGINAL_NAME = "Original";
+ private static final String LOCAL_NAME = "Local";
+ private static final String UPSTREAM_NAME = "Upstream";
+
+ private NameChangedRebaseHandler handler;
+
+ @BeforeEach
+ void setup() {
+ handler = new NameChangedRebaseHandler();
+ }
+
+ @Test
+ void testNoUpstreamChange() {
+ final FlowDifference localDifference = createNameDifference(PROC_A,
LOCAL_NAME);
+ final FlowDifference upstreamOtherComponent =
createNameDifference(PROC_B, UPSTREAM_NAME);
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, Set.of(upstreamOtherComponent), createRoot());
+
+ assertEquals(RebaseClassification.COMPATIBLE,
result.getClassification());
+ }
+
+ @Test
+ void testUpstreamChangedToSameName() {
+ final FlowDifference localDifference = createNameDifference(PROC_A,
LOCAL_NAME);
+ final FlowDifference upstreamDifference = createNameDifference(PROC_A,
LOCAL_NAME);
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, Set.of(upstreamDifference), createRoot());
+
+ assertEquals(RebaseClassification.COMPATIBLE,
result.getClassification());
+ }
+
+ @Test
+ void testUpstreamChangedToDifferentName() {
+ final FlowDifference localDifference = createNameDifference(PROC_A,
LOCAL_NAME);
+ final FlowDifference upstreamDifference = createNameDifference(PROC_A,
UPSTREAM_NAME);
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, Set.of(upstreamDifference), createRoot());
+
+ assertEquals(RebaseClassification.CONFLICTING,
result.getClassification());
+ assertEquals(RebaseConflictCode.SAME_COMPONENT_NAME,
result.getConflictCode());
+ }
+
+ @Test
+ void testApplyProcessorName() {
+ final VersionedProcessGroup mergedFlow = createRoot();
+ final VersionedProcessor processor = createProcessor(PROC_A,
ORIGINAL_NAME);
+ mergedFlow.getProcessors().add(processor);
+
+ handler.apply(createNameDifference(PROC_A, LOCAL_NAME), mergedFlow);
+
+ assertEquals(LOCAL_NAME, processor.getName());
+ }
+
+ @Test
+ void testApplyProcessGroupName() {
+ final VersionedProcessGroup mergedFlow = createRoot();
+ final VersionedProcessGroup childGroup = createGroup(CHILD_GROUP,
ORIGINAL_NAME);
+ mergedFlow.getProcessGroups().add(childGroup);
+
+ final VersionedProcessGroup localGroup = createGroup(CHILD_GROUP,
LOCAL_NAME);
+ final FlowDifference localDifference = new
StandardFlowDifference(DifferenceType.NAME_CHANGED, createGroup(CHILD_GROUP,
ORIGINAL_NAME), localGroup,
+ ORIGINAL_NAME, LOCAL_NAME, "Name changed");
+
+ assertEquals(RebaseClassification.COMPATIBLE,
handler.classify(localDifference, Collections.emptySet(),
mergedFlow).getClassification());
+ handler.apply(localDifference, mergedFlow);
+
+ assertEquals(LOCAL_NAME, childGroup.getName());
+ }
+
+ private FlowDifference createNameDifference(final String identifier, final
String newName) {
+ return new StandardFlowDifference(DifferenceType.NAME_CHANGED,
createProcessor(identifier, ORIGINAL_NAME), createProcessor(identifier,
newName),
+ ORIGINAL_NAME, newName, "Name changed");
+ }
+
+ private VersionedProcessor createProcessor(final String identifier, final
String name) {
+ final VersionedProcessor processor = new VersionedProcessor();
+ processor.setIdentifier(identifier);
+ processor.setName(name);
+ return processor;
+ }
+
+ private VersionedProcessGroup createGroup(final String identifier, final
String name) {
+ final VersionedProcessGroup group = new VersionedProcessGroup();
+ group.setIdentifier(identifier);
+ group.setName(name);
+ return group;
+ }
+
+ private VersionedProcessGroup createRoot() {
+ return createGroup(ROOT, ROOT);
+ }
+}
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/RebaseEngineTest.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/RebaseEngineTest.java
index 66b9287b3a6..5bdd3e93a37 100644
---
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/RebaseEngineTest.java
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/RebaseEngineTest.java
@@ -415,10 +415,22 @@ class RebaseEngineTest {
"oldVal", "newVal", "Property propX changed"));
localDifferences.add(new
StandardFlowDifference(DifferenceType.COMMENTS_CHANGED, processorA,
localProcessorA,
"old comments", "new comments", "Comments changed"));
+ localDifferences.add(new
StandardFlowDifference(DifferenceType.NAME_CHANGED, processorA, localProcessorA,
+ "ProcessorA", "Renamed ProcessorA", "Name changed"));
+ localDifferences.add(new
StandardFlowDifference(DifferenceType.CONCURRENT_TASKS_CHANGED, processorA,
localProcessorA,
+ 1, 6, "Concurrent tasks changed"));
+ localDifferences.add(new
StandardFlowDifference(DifferenceType.SCHEDULED_STATE_CHANGED, processorA,
localProcessorA,
+ ScheduledState.DISABLED, ScheduledState.ENABLED, "Scheduled
state changed"));
+ localDifferences.add(new
StandardFlowDifference(DifferenceType.SCHEDULING_STRATEGY_CHANGED, processorA,
localProcessorA,
+ "TIMER_DRIVEN", "CRON_DRIVEN", "Scheduling strategy changed"));
+ localDifferences.add(new
StandardFlowDifference(DifferenceType.RUN_SCHEDULE_CHANGED, processorA,
localProcessorA,
+ "1 min", "0 0 1 * * ?", "Run schedule changed"));
final VersionedProcessor targetProcessor =
createProcessorWithProperty("proc-a", "ProcessorA", "propX", "oldVal");
targetProcessor.setPosition(new Position(10.0, 20.0));
targetProcessor.setComments("old comments");
+ targetProcessor.setConcurrentlySchedulableTaskCount(1);
+ targetProcessor.setScheduledState(ScheduledState.DISABLED);
final VersionedProcessGroup targetSnapshot = new
VersionedProcessGroup();
targetSnapshot.setIdentifier("root");
@@ -427,7 +439,7 @@ class RebaseEngineTest {
final RebaseAnalysis analysis = engine.analyze(localDifferences,
Collections.emptySet(), targetSnapshot);
assertTrue(analysis.isRebaseAllowed());
- assertEquals(3, analysis.getClassifiedLocalChanges().size());
+ assertEquals(8, analysis.getClassifiedLocalChanges().size());
for (final RebaseAnalysis.ClassifiedDifference classified :
analysis.getClassifiedLocalChanges()) {
assertEquals(RebaseClassification.COMPATIBLE,
classified.getClassification());
}
@@ -440,6 +452,11 @@ class RebaseEngineTest {
assertEquals(60.0, mergedProc.getPosition().getY());
assertEquals("newVal", mergedProc.getProperties().get("propX"));
assertEquals("new comments", mergedProc.getComments());
+ assertEquals("Renamed ProcessorA", mergedProc.getName());
+ assertEquals(6, mergedProc.getConcurrentlySchedulableTaskCount());
+ assertEquals(ScheduledState.ENABLED, mergedProc.getScheduledState());
+ assertEquals("CRON_DRIVEN", mergedProc.getSchedulingStrategy());
+ assertEquals("0 0 1 * * ?", mergedProc.getSchedulingPeriod());
}
@Test
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/ScheduledStateChangedRebaseHandlerTest.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/ScheduledStateChangedRebaseHandlerTest.java
new file mode 100644
index 00000000000..3679a50c9ec
--- /dev/null
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/ScheduledStateChangedRebaseHandlerTest.java
@@ -0,0 +1,145 @@
+/*
+ * 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.registry.flow.diff;
+
+import org.apache.nifi.flow.ScheduledState;
+import org.apache.nifi.flow.VersionedComponent;
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
+import org.apache.nifi.flow.VersionedRemoteGroupPort;
+import org.apache.nifi.flow.VersionedRemoteProcessGroup;
+import org.apache.nifi.flow.VersionedReportingTask;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class ScheduledStateChangedRebaseHandlerTest {
+
+ private static final String ROOT = "root";
+ private static final String PROC_A = "proc-a";
+ private static final String PROC_B = "proc-b";
+ private static final String TASK_A = "task-a";
+ private static final String REMOTE_GROUP = "remote-group";
+ private static final String REMOTE_OUTPUT_PORT = "remote-output-port";
+
+ private ScheduledStateChangedRebaseHandler handler;
+
+ @BeforeEach
+ void setup() {
+ handler = new ScheduledStateChangedRebaseHandler();
+ }
+
+ @Test
+ void testNoUpstreamChange() {
+ final FlowDifference localDifference =
createDifference(createProcessor(PROC_A), ScheduledState.ENABLED);
+ final FlowDifference upstreamOtherComponent =
createDifference(createProcessor(PROC_B), ScheduledState.RUNNING);
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, Set.of(upstreamOtherComponent), createRoot());
+
+ assertEquals(RebaseClassification.COMPATIBLE,
result.getClassification());
+ }
+
+ @Test
+ void testUpstreamChangedToSameState() {
+ final FlowDifference localDifference =
createDifference(createProcessor(PROC_A), ScheduledState.ENABLED);
+ final FlowDifference upstreamDifference =
createDifference(createProcessor(PROC_A), ScheduledState.ENABLED);
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, Set.of(upstreamDifference), createRoot());
+
+ assertEquals(RebaseClassification.COMPATIBLE,
result.getClassification());
+ }
+
+ @Test
+ void testUpstreamChangedToDifferentState() {
+ final FlowDifference localDifference =
createDifference(createProcessor(PROC_A), ScheduledState.ENABLED);
+ final FlowDifference upstreamDifference =
createDifference(createProcessor(PROC_A), ScheduledState.RUNNING);
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, Set.of(upstreamDifference), createRoot());
+
+ assertEquals(RebaseClassification.CONFLICTING,
result.getClassification());
+ assertEquals(RebaseConflictCode.SAME_SCHEDULED_STATE,
result.getConflictCode());
+ }
+
+ @Test
+ void testUnsupportedComponentType() {
+ final VersionedReportingTask reportingTask = new
VersionedReportingTask();
+ reportingTask.setIdentifier(TASK_A);
+ final FlowDifference localDifference = createDifference(reportingTask,
ScheduledState.ENABLED);
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, Collections.emptySet(), createRoot());
+
+ assertEquals(RebaseClassification.UNSUPPORTED,
result.getClassification());
+ assertEquals(RebaseConflictCode.UNSUPPORTED_COMPONENT_TYPE,
result.getConflictCode());
+ }
+
+ @Test
+ void testApplyProcessor() {
+ final VersionedProcessGroup mergedFlow = createRoot();
+ final VersionedProcessor processor = createProcessor(PROC_A);
+ processor.setScheduledState(ScheduledState.DISABLED);
+ mergedFlow.getProcessors().add(processor);
+
+ handler.apply(createDifference(createProcessor(PROC_A),
ScheduledState.ENABLED), mergedFlow);
+
+ assertEquals(ScheduledState.ENABLED, processor.getScheduledState());
+ }
+
+ @Test
+ void testApplyRemoteGroupPort() {
+ final VersionedProcessGroup mergedFlow = createRoot();
+ final VersionedRemoteGroupPort remotePort = createRemoteGroupPort();
+ remotePort.setScheduledState(ScheduledState.DISABLED);
+ final VersionedRemoteProcessGroup remoteProcessGroup = new
VersionedRemoteProcessGroup();
+ remoteProcessGroup.setIdentifier(REMOTE_GROUP);
+ remoteProcessGroup.setOutputPorts(Set.of(remotePort));
+ mergedFlow.getRemoteProcessGroups().add(remoteProcessGroup);
+
+ final FlowDifference localDifference =
createDifference(createRemoteGroupPort(), ScheduledState.ENABLED);
+ assertEquals(RebaseClassification.COMPATIBLE,
handler.classify(localDifference, Collections.emptySet(),
mergedFlow).getClassification());
+ handler.apply(localDifference, mergedFlow);
+
+ assertEquals(ScheduledState.ENABLED, remotePort.getScheduledState());
+ }
+
+ private FlowDifference createDifference(final VersionedComponent
component, final ScheduledState newState) {
+ return new
StandardFlowDifference(DifferenceType.SCHEDULED_STATE_CHANGED, component,
component,
+ ScheduledState.DISABLED, newState, "Scheduled state changed");
+ }
+
+ private VersionedProcessor createProcessor(final String identifier) {
+ final VersionedProcessor processor = new VersionedProcessor();
+ processor.setIdentifier(identifier);
+ return processor;
+ }
+
+ private VersionedRemoteGroupPort createRemoteGroupPort() {
+ final VersionedRemoteGroupPort remotePort = new
VersionedRemoteGroupPort();
+ remotePort.setIdentifier(REMOTE_OUTPUT_PORT);
+ return remotePort;
+ }
+
+ private VersionedProcessGroup createRoot() {
+ final VersionedProcessGroup root = new VersionedProcessGroup();
+ root.setIdentifier(ROOT);
+ return root;
+ }
+}
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/SchedulingChangedRebaseHandlerTest.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/SchedulingChangedRebaseHandlerTest.java
new file mode 100644
index 00000000000..80ec4c6eb09
--- /dev/null
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/SchedulingChangedRebaseHandlerTest.java
@@ -0,0 +1,143 @@
+/*
+ * 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.registry.flow.diff;
+
+import org.apache.nifi.flow.VersionedPort;
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class SchedulingChangedRebaseHandlerTest {
+
+ private static final String ROOT = "root";
+ private static final String PROC_A = "proc-a";
+ private static final String PROC_B = "proc-b";
+ private static final String PORT_A = "port-a";
+ private static final String TIMER_DRIVEN = "TIMER_DRIVEN";
+ private static final String CRON_DRIVEN = "CRON_DRIVEN";
+ private static final String TIMER_PERIOD = "1 min";
+ private static final String UPSTREAM_TIMER_PERIOD = "5 min";
+ private static final String CRON_EXPRESSION = "0 0 1 * * ?";
+
+ private SchedulingChangedRebaseHandler strategyHandler;
+ private SchedulingChangedRebaseHandler scheduleHandler;
+
+ @BeforeEach
+ void setup() {
+ strategyHandler = new
SchedulingChangedRebaseHandler(DifferenceType.SCHEDULING_STRATEGY_CHANGED);
+ scheduleHandler = new
SchedulingChangedRebaseHandler(DifferenceType.RUN_SCHEDULE_CHANGED);
+ }
+
+ @Test
+ void testNoUpstreamChange() {
+ final VersionedProcessor original = createProcessor(PROC_A,
TIMER_DRIVEN, TIMER_PERIOD);
+ final VersionedProcessor local = createProcessor(PROC_A, CRON_DRIVEN,
CRON_EXPRESSION);
+ final VersionedProcessor otherOriginal = createProcessor(PROC_B,
TIMER_DRIVEN, TIMER_PERIOD);
+ final VersionedProcessor otherUpstream = createProcessor(PROC_B,
TIMER_DRIVEN, UPSTREAM_TIMER_PERIOD);
+ final Set<FlowDifference> upstreamDifferences =
Set.of(scheduleDifference(otherOriginal, otherUpstream));
+
+ assertEquals(RebaseClassification.COMPATIBLE,
+ strategyHandler.classify(strategyDifference(original, local),
upstreamDifferences, createRoot()).getClassification());
+ assertEquals(RebaseClassification.COMPATIBLE,
+ scheduleHandler.classify(scheduleDifference(original, local),
upstreamDifferences, createRoot()).getClassification());
+ }
+
+ @Test
+ void testUpstreamChangedRunScheduleOnlyConflictsWithLocalStrategyChange() {
+ final VersionedProcessor original = createProcessor(PROC_A,
TIMER_DRIVEN, TIMER_PERIOD);
+ final VersionedProcessor local = createProcessor(PROC_A, CRON_DRIVEN,
CRON_EXPRESSION);
+ final VersionedProcessor upstream = createProcessor(PROC_A,
TIMER_DRIVEN, UPSTREAM_TIMER_PERIOD);
+ final Set<FlowDifference> upstreamDifferences =
Set.of(scheduleDifference(original, upstream));
+
+ final RebaseAnalysis.ClassifiedDifference result =
strategyHandler.classify(strategyDifference(original, local),
upstreamDifferences, createRoot());
+
+ assertEquals(RebaseClassification.CONFLICTING,
result.getClassification());
+ assertEquals(RebaseConflictCode.SAME_SCHEDULING,
result.getConflictCode());
+ }
+
+ @Test
+ void testUpstreamChangedToSameScheduling() {
+ final VersionedProcessor original = createProcessor(PROC_A,
TIMER_DRIVEN, TIMER_PERIOD);
+ final VersionedProcessor local = createProcessor(PROC_A, CRON_DRIVEN,
CRON_EXPRESSION);
+ final VersionedProcessor upstream = createProcessor(PROC_A,
CRON_DRIVEN, CRON_EXPRESSION);
+ final Set<FlowDifference> upstreamDifferences =
Set.of(strategyDifference(original, upstream), scheduleDifference(original,
upstream));
+
+ assertEquals(RebaseClassification.COMPATIBLE,
+ strategyHandler.classify(strategyDifference(original, local),
upstreamDifferences, createRoot()).getClassification());
+ assertEquals(RebaseClassification.COMPATIBLE,
+ scheduleHandler.classify(scheduleDifference(original, local),
upstreamDifferences, createRoot()).getClassification());
+ }
+
+ @Test
+ void testUnsupportedComponentType() {
+ final VersionedPort port = new VersionedPort();
+ port.setIdentifier(PORT_A);
+ final FlowDifference localDifference = new
StandardFlowDifference(DifferenceType.RUN_SCHEDULE_CHANGED, port, port,
+ TIMER_PERIOD, UPSTREAM_TIMER_PERIOD, "Run schedule changed");
+
+ final RebaseAnalysis.ClassifiedDifference result =
scheduleHandler.classify(localDifference, Collections.emptySet(), createRoot());
+
+ assertEquals(RebaseClassification.UNSUPPORTED,
result.getClassification());
+ assertEquals(RebaseConflictCode.UNSUPPORTED_COMPONENT_TYPE,
result.getConflictCode());
+ }
+
+ @Test
+ void testApplyTimerDrivenToCronDriven() {
+ final VersionedProcessor original = createProcessor(PROC_A,
TIMER_DRIVEN, TIMER_PERIOD);
+ final VersionedProcessor local = createProcessor(PROC_A, CRON_DRIVEN,
CRON_EXPRESSION);
+ final VersionedProcessor merged = createProcessor(PROC_A,
TIMER_DRIVEN, TIMER_PERIOD);
+ final VersionedProcessGroup mergedFlow = createRoot();
+ mergedFlow.getProcessors().add(merged);
+
+ strategyHandler.apply(strategyDifference(original, local), mergedFlow);
+ scheduleHandler.apply(scheduleDifference(original, local), mergedFlow);
+
+ assertEquals(CRON_DRIVEN, merged.getSchedulingStrategy());
+ assertEquals(CRON_EXPRESSION, merged.getSchedulingPeriod());
+ }
+
+ private FlowDifference strategyDifference(final VersionedProcessor before,
final VersionedProcessor after) {
+ return new
StandardFlowDifference(DifferenceType.SCHEDULING_STRATEGY_CHANGED, before,
after,
+ before.getSchedulingStrategy(), after.getSchedulingStrategy(),
"Scheduling strategy changed");
+ }
+
+ private FlowDifference scheduleDifference(final VersionedProcessor before,
final VersionedProcessor after) {
+ return new StandardFlowDifference(DifferenceType.RUN_SCHEDULE_CHANGED,
before, after,
+ before.getSchedulingPeriod(), after.getSchedulingPeriod(),
"Run schedule changed");
+ }
+
+ private VersionedProcessor createProcessor(final String identifier, final
String strategy, final String period) {
+ final VersionedProcessor processor = new VersionedProcessor();
+ processor.setIdentifier(identifier);
+ processor.setSchedulingStrategy(strategy);
+ processor.setSchedulingPeriod(period);
+ return processor;
+ }
+
+ private VersionedProcessGroup createRoot() {
+ final VersionedProcessGroup root = new VersionedProcessGroup();
+ root.setIdentifier(ROOT);
+ return root;
+ }
+}
