You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ex...@apache.org on 2022/07/27 19:48:29 UTC
[nifi] branch main updated: NIFI-10070 Updated status merging for ControllerService and ReportingTask entities
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 c77f85aafb NIFI-10070 Updated status merging for ControllerService and ReportingTask entities
c77f85aafb is described below
commit c77f85aafb7d055caf2215fc6d382cb3faed1544
Author: Nathan Gough <th...@gmail.com>
AuthorDate: Thu Jun 23 17:41:44 2022 -0400
NIFI-10070 Updated status merging for ControllerService and ReportingTask entities
- Corrected node identifier selection in multiple Mergers
This closes #6154
Signed-off-by: David Handermann <ex...@apache.org>
---
.../web/api/dto/status/ReportingTaskStatusDTO.java | 4 +
.../ControllerServicesEndpointMerger.java | 8 +-
.../http/endpoints/InputPortsEndpointMerger.java | 8 +-
.../http/endpoints/OutputPortsEndpointMerger.java | 8 +-
.../endpoints/ProcessGroupsEndpointMerger.java | 8 +-
.../http/endpoints/ProcessorsEndpointMerger.java | 8 +-
.../RemoteProcessGroupsEndpointMerger.java | 8 +-
.../endpoints/ReportingTasksEndpointMerger.java | 8 +-
.../endpoints/RuntimeManifestEndpointMerger.java | 2 +-
.../http/endpoints/UserGroupsEndpointMerger.java | 8 +-
.../http/endpoints/UsersEndpointMerger.java | 8 +-
.../manager/ControllerServiceEntityMerger.java | 11 +++
.../cluster/manager/ReportingTaskEntityMerger.java | 11 +++
.../apache/nifi/cluster/manager/StatusMerger.java | 36 +++++++++
.../manager/ControllerServiceEntityMergerTest.java | 86 +++++++++++++++++++++
.../manager/ReportingTaskEntityMergerTest.java | 87 ++++++++++++++++++++++
16 files changed, 245 insertions(+), 64 deletions(-)
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ReportingTaskStatusDTO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ReportingTaskStatusDTO.java
index 7426557ee4..e518e0c27c 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ReportingTaskStatusDTO.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ReportingTaskStatusDTO.java
@@ -27,6 +27,10 @@ import javax.xml.bind.annotation.XmlType;
@XmlType(name = "reportingTaskStatus")
public class ReportingTaskStatusDTO extends ComponentStatusDTO {
+ public static final String RUNNING = "RUNNING";
+ public static final String STOPPED = "STOPPED";
+ public static final String DISABLED = "DISABLED";
+
@ApiModelProperty(value = "The run status of this ReportingTask",
accessMode = ApiModelProperty.AccessMode.READ_ONLY,
allowableValues = "RUNNING, STOPPED, DISABLED")
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServicesEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServicesEndpointMerger.java
index b63d0eeb09..78c1aade0e 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServicesEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServicesEndpointMerger.java
@@ -54,13 +54,7 @@ public class ControllerServicesEndpointMerger implements EndpointResponseMerger
final Set<ControllerServiceEntity> nodeControllerServiceEntities = nodeResponseEntity.getControllerServices();
for (final ControllerServiceEntity nodeControllerServiceEntity : nodeControllerServiceEntities) {
- final NodeIdentifier nodeId = nodeResponse.getNodeId();
- Map<NodeIdentifier, ControllerServiceEntity> innerMap = entityMap.get(nodeId);
- if (innerMap == null) {
- innerMap = new HashMap<>();
- entityMap.put(nodeControllerServiceEntity.getId(), innerMap);
- }
-
+ Map<NodeIdentifier, ControllerServiceEntity> innerMap = entityMap.computeIfAbsent(nodeControllerServiceEntity.getId(), k -> new HashMap<>());
innerMap.put(nodeResponse.getNodeId(), nodeControllerServiceEntity);
}
}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/InputPortsEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/InputPortsEndpointMerger.java
index 22d697e764..e5a6daa0f5 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/InputPortsEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/InputPortsEndpointMerger.java
@@ -53,13 +53,7 @@ public class InputPortsEndpointMerger implements EndpointResponseMerger {
final Set<PortEntity> nodePortEntities = nodeResponseEntity.getInputPorts();
for (final PortEntity nodePortEntity : nodePortEntities) {
- final NodeIdentifier nodeId = nodeResponse.getNodeId();
- Map<NodeIdentifier, PortEntity> innerMap = entityMap.get(nodeId);
- if (innerMap == null) {
- innerMap = new HashMap<>();
- entityMap.put(nodePortEntity.getId(), innerMap);
- }
-
+ Map<NodeIdentifier, PortEntity> innerMap = entityMap.computeIfAbsent(nodePortEntity.getId(), k -> new HashMap<>());
innerMap.put(nodeResponse.getNodeId(), nodePortEntity);
}
}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/OutputPortsEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/OutputPortsEndpointMerger.java
index 1ab5236c74..562a90f514 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/OutputPortsEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/OutputPortsEndpointMerger.java
@@ -53,13 +53,7 @@ public class OutputPortsEndpointMerger implements EndpointResponseMerger {
final Set<PortEntity> nodePortEntities = nodeResponseEntity.getOutputPorts();
for (final PortEntity nodePortEntity : nodePortEntities) {
- final NodeIdentifier nodeId = nodeResponse.getNodeId();
- Map<NodeIdentifier, PortEntity> innerMap = entityMap.get(nodeId);
- if (innerMap == null) {
- innerMap = new HashMap<>();
- entityMap.put(nodePortEntity.getId(), innerMap);
- }
-
+ Map<NodeIdentifier, PortEntity> innerMap = entityMap.computeIfAbsent(nodePortEntity.getId(), k -> new HashMap<>());
innerMap.put(nodeResponse.getNodeId(), nodePortEntity);
}
}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessGroupsEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessGroupsEndpointMerger.java
index 7cd6a44c8f..26bb7b4154 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessGroupsEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessGroupsEndpointMerger.java
@@ -53,13 +53,7 @@ public class ProcessGroupsEndpointMerger implements EndpointResponseMerger {
final Set<ProcessGroupEntity> nodeProcessGroupEntities = nodeResponseEntity.getProcessGroups();
for (final ProcessGroupEntity nodeProcessGroupEntity : nodeProcessGroupEntities) {
- final NodeIdentifier nodeId = nodeResponse.getNodeId();
- Map<NodeIdentifier, ProcessGroupEntity> innerMap = entityMap.get(nodeId);
- if (innerMap == null) {
- innerMap = new HashMap<>();
- entityMap.put(nodeProcessGroupEntity.getId(), innerMap);
- }
-
+ Map<NodeIdentifier, ProcessGroupEntity> innerMap = entityMap.computeIfAbsent(nodeProcessGroupEntity.getId(), k -> new HashMap<>());
innerMap.put(nodeResponse.getNodeId(), nodeProcessGroupEntity);
}
}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorsEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorsEndpointMerger.java
index 0a9b193d7d..2caea60c3e 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorsEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorsEndpointMerger.java
@@ -53,13 +53,7 @@ public class ProcessorsEndpointMerger implements EndpointResponseMerger {
final Set<ProcessorEntity> nodeProcessorEntities = nodeResponseEntity.getProcessors();
for (final ProcessorEntity nodeProcessorEntity : nodeProcessorEntities) {
- final NodeIdentifier nodeId = nodeResponse.getNodeId();
- Map<NodeIdentifier, ProcessorEntity> innerMap = entityMap.get(nodeId);
- if (innerMap == null) {
- innerMap = new HashMap<>();
- entityMap.put(nodeProcessorEntity.getId(), innerMap);
- }
-
+ Map<NodeIdentifier, ProcessorEntity> innerMap = entityMap.computeIfAbsent(nodeProcessorEntity.getId(), k -> new HashMap<>());
innerMap.put(nodeResponse.getNodeId(), nodeProcessorEntity);
}
}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/RemoteProcessGroupsEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/RemoteProcessGroupsEndpointMerger.java
index ce5be06886..8329681df9 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/RemoteProcessGroupsEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/RemoteProcessGroupsEndpointMerger.java
@@ -65,13 +65,7 @@ public class RemoteProcessGroupsEndpointMerger implements EndpointResponseMerger
if (nodeRpgEntities != null) {
for (final RemoteProcessGroupEntity nodeRpgEntity : nodeRpgEntities) {
- final NodeIdentifier nodeId = nodeResponse.getNodeId();
- Map<NodeIdentifier, RemoteProcessGroupEntity> innerMap = entityMap.get(nodeId);
- if (innerMap == null) {
- innerMap = new HashMap<>();
- entityMap.put(nodeRpgEntity.getId(), innerMap);
- }
-
+ Map<NodeIdentifier, RemoteProcessGroupEntity> innerMap = entityMap.computeIfAbsent(nodeRpgEntity.getId(), k -> new HashMap<>());
innerMap.put(nodeResponse.getNodeId(), nodeRpgEntity);
}
}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ReportingTasksEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ReportingTasksEndpointMerger.java
index 643630bf22..47403d1ba2 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ReportingTasksEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ReportingTasksEndpointMerger.java
@@ -52,13 +52,7 @@ public class ReportingTasksEndpointMerger implements EndpointResponseMerger {
final Set<ReportingTaskEntity> nodeReportingTaskEntities = nodeResponseEntity.getReportingTasks();
for (final ReportingTaskEntity nodeReportingTaskEntity : nodeReportingTaskEntities) {
- final NodeIdentifier nodeId = nodeResponse.getNodeId();
- Map<NodeIdentifier, ReportingTaskEntity> innerMap = entityMap.get(nodeId);
- if (innerMap == null) {
- innerMap = new HashMap<>();
- entityMap.put(nodeReportingTaskEntity.getId(), innerMap);
- }
-
+ Map<NodeIdentifier, ReportingTaskEntity> innerMap = entityMap.computeIfAbsent(nodeReportingTaskEntity.getId(), k -> new HashMap<>());
innerMap.put(nodeResponse.getNodeId(), nodeReportingTaskEntity);
}
}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/RuntimeManifestEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/RuntimeManifestEndpointMerger.java
index c673299add..070294d7a5 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/RuntimeManifestEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/RuntimeManifestEndpointMerger.java
@@ -48,7 +48,7 @@ public class RuntimeManifestEndpointMerger implements EndpointResponseMerger {
final RuntimeManifestEntity responseEntity = clientResponse.getClientResponse().readEntity(RuntimeManifestEntity.class);
final RuntimeManifest responseManifest = responseEntity.getRuntimeManifest();
- final Set<Bundle> responseBundles = responseManifest.getBundles() == null ? Collections.emptySet() : new LinkedHashSet<>(responseManifest.getBundles());
+ final Set<Bundle> responseBundles = responseManifest.getBundles() == null ? new LinkedHashSet<>() : new LinkedHashSet<>(responseManifest.getBundles());
for (final NodeResponse nodeResponse : successfulResponses) {
final RuntimeManifestEntity nodeResponseEntity = nodeResponse.getClientResponse().readEntity(RuntimeManifestEntity.class);
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/UserGroupsEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/UserGroupsEndpointMerger.java
index 51cc915b77..6019680206 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/UserGroupsEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/UserGroupsEndpointMerger.java
@@ -57,13 +57,7 @@ public class UserGroupsEndpointMerger implements EndpointResponseMerger {
userGroupEntities.retainAll(nodeUserGroupEntities);
for (final UserGroupEntity nodeUserGroupEntity : nodeUserGroupEntities) {
- final NodeIdentifier nodeId = nodeResponse.getNodeId();
- Map<NodeIdentifier, UserGroupEntity> innerMap = entityMap.get(nodeId);
- if (innerMap == null) {
- innerMap = new HashMap<>();
- entityMap.put(nodeUserGroupEntity.getId(), innerMap);
- }
-
+ Map<NodeIdentifier, UserGroupEntity> innerMap = entityMap.computeIfAbsent(nodeUserGroupEntity.getId(), k -> new HashMap<>());
innerMap.put(nodeResponse.getNodeId(), nodeUserGroupEntity);
}
}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/UsersEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/UsersEndpointMerger.java
index 7984197694..9356ba6f01 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/UsersEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/UsersEndpointMerger.java
@@ -57,13 +57,7 @@ public class UsersEndpointMerger implements EndpointResponseMerger {
userEntities.retainAll(nodeUserEntities);
for (final UserEntity nodeUserEntity : nodeUserEntities) {
- final NodeIdentifier nodeId = nodeResponse.getNodeId();
- Map<NodeIdentifier, UserEntity> innerMap = entityMap.get(nodeId);
- if (innerMap == null) {
- innerMap = new HashMap<>();
- entityMap.put(nodeUserEntity.getId(), innerMap);
- }
-
+ Map<NodeIdentifier, UserEntity> innerMap = entityMap.computeIfAbsent(nodeUserEntity.getId(), k -> new HashMap<>());
innerMap.put(nodeResponse.getNodeId(), nodeUserEntity);
}
}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ControllerServiceEntityMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ControllerServiceEntityMerger.java
index 86d39ab8f5..18c8727ec4 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ControllerServiceEntityMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ControllerServiceEntityMerger.java
@@ -36,6 +36,17 @@ import java.util.stream.Collectors;
public class ControllerServiceEntityMerger implements ComponentEntityMerger<ControllerServiceEntity> {
+ @Override
+ public void merge(ControllerServiceEntity clientEntity, Map<NodeIdentifier, ControllerServiceEntity> entityMap) {
+ ComponentEntityMerger.super.merge(clientEntity, entityMap);
+ for (Map.Entry<NodeIdentifier, ControllerServiceEntity> entry : entityMap.entrySet()) {
+ final ControllerServiceEntity entityStatus = entry.getValue();
+ if (clientEntity != entityStatus) {
+ StatusMerger.merge(clientEntity.getStatus(), entry.getValue().getStatus());
+ }
+ }
+ }
+
/**
* Merges the ControllerServiceEntity responses.
*
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ReportingTaskEntityMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ReportingTaskEntityMerger.java
index 2784776e29..7617969523 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ReportingTaskEntityMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ReportingTaskEntityMerger.java
@@ -29,6 +29,17 @@ import java.util.stream.Collectors;
public class ReportingTaskEntityMerger implements ComponentEntityMerger<ReportingTaskEntity> {
+ @Override
+ public void merge(ReportingTaskEntity clientEntity, Map<NodeIdentifier, ReportingTaskEntity> entityMap) {
+ ComponentEntityMerger.super.merge(clientEntity, entityMap);
+ for (Map.Entry<NodeIdentifier, ReportingTaskEntity> entry : entityMap.entrySet()) {
+ final ReportingTaskEntity entityStatus = entry.getValue();
+ if (clientEntity != entityStatus) {
+ StatusMerger.merge(clientEntity.getStatus(), entityStatus.getStatus());
+ }
+ }
+ }
+
/**
* Merges the ReportingTaskEntity responses.
*
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/StatusMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/StatusMerger.java
index 0ad131fe79..8be82d919a 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/StatusMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/StatusMerger.java
@@ -17,6 +17,8 @@
package org.apache.nifi.cluster.manager;
+import org.apache.nifi.components.validation.ValidationStatus;
+import org.apache.nifi.controller.service.ControllerServiceState;
import org.apache.nifi.controller.status.FlowFileAvailability;
import org.apache.nifi.controller.status.ProcessGroupStatus;
import org.apache.nifi.controller.status.RunStatus;
@@ -41,6 +43,7 @@ import org.apache.nifi.web.api.dto.diagnostics.JVMSystemDiagnosticsSnapshotDTO;
import org.apache.nifi.web.api.dto.status.ConnectionStatusDTO;
import org.apache.nifi.web.api.dto.status.ConnectionStatusPredictionsSnapshotDTO;
import org.apache.nifi.web.api.dto.status.ConnectionStatusSnapshotDTO;
+import org.apache.nifi.web.api.dto.status.ControllerServiceStatusDTO;
import org.apache.nifi.web.api.dto.status.ControllerStatusDTO;
import org.apache.nifi.web.api.dto.status.NodeConnectionStatusSnapshotDTO;
import org.apache.nifi.web.api.dto.status.NodePortStatusSnapshotDTO;
@@ -55,6 +58,7 @@ import org.apache.nifi.web.api.dto.status.ProcessorStatusDTO;
import org.apache.nifi.web.api.dto.status.ProcessorStatusSnapshotDTO;
import org.apache.nifi.web.api.dto.status.RemoteProcessGroupStatusDTO;
import org.apache.nifi.web.api.dto.status.RemoteProcessGroupStatusSnapshotDTO;
+import org.apache.nifi.web.api.dto.status.ReportingTaskStatusDTO;
import org.apache.nifi.web.api.entity.ConnectionStatusSnapshotEntity;
import org.apache.nifi.web.api.entity.PortStatusSnapshotEntity;
import org.apache.nifi.web.api.entity.ProcessGroupStatusSnapshotEntity;
@@ -1015,4 +1019,36 @@ public class StatusMerger {
return formatCount(count) + " (" + formatDataSize(bytes) + ")";
}
+ public static void merge(final ControllerServiceStatusDTO target, final ControllerServiceStatusDTO toMerge) {
+ if (target == null || toMerge == null) {
+ return;
+ }
+
+ // RunStatus for ControllerServiceStatusDTO can be one of [ENABLED, ENABLING, DISABLED, DISABLING]
+ if (ControllerServiceState.DISABLING.name().equalsIgnoreCase(toMerge.getRunStatus())) {
+ target.setRunStatus(ControllerServiceState.DISABLING.name());
+ } else if (ControllerServiceState.ENABLING.name().equalsIgnoreCase(toMerge.getRunStatus())) {
+ target.setRunStatus(ControllerServiceState.ENABLING.name());
+ }
+
+ if (ValidationStatus.VALIDATING.name().equalsIgnoreCase(toMerge.getValidationStatus())) {
+ target.setValidationStatus(ValidationStatus.VALIDATING.name());
+ } else if (ValidationStatus.INVALID.name().equalsIgnoreCase(toMerge.getRunStatus())) {
+ target.setValidationStatus(ValidationStatus.INVALID.name());
+ }
+ }
+
+ public static void merge(final ReportingTaskStatusDTO target, final ReportingTaskStatusDTO toMerge) {
+ if (target == null || toMerge == null) {
+ return;
+ }
+
+ target.setActiveThreadCount(target.getActiveThreadCount() + toMerge.getActiveThreadCount());
+
+ if (ValidationStatus.VALIDATING.name().equalsIgnoreCase(toMerge.getValidationStatus())) {
+ target.setValidationStatus(ValidationStatus.VALIDATING.name());
+ } else if (ValidationStatus.INVALID.name().equalsIgnoreCase(toMerge.getRunStatus())) {
+ target.setValidationStatus(ValidationStatus.INVALID.name());
+ }
+ }
}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/manager/ControllerServiceEntityMergerTest.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/manager/ControllerServiceEntityMergerTest.java
new file mode 100644
index 0000000000..cd31430aae
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/manager/ControllerServiceEntityMergerTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.cluster.manager;
+
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.components.validation.ValidationStatus;
+import org.apache.nifi.controller.ScheduledState;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ControllerServiceDTO;
+import org.apache.nifi.web.api.dto.PermissionsDTO;
+import org.apache.nifi.web.api.dto.RevisionDTO;
+import org.apache.nifi.web.api.dto.status.ControllerServiceStatusDTO;
+import org.apache.nifi.web.api.entity.ControllerServiceEntity;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class ControllerServiceEntityMergerTest {
+
+ @Test
+ void testMergeStatusFields() {
+ final ControllerServiceEntity nodeOneControllerserviceEntity = getControllerServiceEntity("id1", RunStatus.Stopped.name(), ValidationStatus.VALIDATING.name());
+ final ControllerServiceEntity nodeTwoControllerServiceEntity = getControllerServiceEntity("id2", RunStatus.Validating.name(), ValidationStatus.INVALID.name());
+ final Map<NodeIdentifier, ControllerServiceEntity> entityMap = new HashMap();
+ entityMap.put(getNodeIdentifier("node1", 8000), nodeOneControllerserviceEntity);
+ entityMap.put(getNodeIdentifier("node2", 8010), nodeTwoControllerServiceEntity);
+
+ ControllerServiceEntityMerger merger = new ControllerServiceEntityMerger();
+ merger.merge(nodeOneControllerserviceEntity, entityMap);
+ assertEquals("Stopped", nodeOneControllerserviceEntity.getStatus().getRunStatus());
+ }
+
+ private NodeIdentifier getNodeIdentifier(final String id, final int port) {
+ return new NodeIdentifier(id, "localhost", port, "localhost", port+1, "localhost", port+2, port+3, true);
+ }
+
+ private ControllerServiceEntity getControllerServiceEntity(final String id, final String runStatus, final String validationStatus) {
+ final ControllerServiceDTO dto = new ControllerServiceDTO();
+ dto.setId(id);
+ dto.setState(ScheduledState.STOPPED.name());
+ dto.setValidationStatus(ValidationStatus.VALIDATING.name());
+
+ final ControllerServiceStatusDTO statusDto = new ControllerServiceStatusDTO();
+ statusDto.setRunStatus(RunStatus.Stopped.name());
+ statusDto.setActiveThreadCount(1);
+ statusDto.setValidationStatus(ValidationStatus.VALIDATING.name());
+
+ final PermissionsDTO permissed = new PermissionsDTO();
+ permissed.setCanRead(true);
+ permissed.setCanWrite(true);
+
+ final ControllerServiceEntity entity = new ControllerServiceEntity();
+ entity.setComponent(dto);
+ entity.setRevision(createNewRevision());
+ entity.setDisconnectedNodeAcknowledged(true);
+ entity.setPermissions(permissed);
+ entity.setStatus(statusDto);
+
+ return entity;
+ }
+
+ public RevisionDTO createNewRevision() {
+ final RevisionDTO revisionDto = new RevisionDTO();
+ revisionDto.setClientId(getClass().getName());
+ revisionDto.setVersion(0L);
+ return revisionDto;
+ }
+}
\ No newline at end of file
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/manager/ReportingTaskEntityMergerTest.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/manager/ReportingTaskEntityMergerTest.java
new file mode 100644
index 0000000000..8554587368
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/manager/ReportingTaskEntityMergerTest.java
@@ -0,0 +1,87 @@
+/*
+ * 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.cluster.manager;
+
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.components.validation.ValidationStatus;
+import org.apache.nifi.controller.ScheduledState;
+import org.apache.nifi.web.api.dto.PermissionsDTO;
+import org.apache.nifi.web.api.dto.ReportingTaskDTO;
+import org.apache.nifi.web.api.dto.RevisionDTO;
+import org.apache.nifi.web.api.dto.status.ReportingTaskStatusDTO;
+import org.apache.nifi.web.api.entity.ReportingTaskEntity;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class ReportingTaskEntityMergerTest {
+
+ @Test
+ void testMergeStatusFields() {
+ final ReportingTaskEntity nodeOneReportingTaskEntity = getReportingTaskEntity("id1", ReportingTaskStatusDTO.RUNNING, ValidationStatus.VALID.name());
+ final ReportingTaskEntity nodeTwoReportingTaskEntity = getReportingTaskEntity("id2", ReportingTaskStatusDTO.RUNNING, ValidationStatus.VALIDATING.name());
+ final Map<NodeIdentifier, ReportingTaskEntity> entityMap = new HashMap();
+ entityMap.put(getNodeIdentifier("node1", 8000), nodeOneReportingTaskEntity);
+ entityMap.put(getNodeIdentifier("node2", 8010), nodeTwoReportingTaskEntity);
+
+ ReportingTaskEntityMerger merger = new ReportingTaskEntityMerger();
+ merger.merge(nodeOneReportingTaskEntity, entityMap);
+ assertEquals(2, nodeOneReportingTaskEntity.getStatus().getActiveThreadCount());
+ assertEquals(ReportingTaskStatusDTO.RUNNING, nodeOneReportingTaskEntity.getStatus().getRunStatus());
+ assertEquals(ValidationStatus.VALIDATING.name(), nodeOneReportingTaskEntity.getStatus().getValidationStatus());
+ }
+
+ private NodeIdentifier getNodeIdentifier(final String id, final int port) {
+ return new NodeIdentifier(id, "localhost", port, "localhost", port+1, "localhost", port+2, port+3, true);
+ }
+
+ private ReportingTaskEntity getReportingTaskEntity(final String id, final String runStatus, final String validationStatus) {
+ final ReportingTaskDTO dto = new ReportingTaskDTO();
+ dto.setId(id);
+ dto.setState(ScheduledState.STOPPED.name());
+ dto.setValidationStatus(validationStatus);
+
+ final ReportingTaskStatusDTO statusDto = new ReportingTaskStatusDTO();
+ statusDto.setRunStatus(runStatus);
+ statusDto.setActiveThreadCount(1);
+ statusDto.setValidationStatus(validationStatus);
+
+ final PermissionsDTO permissed = new PermissionsDTO();
+ permissed.setCanRead(true);
+ permissed.setCanWrite(true);
+
+ final ReportingTaskEntity entity = new ReportingTaskEntity();
+ entity.setComponent(dto);
+ entity.setRevision(createNewRevision());
+ entity.setDisconnectedNodeAcknowledged(true);
+ entity.setPermissions(permissed);
+ entity.setStatus(statusDto);
+
+ return entity;
+ }
+
+ public RevisionDTO createNewRevision() {
+ final RevisionDTO revisionDto = new RevisionDTO();
+ revisionDto.setClientId(getClass().getName());
+ revisionDto.setVersion(0L);
+ return revisionDto;
+ }
+}
\ No newline at end of file