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 <[email protected]>
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 <[email protected]>
---
.../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
