You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2016/06/17 17:21:45 UTC
[4/5] nifi git commit: NIFI-2007: - Restoring bulletin functionality.
- Ensuring appropriate merging of bulletins in clustered responses.
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessGroupEndpointMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessGroupEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessGroupEndpointMerger.java
index bef75a0..6b9b080 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessGroupEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessGroupEndpointMerger.java
@@ -17,22 +17,18 @@
package org.apache.nifi.cluster.coordination.http.endpoints;
-import java.net.URI;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-import java.util.regex.Pattern;
-
import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.ProcessGroupEntityMerger;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
-import org.apache.nifi.web.api.dto.FlowSnippetDTO;
-import org.apache.nifi.web.api.dto.ProcessGroupDTO;
-import org.apache.nifi.web.api.dto.ProcessorDTO;
-import org.apache.nifi.web.api.dto.RemoteProcessGroupDTO;
import org.apache.nifi.web.api.entity.ProcessGroupEntity;
-public class ProcessGroupEndpointMerger implements EndpointResponseMerger {
+import java.net.URI;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+public class ProcessGroupEndpointMerger extends AbstractSingleEntityEndpoint<ProcessGroupEntity> implements EndpointResponseMerger {
public static final Pattern PROCESS_GROUP_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))");
@Override
@@ -41,67 +37,12 @@ public class ProcessGroupEndpointMerger implements EndpointResponseMerger {
}
@Override
- public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
- if (!canHandle(uri, method)) {
- throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
- }
-
- final ProcessGroupEntity responseEntity = clientResponse.getClientResponse().getEntity(ProcessGroupEntity.class);
- final ProcessGroupDTO responseDto = responseEntity.getComponent();
-
- final FlowSnippetDTO contents = responseDto.getContents();
- if (contents == null) {
- return new NodeResponse(clientResponse, responseEntity);
- } else {
- final Map<String, Map<NodeIdentifier, ProcessorDTO>> processorMap = new HashMap<>();
- final Map<String, Map<NodeIdentifier, RemoteProcessGroupDTO>> remoteProcessGroupMap = new HashMap<>();
-
- for (final NodeResponse nodeResponse : successfulResponses) {
- final ProcessGroupEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : nodeResponse.getClientResponse().getEntity(ProcessGroupEntity.class);
- final ProcessGroupDTO nodeProcessGroup = nodeResponseEntity.getComponent();
-
- for (final ProcessorDTO nodeProcessor : nodeProcessGroup.getContents().getProcessors()) {
- Map<NodeIdentifier, ProcessorDTO> innerMap = processorMap.get(nodeProcessor.getId());
- if (innerMap == null) {
- innerMap = new HashMap<>();
- processorMap.put(nodeProcessor.getId(), innerMap);
- }
-
- innerMap.put(nodeResponse.getNodeId(), nodeProcessor);
- }
-
- for (final RemoteProcessGroupDTO nodeRemoteProcessGroup : nodeProcessGroup.getContents().getRemoteProcessGroups()) {
- Map<NodeIdentifier, RemoteProcessGroupDTO> innerMap = remoteProcessGroupMap.get(nodeRemoteProcessGroup.getId());
- if (innerMap == null) {
- innerMap = new HashMap<>();
- remoteProcessGroupMap.put(nodeRemoteProcessGroup.getId(), innerMap);
- }
-
- innerMap.put(nodeResponse.getNodeId(), nodeRemoteProcessGroup);
- }
- }
-
- final ProcessorEndpointMerger procMerger = new ProcessorEndpointMerger();
- for (final ProcessorDTO processor : contents.getProcessors()) {
- final String procId = processor.getId();
- final Map<NodeIdentifier, ProcessorDTO> mergeMap = processorMap.get(procId);
-
- procMerger.mergeResponses(processor, mergeMap, successfulResponses, problematicResponses);
- }
-
- final RemoteProcessGroupEndpointMerger rpgMerger = new RemoteProcessGroupEndpointMerger();
- for (final RemoteProcessGroupDTO remoteProcessGroup : contents.getRemoteProcessGroups()) {
- if (remoteProcessGroup.getContents() != null) {
- final String remoteProcessGroupId = remoteProcessGroup.getId();
- final Map<NodeIdentifier, RemoteProcessGroupDTO> mergeMap = remoteProcessGroupMap.get(remoteProcessGroupId);
-
- rpgMerger.mergeResponses(remoteProcessGroup, mergeMap, successfulResponses, problematicResponses);
- }
- }
- }
-
- // create a new client response
- return new NodeResponse(clientResponse, responseEntity);
+ protected Class<ProcessGroupEntity> getEntityClass() {
+ return ProcessGroupEntity.class;
}
+ @Override
+ protected void mergeResponses(ProcessGroupEntity clientEntity, Map<NodeIdentifier, ProcessGroupEntity> entityMap, Set<NodeResponse> successfulResponses, Set<NodeResponse> problematicResponses) {
+ ProcessGroupEntityMerger.mergeProcessGroups(clientEntity, entityMap);
+ }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessGroupsEndpointMerger.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..d4f047e
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessGroupsEndpointMerger.java
@@ -0,0 +1,72 @@
+/*
+ * 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.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.ProcessGroupsEntityMerger;
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.web.api.entity.ProcessGroupEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupsEntity;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+public class ProcessGroupsEndpointMerger implements EndpointResponseMerger {
+ public static final Pattern PROCESS_GROUPS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/process-groups");
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "GET".equalsIgnoreCase(method) && PROCESS_GROUPS_URI_PATTERN.matcher(uri.getPath()).matches();
+ }
+
+ @Override
+ public final NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessGroupsEntity responseEntity = clientResponse.getClientResponse().getEntity(ProcessGroupsEntity.class);
+ final Set<ProcessGroupEntity> processGroupEntities = responseEntity.getProcessGroups();
+
+ final Map<String, Map<NodeIdentifier, ProcessGroupEntity>> entityMap = new HashMap<>();
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ProcessGroupsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : nodeResponse.getClientResponse().getEntity(ProcessGroupsEntity.class);
+ 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);
+ }
+
+ innerMap.put(nodeResponse.getNodeId(), nodeProcessGroupEntity);
+ }
+ }
+
+ ProcessGroupsEntityMerger.mergeProcessGroups(processGroupEntities, entityMap);
+
+ // create a new client response
+ return new NodeResponse(clientResponse, responseEntity);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorEndpointMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorEndpointMerger.java
index bd50eca..b8202c8 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorEndpointMerger.java
@@ -17,28 +17,24 @@
package org.apache.nifi.cluster.coordination.http.endpoints;
-import java.net.URI;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-import java.util.regex.Pattern;
-
import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
import org.apache.nifi.cluster.manager.NodeResponse;
-import org.apache.nifi.cluster.manager.StatusMerger;
+import org.apache.nifi.cluster.manager.ProcessorEntityMerger;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
-import org.apache.nifi.web.api.dto.ProcessorDTO;
import org.apache.nifi.web.api.entity.ProcessorEntity;
+import java.net.URI;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
public class ProcessorEndpointMerger extends AbstractSingleEntityEndpoint<ProcessorEntity> implements EndpointResponseMerger {
- public static final Pattern PROCESSORS_URI_PATTERN = Pattern.compile("/nifi-api/processors");
+ public static final Pattern PROCESSORS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/processors");
public static final Pattern PROCESSOR_URI_PATTERN = Pattern.compile("/nifi-api/processors/[a-f0-9\\-]{36}");
- public static final Pattern CLUSTER_PROCESSOR_URI_PATTERN = Pattern.compile("/nifi-api/cluster/processors/[a-f0-9\\-]{36}");
@Override
public boolean canHandle(final URI uri, final String method) {
- if (("GET".equalsIgnoreCase(method) || "PUT".equalsIgnoreCase(method))
- && (PROCESSOR_URI_PATTERN.matcher(uri.getPath()).matches() || CLUSTER_PROCESSOR_URI_PATTERN.matcher(uri.getPath()).matches())) {
+ if (("GET".equalsIgnoreCase(method) || "PUT".equalsIgnoreCase(method)) && (PROCESSOR_URI_PATTERN.matcher(uri.getPath()).matches())) {
return true;
} else if ("POST".equalsIgnoreCase(method) && PROCESSORS_URI_PATTERN.matcher(uri.getPath()).matches()) {
return true;
@@ -53,42 +49,10 @@ public class ProcessorEndpointMerger extends AbstractSingleEntityEndpoint<Proces
}
- protected void mergeResponses(final ProcessorDTO clientDto, final Map<NodeIdentifier, ProcessorDTO> dtoMap, final Set<NodeResponse> successfulResponses,
- final Set<NodeResponse> problematicResponses) {
- final Map<String, Set<NodeIdentifier>> validationErrorMap = new HashMap<>();
-
- for (final Map.Entry<NodeIdentifier, ProcessorDTO> nodeEntry : dtoMap.entrySet()) {
- final NodeIdentifier nodeId = nodeEntry.getKey();
- final ProcessorDTO nodeProcessor = nodeEntry.getValue();
-
- // merge the validation errors
- mergeValidationErrors(validationErrorMap, nodeId, nodeProcessor.getValidationErrors());
- }
-
- // set the merged the validation errors
- clientDto.setValidationErrors(normalizedMergedValidationErrors(validationErrorMap, dtoMap.size()));
- }
-
@Override
protected void mergeResponses(final ProcessorEntity clientEntity, final Map<NodeIdentifier, ProcessorEntity> entityMap, final Set<NodeResponse> successfulResponses,
final Set<NodeResponse> problematicResponses) {
- final ProcessorDTO clientDto = clientEntity.getComponent();
- final Map<NodeIdentifier, ProcessorDTO> dtoMap = new HashMap<>();
- for (final Map.Entry<NodeIdentifier, ProcessorEntity> entry : entityMap.entrySet()) {
- final ProcessorEntity nodeProcEntity = entry.getValue();
- final ProcessorDTO nodeProcDto = nodeProcEntity.getComponent();
- dtoMap.put(entry.getKey(), nodeProcDto);
- }
-
- for (final Map.Entry<NodeIdentifier, ProcessorEntity> entry : entityMap.entrySet()) {
- final NodeIdentifier nodeId = entry.getKey();
- final ProcessorEntity entity = entry.getValue();
- if (entity != clientEntity) {
- StatusMerger.merge(clientEntity.getStatus(), entity.getStatus(), nodeId.getId(), nodeId.getApiAddress(), nodeId.getApiPort());
- }
- }
-
- mergeResponses(clientDto, dtoMap, successfulResponses, problematicResponses);
+ ProcessorEntityMerger.mergeProcessors(clientEntity, entityMap);
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorsEndpointMerger.java
----------------------------------------------------------------------
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 fa076b9..da84c58 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
@@ -17,21 +17,21 @@
package org.apache.nifi.cluster.coordination.http.endpoints;
-import java.net.URI;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-import java.util.regex.Pattern;
-
import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.ProcessorsEntityMerger;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
-import org.apache.nifi.web.api.dto.ProcessorDTO;
import org.apache.nifi.web.api.entity.ProcessorEntity;
import org.apache.nifi.web.api.entity.ProcessorsEntity;
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
public class ProcessorsEndpointMerger implements EndpointResponseMerger {
- public static final Pattern PROCESSORS_URI_PATTERN = Pattern.compile("/nifi-api/controller/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/processors");
+ public static final Pattern PROCESSORS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/processors");
@Override
public boolean canHandle(final URI uri, final String method) {
@@ -47,30 +47,24 @@ public class ProcessorsEndpointMerger implements EndpointResponseMerger {
final ProcessorsEntity responseEntity = clientResponse.getClientResponse().getEntity(ProcessorsEntity.class);
final Set<ProcessorEntity> processorEntities = responseEntity.getProcessors();
- final Map<String, Map<NodeIdentifier, ProcessorDTO>> dtoMap = new HashMap<>();
+ final Map<String, Map<NodeIdentifier, ProcessorEntity>> entityMap = new HashMap<>();
for (final NodeResponse nodeResponse : successfulResponses) {
final ProcessorsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : nodeResponse.getClientResponse().getEntity(ProcessorsEntity.class);
final Set<ProcessorEntity> nodeProcessorEntities = nodeResponseEntity.getProcessors();
for (final ProcessorEntity nodeProcessorEntity : nodeProcessorEntities) {
final NodeIdentifier nodeId = nodeResponse.getNodeId();
- Map<NodeIdentifier, ProcessorDTO> innerMap = dtoMap.get(nodeId);
+ Map<NodeIdentifier, ProcessorEntity> innerMap = entityMap.get(nodeId);
if (innerMap == null) {
innerMap = new HashMap<>();
- dtoMap.put(nodeProcessorEntity.getId(), innerMap);
+ entityMap.put(nodeProcessorEntity.getId(), innerMap);
}
- innerMap.put(nodeResponse.getNodeId(), nodeProcessorEntity.getComponent());
+ innerMap.put(nodeResponse.getNodeId(), nodeProcessorEntity);
}
}
- final ProcessorEndpointMerger procMerger = new ProcessorEndpointMerger();
- for (final ProcessorEntity entity : processorEntities) {
- final String componentId = entity.getId();
- final Map<NodeIdentifier, ProcessorDTO> mergeMap = dtoMap.get(componentId);
-
- procMerger.mergeResponses(entity.getComponent(), mergeMap, successfulResponses, problematicResponses);
- }
+ ProcessorsEntityMerger.mergeProcessors(processorEntities, entityMap);
// create a new client response
return new NodeResponse(clientResponse, responseEntity);
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/RemoteProcessGroupEndpointMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/RemoteProcessGroupEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/RemoteProcessGroupEndpointMerger.java
index 732d527..e130e5e 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/RemoteProcessGroupEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/RemoteProcessGroupEndpointMerger.java
@@ -17,24 +17,19 @@
package org.apache.nifi.cluster.coordination.http.endpoints;
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.RemoteProcessGroupEntityMerger;
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.web.api.entity.RemoteProcessGroupEntity;
+
import java.net.URI;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;
-import org.apache.nifi.cluster.manager.NodeResponse;
-import org.apache.nifi.cluster.protocol.NodeIdentifier;
-import org.apache.nifi.web.api.dto.RemoteProcessGroupContentsDTO;
-import org.apache.nifi.web.api.dto.RemoteProcessGroupDTO;
-import org.apache.nifi.web.api.dto.RemoteProcessGroupPortDTO;
-import org.apache.nifi.web.api.entity.RemoteProcessGroupEntity;
-
-public class RemoteProcessGroupEndpointMerger extends AbstractSingleDTOEndpoint<RemoteProcessGroupEntity, RemoteProcessGroupDTO> {
- public static final Pattern REMOTE_PROCESS_GROUPS_URI_PATTERN = Pattern.compile("/nifi-api/remote-process-groups");
+public class RemoteProcessGroupEndpointMerger extends AbstractSingleEntityEndpoint<RemoteProcessGroupEntity> implements EndpointResponseMerger {
+ public static final Pattern REMOTE_PROCESS_GROUPS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/remote-process-groups");
public static final Pattern REMOTE_PROCESS_GROUP_URI_PATTERN = Pattern.compile("/nifi-api/remote-process-groups/[a-f0-9\\-]{36}");
@Override
@@ -54,76 +49,9 @@ public class RemoteProcessGroupEndpointMerger extends AbstractSingleDTOEndpoint<
}
@Override
- protected RemoteProcessGroupDTO getDto(final RemoteProcessGroupEntity entity) {
- return entity.getComponent();
- }
-
- @Override
- protected void mergeResponses(RemoteProcessGroupDTO clientDto, Map<NodeIdentifier, RemoteProcessGroupDTO> dtoMap, Set<NodeResponse> successfulResponses, Set<NodeResponse> problematicResponses) {
- final RemoteProcessGroupContentsDTO remoteProcessGroupContents = clientDto.getContents();
-
- Boolean mergedIsTargetSecure = null;
- final List<String> mergedAuthorizationIssues = new ArrayList<>();
- final Set<RemoteProcessGroupPortDTO> mergedInputPorts = new HashSet<>();
- final Set<RemoteProcessGroupPortDTO> mergedOutputPorts = new HashSet<>();
-
- for (final Map.Entry<NodeIdentifier, RemoteProcessGroupDTO> nodeEntry : dtoMap.entrySet()) {
- final NodeIdentifier nodeId = nodeEntry.getKey();
- final RemoteProcessGroupDTO nodeRemoteProcessGroupDto = nodeEntry.getValue();
-
- // merge the issues
- final List<String> nodeAuthorizationIssues = nodeRemoteProcessGroupDto.getAuthorizationIssues();
- if (nodeAuthorizationIssues != null && !nodeAuthorizationIssues.isEmpty()) {
- for (final String nodeAuthorizationIssue : nodeAuthorizationIssues) {
- mergedAuthorizationIssues.add(nodeId.getApiAddress() + ":" + nodeId.getApiPort() + " -- " + nodeAuthorizationIssue);
- }
- }
-
- // use the first target secure flag since they will all be the same
- final Boolean nodeIsTargetSecure = nodeRemoteProcessGroupDto.isTargetSecure();
- if (mergedIsTargetSecure == null) {
- mergedIsTargetSecure = nodeIsTargetSecure;
- }
-
- // merge the ports in the contents
- final RemoteProcessGroupContentsDTO nodeRemoteProcessGroupContentsDto = nodeRemoteProcessGroupDto.getContents();
- if (remoteProcessGroupContents != null && nodeRemoteProcessGroupContentsDto != null) {
- if (nodeRemoteProcessGroupContentsDto.getInputPorts() != null) {
- mergedInputPorts.addAll(nodeRemoteProcessGroupContentsDto.getInputPorts());
- }
- if (nodeRemoteProcessGroupContentsDto.getOutputPorts() != null) {
- mergedOutputPorts.addAll(nodeRemoteProcessGroupContentsDto.getOutputPorts());
- }
- }
- }
-
- if (remoteProcessGroupContents != null) {
- if (!mergedInputPorts.isEmpty()) {
- remoteProcessGroupContents.setInputPorts(mergedInputPorts);
- }
- if (!mergedOutputPorts.isEmpty()) {
- remoteProcessGroupContents.setOutputPorts(mergedOutputPorts);
- }
- }
-
- if (mergedIsTargetSecure != null) {
- clientDto.setTargetSecure(mergedIsTargetSecure);
- }
-
- if (!mergedAuthorizationIssues.isEmpty()) {
- clientDto.setAuthorizationIssues(mergedAuthorizationIssues);
- }
- }
-
protected void mergeResponses(RemoteProcessGroupEntity clientEntity, Map<NodeIdentifier, RemoteProcessGroupEntity> entityMap,
- Set<NodeResponse> successfulResponses, Set<NodeResponse> problematicResponses) {
-
- final RemoteProcessGroupDTO clientDto = clientEntity.getComponent();
- final Map<NodeIdentifier, RemoteProcessGroupDTO> dtoMap = new HashMap<>();
- for (final Map.Entry<NodeIdentifier, RemoteProcessGroupEntity> entry : entityMap.entrySet()) {
- dtoMap.put(entry.getKey(), entry.getValue().getComponent());
- }
+ Set<NodeResponse> successfulResponses, Set<NodeResponse> problematicResponses) {
- mergeResponses(clientDto, dtoMap, successfulResponses, problematicResponses);
+ RemoteProcessGroupEntityMerger.mergeRemoteProcessGroups(clientEntity, entityMap);
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/RemoteProcessGroupsEndpointMerger.java
----------------------------------------------------------------------
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 c387951..c95aa9b 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
@@ -17,21 +17,21 @@
package org.apache.nifi.cluster.coordination.http.endpoints;
-import java.net.URI;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-import java.util.regex.Pattern;
-
import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.RemoteProcessGroupsEntityMerger;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
-import org.apache.nifi.web.api.dto.RemoteProcessGroupDTO;
import org.apache.nifi.web.api.entity.RemoteProcessGroupEntity;
import org.apache.nifi.web.api.entity.RemoteProcessGroupsEntity;
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
public class RemoteProcessGroupsEndpointMerger implements EndpointResponseMerger {
- public static final Pattern REMOTE_PROCESS_GROUPS_URI_PATTERN = Pattern.compile("/nifi-api/controller/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/remote-process-groups");
+ public static final Pattern REMOTE_PROCESS_GROUPS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/remote-process-groups");
@Override
public boolean canHandle(final URI uri, final String method) {
@@ -47,30 +47,24 @@ public class RemoteProcessGroupsEndpointMerger implements EndpointResponseMerger
final RemoteProcessGroupsEntity responseEntity = clientResponse.getClientResponse().getEntity(RemoteProcessGroupsEntity.class);
final Set<RemoteProcessGroupEntity> rpgEntities = responseEntity.getRemoteProcessGroups();
- final Map<String, Map<NodeIdentifier, RemoteProcessGroupDTO>> dtoMap = new HashMap<>();
+ final Map<String, Map<NodeIdentifier, RemoteProcessGroupEntity>> entityMap = new HashMap<>();
for (final NodeResponse nodeResponse : successfulResponses) {
final RemoteProcessGroupsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : nodeResponse.getClientResponse().getEntity(RemoteProcessGroupsEntity.class);
final Set<RemoteProcessGroupEntity> nodeRpgEntities = nodeResponseEntity.getRemoteProcessGroups();
for (final RemoteProcessGroupEntity nodeRpgEntity : nodeRpgEntities) {
final NodeIdentifier nodeId = nodeResponse.getNodeId();
- Map<NodeIdentifier, RemoteProcessGroupDTO> innerMap = dtoMap.get(nodeId);
+ Map<NodeIdentifier, RemoteProcessGroupEntity> innerMap = entityMap.get(nodeId);
if (innerMap == null) {
innerMap = new HashMap<>();
- dtoMap.put(nodeRpgEntity.getId(), innerMap);
+ entityMap.put(nodeRpgEntity.getId(), innerMap);
}
- innerMap.put(nodeResponse.getNodeId(), nodeRpgEntity.getComponent());
+ innerMap.put(nodeResponse.getNodeId(), nodeRpgEntity);
}
}
- final RemoteProcessGroupEndpointMerger rpgMerger = new RemoteProcessGroupEndpointMerger();
- for (final RemoteProcessGroupEntity entity : rpgEntities) {
- final String componentId = entity.getId();
- final Map<NodeIdentifier, RemoteProcessGroupDTO> mergeMap = dtoMap.get(componentId);
-
- rpgMerger.mergeResponses(entity.getComponent(), mergeMap, successfulResponses, problematicResponses);
- }
+ RemoteProcessGroupsEntityMerger.mergeRemoteProcessGroups(rpgEntities, entityMap);
// create a new client response
return new NodeResponse(clientResponse, responseEntity);
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ReportingTaskEndpointMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ReportingTaskEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ReportingTaskEndpointMerger.java
index 245d3bd..b3a7ccc 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ReportingTaskEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ReportingTaskEndpointMerger.java
@@ -17,20 +17,20 @@
package org.apache.nifi.cluster.coordination.http.endpoints;
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.ReportingTaskEntityMerger;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
-import org.apache.nifi.web.api.dto.ReportingTaskDTO;
import org.apache.nifi.web.api.entity.ReportingTaskEntity;
import java.net.URI;
-import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;
-public class ReportingTaskEndpointMerger extends AbstractSingleDTOEndpoint<ReportingTaskEntity, ReportingTaskDTO> {
- public static final String REPORTING_TASKS_URI = "/nifi-api/controller/reporting-tasks/node";
- public static final Pattern REPORTING_TASK_URI_PATTERN = Pattern.compile("/nifi-api/reporting-tasks/node/[a-f0-9\\-]{36}");
+public class ReportingTaskEndpointMerger extends AbstractSingleEntityEndpoint<ReportingTaskEntity> implements EndpointResponseMerger {
+ public static final String REPORTING_TASKS_URI = "/nifi-api/controller/reporting-tasks";
+ public static final Pattern REPORTING_TASK_URI_PATTERN = Pattern.compile("/nifi-api/reporting-tasks/[a-f0-9\\-]{36}");
@Override
public boolean canHandle(URI uri, String method) {
@@ -49,32 +49,7 @@ public class ReportingTaskEndpointMerger extends AbstractSingleDTOEndpoint<Repor
}
@Override
- protected ReportingTaskDTO getDto(ReportingTaskEntity entity) {
- return entity.getComponent();
+ protected void mergeResponses(ReportingTaskEntity clientEntity, Map<NodeIdentifier, ReportingTaskEntity> entityMap, Set<NodeResponse> successfulResponses, Set<NodeResponse> problematicResponses) {
+ ReportingTaskEntityMerger.mergeReportingTasks(clientEntity, entityMap);
}
-
- @Override
- protected void mergeResponses(ReportingTaskDTO clientDto, Map<NodeIdentifier, ReportingTaskDTO> dtoMap, Set<NodeResponse> successfulResponses, Set<NodeResponse> problematicResponses) {
- final Map<String, Set<NodeIdentifier>> validationErrorMap = new HashMap<>();
-
- int activeThreadCount = 0;
- for (final Map.Entry<NodeIdentifier, ReportingTaskDTO> nodeEntry : dtoMap.entrySet()) {
- final NodeIdentifier nodeId = nodeEntry.getKey();
- final ReportingTaskDTO nodeReportingTask = nodeEntry.getValue();
-
- if (nodeReportingTask.getActiveThreadCount() != null) {
- activeThreadCount += nodeReportingTask.getActiveThreadCount();
- }
-
- // merge the validation errors
- mergeValidationErrors(validationErrorMap, nodeId, nodeReportingTask.getValidationErrors());
- }
-
- // set the merged active thread counts
- clientDto.setActiveThreadCount(activeThreadCount);
-
- // set the merged the validation errors
- clientDto.setValidationErrors(normalizedMergedValidationErrors(validationErrorMap, dtoMap.size()));
- }
-
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ReportingTasksEndpointMerger.java
----------------------------------------------------------------------
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 b82b24a..a5390e8 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
@@ -17,18 +17,20 @@
package org.apache.nifi.cluster.coordination.http.endpoints;
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.ReportingTasksEntityMerger;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
import org.apache.nifi.web.api.entity.ReportingTaskEntity;
import org.apache.nifi.web.api.entity.ReportingTasksEntity;
import java.net.URI;
+import java.util.HashMap;
import java.util.Map;
import java.util.Set;
-import java.util.stream.Collectors;
-public class ReportingTasksEndpointMerger extends AbstractMultiEntityEndpoint<ReportingTasksEntity, ReportingTaskEntity> {
- public static final String REPORTING_TASKS_URI = "/nifi-api/controller/reporting-tasks/node";
+public class ReportingTasksEndpointMerger implements EndpointResponseMerger {
+ public static final String REPORTING_TASKS_URI = "/nifi-api/controller/reporting-tasks";
@Override
public boolean canHandle(URI uri, String method) {
@@ -36,28 +38,34 @@ public class ReportingTasksEndpointMerger extends AbstractMultiEntityEndpoint<Re
}
@Override
- protected Class<ReportingTasksEntity> getEntityClass() {
- return ReportingTasksEntity.class;
- }
+ public final NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
- @Override
- protected Set<ReportingTaskEntity> getDtos(ReportingTasksEntity entity) {
- return entity.getReportingTasks();
- }
+ final ReportingTasksEntity responseEntity = clientResponse.getClientResponse().getEntity(ReportingTasksEntity.class);
+ final Set<ReportingTaskEntity> reportingTasksEntities = responseEntity.getReportingTasks();
- @Override
- protected String getComponentId(ReportingTaskEntity entity) {
- return entity.getComponent().getId();
- }
+ final Map<String, Map<NodeIdentifier, ReportingTaskEntity>> entityMap = new HashMap<>();
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ReportingTasksEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : nodeResponse.getClientResponse().getEntity(ReportingTasksEntity.class);
+ final Set<ReportingTaskEntity> nodeReportingTaskEntities = nodeResponseEntity.getReportingTasks();
- @Override
- protected void mergeResponses(ReportingTaskEntity entity, Map<NodeIdentifier, ReportingTaskEntity> entityMap,
- Set<NodeResponse> successfulResponses, Set<NodeResponse> problematicResponses) {
-
- new ReportingTaskEndpointMerger().mergeResponses(
- entity.getComponent(),
- entityMap.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().getComponent())),
- successfulResponses,
- problematicResponses);
+ 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);
+ }
+
+ innerMap.put(nodeResponse.getNodeId(), nodeReportingTaskEntity);
+ }
+ }
+
+ ReportingTasksEntityMerger.mergeReportingTasks(reportingTasksEntities, entityMap);
+
+ // create a new client response
+ return new NodeResponse(clientResponse, responseEntity);
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/BulletinMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/BulletinMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/BulletinMerger.java
new file mode 100644
index 0000000..53bff3b
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/BulletinMerger.java
@@ -0,0 +1,85 @@
+/*
+ * 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.web.api.dto.BulletinDTO;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public final class BulletinMerger {
+
+ private BulletinMerger() {}
+
+ /**
+ * Merges the validation errors.
+ *
+ * @param bulletins bulletins
+ */
+ public static List<BulletinDTO> mergeBulletins(final Map<NodeIdentifier, List<BulletinDTO>> bulletins) {
+ final List<BulletinDTO> bulletinDtos = new ArrayList<>();
+
+ for (final Map.Entry<NodeIdentifier, List<BulletinDTO>> entry : bulletins.entrySet()) {
+ final NodeIdentifier nodeId = entry.getKey();
+ final List<BulletinDTO> nodeBulletins = entry.getValue();
+ final String nodeAddress = nodeId.getApiAddress() + ":" + nodeId.getApiPort();
+
+ for (final BulletinDTO bulletin : nodeBulletins) {
+ bulletin.setNodeAddress(nodeAddress);
+ bulletinDtos.add(bulletin);
+ }
+ }
+
+ Collections.sort(bulletinDtos, (BulletinDTO o1, BulletinDTO o2) -> {
+ final int timeComparison = o1.getTimestamp().compareTo(o2.getTimestamp());
+ if (timeComparison != 0) {
+ return timeComparison;
+ }
+
+ return o1.getNodeAddress().compareTo(o2.getNodeAddress());
+ });
+
+ return bulletinDtos;
+ }
+
+ /**
+ * Normalizes the validation errors.
+ *
+ * @param validationErrorMap validation errors for each node
+ * @param totalNodes total number of nodes
+ * @return the normalized validation errors
+ */
+ public static Set<String> normalizedMergedValidationErrors(final Map<String, Set<NodeIdentifier>> validationErrorMap, int totalNodes) {
+ final Set<String> normalizedValidationErrors = new HashSet<>();
+ for (final Map.Entry<String, Set<NodeIdentifier>> validationEntry : validationErrorMap.entrySet()) {
+ final String msg = validationEntry.getKey();
+ final Set<NodeIdentifier> nodeIds = validationEntry.getValue();
+
+ if (nodeIds.size() == totalNodes) {
+ normalizedValidationErrors.add(msg);
+ } else {
+ nodeIds.forEach(id -> normalizedValidationErrors.add(id.getApiAddress() + ":" + id.getApiPort() + " -- " + msg));
+ }
+ }
+ return normalizedValidationErrors;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ComponentEntityMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ComponentEntityMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ComponentEntityMerger.java
new file mode 100644
index 0000000..566b1de
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ComponentEntityMerger.java
@@ -0,0 +1,51 @@
+/*
+ * 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.web.api.dto.BulletinDTO;
+import org.apache.nifi.web.api.entity.ComponentEntity;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class ComponentEntityMerger {
+
+ /**
+ * Merges the ComponentEntity responses.
+ *
+ * @param clientEntity the entity being returned to the client
+ * @param entityMap all node responses
+ */
+ public static void mergeComponents(final ComponentEntity clientEntity, final Map<NodeIdentifier, ? extends ComponentEntity> entityMap) {
+ final Map<NodeIdentifier, List<BulletinDTO>> bulletinDtos = new HashMap<>();
+ for (final Map.Entry<NodeIdentifier, ? extends ComponentEntity> entry : entityMap.entrySet()) {
+ final NodeIdentifier nodeIdentifier = entry.getKey();
+ final ComponentEntity entity = entry.getValue();
+
+ // consider the bulletins if present and authorized
+ if (entity.getBulletins() != null) {
+ entity.getBulletins().forEach(bulletin -> {
+ bulletinDtos.computeIfAbsent(nodeIdentifier, nodeId -> new ArrayList<>()).add(bulletin);
+ });
+ }
+ }
+ clientEntity.setBulletins(BulletinMerger.mergeBulletins(bulletinDtos));
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ConnectionEntityMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ConnectionEntityMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ConnectionEntityMerger.java
new file mode 100644
index 0000000..75dea4c
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ConnectionEntityMerger.java
@@ -0,0 +1,41 @@
+/*
+ * 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.web.api.entity.ConnectionEntity;
+
+import java.util.Map;
+
+public class ConnectionEntityMerger {
+
+ /**
+ * Merges the ConnectionEntity responses.
+ *
+ * @param clientEntity the entity being returned to the client
+ * @param entityMap all node responses
+ */
+ public static void mergeConnections(final ConnectionEntity clientEntity, final Map<NodeIdentifier, ConnectionEntity> entityMap) {
+ for (final Map.Entry<NodeIdentifier, ConnectionEntity> entry : entityMap.entrySet()) {
+ final NodeIdentifier nodeId = entry.getKey();
+ final ConnectionEntity entity = entry.getValue();
+ if (entity != clientEntity) {
+ StatusMerger.merge(clientEntity.getStatus(), entity.getStatus(), nodeId.getId(), nodeId.getApiAddress(), nodeId.getApiPort());
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ConnectionsEntityMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ConnectionsEntityMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ConnectionsEntityMerger.java
new file mode 100644
index 0000000..42a9350
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ConnectionsEntityMerger.java
@@ -0,0 +1,38 @@
+/*
+ * 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.web.api.entity.ConnectionEntity;
+
+import java.util.Map;
+import java.util.Set;
+
+public class ConnectionsEntityMerger {
+
+ /**
+ * Merges multiple ConnectionEntity responses.
+ *
+ * @param connectionEntities entities being returned to the client
+ * @param entityMap all node responses
+ */
+ public static void mergeConnections(final Set<ConnectionEntity> connectionEntities, final Map<String, Map<NodeIdentifier, ConnectionEntity>> entityMap) {
+ for (final ConnectionEntity entity : connectionEntities) {
+ ConnectionEntityMerger.mergeConnections(entity, entityMap.get(entity.getId()));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ControllerServiceEntityMerger.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..e9e542e
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ControllerServiceEntityMerger.java
@@ -0,0 +1,146 @@
+/*
+ * 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.controller.service.ControllerServiceState;
+import org.apache.nifi.web.api.dto.ControllerServiceDTO;
+import org.apache.nifi.web.api.dto.ControllerServiceReferencingComponentDTO;
+import org.apache.nifi.web.api.entity.ControllerServiceEntity;
+import org.apache.nifi.web.api.entity.ControllerServiceReferencingComponentEntity;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+public class ControllerServiceEntityMerger {
+
+ /**
+ * Merges the ControllerServiceEntity responses.
+ *
+ * @param clientEntity the entity being returned to the client
+ * @param entityMap all node responses
+ */
+ public static void mergeControllerServices(final ControllerServiceEntity clientEntity, final Map<NodeIdentifier, ControllerServiceEntity> entityMap) {
+ final ControllerServiceDTO clientDto = clientEntity.getComponent();
+ final Map<NodeIdentifier, ControllerServiceDTO> dtoMap = new HashMap<>();
+ for (final Map.Entry<NodeIdentifier, ControllerServiceEntity> entry : entityMap.entrySet()) {
+ final ControllerServiceEntity nodeControllerServiceEntity = entry.getValue();
+ final ControllerServiceDTO nodeControllerServiceDto = nodeControllerServiceEntity.getComponent();
+ dtoMap.put(entry.getKey(), nodeControllerServiceDto);
+ }
+
+ ComponentEntityMerger.mergeComponents(clientEntity, entityMap);
+
+ mergeDtos(clientDto, dtoMap);
+ }
+
+ private static void mergeDtos(final ControllerServiceDTO clientDto, final Map<NodeIdentifier, ControllerServiceDTO> dtoMap) {
+ // if unauthorized for the client dto, simple return
+ if (clientDto == null) {
+ return;
+ }
+
+ final Map<String, Set<NodeIdentifier>> validationErrorMap = new HashMap<>();
+ final Set<ControllerServiceReferencingComponentEntity> referencingComponents = clientDto.getReferencingComponents();
+ final Map<NodeIdentifier, Set<ControllerServiceReferencingComponentEntity>> nodeReferencingComponentsMap = new HashMap<>();
+
+ String state = null;
+ for (final Map.Entry<NodeIdentifier, ControllerServiceDTO> nodeEntry : dtoMap.entrySet()) {
+ final ControllerServiceDTO nodeControllerService = nodeEntry.getValue();
+
+ // consider the node controller service if authorized
+ if (nodeControllerService != null) {
+ final NodeIdentifier nodeId = nodeEntry.getKey();
+
+ if (state == null) {
+ if (ControllerServiceState.DISABLING.name().equals(nodeControllerService.getState())) {
+ state = ControllerServiceState.DISABLING.name();
+ } else if (ControllerServiceState.ENABLING.name().equals(nodeControllerService.getState())) {
+ state = ControllerServiceState.ENABLING.name();
+ }
+ }
+
+ nodeReferencingComponentsMap.put(nodeId, nodeControllerService.getReferencingComponents());
+
+ // merge the validation errors
+ ErrorMerger.mergeErrors(validationErrorMap, nodeId, nodeControllerService.getValidationErrors());
+ }
+ }
+
+ // merge the referencing components
+ mergeControllerServiceReferences(referencingComponents, nodeReferencingComponentsMap);
+
+ // store the 'transition' state is applicable
+ if (state != null) {
+ clientDto.setState(state);
+ }
+
+ // set the merged the validation errors
+ clientDto.setValidationErrors(ErrorMerger.normalizedMergedErrors(validationErrorMap, dtoMap.size()));
+ }
+
+ public static void mergeControllerServiceReferences(Set<ControllerServiceReferencingComponentEntity> referencingComponents,
+ Map<NodeIdentifier, Set<ControllerServiceReferencingComponentEntity>> referencingComponentMap) {
+
+ final Map<String, Integer> activeThreadCounts = new HashMap<>();
+ final Map<String, String> states = new HashMap<>();
+ for (final Map.Entry<NodeIdentifier, Set<ControllerServiceReferencingComponentEntity>> nodeEntry : referencingComponentMap.entrySet()) {
+ final Set<ControllerServiceReferencingComponentEntity> nodeReferencingComponents = nodeEntry.getValue();
+
+ // go through all the nodes referencing components
+ if (nodeReferencingComponents != null) {
+ for (final ControllerServiceReferencingComponentEntity nodeReferencingComponentEntity : nodeReferencingComponents) {
+ final ControllerServiceReferencingComponentDTO nodeReferencingComponent = nodeReferencingComponentEntity.getComponent();
+
+ // handle active thread counts
+ if (nodeReferencingComponent.getActiveThreadCount() != null && nodeReferencingComponent.getActiveThreadCount() > 0) {
+ final Integer current = activeThreadCounts.get(nodeReferencingComponent.getId());
+ if (current == null) {
+ activeThreadCounts.put(nodeReferencingComponent.getId(), nodeReferencingComponent.getActiveThreadCount());
+ } else {
+ activeThreadCounts.put(nodeReferencingComponent.getId(), nodeReferencingComponent.getActiveThreadCount() + current);
+ }
+ }
+
+ // handle controller service state
+ final String state = states.get(nodeReferencingComponent.getId());
+ if (state == null) {
+ if (ControllerServiceState.DISABLING.name().equals(nodeReferencingComponent.getState())) {
+ states.put(nodeReferencingComponent.getId(), ControllerServiceState.DISABLING.name());
+ } else if (ControllerServiceState.ENABLING.name().equals(nodeReferencingComponent.getState())) {
+ states.put(nodeReferencingComponent.getId(), ControllerServiceState.ENABLING.name());
+ }
+ }
+ }
+ }
+ }
+
+ // go through each referencing components
+ for (final ControllerServiceReferencingComponentEntity referencingComponent : referencingComponents) {
+ final Integer activeThreadCount = activeThreadCounts.get(referencingComponent.getId());
+ if (activeThreadCount != null) {
+ referencingComponent.getComponent().setActiveThreadCount(activeThreadCount);
+ }
+
+ final String state = states.get(referencingComponent.getId());
+ if (state != null) {
+ referencingComponent.getComponent().setState(state);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ControllerServicesEntityMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ControllerServicesEntityMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ControllerServicesEntityMerger.java
new file mode 100644
index 0000000..ca97af6
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ControllerServicesEntityMerger.java
@@ -0,0 +1,38 @@
+/*
+ * 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.web.api.entity.ControllerServiceEntity;
+
+import java.util.Map;
+import java.util.Set;
+
+public class ControllerServicesEntityMerger {
+
+ /**
+ * Merges multiple ControllerServiceEntity responses.
+ *
+ * @param controllerServiceEntities entities being returned to the client
+ * @param entityMap all node responses
+ */
+ public static void mergeControllerServices(final Set<ControllerServiceEntity> controllerServiceEntities, final Map<String, Map<NodeIdentifier, ControllerServiceEntity>> entityMap) {
+ for (final ControllerServiceEntity entity : controllerServiceEntities) {
+ ControllerServiceEntityMerger.mergeControllerServices(entity, entityMap.get(entity.getId()));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ErrorMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ErrorMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ErrorMerger.java
new file mode 100644
index 0000000..1b51e16
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ErrorMerger.java
@@ -0,0 +1,66 @@
+/*
+ * 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 java.util.Collection;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public final class ErrorMerger {
+
+ private ErrorMerger() {}
+
+ /**
+ * Merges the validation or authorization errors.
+ *
+ * @param validationErrorMap errors for each node
+ * @param nodeId node id
+ * @param nodeErrors node errors
+ */
+ public static void mergeErrors(final Map<String, Set<NodeIdentifier>> validationErrorMap, final NodeIdentifier nodeId, final Collection<String> nodeErrors) {
+ if (nodeErrors != null) {
+ nodeErrors.stream().forEach(
+ err -> validationErrorMap.computeIfAbsent(err, k -> new HashSet<NodeIdentifier>())
+ .add(nodeId));
+ }
+ }
+
+ /**
+ * Normalizes the validation errors.
+ *
+ * @param errorMap validation errors for each node
+ * @param totalNodes total number of nodes
+ * @return the normalized validation errors
+ */
+ public static Set<String> normalizedMergedErrors(final Map<String, Set<NodeIdentifier>> errorMap, int totalNodes) {
+ final Set<String> normalizedErrors = new HashSet<>();
+ for (final Map.Entry<String, Set<NodeIdentifier>> validationEntry : errorMap.entrySet()) {
+ final String msg = validationEntry.getKey();
+ final Set<NodeIdentifier> nodeIds = validationEntry.getValue();
+
+ if (nodeIds.size() == totalNodes) {
+ normalizedErrors.add(msg);
+ } else {
+ nodeIds.forEach(id -> normalizedErrors.add(id.getApiAddress() + ":" + id.getApiPort() + " -- " + msg));
+ }
+ }
+ return normalizedErrors;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/PortEntityMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/PortEntityMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/PortEntityMerger.java
new file mode 100644
index 0000000..37c922f
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/PortEntityMerger.java
@@ -0,0 +1,78 @@
+/*
+ * 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.web.api.dto.PortDTO;
+import org.apache.nifi.web.api.entity.PortEntity;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+public class PortEntityMerger {
+
+ /**
+ * Merges the PortEntity responses.
+ *
+ * @param clientEntity the entity being returned to the client
+ * @param entityMap all node responses
+ */
+ public static void mergePorts(final PortEntity clientEntity, final Map<NodeIdentifier, PortEntity> entityMap) {
+ final PortDTO clientDto = clientEntity.getComponent();
+ final Map<NodeIdentifier, PortDTO> dtoMap = new HashMap<>();
+ for (final Map.Entry<NodeIdentifier, PortEntity> entry : entityMap.entrySet()) {
+ final PortEntity nodePortEntity = entry.getValue();
+ final PortDTO nodePortDto = nodePortEntity.getComponent();
+ dtoMap.put(entry.getKey(), nodePortDto);
+ }
+
+ for (final Map.Entry<NodeIdentifier, PortEntity> entry : entityMap.entrySet()) {
+ final NodeIdentifier nodeId = entry.getKey();
+ final PortEntity entity = entry.getValue();
+ if (entity != clientEntity) {
+ StatusMerger.merge(clientEntity.getStatus(), entity.getStatus(), nodeId.getId(), nodeId.getApiAddress(), nodeId.getApiPort());
+ }
+ }
+
+ ComponentEntityMerger.mergeComponents(clientEntity, entityMap);
+
+ mergeDtos(clientDto, dtoMap);
+ }
+
+ private static void mergeDtos(final PortDTO clientDto, final Map<NodeIdentifier, PortDTO> dtoMap) {
+ // if unauthorized for the client dto, simple return
+ if (clientDto == null) {
+ return;
+ }
+
+ final Map<String, Set<NodeIdentifier>> validationErrorMap = new HashMap<>();
+
+ for (final Map.Entry<NodeIdentifier, PortDTO> nodeEntry : dtoMap.entrySet()) {
+ final PortDTO nodePort = nodeEntry.getValue();
+
+ // merge the validation errors if authorized
+ if (nodePort != null) {
+ final NodeIdentifier nodeId = nodeEntry.getKey();
+ ErrorMerger.mergeErrors(validationErrorMap, nodeId, nodePort.getValidationErrors());
+ }
+ }
+
+ // set the merged the validation errors
+ clientDto.setValidationErrors(ErrorMerger.normalizedMergedErrors(validationErrorMap, dtoMap.size()));
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/PortsEntityMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/PortsEntityMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/PortsEntityMerger.java
new file mode 100644
index 0000000..894ce5b
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/PortsEntityMerger.java
@@ -0,0 +1,38 @@
+/*
+ * 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.web.api.entity.PortEntity;
+
+import java.util.Map;
+import java.util.Set;
+
+public class PortsEntityMerger {
+
+ /**
+ * Merges multiple PortEntity responses.
+ *
+ * @param portEntities entities being returned to the client
+ * @param entityMap all node responses
+ */
+ public static void mergePorts(final Set<PortEntity> portEntities, final Map<String, Map<NodeIdentifier, PortEntity>> entityMap) {
+ for (final PortEntity entity : portEntities) {
+ PortEntityMerger.mergePorts(entity, entityMap.get(entity.getId()));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ProcessGroupEntityMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ProcessGroupEntityMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ProcessGroupEntityMerger.java
new file mode 100644
index 0000000..a9a1b40
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ProcessGroupEntityMerger.java
@@ -0,0 +1,43 @@
+/*
+ * 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.web.api.entity.ProcessGroupEntity;
+
+import java.util.Map;
+
+public class ProcessGroupEntityMerger {
+
+ /**
+ * Merges the ProcessorGroupEntity responses.
+ *
+ * @param clientEntity the entity being returned to the client
+ * @param entityMap all node responses
+ */
+ public static void mergeProcessGroups(final ProcessGroupEntity clientEntity, final Map<NodeIdentifier, ProcessGroupEntity> entityMap) {
+ for (final Map.Entry<NodeIdentifier, ProcessGroupEntity> entry : entityMap.entrySet()) {
+ final NodeIdentifier nodeId = entry.getKey();
+ final ProcessGroupEntity entity = entry.getValue();
+ if (entity != clientEntity) {
+ StatusMerger.merge(clientEntity.getStatus(), entity.getStatus(), nodeId.getId(), nodeId.getApiAddress(), nodeId.getApiPort());
+ }
+ }
+
+ ComponentEntityMerger.mergeComponents(clientEntity, entityMap);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ProcessGroupsEntityMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ProcessGroupsEntityMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ProcessGroupsEntityMerger.java
new file mode 100644
index 0000000..6a0e519
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ProcessGroupsEntityMerger.java
@@ -0,0 +1,38 @@
+/*
+ * 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.web.api.entity.ProcessGroupEntity;
+
+import java.util.Map;
+import java.util.Set;
+
+public class ProcessGroupsEntityMerger {
+
+ /**
+ * Merges multiple ProcessGroupEntity responses.
+ *
+ * @param processGroupEntities entities being returned to the client
+ * @param entityMap all node responses
+ */
+ public static void mergeProcessGroups(final Set<ProcessGroupEntity> processGroupEntities, final Map<String, Map<NodeIdentifier, ProcessGroupEntity>> entityMap) {
+ for (final ProcessGroupEntity entity : processGroupEntities) {
+ ProcessGroupEntityMerger.mergeProcessGroups(entity, entityMap.get(entity.getId()));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ProcessorEntityMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ProcessorEntityMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ProcessorEntityMerger.java
new file mode 100644
index 0000000..e730791
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ProcessorEntityMerger.java
@@ -0,0 +1,78 @@
+/*
+ * 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.web.api.dto.ProcessorDTO;
+import org.apache.nifi.web.api.entity.ProcessorEntity;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+public class ProcessorEntityMerger {
+
+ /**
+ * Merges the ProcessorEntity responses.
+ *
+ * @param clientEntity the entity being returned to the client
+ * @param entityMap all node responses
+ */
+ public static void mergeProcessors(final ProcessorEntity clientEntity, final Map<NodeIdentifier, ProcessorEntity> entityMap) {
+ final ProcessorDTO clientDto = clientEntity.getComponent();
+ final Map<NodeIdentifier, ProcessorDTO> dtoMap = new HashMap<>();
+ for (final Map.Entry<NodeIdentifier, ProcessorEntity> entry : entityMap.entrySet()) {
+ final ProcessorEntity nodeProcEntity = entry.getValue();
+ final ProcessorDTO nodeProcDto = nodeProcEntity.getComponent();
+ dtoMap.put(entry.getKey(), nodeProcDto);
+ }
+
+ for (final Map.Entry<NodeIdentifier, ProcessorEntity> entry : entityMap.entrySet()) {
+ final NodeIdentifier nodeId = entry.getKey();
+ final ProcessorEntity entity = entry.getValue();
+ if (entity != clientEntity) {
+ StatusMerger.merge(clientEntity.getStatus(), entity.getStatus(), nodeId.getId(), nodeId.getApiAddress(), nodeId.getApiPort());
+ }
+ }
+
+ ComponentEntityMerger.mergeComponents(clientEntity, entityMap);
+
+ mergeDtos(clientDto, dtoMap);
+ }
+
+ private static void mergeDtos(final ProcessorDTO clientDto, final Map<NodeIdentifier, ProcessorDTO> dtoMap) {
+ // if unauthorized for the client dto, simple return
+ if (clientDto == null) {
+ return;
+ }
+
+ final Map<String, Set<NodeIdentifier>> validationErrorMap = new HashMap<>();
+
+ for (final Map.Entry<NodeIdentifier, ProcessorDTO> nodeEntry : dtoMap.entrySet()) {
+ final ProcessorDTO nodeProcessor = nodeEntry.getValue();
+
+ // merge the validation errors, if authorized
+ if (nodeProcessor != null) {
+ final NodeIdentifier nodeId = nodeEntry.getKey();
+ ErrorMerger.mergeErrors(validationErrorMap, nodeId, nodeProcessor.getValidationErrors());
+ }
+ }
+
+ // set the merged the validation errors
+ clientDto.setValidationErrors(ErrorMerger.normalizedMergedErrors(validationErrorMap, dtoMap.size()));
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ProcessorsEntityMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ProcessorsEntityMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ProcessorsEntityMerger.java
new file mode 100644
index 0000000..cf4ef7d
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ProcessorsEntityMerger.java
@@ -0,0 +1,38 @@
+/*
+ * 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.web.api.entity.ProcessorEntity;
+
+import java.util.Map;
+import java.util.Set;
+
+public class ProcessorsEntityMerger {
+
+ /**
+ * Merges multiple ProcessorEntity responses.
+ *
+ * @param processorEntities entities being returned to the client
+ * @param entityMap all node responses
+ */
+ public static void mergeProcessors(final Set<ProcessorEntity> processorEntities, final Map<String, Map<NodeIdentifier, ProcessorEntity>> entityMap) {
+ for (final ProcessorEntity entity : processorEntities) {
+ ProcessorEntityMerger.mergeProcessors(entity, entityMap.get(entity.getId()));
+ }
+ }
+}