You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by al...@apache.org on 2017/03/13 14:16:43 UTC
[1/2] nifi-minifi git commit: MINIFI-176 - RPG Output Port support
Repository: nifi-minifi
Updated Branches:
refs/heads/master 7b57cc816 -> 52998f234
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteInputPortSchemaFunction.java
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteInputPortSchemaFunction.java b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteInputPortSchemaFunction.java
deleted file mode 100644
index 7fe16d0..0000000
--- a/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteInputPortSchemaFunction.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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.minifi.toolkit.configuration.dto;
-
-import org.apache.nifi.minifi.commons.schema.RemoteInputPortSchema;
-import org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys;
-import org.apache.nifi.web.api.dto.RemoteProcessGroupPortDTO;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.function.Function;
-
-import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.ID_KEY;
-import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.NAME_KEY;
-
-public class RemoteInputPortSchemaFunction implements Function<RemoteProcessGroupPortDTO, RemoteInputPortSchema> {
- @Override
- public RemoteInputPortSchema apply(RemoteProcessGroupPortDTO remoteProcessGroupPortDTO) {
- Map<String, Object> map = new HashMap<>();
- map.put(ID_KEY, remoteProcessGroupPortDTO.getId());
- map.put(NAME_KEY, remoteProcessGroupPortDTO.getName());
-
- map.put(CommonPropertyKeys.COMMENT_KEY, remoteProcessGroupPortDTO.getComments());
- map.put(CommonPropertyKeys.MAX_CONCURRENT_TASKS_KEY, remoteProcessGroupPortDTO.getConcurrentlySchedulableTaskCount());
- map.put(CommonPropertyKeys.USE_COMPRESSION_KEY, remoteProcessGroupPortDTO.getUseCompression());
- return new RemoteInputPortSchema(map);
- }
-}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemotePortSchemaFunction.java
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemotePortSchemaFunction.java b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemotePortSchemaFunction.java
new file mode 100644
index 0000000..172365c
--- /dev/null
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemotePortSchemaFunction.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.minifi.toolkit.configuration.dto;
+
+import org.apache.nifi.minifi.commons.schema.RemotePortSchema;
+import org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys;
+import org.apache.nifi.web.api.dto.RemoteProcessGroupPortDTO;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Function;
+
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.ID_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.NAME_KEY;
+
+public class RemotePortSchemaFunction implements Function<RemoteProcessGroupPortDTO, RemotePortSchema> {
+ @Override
+ public RemotePortSchema apply(RemoteProcessGroupPortDTO remoteProcessGroupPortDTO) {
+ Map<String, Object> map = new HashMap<>();
+ map.put(ID_KEY, remoteProcessGroupPortDTO.getId());
+ map.put(NAME_KEY, remoteProcessGroupPortDTO.getName());
+
+ map.put(CommonPropertyKeys.COMMENT_KEY, remoteProcessGroupPortDTO.getComments());
+ map.put(CommonPropertyKeys.MAX_CONCURRENT_TASKS_KEY, remoteProcessGroupPortDTO.getConcurrentlySchedulableTaskCount());
+ map.put(CommonPropertyKeys.USE_COMPRESSION_KEY, remoteProcessGroupPortDTO.getUseCompression());
+ return new RemotePortSchema(map);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteProcessGroupSchemaFunction.java
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteProcessGroupSchemaFunction.java b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteProcessGroupSchemaFunction.java
index e7d5fa0..444088b 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteProcessGroupSchemaFunction.java
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteProcessGroupSchemaFunction.java
@@ -17,7 +17,7 @@
package org.apache.nifi.minifi.toolkit.configuration.dto;
-import org.apache.nifi.minifi.commons.schema.RemoteInputPortSchema;
+import org.apache.nifi.minifi.commons.schema.RemotePortSchema;
import org.apache.nifi.minifi.commons.schema.RemoteProcessGroupSchema;
import org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys;
import org.apache.nifi.web.api.dto.RemoteProcessGroupContentsDTO;
@@ -31,10 +31,10 @@ import java.util.function.Function;
import java.util.stream.Collectors;
public class RemoteProcessGroupSchemaFunction implements Function<RemoteProcessGroupDTO, RemoteProcessGroupSchema> {
- private final RemoteInputPortSchemaFunction remoteInputPortSchemaFunction;
+ private final RemotePortSchemaFunction remotePortSchemaFunction;
- public RemoteProcessGroupSchemaFunction(RemoteInputPortSchemaFunction remoteInputPortSchemaFunction) {
- this.remoteInputPortSchemaFunction = remoteInputPortSchemaFunction;
+ public RemoteProcessGroupSchemaFunction(RemotePortSchemaFunction remotePortSchemaFunction) {
+ this.remotePortSchemaFunction = remotePortSchemaFunction;
}
@Override
@@ -49,8 +49,15 @@ public class RemoteProcessGroupSchemaFunction implements Function<RemoteProcessG
Set<RemoteProcessGroupPortDTO> inputPorts = contents.getInputPorts();
if (inputPorts != null) {
map.put(CommonPropertyKeys.INPUT_PORTS_KEY, inputPorts.stream()
- .map(remoteInputPortSchemaFunction)
- .map(RemoteInputPortSchema::toMap)
+ .map(remotePortSchemaFunction)
+ .map(RemotePortSchema::toMap)
+ .collect(Collectors.toList()));
+ }
+ Set<RemoteProcessGroupPortDTO> outputPorts = contents.getOutputPorts();
+ if (outputPorts != null) {
+ map.put(CommonPropertyKeys.OUTPUT_PORTS_KEY, outputPorts.stream()
+ .map(remotePortSchemaFunction)
+ .map(RemotePortSchema::toMap)
.collect(Collectors.toList()));
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/ConfigMainTest.java
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/ConfigMainTest.java b/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/ConfigMainTest.java
index ae5936c..4d8bc66 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/ConfigMainTest.java
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/ConfigMainTest.java
@@ -22,7 +22,7 @@ import org.apache.nifi.controller.repository.io.LimitedInputStream;
import org.apache.nifi.minifi.commons.schema.ConfigSchema;
import org.apache.nifi.minifi.commons.schema.ConnectionSchema;
import org.apache.nifi.minifi.commons.schema.ProcessorSchema;
-import org.apache.nifi.minifi.commons.schema.RemoteInputPortSchema;
+import org.apache.nifi.minifi.commons.schema.RemotePortSchema;
import org.apache.nifi.minifi.commons.schema.RemoteProcessGroupSchema;
import org.apache.nifi.minifi.commons.schema.common.ConvertableSchema;
import org.apache.nifi.minifi.commons.schema.serialization.SchemaLoader;
@@ -208,6 +208,11 @@ public class ConfigMainTest {
}
@Test
+ public void testTransformRoundTripSimpleRPGToLogAttributes() throws IOException, JAXBException, SchemaLoaderException {
+ transformRoundTrip("SimpleRPGToLogAttributes");
+ }
+
+ @Test
public void testTransformRoundTripNestedControllerServices() throws IOException, JAXBException, SchemaLoaderException {
transformRoundTrip("NestedControllerServices");
}
@@ -348,7 +353,7 @@ public class ConfigMainTest {
v1RPG.setId(currentRPG.getId());
}
- configSchemaUpgradedFromV1.getProcessGroupSchema().getRemoteProcessGroups().stream().flatMap(g -> g.getInputPorts().stream()).map(RemoteInputPortSchema::getId).sequential()
+ configSchemaUpgradedFromV1.getProcessGroupSchema().getRemoteProcessGroups().stream().flatMap(g -> g.getInputPorts().stream()).map(RemotePortSchema::getId).sequential()
.forEach(id -> v1IdToCurrentIdMap.put(id, id));
List<ConnectionSchema> currentConnections = configSchemaFromCurrent.getProcessGroupSchema().getConnections();
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteInputPortSchemaTest.java
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteInputPortSchemaTest.java b/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteInputPortSchemaTest.java
index 94b5beb..61e4b5d 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteInputPortSchemaTest.java
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteInputPortSchemaTest.java
@@ -17,7 +17,7 @@
package org.apache.nifi.minifi.toolkit.configuration.dto;
-import org.apache.nifi.minifi.commons.schema.RemoteInputPortSchema;
+import org.apache.nifi.minifi.commons.schema.RemotePortSchema;
import org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys;
import org.apache.nifi.web.api.dto.RemoteProcessGroupPortDTO;
import org.junit.Before;
@@ -29,7 +29,7 @@ import java.util.UUID;
import static org.junit.Assert.assertEquals;
-public class RemoteInputPortSchemaTest extends BaseSchemaTester<RemoteInputPortSchema, RemoteProcessGroupPortDTO> {
+public class RemoteInputPortSchemaTest extends BaseSchemaTester<RemotePortSchema, RemoteProcessGroupPortDTO> {
private String testId = UUID.nameUUIDFromBytes("testId".getBytes(StandardCharsets.UTF_8)).toString();
private String testName = "testName";
@@ -38,7 +38,7 @@ public class RemoteInputPortSchemaTest extends BaseSchemaTester<RemoteInputPortS
private boolean testUseCompression = false;
public RemoteInputPortSchemaTest() {
- super(new RemoteInputPortSchemaFunction(), RemoteInputPortSchema::new);
+ super(new RemotePortSchemaFunction(), RemotePortSchema::new);
}
@Before
@@ -94,7 +94,7 @@ public class RemoteInputPortSchemaTest extends BaseSchemaTester<RemoteInputPortS
}
@Override
- public void assertSchemaEquals(RemoteInputPortSchema one, RemoteInputPortSchema two) {
+ public void assertSchemaEquals(RemotePortSchema one, RemotePortSchema two) {
assertEquals(one.getId(), two.getId());
assertEquals(one.getName(), two.getName());
assertEquals(one.getComment(), two.getComment());
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteProcessGroupSchemaTest.java
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteProcessGroupSchemaTest.java b/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteProcessGroupSchemaTest.java
index e4d5b46..a07ccea 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteProcessGroupSchemaTest.java
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/dto/RemoteProcessGroupSchemaTest.java
@@ -17,7 +17,7 @@
package org.apache.nifi.minifi.toolkit.configuration.dto;
-import org.apache.nifi.minifi.commons.schema.RemoteInputPortSchema;
+import org.apache.nifi.minifi.commons.schema.RemotePortSchema;
import org.apache.nifi.minifi.commons.schema.RemoteProcessGroupSchema;
import org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys;
import org.apache.nifi.web.api.dto.RemoteProcessGroupContentsDTO;
@@ -47,7 +47,7 @@ public class RemoteProcessGroupSchemaTest extends BaseSchemaTester<RemoteProcess
private String transportProtocol = "HTTP";
public RemoteProcessGroupSchemaTest() {
- super(new RemoteProcessGroupSchemaFunction(new RemoteInputPortSchemaFunction()), RemoteProcessGroupSchema::new);
+ super(new RemoteProcessGroupSchemaFunction(new RemotePortSchemaFunction()), RemoteProcessGroupSchema::new);
remoteInputPortSchemaTest = new RemoteInputPortSchemaTest();
}
@@ -142,8 +142,8 @@ public class RemoteProcessGroupSchemaTest extends BaseSchemaTester<RemoteProcess
assertEquals(one.getName(), two.getName());
assertEquals(one.getUrl(), two.getUrl());
- List<RemoteInputPortSchema> oneInputPorts = one.getInputPorts();
- List<RemoteInputPortSchema> twoInputPorts = two.getInputPorts();
+ List<RemotePortSchema> oneInputPorts = one.getInputPorts();
+ List<RemotePortSchema> twoInputPorts = two.getInputPorts();
if (oneInputPorts == null) {
assertNull(twoInputPorts);
} else {
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/InvokeHttpMiNiFiTemplateTest.yml
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/InvokeHttpMiNiFiTemplateTest.yml b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/InvokeHttpMiNiFiTemplateTest.yml
index d7def45..4287987 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/InvokeHttpMiNiFiTemplateTest.yml
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/InvokeHttpMiNiFiTemplateTest.yml
@@ -300,4 +300,5 @@ Remote Process Groups:
comment: ''
max concurrent tasks: 1
use compression: false
+ Output Ports: []
NiFi Properties Overrides: {}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ProcessGroupsAndRemoteProcessGroups.yml
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ProcessGroupsAndRemoteProcessGroups.yml b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ProcessGroupsAndRemoteProcessGroups.yml
index 8203cac..823e56f 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ProcessGroupsAndRemoteProcessGroups.yml
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ProcessGroupsAndRemoteProcessGroups.yml
@@ -194,6 +194,7 @@ Process Groups:
comment: ''
max concurrent tasks: 1
use compression: false
+ Output Ports: []
Input Ports:
- id: 2078c936-0158-1000-0000-000000000000
name: input
@@ -258,6 +259,7 @@ Process Groups:
comment: ''
max concurrent tasks: 1
use compression: false
+ Output Ports: []
Input Ports: []
Output Ports: []
Funnels: []
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/SimpleRPGToLogAttributes.xml
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/SimpleRPGToLogAttributes.xml b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/SimpleRPGToLogAttributes.xml
new file mode 100644
index 0000000..2871670
--- /dev/null
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/SimpleRPGToLogAttributes.xml
@@ -0,0 +1,149 @@
+<?xml version="1.0" ?>
+<!--
+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.
+ -->
+<template encoding-version="1.0">
+ <description></description>
+ <groupId>6b94b964-015a-1000-17d3-c4f5a6c3dc97</groupId>
+ <name>SimpleRPGToLogAttributes</name>
+ <snippet>
+ <connections>
+ <id>6b97306b-015a-1000-0000-000000000000</id>
+ <parentGroupId>6b94b964-015a-1000-0000-000000000000</parentGroupId>
+ <backPressureDataSizeThreshold>1 GB</backPressureDataSizeThreshold>
+ <backPressureObjectThreshold>10000</backPressureObjectThreshold>
+ <destination>
+ <groupId>6b94b964-015a-1000-0000-000000000000</groupId>
+ <id>6b97126a-015a-1000-0000-000000000000</id>
+ <type>PROCESSOR</type>
+ </destination>
+ <flowFileExpiration>0 sec</flowFileExpiration>
+ <labelIndex>1</labelIndex>
+ <name></name>
+ <source>
+ <groupId>6b95d112-015a-1000-0000-000000000000</groupId>
+ <id>6b965db7-015a-1000-4c43-9d20931a1b9c</id>
+ <type>REMOTE_OUTPUT_PORT</type>
+ </source>
+ <zIndex>0</zIndex>
+ </connections>
+ <processors>
+ <id>6b97126a-015a-1000-0000-000000000000</id>
+ <parentGroupId>6b94b964-015a-1000-0000-000000000000</parentGroupId>
+ <position>
+ <x>20.0</x>
+ <y>289.0</y>
+ </position>
+ <config>
+ <bulletinLevel>WARN</bulletinLevel>
+ <comments></comments>
+ <concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
+ <descriptors>
+ <entry>
+ <key>Log Level</key>
+ <value>
+ <name>Log Level</name>
+ </value>
+ </entry>
+ <entry>
+ <key>Log Payload</key>
+ <value>
+ <name>Log Payload</name>
+ </value>
+ </entry>
+ <entry>
+ <key>Attributes to Log</key>
+ <value>
+ <name>Attributes to Log</name>
+ </value>
+ </entry>
+ <entry>
+ <key>Attributes to Ignore</key>
+ <value>
+ <name>Attributes to Ignore</name>
+ </value>
+ </entry>
+ <entry>
+ <key>Log prefix</key>
+ <value>
+ <name>Log prefix</name>
+ </value>
+ </entry>
+ </descriptors>
+ <executionNode>ALL</executionNode>
+ <lossTolerant>false</lossTolerant>
+ <penaltyDuration>30 sec</penaltyDuration>
+ <properties>
+ <entry>
+ <key>Log Level</key>
+ <value>info</value>
+ </entry>
+ <entry>
+ <key>Log Payload</key>
+ <value>false</value>
+ </entry>
+ <entry>
+ <key>Attributes to Log</key>
+ </entry>
+ <entry>
+ <key>Attributes to Ignore</key>
+ </entry>
+ <entry>
+ <key>Log prefix</key>
+ </entry>
+ </properties>
+ <runDurationMillis>0</runDurationMillis>
+ <schedulingPeriod>0 sec</schedulingPeriod>
+ <schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
+ <yieldDuration>1 sec</yieldDuration>
+ </config>
+ <name>LogAttribute</name>
+ <relationships>
+ <autoTerminate>true</autoTerminate>
+ <name>success</name>
+ </relationships>
+ <style></style>
+ <type>org.apache.nifi.processors.standard.LogAttribute</type>
+ </processors>
+ <remoteProcessGroups>
+ <id>6b95d112-015a-1000-0000-000000000000</id>
+ <parentGroupId>6b94b964-015a-1000-0000-000000000000</parentGroupId>
+ <position>
+ <x>0.0</x>
+ <y>0.0</y>
+ </position>
+ <communicationsTimeout>30 sec</communicationsTimeout>
+ <contents>
+ <outputPorts>
+ <concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
+ <connected>true</connected>
+ <exists>true</exists>
+ <id>6b965db7-015a-1000-4c43-9d20931a1b9c</id>
+ <name>Output</name>
+ <targetRunning>true</targetRunning>
+ <transmitting>true</transmitting>
+ <useCompression>false</useCompression>
+ </outputPorts>
+ </contents>
+ <proxyHost></proxyHost>
+ <proxyUser></proxyUser>
+ <targetUri>http://localhost:8080/nifi</targetUri>
+ <transportProtocol>HTTP</transportProtocol>
+ <yieldDuration>10 sec</yieldDuration>
+ </remoteProcessGroups>
+ </snippet>
+ <timestamp>02/23/2017 10:31:38 EST</timestamp>
+</template>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/SimpleRPGToLogAttributes.yml
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/SimpleRPGToLogAttributes.yml b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/SimpleRPGToLogAttributes.yml
new file mode 100644
index 0000000..e267792
--- /dev/null
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/SimpleRPGToLogAttributes.yml
@@ -0,0 +1,110 @@
+# 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.
+
+MiNiFi Config Version: 3
+Flow Controller:
+ name: SimpleRPGToLogAttributes
+ comment: ''
+Core Properties:
+ flow controller graceful shutdown period: 10 sec
+ flow service write delay interval: 500 ms
+ administrative yield duration: 30 sec
+ bored yield duration: 10 millis
+ max concurrent threads: 1
+FlowFile Repository:
+ partitions: 256
+ checkpoint interval: 2 mins
+ always sync: false
+ Swap:
+ threshold: 20000
+ in period: 5 sec
+ in threads: 1
+ out period: 5 sec
+ out threads: 4
+Content Repository:
+ content claim max appendable size: 10 MB
+ content claim max flow files: 100
+ always sync: false
+Provenance Repository:
+ provenance rollover time: 1 min
+Component Status Repository:
+ buffer size: 1440
+ snapshot frequency: 1 min
+Security Properties:
+ keystore: ''
+ keystore type: ''
+ keystore password: ''
+ key password: ''
+ truststore: ''
+ truststore type: ''
+ truststore password: ''
+ ssl protocol: ''
+ Sensitive Props:
+ key:
+ algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
+ provider: BC
+Processors:
+- id: 6b97126a-015a-1000-0000-000000000000
+ name: LogAttribute
+ class: org.apache.nifi.processors.standard.LogAttribute
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 0 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ - success
+ Properties:
+ Attributes to Ignore:
+ Attributes to Log:
+ Log Level: info
+ Log Payload: 'false'
+ Log prefix:
+Controller Services: []
+Process Groups: []
+Input Ports: []
+Output Ports: []
+Funnels: []
+Connections:
+- id: 6b97306b-015a-1000-0000-000000000000
+ name: 6b965db7-015a-1000-4c43-9d20931a1b9c//LogAttribute
+ source id: 6b965db7-015a-1000-4c43-9d20931a1b9c
+ source relationship names: []
+ destination id: 6b97126a-015a-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+Remote Process Groups:
+- id: 6b95d112-015a-1000-0000-000000000000
+ name: ''
+ url: http://localhost:8080/nifi
+ comment: ''
+ timeout: 30 sec
+ yield period: 10 sec
+ transport protocol: HTTP
+ proxy host: ''
+ proxy port: ''
+ proxy user: ''
+ proxy password: ''
+ Input Ports: []
+ Output Ports:
+ - id: 6b965db7-015a-1000-4c43-9d20931a1b9c
+ name: Output
+ comment: ''
+ max concurrent tasks: 1
+ use compression: false
+NiFi Properties Overrides: {}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/SimpleTailFileToRPG.yml
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/SimpleTailFileToRPG.yml b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/SimpleTailFileToRPG.yml
index 09859b7..a9108b9 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/SimpleTailFileToRPG.yml
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/SimpleTailFileToRPG.yml
@@ -116,4 +116,5 @@ Remote Process Groups:
comment: ''
max concurrent tasks: 1
use compression: false
+ Output Ports: []
NiFi Properties Overrides: {}
[2/2] nifi-minifi git commit: MINIFI-176 - RPG Output Port support
Posted by al...@apache.org.
MINIFI-176 - RPG Output Port support
This closes #73.
Signed-off-by: Aldrin Piri <al...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/nifi-minifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi-minifi/commit/52998f23
Tree: http://git-wip-us.apache.org/repos/asf/nifi-minifi/tree/52998f23
Diff: http://git-wip-us.apache.org/repos/asf/nifi-minifi/diff/52998f23
Branch: refs/heads/master
Commit: 52998f2343e864000c959002b7ce7cd8c95ed8f5
Parents: 7b57cc8
Author: Bryan Rosander <br...@apache.org>
Authored: Thu Feb 23 11:52:02 2017 -0500
Committer: Aldrin Piri <al...@apache.org>
Committed: Mon Mar 13 10:15:19 2017 -0400
----------------------------------------------------------------------
.../bootstrap/util/ConfigTransformer.java | 48 +++---
.../bootstrap/util/ParentGroupIdResolver.java | 24 ++-
.../bootstrap/util/ConfigTransformerTest.java | 17 ++-
.../util/ParentGroupIdResolverTest.java | 40 ++++-
.../test/resources/SimpleRPGToLogAttributes.yml | 110 ++++++++++++++
.../minifi/commons/schema/ConfigSchema.java | 10 +-
.../commons/schema/ProcessGroupSchema.java | 2 +
.../commons/schema/RemoteInputPortSchema.java | 66 --------
.../minifi/commons/schema/RemotePortSchema.java | 66 ++++++++
.../schema/RemoteProcessGroupSchema.java | 26 +++-
.../commons/schema/v1/ConfigSchemaV1.java | 6 +-
.../schema/v1/RemoteProcessGroupSchemaV1.java | 10 +-
.../commons/schema/v2/ConfigSchemaV2.java | 4 +-
.../schema/v2/RemoteProcessGroupSchemaV2.java | 8 +-
.../commons/status/rpg/InputPortStatus.java | 81 ----------
.../minifi/commons/status/rpg/PortStatus.java | 81 ++++++++++
.../rpg/RemoteProcessGroupStatusBean.java | 18 ++-
.../status/util/StatusReportPopulator.java | 14 +-
.../minifi/commons/status/TestStatusReport.java | 4 +-
.../src/main/markdown/System_Admin_Guide.md | 16 +-
.../nifi/minifi/status/StatusRequestParser.java | 35 +++--
.../minifi/status/TestStatusConfigReporter.java | 38 ++++-
.../configuration/dto/ConfigSchemaFunction.java | 2 +-
.../dto/RemoteInputPortSchemaFunction.java | 43 ------
.../dto/RemotePortSchemaFunction.java | 43 ++++++
.../dto/RemoteProcessGroupSchemaFunction.java | 19 ++-
.../toolkit/configuration/ConfigMainTest.java | 9 +-
.../dto/RemoteInputPortSchemaTest.java | 8 +-
.../dto/RemoteProcessGroupSchemaTest.java | 8 +-
.../resources/InvokeHttpMiNiFiTemplateTest.yml | 1 +
.../ProcessGroupsAndRemoteProcessGroups.yml | 2 +
.../test/resources/SimpleRPGToLogAttributes.xml | 149 +++++++++++++++++++
.../test/resources/SimpleRPGToLogAttributes.yml | 110 ++++++++++++++
.../src/test/resources/SimpleTailFileToRPG.yml | 1 +
34 files changed, 814 insertions(+), 305 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformer.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformer.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformer.java
index 836214d..64422c4 100644
--- a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformer.java
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformer.java
@@ -36,7 +36,7 @@ import org.apache.nifi.minifi.commons.schema.ProcessGroupSchema;
import org.apache.nifi.minifi.commons.schema.ProcessorSchema;
import org.apache.nifi.minifi.commons.schema.ProvenanceReportingSchema;
import org.apache.nifi.minifi.commons.schema.ProvenanceRepositorySchema;
-import org.apache.nifi.minifi.commons.schema.RemoteInputPortSchema;
+import org.apache.nifi.minifi.commons.schema.RemotePortSchema;
import org.apache.nifi.minifi.commons.schema.RemoteProcessGroupSchema;
import org.apache.nifi.minifi.commons.schema.SecurityPropertiesSchema;
import org.apache.nifi.minifi.commons.schema.SensitivePropsSchema;
@@ -566,9 +566,14 @@ public final class ConfigTransformer {
addTextElement(element, "proxyPassword", remoteProcessGroupProperties.getProxyPassword());
}
- List<RemoteInputPortSchema> remoteInputPorts = remoteProcessGroupProperties.getInputPorts();
- for (RemoteInputPortSchema remoteInputPortSchema : remoteInputPorts) {
- addRemoteGroupPort(element, remoteInputPortSchema);
+ List<RemotePortSchema> remoteInputPorts = remoteProcessGroupProperties.getInputPorts();
+ for (RemotePortSchema remoteInputPortSchema : remoteInputPorts) {
+ addRemoteGroupPort(element, remoteInputPortSchema, "inputPort");
+ }
+
+ List<RemotePortSchema> remoteOutputPorts = remoteProcessGroupProperties.getOutputPorts();
+ for (RemotePortSchema remoteOutputPortSchema : remoteOutputPorts) {
+ addRemoteGroupPort(element, remoteOutputPortSchema, "outputPort");
}
parentElement.appendChild(element);
@@ -577,10 +582,10 @@ public final class ConfigTransformer {
}
}
- protected static void addRemoteGroupPort(final Element parentElement, RemoteInputPortSchema inputPort) throws ConfigurationChangeException {
+ protected static void addRemoteGroupPort(final Element parentElement, RemotePortSchema inputPort, String tagName) throws ConfigurationChangeException {
try {
final Document doc = parentElement.getOwnerDocument();
- final Element element = doc.createElement("inputPort");
+ final Element element = doc.createElement(tagName);
parentElement.appendChild(element);
addTextElement(element, "id", inputPort.getId());
addTextElement(element, "name", inputPort.getName());
@@ -640,29 +645,22 @@ public final class ConfigTransformer {
String groupIdTag = sourceOrDestination + "GroupId";
String typeTag = sourceOrDestination + "Type";
- String parentId = parentGroupIdResolver.getRemoteInputPortParentId(id);
+ String parentId;
String type;
- if (parentId != null) {
+ if ((parentId = parentGroupIdResolver.getRemoteInputPortParentId(id)) != null) {
type = "REMOTE_INPUT_PORT";
+ } else if ((parentId = parentGroupIdResolver.getRemoteOutputPortParentId(id)) != null) {
+ type = "REMOTE_OUTPUT_PORT";
+ } else if ((parentId = parentGroupIdResolver.getInputPortParentId(id)) != null) {
+ type = "INPUT_PORT";
+ } else if ((parentId = parentGroupIdResolver.getOutputPortParentId(id)) != null) {
+ type = "OUTPUT_PORT";
+ } else if ((parentId = parentGroupIdResolver.getFunnelParentId(id)) != null) {
+ type = "FUNNEL";
} else {
- parentId = parentGroupIdResolver.getInputPortParentId(id);
- if (parentId != null) {
- type = "INPUT_PORT";
- } else {
- parentId = parentGroupIdResolver.getOutputPortParentId(id);
- if (parentId != null) {
- type = "OUTPUT_PORT";
- } else {
- parentId = parentGroupIdResolver.getFunnelParentId(id);
- if (parentId != null) {
- type = "FUNNEL";
- } else {
- parentId = parentGroupIdResolver.getProcessorParentId(id);
- type = "PROCESSOR";
- }
- }
- }
+ parentId = parentGroupIdResolver.getProcessorParentId(id);
+ type = "PROCESSOR";
}
addTextElement(element, idTag, id);
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolver.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolver.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolver.java
index 3d266ad..d0a7d29 100644
--- a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolver.java
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolver.java
@@ -20,12 +20,13 @@
package org.apache.nifi.minifi.bootstrap.util;
import org.apache.nifi.minifi.commons.schema.ProcessGroupSchema;
-import org.apache.nifi.minifi.commons.schema.RemoteInputPortSchema;
+import org.apache.nifi.minifi.commons.schema.RemotePortSchema;
import org.apache.nifi.minifi.commons.schema.RemoteProcessGroupSchema;
import org.apache.nifi.minifi.commons.schema.common.BaseSchemaWithId;
import java.util.Collection;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.function.Function;
@@ -35,13 +36,15 @@ public class ParentGroupIdResolver {
private final Map<String, String> outputPortIdToParentIdMap;
private final Map<String, String> funnelIdToParentIdMap;
private final Map<String, String> remoteInputPortIdToParentIdMap;
+ private final Map<String, String> remoteOutputPortIdToParentIdMap;
public ParentGroupIdResolver(ProcessGroupSchema processGroupSchema) {
this.processorIdToParentIdMap = getParentIdMap(processGroupSchema, ProcessGroupSchema::getProcessors);
this.inputPortIdToParentIdMap = getParentIdMap(processGroupSchema, ProcessGroupSchema::getInputPortSchemas);
this.outputPortIdToParentIdMap = getParentIdMap(processGroupSchema, ProcessGroupSchema::getOutputPortSchemas);
this.funnelIdToParentIdMap = getParentIdMap(processGroupSchema, ProcessGroupSchema::getFunnels);
- this.remoteInputPortIdToParentIdMap = getRemoteInputPortParentIdMap(processGroupSchema);
+ this.remoteInputPortIdToParentIdMap = getRemotePortParentIdMap(processGroupSchema, RemoteProcessGroupSchema::getInputPorts);
+ this.remoteOutputPortIdToParentIdMap = getRemotePortParentIdMap(processGroupSchema, RemoteProcessGroupSchema::getOutputPorts);
}
protected static Map<String, String> getParentIdMap(ProcessGroupSchema processGroupSchema, Function<ProcessGroupSchema, Collection<? extends BaseSchemaWithId>> schemaAccessor) {
@@ -56,25 +59,30 @@ public class ParentGroupIdResolver {
processGroupSchema.getProcessGroupSchemas().forEach(p -> getParentIdMap(p, output, schemaAccessor));
}
- protected static Map<String, String> getRemoteInputPortParentIdMap(ProcessGroupSchema processGroupSchema) {
+ protected static Map<String, String> getRemotePortParentIdMap(ProcessGroupSchema processGroupSchema, Function<RemoteProcessGroupSchema, List<RemotePortSchema>> getPortsFunction) {
Map<String, String> result = new HashMap<>();
- getRemoteInputPortParentIdMap(processGroupSchema, result);
+ getRemotePortParentIdMap(processGroupSchema, result, getPortsFunction);
return result;
}
- protected static void getRemoteInputPortParentIdMap(ProcessGroupSchema processGroupSchema, Map<String, String> output) {
+ protected static void getRemotePortParentIdMap(ProcessGroupSchema processGroupSchema, Map<String, String> output, Function<RemoteProcessGroupSchema,
+ List<RemotePortSchema>> getPortsFunction) {
for (RemoteProcessGroupSchema remoteProcessGroupSchema : processGroupSchema.getRemoteProcessGroups()) {
- for (RemoteInputPortSchema remoteInputPortSchema : remoteProcessGroupSchema.getInputPorts()) {
- output.put(remoteInputPortSchema.getId(), remoteProcessGroupSchema.getId());
+ for (RemotePortSchema remotePortSchema : getPortsFunction.apply(remoteProcessGroupSchema)) {
+ output.put(remotePortSchema.getId(), remoteProcessGroupSchema.getId());
}
}
- processGroupSchema.getProcessGroupSchemas().forEach(p -> getRemoteInputPortParentIdMap(p, output));
+ processGroupSchema.getProcessGroupSchemas().forEach(p -> getRemotePortParentIdMap(p, output, getPortsFunction));
}
public String getRemoteInputPortParentId(String id) {
return remoteInputPortIdToParentIdMap.get(id);
}
+ public String getRemoteOutputPortParentId(String id) {
+ return remoteOutputPortIdToParentIdMap.get(id);
+ }
+
public String getInputPortParentId(String id) {
return inputPortIdToParentIdMap.get(id);
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformerTest.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformerTest.java b/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformerTest.java
index 38fc673..ecd9001 100644
--- a/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformerTest.java
+++ b/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformerTest.java
@@ -25,7 +25,7 @@ import org.apache.nifi.minifi.commons.schema.FunnelSchema;
import org.apache.nifi.minifi.commons.schema.PortSchema;
import org.apache.nifi.minifi.commons.schema.ProcessGroupSchema;
import org.apache.nifi.minifi.commons.schema.ProcessorSchema;
-import org.apache.nifi.minifi.commons.schema.RemoteInputPortSchema;
+import org.apache.nifi.minifi.commons.schema.RemotePortSchema;
import org.apache.nifi.minifi.commons.schema.RemoteProcessGroupSchema;
import org.apache.nifi.minifi.commons.schema.common.StringUtil;
import org.apache.nifi.minifi.commons.schema.exception.SchemaLoaderException;
@@ -140,6 +140,11 @@ public class ConfigTransformerTest {
}
@Test
+ public void testRpgOutputPort() throws Exception {
+ testConfigFileTransform("SimpleRPGToLogAttributes.yml");
+ }
+
+ @Test
public void testNifiPropertiesNoOverrides() throws IOException, ConfigurationChangeException, SchemaLoaderException {
Properties pre216Properties = new Properties();
try (InputStream pre216PropertiesStream = ConfigTransformerTest.class.getClassLoader().getResourceAsStream("MINIFI-216/nifi.properties.before")) {
@@ -290,11 +295,17 @@ public class ConfigTransformerTest {
NodeList inputPortElements = (NodeList) xPathFactory.newXPath().evaluate("inputPort", element, XPathConstants.NODESET);
assertEquals(remoteProcessingGroupSchema.getInputPorts().size(), inputPortElements.getLength());
for (int i = 0; i < inputPortElements.getLength(); i++) {
- testRemoteInputPort((Element) inputPortElements.item(i), remoteProcessingGroupSchema.getInputPorts().get(i));
+ testRemotePort((Element) inputPortElements.item(i), remoteProcessingGroupSchema.getInputPorts().get(i));
+ }
+
+ NodeList outputPortElements = (NodeList) xPathFactory.newXPath().evaluate("outputPort", element, XPathConstants.NODESET);
+ assertEquals(remoteProcessingGroupSchema.getOutputPorts().size(), outputPortElements.getLength());
+ for (int i = 0; i < outputPortElements.getLength(); i++) {
+ testRemotePort((Element) outputPortElements.item(i), remoteProcessingGroupSchema.getOutputPorts().get(i));
}
}
- private void testRemoteInputPort(Element element, RemoteInputPortSchema remoteInputPortSchema) throws XPathExpressionException {
+ private void testRemotePort(Element element, RemotePortSchema remoteInputPortSchema) throws XPathExpressionException {
assertEquals(remoteInputPortSchema.getId(), getText(element, "id"));
assertEquals(remoteInputPortSchema.getName(), getText(element, "name"));
assertEquals(remoteInputPortSchema.getComment(), getText(element, "comment"));
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolverTest.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolverTest.java b/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolverTest.java
index 2855c29..cdcd47b 100644
--- a/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolverTest.java
+++ b/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolverTest.java
@@ -53,6 +53,37 @@ public class ParentGroupIdResolverTest {
ParentGroupIdResolver parentGroupIdResolver = createParentGroupIdResolver(configLines);
assertEquals("rpgOneId", parentGroupIdResolver.getRemoteInputPortParentId("one"));
assertEquals("rpgTwoId", parentGroupIdResolver.getRemoteInputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getRemoteOutputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getRemoteOutputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getInputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getInputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getOutputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getOutputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getProcessorParentId("one"));
+ assertNull(parentGroupIdResolver.getProcessorParentId("two"));
+ assertNull(parentGroupIdResolver.getFunnelParentId("one"));
+ assertNull(parentGroupIdResolver.getFunnelParentId("two"));
+ }
+ @Test
+ public void testRemoteOutputPortParentId() throws IOException, SchemaLoaderException {
+ List<String> configLines = new ArrayList<>();
+ configLines.add("MiNiFi Config Version: 2");
+ configLines.add("Remote Process Groups:");
+ configLines.add("- name: rpgOne");
+ configLines.add(" id: rpgOneId");
+ configLines.add(" Output Ports:");
+ configLines.add(" - id: one");
+ configLines.add("Process Groups:");
+ configLines.add("- Remote Process Groups:");
+ configLines.add(" - name: rpgTwo");
+ configLines.add(" id: rpgTwoId");
+ configLines.add(" Output Ports:");
+ configLines.add(" - id: two");
+ ParentGroupIdResolver parentGroupIdResolver = createParentGroupIdResolver(configLines);
+ assertNull(parentGroupIdResolver.getRemoteInputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getRemoteInputPortParentId("two"));
+ assertEquals("rpgOneId", parentGroupIdResolver.getRemoteOutputPortParentId("one"));
+ assertEquals("rpgTwoId", parentGroupIdResolver.getRemoteOutputPortParentId("two"));
assertNull(parentGroupIdResolver.getInputPortParentId("one"));
assertNull(parentGroupIdResolver.getInputPortParentId("two"));
assertNull(parentGroupIdResolver.getOutputPortParentId("one"));
@@ -76,6 +107,8 @@ public class ParentGroupIdResolverTest {
ParentGroupIdResolver parentGroupIdResolver = createParentGroupIdResolver(configLines);
assertNull(parentGroupIdResolver.getRemoteInputPortParentId("one"));
assertNull(parentGroupIdResolver.getRemoteInputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getRemoteOutputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getRemoteOutputPortParentId("two"));
assertEquals(ConfigTransformer.ROOT_GROUP, parentGroupIdResolver.getInputPortParentId("one"));
assertEquals("pgTwo", parentGroupIdResolver.getInputPortParentId("two"));
assertNull(parentGroupIdResolver.getOutputPortParentId("one"));
@@ -99,6 +132,8 @@ public class ParentGroupIdResolverTest {
ParentGroupIdResolver parentGroupIdResolver = createParentGroupIdResolver(configLines);
assertNull(parentGroupIdResolver.getRemoteInputPortParentId("one"));
assertNull(parentGroupIdResolver.getRemoteInputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getRemoteOutputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getRemoteOutputPortParentId("two"));
assertNull(parentGroupIdResolver.getInputPortParentId("one"));
assertNull(parentGroupIdResolver.getInputPortParentId("two"));
assertEquals(ConfigTransformer.ROOT_GROUP, parentGroupIdResolver.getOutputPortParentId("one"));
@@ -122,6 +157,8 @@ public class ParentGroupIdResolverTest {
ParentGroupIdResolver parentGroupIdResolver = createParentGroupIdResolver(configLines);
assertNull(parentGroupIdResolver.getRemoteInputPortParentId("one"));
assertNull(parentGroupIdResolver.getRemoteInputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getRemoteOutputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getRemoteOutputPortParentId("two"));
assertNull(parentGroupIdResolver.getInputPortParentId("one"));
assertNull(parentGroupIdResolver.getInputPortParentId("two"));
assertNull(parentGroupIdResolver.getOutputPortParentId("one"));
@@ -134,7 +171,6 @@ public class ParentGroupIdResolverTest {
@Test
public void testFunnelParentId() throws IOException, SchemaLoaderException {
- StringBuilder sb = new StringBuilder();
List<String> configLines = new ArrayList<>();
configLines.add("MiNiFi Config Version: 2");
configLines.add("Funnels:");
@@ -146,6 +182,8 @@ public class ParentGroupIdResolverTest {
ParentGroupIdResolver parentGroupIdResolver = createParentGroupIdResolver(configLines);
assertNull(parentGroupIdResolver.getRemoteInputPortParentId("one"));
assertNull(parentGroupIdResolver.getRemoteInputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getRemoteOutputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getRemoteOutputPortParentId("two"));
assertNull(parentGroupIdResolver.getInputPortParentId("one"));
assertNull(parentGroupIdResolver.getInputPortParentId("two"));
assertNull(parentGroupIdResolver.getOutputPortParentId("one"));
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-bootstrap/src/test/resources/SimpleRPGToLogAttributes.yml
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/resources/SimpleRPGToLogAttributes.yml b/minifi-bootstrap/src/test/resources/SimpleRPGToLogAttributes.yml
new file mode 100644
index 0000000..578143a
--- /dev/null
+++ b/minifi-bootstrap/src/test/resources/SimpleRPGToLogAttributes.yml
@@ -0,0 +1,110 @@
+# 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.
+
+MiNiFi Config Version: 3
+Flow Controller:
+ name: SimpleRPGToTailFile
+ comment: ''
+Core Properties:
+ flow controller graceful shutdown period: 10 sec
+ flow service write delay interval: 500 ms
+ administrative yield duration: 30 sec
+ bored yield duration: 10 millis
+ max concurrent threads: 1
+FlowFile Repository:
+ partitions: 256
+ checkpoint interval: 2 mins
+ always sync: false
+ Swap:
+ threshold: 20000
+ in period: 5 sec
+ in threads: 1
+ out period: 5 sec
+ out threads: 4
+Content Repository:
+ content claim max appendable size: 10 MB
+ content claim max flow files: 100
+ always sync: false
+Provenance Repository:
+ provenance rollover time: 1 min
+Component Status Repository:
+ buffer size: 1440
+ snapshot frequency: 1 min
+Security Properties:
+ keystore: ''
+ keystore type: ''
+ keystore password: ''
+ key password: ''
+ truststore: ''
+ truststore type: ''
+ truststore password: ''
+ ssl protocol: ''
+ Sensitive Props:
+ key:
+ algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
+ provider: BC
+Processors:
+- id: 6b97126a-015a-1000-0000-000000000000
+ name: LogAttribute
+ class: org.apache.nifi.processors.standard.LogAttribute
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 0 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ - success
+ Properties:
+ Attributes to Ignore:
+ Attributes to Log:
+ Log Level: info
+ Log Payload: 'false'
+ Log prefix:
+Controller Services: []
+Process Groups: []
+Input Ports: []
+Output Ports: []
+Funnels: []
+Connections:
+- id: 6b97306b-015a-1000-0000-000000000000
+ name: 6b965db7-015a-1000-4c43-9d20931a1b9c//LogAttribute
+ source id: 6b965db7-015a-1000-4c43-9d20931a1b9c
+ source relationship names: []
+ destination id: 6b97126a-015a-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+Remote Process Groups:
+- id: 6b95d112-015a-1000-0000-000000000000
+ name: ''
+ url: http://localhost:8080/nifi
+ comment: ''
+ timeout: 30 sec
+ yield period: 10 sec
+ transport protocol: HTTP
+ proxy host: ''
+ proxy port: ''
+ proxy user: ''
+ proxy password: ''
+ Input Ports: []
+ Output Ports:
+ - id: 6b965db7-015a-1000-4c43-9d20931a1b9c
+ name: Output
+ comment: ''
+ max concurrent tasks: 1
+ use compression: false
+NiFi Properties Overrides: {}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ConfigSchema.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ConfigSchema.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ConfigSchema.java
index 1d06974..0b33825 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ConfigSchema.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ConfigSchema.java
@@ -45,6 +45,7 @@ public class ConfigSchema extends BaseSchema implements WritableSchema, Converta
public static final int CONFIG_VERSION = 3;
public static final String VERSION = "MiNiFi Config Version";
public static final String FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_INPUT_PORT_IDS = "Found the following duplicate remote input port ids: ";
+ public static final String FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_OUTPUT_PORT_IDS = "Found the following duplicate remote output port ids: ";
public static final String FOUND_THE_FOLLOWING_DUPLICATE_INPUT_PORT_IDS = "Found the following duplicate input port ids: ";
public static final String FOUND_THE_FOLLOWING_DUPLICATE_OUTPUT_PORT_IDS = "Found the following duplicate output port ids: ";
public static final String FOUND_THE_FOLLOWING_DUPLICATE_IDS = "Found the following ids that occur both in more than one Processor(s), Input Port(s), Output Port(s) and/or Remote Input Port(s): ";
@@ -114,7 +115,9 @@ public class ConfigSchema extends BaseSchema implements WritableSchema, Converta
List<String> allConnectionIds = allConnectionSchemas.stream().map(ConnectionSchema::getId).collect(Collectors.toList());
List<String> allRemoteProcessGroupNames = allRemoteProcessGroups.stream().map(RemoteProcessGroupSchema::getName).collect(Collectors.toList());
List<String> allRemoteInputPortIds = allRemoteProcessGroups.stream().filter(r -> r.getInputPorts() != null)
- .flatMap(r -> r.getInputPorts().stream()).map(RemoteInputPortSchema::getId).collect(Collectors.toList());
+ .flatMap(r -> r.getInputPorts().stream()).map(RemotePortSchema::getId).collect(Collectors.toList());
+ List<String> allRemoteOutputPortIds = allRemoteProcessGroups.stream().filter(r -> r.getOutputPorts() != null)
+ .flatMap(r -> r.getOutputPorts().stream()).map(RemotePortSchema::getId).collect(Collectors.toList());
List<String> allInputPortIds = allProcessGroups.stream().flatMap(p -> p.getInputPortSchemas().stream()).map(PortSchema::getId).collect(Collectors.toList());
List<String> allOutputPortIds = allProcessGroups.stream().flatMap(p -> p.getOutputPortSchemas().stream()).map(PortSchema::getId).collect(Collectors.toList());
@@ -124,12 +127,13 @@ public class ConfigSchema extends BaseSchema implements WritableSchema, Converta
checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_CONNECTION_IDS, allConnectionIds);
checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_PROCESS_GROUP_NAMES, allRemoteProcessGroupNames);
checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_INPUT_PORT_IDS, allRemoteInputPortIds);
+ checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_OUTPUT_PORT_IDS, allRemoteOutputPortIds);
checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_INPUT_PORT_IDS, allInputPortIds);
checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_OUTPUT_PORT_IDS, allOutputPortIds);
// Potential connection sources and destinations need to have unique ids
- CollectionOverlap<String> overlapResults = new CollectionOverlap<>(new HashSet<>(allProcessorIds), new HashSet<>(allRemoteInputPortIds), new HashSet<>(allInputPortIds),
- new HashSet<>(allOutputPortIds), new HashSet<>(allFunnelIds));
+ CollectionOverlap<String> overlapResults = new CollectionOverlap<>(new HashSet<>(allProcessorIds), new HashSet<>(allRemoteInputPortIds), new HashSet<>(allRemoteOutputPortIds),
+ new HashSet<>(allInputPortIds), new HashSet<>(allOutputPortIds), new HashSet<>(allFunnelIds));
if (overlapResults.getDuplicates().size() > 0) {
addValidationIssue(FOUND_THE_FOLLOWING_DUPLICATE_IDS + overlapResults.getDuplicates().stream().sorted().collect(Collectors.joining(", ")));
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessGroupSchema.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessGroupSchema.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessGroupSchema.java
index 56e6af8..08a3acb 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessGroupSchema.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessGroupSchema.java
@@ -139,6 +139,8 @@ public class ProcessGroupSchema extends BaseSchemaWithIdAndName implements Writa
Set<String> result = new HashSet<>();
inputPortSchemas.stream().map(PortSchema::getId).forEachOrdered(result::add);
outputPortSchemas.stream().map(PortSchema::getId).forEachOrdered(result::add);
+ remoteProcessGroups.stream().flatMap(r -> r.getInputPorts().stream()).map(RemotePortSchema::getId).forEachOrdered(result::add);
+ remoteProcessGroups.stream().flatMap(r -> r.getOutputPorts().stream()).map(RemotePortSchema::getId).forEachOrdered(result::add);
processGroupSchemas.stream().flatMap(p -> p.getPortIds().stream()).forEachOrdered(result::add);
return result;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteInputPortSchema.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteInputPortSchema.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteInputPortSchema.java
deleted file mode 100644
index 736c63f..0000000
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteInputPortSchema.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * 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.minifi.commons.schema;
-
-import org.apache.nifi.minifi.commons.schema.common.BaseSchemaWithIdAndName;
-
-import java.util.Map;
-
-import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.COMMENT_KEY;
-import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.MAX_CONCURRENT_TASKS_KEY;
-import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.USE_COMPRESSION_KEY;
-
-public class RemoteInputPortSchema extends BaseSchemaWithIdAndName {
- public static final String DEFAULT_COMMENT = "";
- public static final int DEFAULT_MAX_CONCURRENT_TASKS = 1;
- public static final boolean DEFAULT_USE_COMPRESSION = true;
-
- private String comment = DEFAULT_COMMENT;
- private Number maxConcurrentTasks = DEFAULT_MAX_CONCURRENT_TASKS;
- private Boolean useCompression = DEFAULT_USE_COMPRESSION;
-
- public RemoteInputPortSchema(Map map) {
- super(map, "RemoteInputPort(id: {id}, name: {name})");
- String wrapperName = getWrapperName();
-
- comment = getOptionalKeyAsType(map, COMMENT_KEY, String.class, wrapperName, DEFAULT_COMMENT);
- maxConcurrentTasks = getOptionalKeyAsType(map, MAX_CONCURRENT_TASKS_KEY, Number.class, wrapperName, DEFAULT_MAX_CONCURRENT_TASKS);
- useCompression = getOptionalKeyAsType(map, USE_COMPRESSION_KEY, Boolean.class, wrapperName, DEFAULT_USE_COMPRESSION);
- }
-
- @Override
- public Map<String, Object> toMap() {
- Map<String, Object> result = super.toMap();
- result.put(COMMENT_KEY, comment);
- result.put(MAX_CONCURRENT_TASKS_KEY, maxConcurrentTasks);
- result.put(USE_COMPRESSION_KEY, useCompression);
- return result;
- }
-
- public String getComment() {
- return comment;
- }
-
- public Number getMax_concurrent_tasks() {
- return maxConcurrentTasks;
- }
-
- public boolean getUseCompression() {
- return useCompression;
- }
-}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemotePortSchema.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemotePortSchema.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemotePortSchema.java
new file mode 100644
index 0000000..e73941b
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemotePortSchema.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.minifi.commons.schema;
+
+import org.apache.nifi.minifi.commons.schema.common.BaseSchemaWithIdAndName;
+
+import java.util.Map;
+
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.COMMENT_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.MAX_CONCURRENT_TASKS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.USE_COMPRESSION_KEY;
+
+public class RemotePortSchema extends BaseSchemaWithIdAndName {
+ public static final String DEFAULT_COMMENT = "";
+ public static final int DEFAULT_MAX_CONCURRENT_TASKS = 1;
+ public static final boolean DEFAULT_USE_COMPRESSION = true;
+
+ private String comment = DEFAULT_COMMENT;
+ private Number maxConcurrentTasks = DEFAULT_MAX_CONCURRENT_TASKS;
+ private Boolean useCompression = DEFAULT_USE_COMPRESSION;
+
+ public RemotePortSchema(Map map) {
+ super(map, "RemoteInputPort(id: {id}, name: {name})");
+ String wrapperName = getWrapperName();
+
+ comment = getOptionalKeyAsType(map, COMMENT_KEY, String.class, wrapperName, DEFAULT_COMMENT);
+ maxConcurrentTasks = getOptionalKeyAsType(map, MAX_CONCURRENT_TASKS_KEY, Number.class, wrapperName, DEFAULT_MAX_CONCURRENT_TASKS);
+ useCompression = getOptionalKeyAsType(map, USE_COMPRESSION_KEY, Boolean.class, wrapperName, DEFAULT_USE_COMPRESSION);
+ }
+
+ @Override
+ public Map<String, Object> toMap() {
+ Map<String, Object> result = super.toMap();
+ result.put(COMMENT_KEY, comment);
+ result.put(MAX_CONCURRENT_TASKS_KEY, maxConcurrentTasks);
+ result.put(USE_COMPRESSION_KEY, useCompression);
+ return result;
+ }
+
+ public String getComment() {
+ return comment;
+ }
+
+ public Number getMax_concurrent_tasks() {
+ return maxConcurrentTasks;
+ }
+
+ public boolean getUseCompression() {
+ return useCompression;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteProcessGroupSchema.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteProcessGroupSchema.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteProcessGroupSchema.java
index e73ff73..5a7593a 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteProcessGroupSchema.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteProcessGroupSchema.java
@@ -20,11 +20,13 @@ package org.apache.nifi.minifi.commons.schema;
import org.apache.nifi.minifi.commons.schema.common.BaseSchemaWithIdAndName;
import org.apache.nifi.minifi.commons.schema.common.StringUtil;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.COMMENT_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.INPUT_PORTS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.OUTPUT_PORTS_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.YIELD_PERIOD_KEY;
public class RemoteProcessGroupSchema extends BaseSchemaWithIdAndName {
@@ -60,7 +62,8 @@ public class RemoteProcessGroupSchema extends BaseSchemaWithIdAndName {
public static final String DEFAULT_PROXY_PASSWORD = "";
private String url;
- private List<RemoteInputPortSchema> inputPorts;
+ private List<RemotePortSchema> inputPorts;
+ private List<RemotePortSchema> outputPorts;
private String comment = DEFAULT_COMMENT;
private String timeout = DEFAULT_TIMEOUT;
@@ -75,11 +78,15 @@ public class RemoteProcessGroupSchema extends BaseSchemaWithIdAndName {
super(map, "RemoteProcessGroup(id: {id}, name: {name})");
String wrapperName = getWrapperName();
url = getRequiredKeyAsType(map, URL_KEY, String.class, wrapperName);
- inputPorts = convertListToType(getRequiredKeyAsType(map, INPUT_PORTS_KEY, List.class, wrapperName), "input port", RemoteInputPortSchema.class, INPUT_PORTS_KEY);
- if (inputPorts != null) {
- for (RemoteInputPortSchema remoteInputPortSchema: inputPorts) {
- addIssuesIfNotNull(remoteInputPortSchema);
- }
+
+ inputPorts = convertListToType(getOptionalKeyAsType(map, INPUT_PORTS_KEY, List.class, wrapperName, new ArrayList<>()), "input port", RemotePortSchema.class, INPUT_PORTS_KEY);
+ addIssuesIfNotNull(inputPorts);
+
+ outputPorts = convertListToType(getOptionalKeyAsType(map, OUTPUT_PORTS_KEY, List.class, wrapperName, new ArrayList<>()), "output port", RemotePortSchema.class, OUTPUT_PORTS_KEY);
+ addIssuesIfNotNull(outputPorts);
+
+ if (inputPorts.size() == 0 && outputPorts.size() == 0) {
+ addValidationIssue("Expected either '" + INPUT_PORTS_KEY + "', '" + OUTPUT_PORTS_KEY + "' in section '" + wrapperName + "' to have value(s)");
}
comment = getOptionalKeyAsType(map, COMMENT_KEY, String.class, wrapperName, DEFAULT_COMMENT);
@@ -129,6 +136,7 @@ public class RemoteProcessGroupSchema extends BaseSchemaWithIdAndName {
result.put(PROXY_USER_KEY, proxyUser);
result.put(PROXY_PASSWORD_KEY, proxyPassword);
putListIfNotNull(result, INPUT_PORTS_KEY, inputPorts);
+ putListIfNotNull(result, OUTPUT_PORTS_KEY, outputPorts);
return result;
}
@@ -148,10 +156,14 @@ public class RemoteProcessGroupSchema extends BaseSchemaWithIdAndName {
return yieldPeriod;
}
- public List<RemoteInputPortSchema> getInputPorts() {
+ public List<RemotePortSchema> getInputPorts() {
return inputPorts;
}
+ public List<RemotePortSchema> getOutputPorts() {
+ return outputPorts;
+ }
+
public String getTransportProtocol() {
return transportProtocol;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1.java
index 1989b88..303a6a8 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1.java
@@ -29,7 +29,7 @@ import org.apache.nifi.minifi.commons.schema.FlowFileRepositorySchema;
import org.apache.nifi.minifi.commons.schema.ProcessorSchema;
import org.apache.nifi.minifi.commons.schema.ProvenanceReportingSchema;
import org.apache.nifi.minifi.commons.schema.ProvenanceRepositorySchema;
-import org.apache.nifi.minifi.commons.schema.RemoteInputPortSchema;
+import org.apache.nifi.minifi.commons.schema.RemotePortSchema;
import org.apache.nifi.minifi.commons.schema.RemoteProcessGroupSchema;
import org.apache.nifi.minifi.commons.schema.SecurityPropertiesSchema;
import org.apache.nifi.minifi.commons.schema.common.BaseSchema;
@@ -125,7 +125,7 @@ public class ConfigSchemaV1 extends BaseSchema implements ConvertableSchema<Conf
.collect(Collectors.toList()));
Set<String> connectableNames = new HashSet<>(processorNames);
- connectableNames.addAll(remoteProcessingGroups.stream().flatMap(r -> r.getInputPorts().stream()).map(RemoteInputPortSchema::getId).collect(Collectors.toList()));
+ connectableNames.addAll(remoteProcessingGroups.stream().flatMap(r -> r.getInputPorts().stream()).map(RemotePortSchema::getId).collect(Collectors.toList()));
connections.forEach(c -> {
String destinationName = c.getDestinationName();
if (!StringUtil.isNullOrEmpty(destinationName) && !connectableNames.contains(destinationName)) {
@@ -167,7 +167,7 @@ public class ConfigSchemaV1 extends BaseSchema implements ConvertableSchema<Conf
Set<String> remoteInputPortIds = new HashSet<>();
if (remoteProcessingGroups != null) {
remoteInputPortIds.addAll(remoteProcessingGroups.stream().filter(r -> r.getInputPorts() != null)
- .flatMap(r -> r.getInputPorts().stream()).map(RemoteInputPortSchema::getId).collect(Collectors.toSet()));
+ .flatMap(r -> r.getInputPorts().stream()).map(RemotePortSchema::getId).collect(Collectors.toSet()));
}
Set<String> problematicDuplicateNames = new HashSet<>();
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/RemoteProcessGroupSchemaV1.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/RemoteProcessGroupSchemaV1.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/RemoteProcessGroupSchemaV1.java
index fe1fd4f..720d1c5 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/RemoteProcessGroupSchemaV1.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/RemoteProcessGroupSchemaV1.java
@@ -17,7 +17,7 @@
package org.apache.nifi.minifi.commons.schema.v1;
-import org.apache.nifi.minifi.commons.schema.RemoteInputPortSchema;
+import org.apache.nifi.minifi.commons.schema.RemotePortSchema;
import org.apache.nifi.minifi.commons.schema.RemoteProcessGroupSchema;
import org.apache.nifi.minifi.commons.schema.common.BaseSchema;
import org.apache.nifi.minifi.commons.schema.common.ConvertableSchema;
@@ -42,7 +42,7 @@ public class RemoteProcessGroupSchemaV1 extends BaseSchema implements Convertabl
private String name;
private String url;
- private List<RemoteInputPortSchema> inputPorts;
+ private List<RemotePortSchema> inputPorts;
private String comment = DEFAULT_COMMENT;
private String timeout = DEFAULT_TIMEOUT;
@@ -52,9 +52,9 @@ public class RemoteProcessGroupSchemaV1 extends BaseSchema implements Convertabl
name = getRequiredKeyAsType(map, NAME_KEY, String.class, REMOTE_PROCESS_GROUPS_KEY);
String wrapperName = new StringBuilder("RemoteProcessingGroup(name: ").append(StringUtil.isNullOrEmpty(name) ? "unknown" : name).append(")").toString();
url = getRequiredKeyAsType(map, URL_KEY, String.class, wrapperName);
- inputPorts = convertListToType(getRequiredKeyAsType(map, INPUT_PORTS_KEY, List.class, wrapperName), "input port", RemoteInputPortSchema.class, INPUT_PORTS_KEY);
+ inputPorts = convertListToType(getRequiredKeyAsType(map, INPUT_PORTS_KEY, List.class, wrapperName), "input port", RemotePortSchema.class, INPUT_PORTS_KEY);
if (inputPorts != null) {
- for (RemoteInputPortSchema remoteInputPortSchema : inputPorts) {
+ for (RemotePortSchema remoteInputPortSchema : inputPorts) {
addIssuesIfNotNull(remoteInputPortSchema);
}
}
@@ -96,7 +96,7 @@ public class RemoteProcessGroupSchemaV1 extends BaseSchema implements Convertabl
return yieldPeriod;
}
- public List<RemoteInputPortSchema> getInputPorts() {
+ public List<RemotePortSchema> getInputPorts() {
return inputPorts;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v2/ConfigSchemaV2.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v2/ConfigSchemaV2.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v2/ConfigSchemaV2.java
index d73d8ae..ef95988 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v2/ConfigSchemaV2.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v2/ConfigSchemaV2.java
@@ -29,7 +29,7 @@ import org.apache.nifi.minifi.commons.schema.PortSchema;
import org.apache.nifi.minifi.commons.schema.ProcessorSchema;
import org.apache.nifi.minifi.commons.schema.ProvenanceReportingSchema;
import org.apache.nifi.minifi.commons.schema.ProvenanceRepositorySchema;
-import org.apache.nifi.minifi.commons.schema.RemoteInputPortSchema;
+import org.apache.nifi.minifi.commons.schema.RemotePortSchema;
import org.apache.nifi.minifi.commons.schema.RemoteProcessGroupSchema;
import org.apache.nifi.minifi.commons.schema.SecurityPropertiesSchema;
import org.apache.nifi.minifi.commons.schema.common.BaseSchema;
@@ -117,7 +117,7 @@ public class ConfigSchemaV2 extends BaseSchema implements ConvertableSchema<Conf
List<String> allConnectionIds = allConnectionSchemas.stream().map(ConnectionSchema::getId).collect(Collectors.toList());
List<String> allRemoteProcessGroupNames = allRemoteProcessGroups.stream().map(RemoteProcessGroupSchema::getName).collect(Collectors.toList());
List<String> allRemoteInputPortIds = allRemoteProcessGroups.stream().filter(r -> r.getInputPorts() != null)
- .flatMap(r -> r.getInputPorts().stream()).map(RemoteInputPortSchema::getId).collect(Collectors.toList());
+ .flatMap(r -> r.getInputPorts().stream()).map(RemotePortSchema::getId).collect(Collectors.toList());
List<String> allInputPortIds = allProcessGroups.stream().flatMap(p -> p.getInputPortSchemas().stream()).map(PortSchema::getId).collect(Collectors.toList());
List<String> allOutputPortIds = allProcessGroups.stream().flatMap(p -> p.getOutputPortSchemas().stream()).map(PortSchema::getId).collect(Collectors.toList());
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v2/RemoteProcessGroupSchemaV2.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v2/RemoteProcessGroupSchemaV2.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v2/RemoteProcessGroupSchemaV2.java
index 3e1b762..447a848 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v2/RemoteProcessGroupSchemaV2.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v2/RemoteProcessGroupSchemaV2.java
@@ -17,7 +17,7 @@
package org.apache.nifi.minifi.commons.schema.v2;
-import org.apache.nifi.minifi.commons.schema.RemoteInputPortSchema;
+import org.apache.nifi.minifi.commons.schema.RemotePortSchema;
import org.apache.nifi.minifi.commons.schema.RemoteProcessGroupSchema;
import org.apache.nifi.minifi.commons.schema.RemoteProcessGroupSchema.TransportProtocolOptions;
import org.apache.nifi.minifi.commons.schema.common.BaseSchema;
@@ -42,7 +42,7 @@ import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.YI
public class RemoteProcessGroupSchemaV2 extends BaseSchema implements ConvertableSchema<RemoteProcessGroupSchema> {
private BaseSchemaWithIdAndName idAndName;
private String url;
- private List<RemoteInputPortSchema> inputPorts;
+ private List<RemotePortSchema> inputPorts;
private String comment = DEFAULT_COMMENT;
private String timeout = DEFAULT_TIMEOUT;
@@ -54,9 +54,9 @@ public class RemoteProcessGroupSchemaV2 extends BaseSchema implements Convertabl
String wrapperName = idAndName.getWrapperName();
url = getRequiredKeyAsType(map, URL_KEY, String.class, wrapperName);
- inputPorts = convertListToType(getRequiredKeyAsType(map, INPUT_PORTS_KEY, List.class, wrapperName), "input port", RemoteInputPortSchema.class, INPUT_PORTS_KEY);
+ inputPorts = convertListToType(getRequiredKeyAsType(map, INPUT_PORTS_KEY, List.class, wrapperName), "input port", RemotePortSchema.class, INPUT_PORTS_KEY);
if (inputPorts != null) {
- for (RemoteInputPortSchema remoteInputPortSchema: inputPorts) {
+ for (RemotePortSchema remoteInputPortSchema: inputPorts) {
addIssuesIfNotNull(remoteInputPortSchema);
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/rpg/InputPortStatus.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/rpg/InputPortStatus.java b/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/rpg/InputPortStatus.java
deleted file mode 100644
index 634580e..0000000
--- a/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/rpg/InputPortStatus.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * 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.minifi.commons.status.rpg;
-
-public class InputPortStatus implements java.io.Serializable {
- private String name;
- private boolean targetExists;
- private boolean targetRunning;
-
- public InputPortStatus() {
- }
-
- public String getName() {
- return name;
- }
-
- public void setName(String name) {
- this.name = name;
- }
-
- public boolean isTargetExists() {
- return targetExists;
- }
-
- public void setTargetExists(boolean targetExists) {
- this.targetExists = targetExists;
- }
-
- public boolean isTargetRunning() {
- return targetRunning;
- }
-
- public void setTargetRunning(boolean targetRunning) {
- this.targetRunning = targetRunning;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
-
- InputPortStatus inputPortStatus = (InputPortStatus) o;
-
- if (isTargetExists() != inputPortStatus.isTargetExists()) return false;
- if (isTargetRunning() != inputPortStatus.isTargetRunning()) return false;
- return getName() != null ? getName().equals(inputPortStatus.getName()) : inputPortStatus.getName() == null;
-
- }
-
- @Override
- public int hashCode() {
- int result = getName() != null ? getName().hashCode() : 0;
- result = 31 * result + (isTargetExists() ? 1 : 0);
- result = 31 * result + (isTargetRunning() ? 1 : 0);
- return result;
- }
-
- @Override
- public String toString() {
- return "{" +
- "name='" + name + '\'' +
- ", targetExists=" + targetExists +
- ", targetRunning=" + targetRunning +
- '}';
- }
-}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/rpg/PortStatus.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/rpg/PortStatus.java b/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/rpg/PortStatus.java
new file mode 100644
index 0000000..b2263b6
--- /dev/null
+++ b/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/rpg/PortStatus.java
@@ -0,0 +1,81 @@
+/*
+ * 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.minifi.commons.status.rpg;
+
+public class PortStatus implements java.io.Serializable {
+ private String name;
+ private boolean targetExists;
+ private boolean targetRunning;
+
+ public PortStatus() {
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public boolean isTargetExists() {
+ return targetExists;
+ }
+
+ public void setTargetExists(boolean targetExists) {
+ this.targetExists = targetExists;
+ }
+
+ public boolean isTargetRunning() {
+ return targetRunning;
+ }
+
+ public void setTargetRunning(boolean targetRunning) {
+ this.targetRunning = targetRunning;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ PortStatus inputPortStatus = (PortStatus) o;
+
+ if (isTargetExists() != inputPortStatus.isTargetExists()) return false;
+ if (isTargetRunning() != inputPortStatus.isTargetRunning()) return false;
+ return getName() != null ? getName().equals(inputPortStatus.getName()) : inputPortStatus.getName() == null;
+
+ }
+
+ @Override
+ public int hashCode() {
+ int result = getName() != null ? getName().hashCode() : 0;
+ result = 31 * result + (isTargetExists() ? 1 : 0);
+ result = 31 * result + (isTargetRunning() ? 1 : 0);
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ return "{" +
+ "name='" + name + '\'' +
+ ", targetExists=" + targetExists +
+ ", targetRunning=" + targetRunning +
+ '}';
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/rpg/RemoteProcessGroupStatusBean.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/rpg/RemoteProcessGroupStatusBean.java b/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/rpg/RemoteProcessGroupStatusBean.java
index 3acedc0..ceab031 100644
--- a/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/rpg/RemoteProcessGroupStatusBean.java
+++ b/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/rpg/RemoteProcessGroupStatusBean.java
@@ -26,7 +26,8 @@ public class RemoteProcessGroupStatusBean implements Serializable {
private String name;
private RemoteProcessGroupHealth remoteProcessGroupHealth;
private List<BulletinStatus> bulletinList;
- private List<InputPortStatus> inputPortStatusList;
+ private List<PortStatus> inputPortStatusList;
+ private List<PortStatus> outputPortStatusList;
private RemoteProcessGroupStats remoteProcessGroupStats;
public RemoteProcessGroupStatusBean() {
@@ -56,14 +57,22 @@ public class RemoteProcessGroupStatusBean implements Serializable {
this.bulletinList = bulletinList;
}
- public List<InputPortStatus> getInputPortStatusList() {
+ public List<PortStatus> getInputPortStatusList() {
return inputPortStatusList;
}
- public void setInputPortStatusList(List<InputPortStatus> inputPortStatusList) {
+ public void setInputPortStatusList(List<PortStatus> inputPortStatusList) {
this.inputPortStatusList = inputPortStatusList;
}
+ public List<PortStatus> getOutputPortStatusList() {
+ return outputPortStatusList;
+ }
+
+ public void setOutputPortStatusList(List<PortStatus> outputPortStatusList) {
+ this.outputPortStatusList = outputPortStatusList;
+ }
+
public RemoteProcessGroupStats getRemoteProcessGroupStats() {
return remoteProcessGroupStats;
}
@@ -83,6 +92,7 @@ public class RemoteProcessGroupStatusBean implements Serializable {
if (getRemoteProcessGroupHealth() != null ? !getRemoteProcessGroupHealth().equals(that.getRemoteProcessGroupHealth()) : that.getRemoteProcessGroupHealth() != null) return false;
if (getBulletinList() != null ? !getBulletinList().equals(that.getBulletinList()) : that.getBulletinList() != null) return false;
if (getInputPortStatusList() != null ? !getInputPortStatusList().equals(that.getInputPortStatusList()) : that.getInputPortStatusList() != null) return false;
+ if (getOutputPortStatusList() != null ? !getOutputPortStatusList().equals(that.getOutputPortStatusList()) : that.getOutputPortStatusList() != null) return false;
return getRemoteProcessGroupStats() != null ? getRemoteProcessGroupStats().equals(that.getRemoteProcessGroupStats()) : that.getRemoteProcessGroupStats() == null;
}
@@ -93,6 +103,7 @@ public class RemoteProcessGroupStatusBean implements Serializable {
result = 31 * result + (getRemoteProcessGroupHealth() != null ? getRemoteProcessGroupHealth().hashCode() : 0);
result = 31 * result + (getBulletinList() != null ? getBulletinList().hashCode() : 0);
result = 31 * result + (getInputPortStatusList() != null ? getInputPortStatusList().hashCode() : 0);
+ result = 31 * result + (getOutputPortStatusList() != null ? getOutputPortStatusList().hashCode() : 0);
result = 31 * result + (getRemoteProcessGroupStats() != null ? getRemoteProcessGroupStats().hashCode() : 0);
return result;
}
@@ -104,6 +115,7 @@ public class RemoteProcessGroupStatusBean implements Serializable {
", remoteProcessGroupHealth=" + remoteProcessGroupHealth +
", bulletinList=" + bulletinList +
", inputPortStatusList=" + inputPortStatusList +
+ ", inputPortStatusList=" + outputPortStatusList +
", remoteProcessGroupStats=" + remoteProcessGroupStats +
'}';
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/util/StatusReportPopulator.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/util/StatusReportPopulator.java b/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/util/StatusReportPopulator.java
index 8bdee99..5e81ea6 100644
--- a/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/util/StatusReportPopulator.java
+++ b/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/util/StatusReportPopulator.java
@@ -33,7 +33,7 @@ import org.apache.nifi.minifi.commons.status.processor.ProcessorStats;
import org.apache.nifi.minifi.commons.status.processor.ProcessorStatusBean;
import org.apache.nifi.minifi.commons.status.reportingTask.ReportingTaskHealth;
import org.apache.nifi.minifi.commons.status.reportingTask.ReportingTaskStatus;
-import org.apache.nifi.minifi.commons.status.rpg.InputPortStatus;
+import org.apache.nifi.minifi.commons.status.rpg.PortStatus;
import org.apache.nifi.minifi.commons.status.rpg.RemoteProcessGroupHealth;
import org.apache.nifi.minifi.commons.status.rpg.RemoteProcessGroupStats;
import org.apache.nifi.minifi.commons.status.rpg.RemoteProcessGroupStatusBean;
@@ -134,7 +134,7 @@ public class StatusReportPopulator {
flowStatusReport.setConnectionStatusList(Collections.singletonList(expectedConnectionStatus));
}
- public static void addExpectedRemoteProcessGroupStatus(FlowStatusReport flowStatusReport, boolean addHealth, boolean addInputPort,
+ public static void addExpectedRemoteProcessGroupStatus(FlowStatusReport flowStatusReport, boolean addHealth, boolean addInputPort, boolean addOutputPort,
boolean addStats, boolean addBulletins, boolean populateBulletins) {
RemoteProcessGroupStatusBean expectedRemoteProcessGroupStatus = new RemoteProcessGroupStatusBean();
expectedRemoteProcessGroupStatus.setName("rpg1");
@@ -160,13 +160,21 @@ public class StatusReportPopulator {
}
if (addInputPort) {
- InputPortStatus expectedInputPortStatus = new InputPortStatus();
+ PortStatus expectedInputPortStatus = new PortStatus();
expectedInputPortStatus.setName("inputPort");
expectedInputPortStatus.setTargetExists(true);
expectedInputPortStatus.setTargetRunning(false);
expectedRemoteProcessGroupStatus.setInputPortStatusList(Collections.singletonList(expectedInputPortStatus));
}
+ if (addOutputPort) {
+ PortStatus expectedOutputPortStatus = new PortStatus();
+ expectedOutputPortStatus.setName("outputPort");
+ expectedOutputPortStatus.setTargetExists(true);
+ expectedOutputPortStatus.setTargetRunning(false);
+ expectedRemoteProcessGroupStatus.setOutputPortStatusList(Collections.singletonList(expectedOutputPortStatus));
+ }
+
if (addStats) {
RemoteProcessGroupStats remoteProcessGroupStats = new RemoteProcessGroupStats();
remoteProcessGroupStats.setActiveThreads(3);
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-commons/minifi-utils/src/test/java/org/apache/nifi/minifi/commons/status/TestStatusReport.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-utils/src/test/java/org/apache/nifi/minifi/commons/status/TestStatusReport.java b/minifi-commons/minifi-utils/src/test/java/org/apache/nifi/minifi/commons/status/TestStatusReport.java
index cae5b3a..fc697e3 100644
--- a/minifi-commons/minifi-utils/src/test/java/org/apache/nifi/minifi/commons/status/TestStatusReport.java
+++ b/minifi-commons/minifi-utils/src/test/java/org/apache/nifi/minifi/commons/status/TestStatusReport.java
@@ -47,7 +47,7 @@ public class TestStatusReport {
addReportingTaskStatus(original, true, true, true, true);
addConnectionStatus(original, true, true);
addProcessorStatus(original, true, true, true, true, true);
- addExpectedRemoteProcessGroupStatus(original, true, true, true, true, true);
+ addExpectedRemoteProcessGroupStatus(original, true, true, true, true, true, true);
byte[] byteArrayCopy = serialize(original);
FlowStatusReport copy = unSerialize(byteArrayCopy, FlowStatusReport.class);
@@ -63,7 +63,7 @@ public class TestStatusReport {
addInstanceStatus(original, true, true, true, true);
addSystemDiagnosticStatus(original, true, true, true, true, true);
addProcessorStatus(original, true, true, true, true, true);
- addExpectedRemoteProcessGroupStatus(original, true, true, true, true, true);
+ addExpectedRemoteProcessGroupStatus(original, true, true, true, true, true, true);
byte[] byteArrayCopy = serialize(original);
FlowStatusReport copy = unSerialize(byteArrayCopy, FlowStatusReport.class);
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-docs/src/main/markdown/System_Admin_Guide.md
----------------------------------------------------------------------
diff --git a/minifi-docs/src/main/markdown/System_Admin_Guide.md b/minifi-docs/src/main/markdown/System_Admin_Guide.md
index 6d7f14e..f1d85c3 100644
--- a/minifi-docs/src/main/markdown/System_Admin_Guide.md
+++ b/minifi-docs/src/main/markdown/System_Admin_Guide.md
@@ -229,7 +229,8 @@ Option | Description
------ | -----------
health | The connections's queued bytes and queued FlowFile count.
bulletins | A list of all the current bulletins (if there are any).
-inputPorts | A list of every input port for this RPG and their status. Their status includes it's name, whether the target exit and whether it's currently running.
+inputPorts | A list of every input port for this RPG and their status. Their status includes its name, whether the target exists and whether it's currently running.
+outputPorts | A list of every output port for this RPG and their status. Their status includes its name, whether the target exists and whether it's currently running.
stats | The current stats of the RPG. This includes the active threads, sent content size and count.
An example query to get the health, bulletins, input ports and stats of all the RPGS is below.
@@ -350,6 +351,7 @@ parses and upconverts to the current version without issue.
1. Added support for Controller Services.
2. Added support for Site-To-Site over proxy.
3. Added support for overriding nifi.properties values
+4. Added support for Output Ports to Remote Process Groups
## Flow Controller
@@ -597,17 +599,19 @@ proxy host | The hostname of the proxy server
proxy port | The port to connect to on the proxy server
proxy user | The user name on the proxy server
proxy password | The password for the proxy server
+Input Ports | The Input Ports for this Remote Process Group (See below)
+Output Ports | The Output Ports for this Remote Process Group (See below)
-#### Input Ports Subsection
+#### Input/Output Ports Subsection
-When connecting via Site to Site, MiNiFi needs to know which input port to communicate to of the core NiFi instance. These properties designate and configure communication with that port.
+When connecting via Site to Site, MiNiFi needs to know which input or output port to communicate to of the core NiFi instance. These properties designate and configure communication with that port.
*Property* | *Description*
-------------------- | -------------
-id | The id of the input port as it exists on the core NiFi instance. To get this information access the UI of the core instance, right the input port that is desired to be connect to and select "configure". The id of the port should under the "Id" section.
-name | The name of the input port as it exists on the core NiFi instance. To get this information access the UI of the core instance, right the input port that is desired to be connect to and select "configure". The id of the port should under the "Port name" section.
-comments: | A comment about the Input Port. This is not used for any underlying implementation but solely for the users of this configuration and MiNiFi agent.
+id | The id of the port as it exists on the core NiFi instance. To get this information access the UI of the core instance, right click the port that is desired to be connect to and select "configure". The id of the port should under the "Id" section.
+name | The name of the port as it exists on the core NiFi instance. To get this information access the UI of the core instance, right click the port that is desired to be connect to and select "configure". The name of the port should under the "Port name" section.
+comments: | A comment about the Port. This is not used for any underlying implementation but solely for the users of this configuration and MiNiFi agent.
max concurrent tasks | The number of tasks that this port should be scheduled for at maximum.
use compression | Whether or not compression should be used when communicating with the port. This is a boolean value of either "true" or "false"
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/status/StatusRequestParser.java
----------------------------------------------------------------------
diff --git a/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/status/StatusRequestParser.java b/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/status/StatusRequestParser.java
index 74d3b1f..9f09d06 100644
--- a/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/status/StatusRequestParser.java
+++ b/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/status/StatusRequestParser.java
@@ -44,7 +44,7 @@ import org.apache.nifi.minifi.commons.status.processor.ProcessorStats;
import org.apache.nifi.minifi.commons.status.processor.ProcessorStatusBean;
import org.apache.nifi.minifi.commons.status.reportingTask.ReportingTaskHealth;
import org.apache.nifi.minifi.commons.status.reportingTask.ReportingTaskStatus;
-import org.apache.nifi.minifi.commons.status.rpg.InputPortStatus;
+import org.apache.nifi.minifi.commons.status.rpg.PortStatus;
import org.apache.nifi.minifi.commons.status.rpg.RemoteProcessGroupHealth;
import org.apache.nifi.minifi.commons.status.rpg.RemoteProcessGroupStats;
import org.apache.nifi.minifi.commons.status.rpg.RemoteProcessGroupStatusBean;
@@ -64,6 +64,8 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
public final class StatusRequestParser {
private StatusRequestParser() {
@@ -139,20 +141,10 @@ public final class StatusRequestParser {
remoteProcessGroupStatusBean.setBulletinList(transformBulletins(bulletinList));
break;
case "inputports":
- List<InputPortStatus> inputPortStatusList = new LinkedList<>();
- RemoteProcessGroup remoteProcessGroup = flowController.getGroup(rootGroupId).getRemoteProcessGroup(inputRemoteProcessGroupStatus.getId());
- Set<RemoteGroupPort> inputPorts = remoteProcessGroup.getInputPorts();
-
- for (RemoteGroupPort inputPort : inputPorts) {
- InputPortStatus inputPortStatus = new InputPortStatus();
-
- inputPortStatus.setName(inputPort.getName());
- inputPortStatus.setTargetExists(inputPort.getTargetExists());
- inputPortStatus.setTargetRunning(inputPort.isTargetRunning());
-
- inputPortStatusList.add(inputPortStatus);
- }
- remoteProcessGroupStatusBean.setInputPortStatusList(inputPortStatusList);
+ remoteProcessGroupStatusBean.setInputPortStatusList(getPortStatusList(inputRemoteProcessGroupStatus, flowController, rootGroupId, RemoteProcessGroup::getInputPorts));
+ break;
+ case "outputports":
+ remoteProcessGroupStatusBean.setOutputPortStatusList(getPortStatusList(inputRemoteProcessGroupStatus, flowController, rootGroupId, RemoteProcessGroup::getOutputPorts));
break;
case "stats":
RemoteProcessGroupStats remoteProcessGroupStats = new RemoteProcessGroupStats();
@@ -168,6 +160,19 @@ public final class StatusRequestParser {
return remoteProcessGroupStatusBean;
}
+ private static List<PortStatus> getPortStatusList(RemoteProcessGroupStatus inputRemoteProcessGroupStatus, FlowController flowController, String rootGroupId,
+ Function<RemoteProcessGroup, Set<RemoteGroupPort>> portFunction) {
+ return portFunction.apply(flowController.getGroup(rootGroupId).getRemoteProcessGroup(inputRemoteProcessGroupStatus.getId())).stream().map(r -> {
+ PortStatus portStatus = new PortStatus();
+
+ portStatus.setName(r.getName());
+ portStatus.setTargetExists(r.getTargetExists());
+ portStatus.setTargetRunning(r.isTargetRunning());
+
+ return portStatus;
+ }).collect(Collectors.toList());
+ }
+
static ConnectionStatusBean parseConnectionStatusRequest(ConnectionStatus inputConnectionStatus, String statusTypes, Logger logger) {
ConnectionStatusBean connectionStatusBean = new ConnectionStatusBean();
connectionStatusBean.setName(inputConnectionStatus.getName());
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/status/TestStatusConfigReporter.java
----------------------------------------------------------------------
diff --git a/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/status/TestStatusConfigReporter.java b/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/status/TestStatusConfigReporter.java
index 37483af..ee2f115 100644
--- a/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/status/TestStatusConfigReporter.java
+++ b/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/status/TestStatusConfigReporter.java
@@ -435,7 +435,7 @@ public class TestStatusConfigReporter {
FlowStatusReport expected = new FlowStatusReport();
expected.setErrorsGeneratingReport(Collections.EMPTY_LIST);
- addExpectedRemoteProcessGroupStatus(expected, true, false, false, false, false);
+ addExpectedRemoteProcessGroupStatus(expected, true, false, false, false, false, false);
assertEquals(expected, actual);
}
@@ -450,7 +450,7 @@ public class TestStatusConfigReporter {
FlowStatusReport expected = new FlowStatusReport();
expected.setErrorsGeneratingReport(Collections.EMPTY_LIST);
- addExpectedRemoteProcessGroupStatus(expected, false, false, false, true, true);
+ addExpectedRemoteProcessGroupStatus(expected, false, false, false, false, true, true);
assertEquals(expected, actual);
}
@@ -465,7 +465,22 @@ public class TestStatusConfigReporter {
FlowStatusReport expected = new FlowStatusReport();
expected.setErrorsGeneratingReport(Collections.EMPTY_LIST);
- addExpectedRemoteProcessGroupStatus(expected, false, true, false, false, false);
+ addExpectedRemoteProcessGroupStatus(expected, false, true, false, false, false, false);
+
+ assertEquals(expected, actual);
+ }
+
+ @Test
+ public void remoteProcessGroupStatusOutputPorts() throws Exception {
+ populateRemoteProcessGroup(false, false);
+
+ String statusRequest = "remoteProcessGroup:all:outputPorts";
+ FlowStatusReport actual = StatusConfigReporter.getStatus(mockFlowController, statusRequest, LoggerFactory.getLogger(TestStatusConfigReporter.class));
+
+ FlowStatusReport expected = new FlowStatusReport();
+ expected.setErrorsGeneratingReport(Collections.EMPTY_LIST);
+
+ addExpectedRemoteProcessGroupStatus(expected, false, false, true, false, false, false);
assertEquals(expected, actual);
}
@@ -480,7 +495,7 @@ public class TestStatusConfigReporter {
FlowStatusReport expected = new FlowStatusReport();
expected.setErrorsGeneratingReport(Collections.EMPTY_LIST);
- addExpectedRemoteProcessGroupStatus(expected, false, false, true, false, false);
+ addExpectedRemoteProcessGroupStatus(expected, false, false, false, true, false, false);
assertEquals(expected, actual);
}
@@ -490,13 +505,13 @@ public class TestStatusConfigReporter {
public void remoteProcessGroupStatusAll() throws Exception {
populateRemoteProcessGroup(true, true);
- String statusRequest = "remoteProcessGroup:all:health, bulletins, inputPorts, stats";
+ String statusRequest = "remoteProcessGroup:all:health, bulletins, inputPorts, outputPorts, stats";
FlowStatusReport actual = StatusConfigReporter.getStatus(mockFlowController, statusRequest, LoggerFactory.getLogger(TestStatusConfigReporter.class));
FlowStatusReport expected = new FlowStatusReport();
expected.setErrorsGeneratingReport(Collections.EMPTY_LIST);
- addExpectedRemoteProcessGroupStatus(expected, true, true, true, true, true);
+ addExpectedRemoteProcessGroupStatus(expected, true, true, true, true, true, true);
assertEquals(expected, actual);
}
@@ -515,7 +530,7 @@ public class TestStatusConfigReporter {
String statusRequest = "controllerServices:bulletins,health; processor:all:health,stats,bulletins; instance:bulletins,health,stats ; systemDiagnostics:garbagecollection, heap, " +
"processorstats, contentrepositoryusage, flowfilerepositoryusage; connection:all:health,stats; provenanceReporting:health,bulletins; remoteProcessGroup:all:health, " +
- "bulletins, inputPorts, stats";
+ "bulletins, inputPorts, outputPorts, stats";
FlowStatusReport actual = StatusConfigReporter.getStatus(mockFlowController, statusRequest, LoggerFactory.getLogger(TestStatusConfigReporter.class));
@@ -528,7 +543,7 @@ public class TestStatusConfigReporter {
addReportingTaskStatus(expected, true, true, true, false);
addConnectionStatus(expected, true, true);
addProcessorStatus(expected, true, true, true, true, false);
- addExpectedRemoteProcessGroupStatus(expected, true, true, true, true, false);
+ addExpectedRemoteProcessGroupStatus(expected, true, true, true, true, true, false);
assertEquals(expected, actual);
}
@@ -697,6 +712,13 @@ public class TestStatusConfigReporter {
when(remoteProcessGroup.getInputPorts()).thenReturn(Collections.singleton(remoteGroupPort));
+ remoteGroupPort = mock(RemoteGroupPort.class);
+ when(remoteGroupPort.getName()).thenReturn("outputPort");
+ when(remoteGroupPort.getTargetExists()).thenReturn(true);
+ when(remoteGroupPort.isTargetRunning()).thenReturn(false);
+
+ when(remoteProcessGroup.getOutputPorts()).thenReturn(Collections.singleton(remoteGroupPort));
+
RemoteProcessGroupStatus remoteProcessGroupStatus = new RemoteProcessGroupStatus();
addRemoteProcessGroupStatus(remoteProcessGroupStatus);
if (addBulletins) {
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/52998f23/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ConfigSchemaFunction.java
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ConfigSchemaFunction.java b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ConfigSchemaFunction.java
index 5665430..221c4aa 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ConfigSchemaFunction.java
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ConfigSchemaFunction.java
@@ -55,7 +55,7 @@ public class ConfigSchemaFunction implements Function<TemplateDTO, ConfigSchema>
public ConfigSchemaFunction() {
this(new FlowControllerSchemaFunction(), new ProcessorSchemaFunction(), new ConnectionSchemaFunction(), new FunnelSchemaFunction(), new RemoteProcessGroupSchemaFunction(
- new RemoteInputPortSchemaFunction()), new PortSchemaFunction(INPUT_PORTS_KEY), new PortSchemaFunction(OUTPUT_PORTS_KEY), new ControllerServiceSchemaFunction());
+ new RemotePortSchemaFunction()), new PortSchemaFunction(INPUT_PORTS_KEY), new PortSchemaFunction(OUTPUT_PORTS_KEY), new ControllerServiceSchemaFunction());
}
public ConfigSchemaFunction(FlowControllerSchemaFunction flowControllerSchemaFunction, ProcessorSchemaFunction processorSchemaFunction, ConnectionSchemaFunction connectionSchemaFunction,