You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by mc...@apache.org on 2017/11/13 18:55:04 UTC
nifi git commit: NIFI-3155, NIFI-4571: Fix issue of Remote Group Port
having ID the same as on the remote system, by adding an additional targetId
field RemoteGroupPort's;
this also fixes issue of stats being mixed together for RPG's with the same
URL. T
Repository: nifi
Updated Branches:
refs/heads/master c70a86eac -> 2d3e5abf8
NIFI-3155, NIFI-4571: Fix issue of Remote Group Port having ID the same as on the remote system, by adding an additional targetId field RemoteGroupPort's; this also fixes issue of stats being mixed together for RPG's with the same URL. This closes #2253
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/2d3e5abf
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/2d3e5abf
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/2d3e5abf
Branch: refs/heads/master
Commit: 2d3e5abf81d5fa70cd6efd60644755b51c785369
Parents: c70a86e
Author: Mark Payne <ma...@hotmail.com>
Authored: Mon Nov 6 10:10:29 2017 -0500
Committer: Matt Gilman <ma...@gmail.com>
Committed: Mon Nov 13 13:54:34 2017 -0500
----------------------------------------------------------------------
.../web/api/dto/RemoteProcessGroupPortDTO.java | 12 ++++-
.../RemoteProcessGroupPortDescriptor.java | 7 ++-
.../org/apache/nifi/remote/RemoteGroupPort.java | 2 +
.../apache/nifi/controller/FlowController.java | 1 +
.../serialization/FlowFromDOMFactory.java | 4 ++
.../serialization/StandardFlowSerializer.java | 1 +
.../nifi/fingerprint/FingerprintFactory.java | 2 +-
.../nifi/remote/StandardRemoteProcessGroup.java | 51 +++++++++++++-------
...tandardRemoteProcessGroupPortDescriptor.java | 10 ++++
.../src/main/resources/FlowConfiguration.xsd | 1 +
.../nifi/remote/StandardRemoteGroupPort.java | 11 ++++-
.../remote/TestStandardRemoteGroupPort.java | 2 +-
.../org/apache/nifi/web/api/dto/DtoFactory.java | 2 +
.../org/apache/nifi/web/util/SnippetUtils.java | 15 +++++-
.../audit/TestRemoteProcessGroupAuditor.java | 1 +
.../impl/TestStandardRemoteProcessGroupDAO.java | 1 +
16 files changed, 98 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/2d3e5abf/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/RemoteProcessGroupPortDTO.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/RemoteProcessGroupPortDTO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/RemoteProcessGroupPortDTO.java
index 83bd59e..1ebb70e 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/RemoteProcessGroupPortDTO.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/RemoteProcessGroupPortDTO.java
@@ -27,6 +27,7 @@ import javax.xml.bind.annotation.XmlType;
public class RemoteProcessGroupPortDTO {
private String id;
+ private String targetId;
private String groupId;
private String name;
private String comments;
@@ -70,7 +71,7 @@ public class RemoteProcessGroupPortDTO {
* @return id of the target port
*/
@ApiModelProperty(
- value = "The id of the target port."
+ value = "The id of the port."
)
public String getId() {
return id;
@@ -79,6 +80,15 @@ public class RemoteProcessGroupPortDTO {
public void setId(String id) {
this.id = id;
}
+
+ @ApiModelProperty("The id of the target port.")
+ public String getTargetId() {
+ return targetId;
+ }
+
+ public void setTargetId(String targetId) {
+ this.targetId = targetId;
+ }
/**
* @return id of the remote process group that this port resides in
http://git-wip-us.apache.org/repos/asf/nifi/blob/2d3e5abf/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/RemoteProcessGroupPortDescriptor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/RemoteProcessGroupPortDescriptor.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/RemoteProcessGroupPortDescriptor.java
index c330c13..b797749 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/RemoteProcessGroupPortDescriptor.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/RemoteProcessGroupPortDescriptor.java
@@ -30,11 +30,16 @@ public interface RemoteProcessGroupPortDescriptor {
Integer getConcurrentlySchedulableTaskCount();
/**
- * @return id of the target port
+ * @return id of the port
*/
String getId();
/**
+ * @return the id of the target port
+ */
+ String getTargetId();
+
+ /**
* @return id of the remote process group that this port resides in
*/
String getGroupId();
http://git-wip-us.apache.org/repos/asf/nifi/blob/2d3e5abf/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/remote/RemoteGroupPort.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/remote/RemoteGroupPort.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/remote/RemoteGroupPort.java
index 07faf42..5d5fe7d 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/remote/RemoteGroupPort.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/remote/RemoteGroupPort.java
@@ -40,6 +40,8 @@ public abstract class RemoteGroupPort extends AbstractPort implements Port, Remo
public abstract boolean getTargetExists();
+ public abstract String getTargetIdentifier();
+
public abstract boolean isTargetRunning();
public abstract Integer getBatchCount();
http://git-wip-us.apache.org/repos/asf/nifi/blob/2d3e5abf/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index 99d8ed0..49c1789 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -2061,6 +2061,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
for (final RemoteProcessGroupPortDTO port : ports) {
final StandardRemoteProcessGroupPortDescriptor descriptor = new StandardRemoteProcessGroupPortDescriptor();
descriptor.setId(port.getId());
+ descriptor.setTargetId(port.getTargetId());
descriptor.setName(port.getName());
descriptor.setComments(port.getComments());
descriptor.setTargetRunning(port.isTargetRunning());
http://git-wip-us.apache.org/repos/asf/nifi/blob/2d3e5abf/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/FlowFromDOMFactory.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/FlowFromDOMFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/FlowFromDOMFactory.java
index d455043..f67ecd9 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/FlowFromDOMFactory.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/FlowFromDOMFactory.java
@@ -366,6 +366,10 @@ public class FlowFromDOMFactory {
}
descriptor.setId(id);
+
+ final String targetId = getString(element, "targetId");
+ descriptor.setTargetId(targetId == null ? id : targetId);
+
descriptor.setName(getString(element, "name"));
descriptor.setComments(getString(element, "comments"));
descriptor.setConcurrentlySchedulableTaskCount(getInt(element, "maxConcurrentTasks"));
http://git-wip-us.apache.org/repos/asf/nifi/blob/2d3e5abf/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/StandardFlowSerializer.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/StandardFlowSerializer.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/StandardFlowSerializer.java
index 2a8df96..bc28a25 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/StandardFlowSerializer.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/StandardFlowSerializer.java
@@ -326,6 +326,7 @@ public class StandardFlowSerializer implements FlowSerializer {
addPosition(element, port.getPosition());
addTextElement(element, "comments", port.getComments());
addTextElement(element, "scheduledState", scheduledStateLookup.getScheduledState(port).name());
+ addTextElement(element, "targetId", port.getTargetIdentifier());
addTextElement(element, "maxConcurrentTasks", port.getMaxConcurrentTasks());
addTextElement(element, "useCompression", String.valueOf(port.isUseCompression()));
final Integer batchCount = port.getBatchCount();
http://git-wip-us.apache.org/repos/asf/nifi/blob/2d3e5abf/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/fingerprint/FingerprintFactory.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/fingerprint/FingerprintFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/fingerprint/FingerprintFactory.java
index 3957955..7c68475 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/fingerprint/FingerprintFactory.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/fingerprint/FingerprintFactory.java
@@ -544,7 +544,7 @@ public class FingerprintFactory {
}
private StringBuilder addRemoteGroupPortFingerprint(final StringBuilder builder, final Element remoteGroupPortElement) {
- for (final String childName : new String[] {"id", "maxConcurrentTasks", "useCompression", "batchCount", "batchSize", "batchDuration"}) {
+ for (final String childName : new String[] {"id", "targetId", "maxConcurrentTasks", "useCompression", "batchCount", "batchSize", "batchDuration"}) {
appendFirstValue(builder, DomUtils.getChildNodesByTagName(remoteGroupPortElement, childName));
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/2d3e5abf/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/remote/StandardRemoteProcessGroup.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/remote/StandardRemoteProcessGroup.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/remote/StandardRemoteProcessGroup.java
index 2b2e1fb..49fd90f 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/remote/StandardRemoteProcessGroup.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/remote/StandardRemoteProcessGroup.java
@@ -22,6 +22,7 @@ import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.net.NetworkInterface;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -34,6 +35,7 @@ import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -41,6 +43,8 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Function;
+import java.util.stream.Collectors;
import javax.net.ssl.SSLContext;
import javax.ws.rs.core.Response;
@@ -410,11 +414,14 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup {
public void setInputPorts(final Set<RemoteProcessGroupPortDescriptor> ports) {
writeLock.lock();
try {
- final List<String> newPortIds = new ArrayList<>();
+ final List<String> newPortTargetIds = new ArrayList<>();
for (final RemoteProcessGroupPortDescriptor descriptor : ports) {
- newPortIds.add(descriptor.getId());
+ newPortTargetIds.add(descriptor.getTargetId());
- if (!inputPorts.containsKey(descriptor.getId())) {
+ final Map<String, StandardRemoteGroupPort> inputPortByTargetId = inputPorts.values().stream()
+ .collect(Collectors.toMap(StandardRemoteGroupPort::getTargetIdentifier, Function.identity()));
+
+ if (!inputPortByTargetId.containsKey(descriptor.getTargetId())) {
addInputPort(descriptor);
}
@@ -430,11 +437,10 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup {
// See if we have any ports that no longer exist; cannot be removed within the loop because it would cause
// a ConcurrentModificationException.
- final Iterator<Map.Entry<String, StandardRemoteGroupPort>> itr = inputPorts.entrySet().iterator();
+ final Iterator<StandardRemoteGroupPort> itr = inputPorts.values().iterator();
while (itr.hasNext()) {
- final Map.Entry<String, StandardRemoteGroupPort> entry = itr.next();
- if (!newPortIds.contains(entry.getKey())) {
- final StandardRemoteGroupPort port = entry.getValue();
+ final StandardRemoteGroupPort port = itr.next();
+ if (!newPortTargetIds.contains(port.getTargetIdentifier())) {
port.setTargetExists(false);
port.setTargetRunning(false);
@@ -481,11 +487,14 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup {
public void setOutputPorts(final Set<RemoteProcessGroupPortDescriptor> ports) {
writeLock.lock();
try {
- final List<String> newPortIds = new ArrayList<>();
+ final List<String> newPortTargetIds = new ArrayList<>();
for (final RemoteProcessGroupPortDescriptor descriptor : requireNonNull(ports)) {
- newPortIds.add(descriptor.getId());
+ newPortTargetIds.add(descriptor.getTargetId());
+
+ final Map<String, StandardRemoteGroupPort> outputPortsByTargetId = outputPorts.values().stream()
+ .collect(Collectors.toMap(StandardRemoteGroupPort::getTargetIdentifier, Function.identity()));
- if (!outputPorts.containsKey(descriptor.getId())) {
+ if (!outputPortsByTargetId.containsKey(descriptor.getTargetId())) {
addOutputPort(descriptor);
}
@@ -501,11 +510,10 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup {
// See if we have any ports that no longer exist; cannot be removed within the loop because it would cause
// a ConcurrentModificationException.
- final Iterator<Map.Entry<String, StandardRemoteGroupPort>> itr = outputPorts.entrySet().iterator();
+ final Iterator<StandardRemoteGroupPort> itr = outputPorts.values().iterator();
while (itr.hasNext()) {
- final Map.Entry<String, StandardRemoteGroupPort> entry = itr.next();
- if (!newPortIds.contains(entry.getKey())) {
- final StandardRemoteGroupPort port = entry.getValue();
+ final StandardRemoteGroupPort port = itr.next();
+ if (!newPortTargetIds.contains(port.getTargetIdentifier())) {
port.setTargetExists(false);
port.setTargetRunning(false);
@@ -619,7 +627,7 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup {
throw new IllegalStateException("Output Port with ID " + descriptor.getId() + " already exists");
}
- final StandardRemoteGroupPort port = new StandardRemoteGroupPort(descriptor.getId(), descriptor.getName(), getProcessGroup(),
+ final StandardRemoteGroupPort port = new StandardRemoteGroupPort(descriptor.getId(), descriptor.getTargetId(), descriptor.getName(), getProcessGroup(),
this, TransferDirection.RECEIVE, ConnectableType.REMOTE_OUTPUT_PORT, sslContext, scheduler, nifiProperties);
outputPorts.put(descriptor.getId(), port);
@@ -694,7 +702,11 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup {
throw new IllegalStateException("Input Port with ID " + descriptor.getId() + " already exists");
}
- final StandardRemoteGroupPort port = new StandardRemoteGroupPort(descriptor.getId(), descriptor.getName(), getProcessGroup(), this,
+ // We need to generate the port's UUID deterministically because we need
+ // all nodes in a cluster to use the same UUID. However, we want the ID to be
+ // unique for each Remote Group Port, so that if we have multiple RPG's pointing
+ // to the same target, we have unique ID's for each of those ports.
+ final StandardRemoteGroupPort port = new StandardRemoteGroupPort(descriptor.getId(), descriptor.getTargetId(), descriptor.getName(), getProcessGroup(), this,
TransferDirection.SEND, ConnectableType.REMOTE_INPUT_PORT, sslContext, scheduler, nifiProperties);
if (descriptor.getConcurrentlySchedulableTaskCount() != null) {
@@ -719,6 +731,10 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup {
}
}
+ private String generatePortId(final String targetId) {
+ return UUID.nameUUIDFromBytes((this.getIdentifier() + targetId).getBytes(StandardCharsets.UTF_8)).toString();
+ }
+
@Override
public RemoteGroupPort getOutputPort(final String portIdentifier) {
readLock.lock();
@@ -964,7 +980,8 @@ public class StandardRemoteProcessGroup implements RemoteProcessGroup {
for (final PortDTO port : ports) {
final StandardRemoteProcessGroupPortDescriptor descriptor = new StandardRemoteProcessGroupPortDescriptor();
final ScheduledState scheduledState = ScheduledState.valueOf(port.getState());
- descriptor.setId(port.getId());
+ descriptor.setId(generatePortId(port.getId()));
+ descriptor.setTargetId(port.getId());
descriptor.setName(port.getName());
descriptor.setComments(port.getComments());
descriptor.setTargetRunning(ScheduledState.RUNNING.equals(scheduledState));
http://git-wip-us.apache.org/repos/asf/nifi/blob/2d3e5abf/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/remote/StandardRemoteProcessGroupPortDescriptor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/remote/StandardRemoteProcessGroupPortDescriptor.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/remote/StandardRemoteProcessGroupPortDescriptor.java
index c3a8f5e..7dbcec1 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/remote/StandardRemoteProcessGroupPortDescriptor.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/remote/StandardRemoteProcessGroupPortDescriptor.java
@@ -21,6 +21,7 @@ import org.apache.nifi.groups.RemoteProcessGroupPortDescriptor;
public class StandardRemoteProcessGroupPortDescriptor implements RemoteProcessGroupPortDescriptor {
private String id;
+ private String targetId;
private String groupId;
private String name;
private String comments;
@@ -62,6 +63,15 @@ public class StandardRemoteProcessGroupPortDescriptor implements RemoteProcessGr
}
@Override
+ public String getTargetId() {
+ return targetId;
+ }
+
+ public void setTargetId(String targetId) {
+ this.targetId = targetId;
+ }
+
+ @Override
public String getGroupId() {
return groupId;
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/2d3e5abf/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/resources/FlowConfiguration.xsd
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/resources/FlowConfiguration.xsd b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/resources/FlowConfiguration.xsd
index 247a790..19e1c55 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/resources/FlowConfiguration.xsd
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/resources/FlowConfiguration.xsd
@@ -285,6 +285,7 @@
<xs:complexContent>
<xs:extension base="PortType">
<xs:sequence>
+ <xs:element name="targetId" type="NonEmptyStringType" minOccurs="0" maxOccurs="1" />
<xs:element name="maxConcurrentTasks" type="xs:positiveInteger"></xs:element>
<xs:element name="useCompression" type="xs:boolean"></xs:element>
<xs:element name="batchCount" type="xs:positiveInteger" minOccurs="0" maxOccurs="1" />
http://git-wip-us.apache.org/repos/asf/nifi/blob/2d3e5abf/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/StandardRemoteGroupPort.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/StandardRemoteGroupPort.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/StandardRemoteGroupPort.java
index 1be5b42..606c32a 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/StandardRemoteGroupPort.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/StandardRemoteGroupPort.java
@@ -90,6 +90,7 @@ public class StandardRemoteGroupPort extends RemoteGroupPort {
private final SSLContext sslContext;
private final TransferDirection transferDirection;
private final NiFiProperties nifiProperties;
+ private final String targetId;
private final AtomicReference<SiteToSiteClient> clientRef = new AtomicReference<>();
@@ -97,7 +98,7 @@ public class StandardRemoteGroupPort extends RemoteGroupPort {
return clientRef.get();
}
- public StandardRemoteGroupPort(final String id, final String name, final ProcessGroup processGroup, final RemoteProcessGroup remoteGroup,
+ public StandardRemoteGroupPort(final String id, final String targetId, final String name, final ProcessGroup processGroup, final RemoteProcessGroup remoteGroup,
final TransferDirection direction, final ConnectableType type, final SSLContext sslContext, final ProcessScheduler scheduler,
final NiFiProperties nifiProperties) {
// remote group port id needs to be unique but cannot just be the id of the port
@@ -105,6 +106,7 @@ public class StandardRemoteGroupPort extends RemoteGroupPort {
// instance more than once.
super(id, name, processGroup, type, scheduler);
+ this.targetId = targetId;
this.remoteGroup = remoteGroup;
this.transferDirection = direction;
this.sslContext = sslContext;
@@ -112,6 +114,11 @@ public class StandardRemoteGroupPort extends RemoteGroupPort {
setScheduldingPeriod(MINIMUM_SCHEDULING_NANOS + " nanos");
}
+ @Override
+ public String getTargetIdentifier() {
+ return targetId == null ? getIdentifier() : targetId;
+ }
+
private static File getPeerPersistenceFile(final String portId, final NiFiProperties nifiProperties) {
final File stateDir = nifiProperties.getPersistentStateDirectory();
return new File(stateDir, portId + ".peers");
@@ -164,7 +171,7 @@ public class StandardRemoteGroupPort extends RemoteGroupPort {
final SiteToSiteClient.Builder clientBuilder = new SiteToSiteClient.Builder()
.urls(SiteToSiteRestApiClient.parseClusterUrls(remoteGroup.getTargetUris()))
- .portIdentifier(getIdentifier())
+ .portIdentifier(getTargetIdentifier())
.sslContext(sslContext)
.useCompression(isUseCompression())
.eventReporter(remoteGroup.getEventReporter())
http://git-wip-us.apache.org/repos/asf/nifi/blob/2d3e5abf/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestStandardRemoteGroupPort.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestStandardRemoteGroupPort.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestStandardRemoteGroupPort.java
index f677b88..1b41943 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestStandardRemoteGroupPort.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestStandardRemoteGroupPort.java
@@ -120,7 +120,7 @@ public class TestStandardRemoteGroupPort {
break;
}
- port = spy(new StandardRemoteGroupPort(ID, NAME,
+ port = spy(new StandardRemoteGroupPort(ID, ID, NAME,
processGroup, remoteGroup, direction, connectableType, null, scheduler, NiFiProperties.createBasicNiFiProperties(null, null)));
doReturn(true).when(remoteGroup).isTransmitting();
http://git-wip-us.apache.org/repos/asf/nifi/blob/2d3e5abf/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
index f995f6b..37e146c 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
@@ -1502,6 +1502,7 @@ public final class DtoFactory {
final RemoteProcessGroupPortDTO dto = new RemoteProcessGroupPortDTO();
dto.setId(port.getIdentifier());
+ dto.setTargetId(port.getTargetIdentifier());
dto.setName(port.getName());
dto.setComments(port.getComments());
dto.setTransmitting(port.isRunning());
@@ -3168,6 +3169,7 @@ public final class DtoFactory {
public RemoteProcessGroupPortDTO copy(final RemoteProcessGroupPortDTO original) {
final RemoteProcessGroupPortDTO copy = new RemoteProcessGroupPortDTO();
copy.setId(original.getId());
+ copy.setTargetId(original.getTargetId());
copy.setGroupId(original.getGroupId());
copy.setName(original.getName());
copy.setComments(original.getComments());
http://git-wip-us.apache.org/repos/asf/nifi/blob/2d3e5abf/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/SnippetUtils.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/SnippetUtils.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/SnippetUtils.java
index db7a5b0..4469ea1 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/SnippetUtils.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/SnippetUtils.java
@@ -658,13 +658,24 @@ public final class SnippetUtils {
if (contents != null && contents.getInputPorts() != null) {
for (final RemoteProcessGroupPortDTO remotePort : contents.getInputPorts()) {
remotePort.setGroupId(cp.getId());
- connectableMap.put(remoteGroupDTO.getId() + "-" + remotePort.getId(), dtoFactory.createConnectableDto(remotePort, ConnectableType.REMOTE_INPUT_PORT));
+ final String originalId = remotePort.getId();
+ if (remotePort.getTargetId() == null) {
+ remotePort.setTargetId(originalId);
+ }
+ remotePort.setId(generateId(remotePort.getId(), idGenerationSeed, isCopy));
+
+ connectableMap.put(remoteGroupDTO.getId() + "-" + originalId, dtoFactory.createConnectableDto(remotePort, ConnectableType.REMOTE_INPUT_PORT));
}
}
if (contents != null && contents.getOutputPorts() != null) {
for (final RemoteProcessGroupPortDTO remotePort : contents.getOutputPorts()) {
remotePort.setGroupId(cp.getId());
- connectableMap.put(remoteGroupDTO.getId() + "-" + remotePort.getId(), dtoFactory.createConnectableDto(remotePort, ConnectableType.REMOTE_OUTPUT_PORT));
+ final String originalId = remotePort.getId();
+ if (remotePort.getTargetId() == null) {
+ remotePort.setTargetId(originalId);
+ }
+ remotePort.setId(generateId(remotePort.getId(), idGenerationSeed, isCopy));
+ connectableMap.put(remoteGroupDTO.getId() + "-" + originalId, dtoFactory.createConnectableDto(remotePort, ConnectableType.REMOTE_OUTPUT_PORT));
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/2d3e5abf/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/audit/TestRemoteProcessGroupAuditor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/audit/TestRemoteProcessGroupAuditor.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/audit/TestRemoteProcessGroupAuditor.java
index 68be1cd..fb9e43c 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/audit/TestRemoteProcessGroupAuditor.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/audit/TestRemoteProcessGroupAuditor.java
@@ -415,6 +415,7 @@ public class TestRemoteProcessGroupAuditor {
final ProceedingJoinPoint joinPoint = mock(ProceedingJoinPoint.class);
final String remoteProcessGroupId = "remote-process-group-id";
inputRPGPortDTO.setId(remoteProcessGroupId);
+ inputRPGPortDTO.setTargetId(remoteProcessGroupId);
final String targetUrl = "http://localhost:8080/nifi";
when(existingRPG.getIdentifier()).thenReturn(remoteProcessGroupId);
http://git-wip-us.apache.org/repos/asf/nifi/blob/2d3e5abf/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/dao/impl/TestStandardRemoteProcessGroupDAO.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/dao/impl/TestStandardRemoteProcessGroupDAO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/dao/impl/TestStandardRemoteProcessGroupDAO.java
index f68a115..d8840ec 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/dao/impl/TestStandardRemoteProcessGroupDAO.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/dao/impl/TestStandardRemoteProcessGroupDAO.java
@@ -79,6 +79,7 @@ public class TestStandardRemoteProcessGroupDAO {
final RemoteProcessGroupPortDTO dto = new RemoteProcessGroupPortDTO();
dto.setGroupId(remoteProcessGroupId);
dto.setId(remoteProcessGroupInputPortId);
+ dto.setTargetId(remoteProcessGroupInputPortId);
final BatchSettingsDTO batchSettings = new BatchSettingsDTO();
dto.setBatchSettings(batchSettings);