You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jp...@apache.org on 2016/07/20 23:45:19 UTC
nifi git commit: NIFI-2332: Ensure that bulletin node address is set
when clustered but not overwritten if already set
Repository: nifi
Updated Branches:
refs/heads/master 7f2bda29d -> 10b91ffe6
NIFI-2332: Ensure that bulletin node address is set when clustered but not overwritten if already set
This closes #692
Signed-off-by: jpercivall <jo...@yahoo.com>
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/10b91ffe
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/10b91ffe
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/10b91ffe
Branch: refs/heads/master
Commit: 10b91ffe6c192b15c643f6d17c29924676208cc0
Parents: 7f2bda2
Author: Mark Payne <ma...@hotmail.com>
Authored: Wed Jul 20 14:59:58 2016 -0400
Committer: jpercivall <jo...@yahoo.com>
Committed: Wed Jul 20 19:44:32 2016 -0400
----------------------------------------------------------------------
.../ControllerBulletinsEndpointMerger.java | 16 +++++++++++++++-
.../endpoints/ControllerStatusEndpointMerger.java | 1 -
.../apache/nifi/cluster/manager/BulletinMerger.java | 5 ++++-
3 files changed, 19 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/10b91ffe/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerBulletinsEndpointMerger.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/ControllerBulletinsEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerBulletinsEndpointMerger.java
index 6e79bc1..690f1cc 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerBulletinsEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerBulletinsEndpointMerger.java
@@ -38,7 +38,7 @@ import static org.apache.nifi.reporting.BulletinRepository.MAX_BULLETINS_FOR_CON
import static org.apache.nifi.reporting.BulletinRepository.MAX_BULLETINS_PER_COMPONENT;
public class ControllerBulletinsEndpointMerger extends AbstractSingleEntityEndpoint<ControllerBulletinsEntity> implements EndpointResponseMerger {
- public static final Pattern CONTROLLER_BULLETINS_URI_PATTERN = Pattern.compile("/nifi-api/controller/bulletins");
+ public static final Pattern CONTROLLER_BULLETINS_URI_PATTERN = Pattern.compile("/nifi-api/flow/controller/bulletins");
@Override
public boolean canHandle(URI uri, String method) {
@@ -60,24 +60,38 @@ public class ControllerBulletinsEndpointMerger extends AbstractSingleEntityEndpo
for (final Map.Entry<NodeIdentifier, ControllerBulletinsEntity> entry : entityMap.entrySet()) {
final NodeIdentifier nodeIdentifier = entry.getKey();
final ControllerBulletinsEntity entity = entry.getValue();
+ final String nodeAddress = nodeIdentifier.getApiAddress() + ":" + nodeIdentifier.getApiPort();
// consider the bulletins if present and authorized
if (entity.getBulletins() != null) {
entity.getBulletins().forEach(bulletin -> {
+ if (bulletin.getNodeAddress() == null) {
+ bulletin.setNodeAddress(nodeAddress);
+ }
+
bulletinDtos.computeIfAbsent(nodeIdentifier, nodeId -> new ArrayList<>()).add(bulletin);
});
}
if (entity.getControllerServiceBulletins() != null) {
entity.getControllerServiceBulletins().forEach(bulletin -> {
+ if (bulletin.getNodeAddress() == null) {
+ bulletin.setNodeAddress(nodeAddress);
+ }
+
controllerServiceBulletinDtos.computeIfAbsent(nodeIdentifier, nodeId -> new ArrayList<>()).add(bulletin);
});
}
if (entity.getReportingTaskBulletins() != null) {
entity.getReportingTaskBulletins().forEach(bulletin -> {
+ if (bulletin.getNodeAddress() == null) {
+ bulletin.setNodeAddress(nodeAddress);
+ }
+
reportingTaskBulletinDtos.computeIfAbsent(nodeIdentifier, nodeId -> new ArrayList<>()).add(bulletin);
});
}
}
+
clientEntity.setBulletins(BulletinMerger.mergeBulletins(bulletinDtos));
clientEntity.setControllerServiceBulletins(BulletinMerger.mergeBulletins(controllerServiceBulletinDtos));
clientEntity.setReportingTaskBulletins(BulletinMerger.mergeBulletins(reportingTaskBulletinDtos));
http://git-wip-us.apache.org/repos/asf/nifi/blob/10b91ffe/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerStatusEndpointMerger.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/ControllerStatusEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerStatusEndpointMerger.java
index 26c7de6..4634624 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerStatusEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerStatusEndpointMerger.java
@@ -50,7 +50,6 @@ public class ControllerStatusEndpointMerger extends AbstractSingleDTOEndpoint<Co
protected void mergeResponses(ControllerStatusDTO clientDto, Map<NodeIdentifier, ControllerStatusDTO> dtoMap, Set<NodeResponse> successfulResponses, Set<NodeResponse> problematicResponses) {
ControllerStatusDTO mergedStatus = clientDto;
for (final Map.Entry<NodeIdentifier, ControllerStatusDTO> entry : dtoMap.entrySet()) {
- final NodeIdentifier nodeId = entry.getKey();
final ControllerStatusDTO nodeStatus = entry.getValue();
if (nodeStatus == mergedStatus) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/10b91ffe/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
index 0d32244..431c834 100644
--- 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
@@ -62,7 +62,10 @@ public final class BulletinMerger {
final String nodeAddress = nodeId.getApiAddress() + ":" + nodeId.getApiPort();
for (final BulletinDTO bulletin : nodeBulletins) {
- bulletin.setNodeAddress(nodeAddress);
+ if (bulletin.getNodeAddress() == null) {
+ bulletin.setNodeAddress(nodeAddress);
+ }
+
bulletinDtos.add(bulletin);
}
}