You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by br...@apache.org on 2016/12/09 19:51:47 UTC
[3/3] nifi-minifi git commit: MINIFI-154 Adding support for
Controller Services
MINIFI-154 Adding support for Controller Services
This closes #63
Signed-off-by: Bryan Rosander <br...@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/cbb2bdd8
Tree: http://git-wip-us.apache.org/repos/asf/nifi-minifi/tree/cbb2bdd8
Diff: http://git-wip-us.apache.org/repos/asf/nifi-minifi/diff/cbb2bdd8
Branch: refs/heads/master
Commit: cbb2bdd8e00b3b96ef947db486a7608363c058a2
Parents: b028f5a
Author: Joseph Percivall <JP...@apache.org>
Authored: Wed Dec 7 13:59:58 2016 -0500
Committer: Bryan Rosander <br...@apache.org>
Committed: Fri Dec 9 14:50:46 2016 -0500
----------------------------------------------------------------------
minifi-assembly/pom.xml | 14 +-
minifi-bootstrap/pom.xml | 1 +
.../bootstrap/util/ConfigTransformer.java | 36 +-
.../bootstrap/util/ConfigTransformerTest.java | 16 +
.../minifi/commons/schema/ConfigSchema.java | 5 +-
.../commons/schema/ControllerServiceSchema.java | 69 ++
.../commons/schema/ProcessGroupSchema.java | 9 +
.../minifi/commons/schema/ProcessorSchema.java | 12 +-
.../schema/common/CommonPropertyKeys.java | 9 +
.../schema/serialization/SchemaLoader.java | 2 +
.../commons/schema/v1/ProcessorSchemaV1.java | 10 +-
.../commons/schema/v2/ConfigSchemaV2.java | 242 ++++
.../commons/schema/v2/ProcessGroupSchemaV2.java | 186 +++
.../schema/serialization/SchemaLoaderTest.java | 10 +-
.../schema/v1/ProcessorSchemaV1Test.java | 10 +-
.../src/test/resources/config-minimal-v3.yml | 38 +
minifi-docs/src/main/assembly/dependencies.xml | 7 +
.../src/main/markdown/System_Admin_Guide.md | 52 +-
.../markdown/minifi-java-agent-quick-start.md | 21 +-
.../minifi-framework-nar/pom.xml | 2 +
.../minifi-provenance-reporting-nar/pom.xml | 5 +-
.../minifi-ssl-context-service-nar/pom.xml | 6 +-
.../src/main/resources/META-INF/NOTICE | 19 -
minifi-nar-bundles/minifi-standard-nar/pom.xml | 5 +-
.../minifi-standard-services-api-nar/pom.xml | 94 --
.../src/main/resources/META-INF/LICENSE | 202 ----
.../src/main/resources/META-INF/NOTICE | 24 -
minifi-nar-bundles/pom.xml | 1 -
.../configuration/dto/ConfigSchemaFunction.java | 13 +-
.../dto/ControllerServiceSchemaFunction.java | 51 +
.../dto/ProcessorSchemaFunction.java | 9 +-
.../toolkit/configuration/ConfigMainTest.java | 5 +
.../configuration/dto/ProcessorSchemaTest.java | 15 +-
.../src/test/resources/CsvToJson.yml | 3 +-
.../resources/DecompressionCircularFlow.yml | 3 +-
.../resources/InvokeHttpMiNiFiTemplateTest.yml | 3 +-
.../test/resources/MultipleRelationships.yml | 3 +-
.../test/resources/NestedControllerServices.xml | 1089 ++++++++++++++++++
.../test/resources/NestedControllerServices.yml | 290 +++++
.../ProcessGroupsAndRemoteProcessGroups.yml | 5 +-
...aceTextExpressionLanguageCSVReformatting.yml | 3 +-
.../src/test/resources/SimpleTailFileToRPG.yml | 3 +-
.../src/test/resources/StressTestFramework.yml | 3 +-
.../resources/StressTestFrameworkFunnel.yml | 3 +-
.../src/test/resources/config.yml | 2 +-
pom.xml | 17 +-
46 files changed, 2210 insertions(+), 417 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-assembly/pom.xml
----------------------------------------------------------------------
diff --git a/minifi-assembly/pom.xml b/minifi-assembly/pom.xml
index f525d01..1ce48a7 100644
--- a/minifi-assembly/pom.xml
+++ b/minifi-assembly/pom.xml
@@ -166,8 +166,8 @@ limitations under the License.
<artifactId>nifi-framework-core-api</artifactId>
</dependency>
<dependency>
- <groupId>org.apache.nifi.minifi</groupId>
- <artifactId>minifi-standard-services-api-nar</artifactId>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-standard-services-api-nar</artifactId>
<type>nar</type>
</dependency>
<dependency>
@@ -187,12 +187,6 @@ limitations under the License.
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-persistent-provenance-repository</artifactId>
</dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-site-to-site-reporting-task</artifactId>
- <version>1.0.0</version>
- <scope>compile</scope>
- </dependency>
<!-- Provided in NiFi so must include here too -->
<dependency>
@@ -215,6 +209,10 @@ limitations under the License.
<artifactId>commons-lang3</artifactId>
</dependency>
<dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ </dependency>
+ <dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
</dependency>
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-bootstrap/pom.xml
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/pom.xml b/minifi-bootstrap/pom.xml
index 099f6e6..cb57cb2 100644
--- a/minifi-bootstrap/pom.xml
+++ b/minifi-bootstrap/pom.xml
@@ -80,6 +80,7 @@ limitations under the License.
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
+ <scope>provided</scope>
</dependency>
</dependencies>
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/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 def5980..55d3285 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
@@ -25,6 +25,7 @@ import org.apache.nifi.minifi.commons.schema.ComponentStatusRepositorySchema;
import org.apache.nifi.minifi.commons.schema.ConfigSchema;
import org.apache.nifi.minifi.commons.schema.ConnectionSchema;
import org.apache.nifi.minifi.commons.schema.ContentRepositorySchema;
+import org.apache.nifi.minifi.commons.schema.ControllerServiceSchema;
import org.apache.nifi.minifi.commons.schema.CorePropertiesSchema;
import org.apache.nifi.minifi.commons.schema.FlowControllerSchema;
import org.apache.nifi.minifi.commons.schema.FlowFileRepositorySchema;
@@ -309,7 +310,11 @@ public final class ConfigTransformer {
SecurityPropertiesSchema securityProperties = configSchema.getSecurityProperties();
if (securityProperties.useSSL()) {
- final Element controllerServicesNode = doc.createElement("controllerServices");
+ Element controllerServicesNode = doc.getElementById("controllerServices");
+ if(controllerServicesNode == null) {
+ controllerServicesNode = doc.createElement("controllerServices");
+ }
+
rootNode.appendChild(controllerServicesNode);
addSSLControllerService(controllerServicesNode, securityProperties);
}
@@ -356,6 +361,31 @@ public final class ConfigTransformer {
}
}
+ protected static void addControllerService(final Element element, ControllerServiceSchema controllerServiceSchema) throws ConfigurationChangeException {
+ try {
+ final Element serviceElement = element.getOwnerDocument().createElement("controllerService");
+ addTextElement(serviceElement, "id", controllerServiceSchema.getId());
+ addTextElement(serviceElement, "name", controllerServiceSchema.getName());
+ addTextElement(serviceElement, "comment", "");
+ addTextElement(serviceElement, "class", controllerServiceSchema.getServiceClass());
+
+ addTextElement(serviceElement, "enabled", "true");
+
+ Map<String, Object> attributes = controllerServiceSchema.getProperties();
+
+ addConfiguration(serviceElement, attributes);
+
+ String annotationData = controllerServiceSchema.getAnnotationData();
+ if(annotationData != null && !annotationData.isEmpty()) {
+ addTextElement(element, "annotationData", annotationData);
+ }
+
+ element.appendChild(serviceElement);
+ } catch (Exception e) {
+ throw new ConfigurationChangeException("Failed to parse the config YAML while trying to create an SSL Controller Service", e);
+ }
+ }
+
protected static void addProcessGroup(Document doc, Element element, ProcessGroupSchema processGroupSchema, ParentGroupIdResolver parentGroupIdResolver) throws ConfigurationChangeException {
try {
String processGroupId = processGroupSchema.getId();
@@ -393,6 +423,10 @@ public final class ConfigTransformer {
for (ConnectionSchema connectionConfig : processGroupSchema.getConnections()) {
addConnection(element, connectionConfig, parentGroupIdResolver);
}
+
+ for (ControllerServiceSchema controllerServiceSchema : processGroupSchema.getControllerServices()) {
+ addControllerService(element, controllerServiceSchema);
+ }
} catch (ConfigurationChangeException e) {
throw e;
} catch (Exception e) {
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/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 726b686..be88cb4 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
@@ -20,6 +20,7 @@ package org.apache.nifi.minifi.bootstrap.util;
import org.apache.nifi.minifi.bootstrap.configuration.ConfigurationChangeException;
import org.apache.nifi.minifi.commons.schema.ConfigSchema;
import org.apache.nifi.minifi.commons.schema.ConnectionSchema;
+import org.apache.nifi.minifi.commons.schema.ControllerServiceSchema;
import org.apache.nifi.minifi.commons.schema.FunnelSchema;
import org.apache.nifi.minifi.commons.schema.PortSchema;
import org.apache.nifi.minifi.commons.schema.ProcessGroupSchema;
@@ -140,6 +141,11 @@ public class ConfigTransformerTest {
for (int i = 0; i < processorElements.getLength(); i++) {
testProcessor((Element) processorElements.item(i), processGroupSchema.getProcessors().get(i));
}
+ NodeList controllerServiceElements = (NodeList) xPathFactory.newXPath().evaluate("controllerService", element, XPathConstants.NODESET);
+ assertEquals(processGroupSchema.getControllerServices().size(), controllerServiceElements.getLength());
+ for (int i = 0; i < controllerServiceElements.getLength(); i++) {
+ testControllerService((Element) controllerServiceElements.item(i), processGroupSchema.getControllerServices().get(i));
+ }
NodeList remoteProcessGroupElements = (NodeList) xPathFactory.newXPath().evaluate("remoteProcessGroup", element, XPathConstants.NODESET);
assertEquals(processGroupSchema.getRemoteProcessGroups().size(), remoteProcessGroupElements.getLength());
@@ -188,10 +194,20 @@ public class ConfigTransformerTest {
assertEquals(processorSchema.getYieldPeriod(), getText(element, "yieldPeriod"));
assertEquals(processorSchema.getSchedulingStrategy(), getText(element, "schedulingStrategy"));
assertEquals(processorSchema.getRunDurationNanos().toString(), getText(element, "runDurationNanos"));
+ assertEquals(processorSchema.getAnnotationData(), getText(element, "annotationData"));
testProperties(element, processorSchema.getProperties());
}
+ private void testControllerService(Element element, ControllerServiceSchema controllerServiceSchema) throws XPathExpressionException {
+ assertEquals(controllerServiceSchema.getId(), getText(element, "id"));
+ assertEquals(controllerServiceSchema.getName(), getText(element, "name"));
+ assertEquals(controllerServiceSchema.getServiceClass(), getText(element, "class"));
+ assertEquals(controllerServiceSchema.getAnnotationData(), getText(element, "annotationData"));
+
+ testProperties(element, controllerServiceSchema.getProperties());
+ }
+
private void testRemoteProcessGroups(Element element, RemoteProcessGroupSchema remoteProcessingGroupSchema) throws XPathExpressionException {
assertEquals(remoteProcessingGroupSchema.getId(), getText(element, "id"));
assertEquals(remoteProcessingGroupSchema.getName(), getText(element, "name"));
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/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 0f5d7e9..7a9bf09 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
@@ -42,7 +42,7 @@ import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PR
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SECURITY_PROPS_KEY;
public class ConfigSchema extends BaseSchema implements WritableSchema, ConvertableSchema<ConfigSchema> {
- public static final int CONFIG_VERSION = 2;
+ 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_INPUT_PORT_IDS = "Found the following duplicate input port ids: ";
@@ -52,6 +52,7 @@ public class ConfigSchema extends BaseSchema implements WritableSchema, Converta
public static final String HAS_INVALID_SOURCE_ID = " has invalid source id ";
public static final String HAS_INVALID_DESTINATION_ID = " has invalid destination id ";
public static final String FOUND_THE_FOLLOWING_DUPLICATE_PROCESSOR_IDS = "Found the following duplicate processor ids: ";
+ public static final String FOUND_THE_FOLLOWING_DUPLICATE_CONTROLLER_SERVICE_IDS = "Found the following duplicate controller service ids: ";
public static final String FOUND_THE_FOLLOWING_DUPLICATE_CONNECTION_IDS = "Found the following duplicate connection ids: ";
public static final String FOUND_THE_FOLLOWING_DUPLICATE_FUNNEL_IDS = "Found the following duplicate funnel ids: ";
public static final String FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_PROCESS_GROUP_NAMES = "Found the following duplicate remote process group names: ";
@@ -101,6 +102,7 @@ public class ConfigSchema extends BaseSchema implements WritableSchema, Converta
List<RemoteProcessGroupSchema> allRemoteProcessGroups = allProcessGroups.stream().flatMap(p -> p.getRemoteProcessGroups().stream()).collect(Collectors.toList());
List<String> allProcessorIds = allProcessGroups.stream().flatMap(p -> p.getProcessors().stream()).map(ProcessorSchema::getId).collect(Collectors.toList());
+ List<String> allControllerServiceIds = allProcessGroups.stream().flatMap(p -> p.getControllerServices().stream()).map(ControllerServiceSchema::getId).collect(Collectors.toList());
List<String> allFunnelIds = allProcessGroups.stream().flatMap(p -> p.getFunnels().stream()).map(FunnelSchema::getId).collect(Collectors.toList());
List<String> allConnectionIds = allConnectionSchemas.stream().map(ConnectionSchema::getId).collect(Collectors.toList());
List<String> allRemoteProcessGroupNames = allRemoteProcessGroups.stream().map(RemoteProcessGroupSchema::getName).collect(Collectors.toList());
@@ -110,6 +112,7 @@ public class ConfigSchema extends BaseSchema implements WritableSchema, Converta
List<String> allOutputPortIds = allProcessGroups.stream().flatMap(p -> p.getOutputPortSchemas().stream()).map(PortSchema::getId).collect(Collectors.toList());
checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_PROCESSOR_IDS, allProcessorIds);
+ checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_CONTROLLER_SERVICE_IDS, allControllerServiceIds);
checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_FUNNEL_IDS, allFunnelIds);
checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_CONNECTION_IDS, allConnectionIds);
checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_PROCESS_GROUP_NAMES, allRemoteProcessGroupNames);
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ControllerServiceSchema.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ControllerServiceSchema.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ControllerServiceSchema.java
new file mode 100644
index 0000000..cc3a85c
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ControllerServiceSchema.java
@@ -0,0 +1,69 @@
+/*
+ * 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 java.util.TreeMap;
+
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.ANNOTATION_DATA_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.DEFAULT_PROPERTIES;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROPERTIES_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.TYPE_KEY;
+
+public class ControllerServiceSchema extends BaseSchemaWithIdAndName {
+
+ private Map<String, Object> properties = DEFAULT_PROPERTIES;
+ private String annotationData = "";
+ private String serviceClass;
+
+ public ControllerServiceSchema(Map map) {
+ super(map, "Controller Service(id: {id}, name: {name})");
+ String wrapperName = getWrapperName();
+ serviceClass = getRequiredKeyAsType(map, TYPE_KEY, String.class, wrapperName);
+ properties = getOptionalKeyAsType(map, PROPERTIES_KEY, Map.class, wrapperName, DEFAULT_PROPERTIES);
+ annotationData = getOptionalKeyAsType(map, ANNOTATION_DATA_KEY, String.class, wrapperName, "");
+ }
+
+
+ @Override
+ public Map<String, Object> toMap() {
+ Map<String, Object> result = super.toMap();
+ result.put(TYPE_KEY, serviceClass);
+ result.put(PROPERTIES_KEY, new TreeMap<>(properties));
+
+ if(annotationData != null && !annotationData.isEmpty()) {
+ result.put(ANNOTATION_DATA_KEY, annotationData);
+ }
+
+ return result;
+ }
+
+ public String getAnnotationData() {
+ return annotationData;
+ }
+
+ public String getServiceClass() {
+ return serviceClass;
+ }
+
+ public Map<String, Object> getProperties() {
+ return properties;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/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 028f666..56e6af8 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
@@ -31,6 +31,7 @@ import java.util.Set;
import java.util.stream.Collectors;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CONNECTIONS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CONTROLLER_SERVICES_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.FUNNELS_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.ID_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.INPUT_PORTS_KEY;
@@ -46,6 +47,7 @@ public class ProcessGroupSchema extends BaseSchemaWithIdAndName implements Writa
private String comment;
private List<ProcessorSchema> processors;
+ private List<ControllerServiceSchema> controllerServiceSchemas;
private List<FunnelSchema> funnels;
private List<ConnectionSchema> connections;
private List<RemoteProcessGroupSchema> remoteProcessGroups;
@@ -57,6 +59,7 @@ public class ProcessGroupSchema extends BaseSchemaWithIdAndName implements Writa
super(map, wrapperName);
processors = getOptionalKeyAsList(map, PROCESSORS_KEY, ProcessorSchema::new, wrapperName);
+ controllerServiceSchemas = getOptionalKeyAsList(map, CONTROLLER_SERVICES_KEY, ControllerServiceSchema::new, wrapperName);
funnels = getOptionalKeyAsList(map, FUNNELS_KEY, FunnelSchema::new, wrapperName);
remoteProcessGroups = getOptionalKeyAsList(map, REMOTE_PROCESS_GROUPS_KEY, RemoteProcessGroupSchema::new, wrapperName);
connections = getOptionalKeyAsList(map, CONNECTIONS_KEY, ConnectionSchema::new, wrapperName);
@@ -83,6 +86,7 @@ public class ProcessGroupSchema extends BaseSchemaWithIdAndName implements Writa
connections.stream().filter(c -> funnelIds.contains(c.getSourceId())).forEachOrdered(c -> c.setNeedsSourceRelationships(false));
addIssuesIfNotNull(processors);
+ addIssuesIfNotNull(controllerServiceSchemas);
addIssuesIfNotNull(remoteProcessGroups);
addIssuesIfNotNull(processGroupSchemas);
addIssuesIfNotNull(funnels);
@@ -97,6 +101,7 @@ public class ProcessGroupSchema extends BaseSchemaWithIdAndName implements Writa
}
StringUtil.doIfNotNullOrEmpty(getName(), name -> result.put(NAME_KEY, name));
putListIfNotNull(result, PROCESSORS_KEY, processors);
+ putListIfNotNull(result, CONTROLLER_SERVICES_KEY, controllerServiceSchemas);
putListIfNotNull(result, PROCESS_GROUPS_KEY, processGroupSchemas);
putListIfNotNull(result, INPUT_PORTS_KEY, inputPortSchemas);
putListIfNotNull(result, OUTPUT_PORTS_KEY, outputPortSchemas);
@@ -110,6 +115,10 @@ public class ProcessGroupSchema extends BaseSchemaWithIdAndName implements Writa
return processors;
}
+ public List<ControllerServiceSchema> getControllerServices() {
+ return controllerServiceSchemas;
+ }
+
public List<FunnelSchema> getFunnels() {
return funnels;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessorSchema.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessorSchema.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessorSchema.java
index 6f2ff8e..00eb9c5 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessorSchema.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessorSchema.java
@@ -25,25 +25,25 @@ import java.util.List;
import java.util.Map;
import java.util.TreeMap;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.ANNOTATION_DATA_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CLASS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.DEFAULT_PROPERTIES;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.MAX_CONCURRENT_TASKS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROPERTIES_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SCHEDULING_PERIOD_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SCHEDULING_STRATEGY_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.YIELD_PERIOD_KEY;
public class ProcessorSchema extends BaseSchemaWithIdAndName {
- public static final String CLASS_KEY = "class";
public static final String PENALIZATION_PERIOD_KEY = "penalization period";
public static final String RUN_DURATION_NANOS_KEY = "run duration nanos";
public static final String AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY = "auto-terminated relationships list";
- public static final String PROCESSOR_PROPS_KEY = "Properties";
- public static final String ANNOTATION_DATA_KEY = "annotation data";
public static final int DEFAULT_MAX_CONCURRENT_TASKS = 1;
public static final String DEFAULT_PENALIZATION_PERIOD = "30 sec";
public static final String DEFAULT_YIELD_DURATION = "1 sec";
public static final long DEFAULT_RUN_DURATION_NANOS = 0;
public static final List<String> DEFAULT_AUTO_TERMINATED_RELATIONSHIPS_LIST = Collections.emptyList();
- public static final Map<String, Object> DEFAULT_PROPERTIES = Collections.emptyMap();
public static final String IT_IS_NOT_A_VALID_SCHEDULING_STRATEGY = "it is not a valid scheduling strategy";
private String processorClass;
@@ -72,7 +72,7 @@ public class ProcessorSchema extends BaseSchemaWithIdAndName {
yieldPeriod = getOptionalKeyAsType(map, YIELD_PERIOD_KEY, String.class, wrapperName, DEFAULT_YIELD_DURATION);
runDurationNanos = getOptionalKeyAsType(map, RUN_DURATION_NANOS_KEY, Number.class, wrapperName, DEFAULT_RUN_DURATION_NANOS);
autoTerminatedRelationshipsList = getOptionalKeyAsType(map, AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY, List.class, wrapperName, DEFAULT_AUTO_TERMINATED_RELATIONSHIPS_LIST);
- properties = getOptionalKeyAsType(map, PROCESSOR_PROPS_KEY, Map.class, wrapperName, DEFAULT_PROPERTIES);
+ properties = getOptionalKeyAsType(map, PROPERTIES_KEY, Map.class, wrapperName, DEFAULT_PROPERTIES);
annotationData = getOptionalKeyAsType(map, ANNOTATION_DATA_KEY, String.class, wrapperName, "");
}
@@ -97,7 +97,7 @@ public class ProcessorSchema extends BaseSchemaWithIdAndName {
result.put(YIELD_PERIOD_KEY, yieldPeriod);
result.put(RUN_DURATION_NANOS_KEY, runDurationNanos);
result.put(AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY, autoTerminatedRelationshipsList);
- result.put(PROCESSOR_PROPS_KEY, new TreeMap<>(properties));
+ result.put(PROPERTIES_KEY, new TreeMap<>(properties));
if(annotationData != null && !annotationData.isEmpty()) {
result.put(ANNOTATION_DATA_KEY, annotationData);
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/CommonPropertyKeys.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/CommonPropertyKeys.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/CommonPropertyKeys.java
index 4efb258..05ad607 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/CommonPropertyKeys.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/CommonPropertyKeys.java
@@ -17,6 +17,9 @@
package org.apache.nifi.minifi.commons.schema.common;
+import java.util.Collections;
+import java.util.Map;
+
public class CommonPropertyKeys {
public static final String CORE_PROPS_KEY = "Core Properties";
public static final String FLOWFILE_REPO_KEY = "FlowFile Repository";
@@ -32,6 +35,7 @@ public class CommonPropertyKeys {
public static final String REMOTE_PROCESS_GROUPS_KEY = "Remote Process Groups";
public static final String INPUT_PORTS_KEY = "Input Ports";
public static final String OUTPUT_PORTS_KEY = "Output Ports";
+ public static final String CONTROLLER_SERVICES_KEY = "Controller Services";
public static final String FUNNELS_KEY = "Funnels";
public static final String PROVENANCE_REPO_KEY = "Provenance Repository";
@@ -46,7 +50,12 @@ public class CommonPropertyKeys {
public static final String SCHEDULING_STRATEGY_KEY = "scheduling strategy";
public static final String SCHEDULING_PERIOD_KEY = "scheduling period";
public static final String USE_COMPRESSION_KEY = "use compression";
+ public static final String PROPERTIES_KEY = "Properties";
+ public static final String CLASS_KEY = "class";
+ public static final String TYPE_KEY = "type";
+ public static final String ANNOTATION_DATA_KEY = "annotation data";
+ public static final Map<String, Object> DEFAULT_PROPERTIES = Collections.emptyMap();
private CommonPropertyKeys() {
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoader.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoader.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoader.java
index 331f40d..d3eb766 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoader.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoader.java
@@ -22,6 +22,7 @@ import org.apache.nifi.minifi.commons.schema.common.ConvertableSchema;
import org.apache.nifi.minifi.commons.schema.common.StringUtil;
import org.apache.nifi.minifi.commons.schema.exception.SchemaLoaderException;
import org.apache.nifi.minifi.commons.schema.v1.ConfigSchemaV1;
+import org.apache.nifi.minifi.commons.schema.v2.ConfigSchemaV2;
import org.yaml.snakeyaml.Yaml;
import org.yaml.snakeyaml.error.YAMLException;
@@ -40,6 +41,7 @@ public class SchemaLoader {
result.put(String.valueOf((Object) null), ConfigSchemaV1::new);
result.put("", ConfigSchemaV1::new);
result.put(Integer.toString(ConfigSchemaV1.CONFIG_VERSION), ConfigSchemaV1::new);
+ result.put(Integer.toString(ConfigSchemaV2.CONFIG_VERSION), ConfigSchemaV2::new);
result.put(Integer.toString(ConfigSchema.CONFIG_VERSION), ConfigSchema::new);
return result;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1.java
index d64c98b..42b4050 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1.java
@@ -28,21 +28,21 @@ import java.util.List;
import java.util.Map;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY;
-import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.CLASS_KEY;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_AUTO_TERMINATED_RELATIONSHIPS_LIST;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_MAX_CONCURRENT_TASKS;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_PENALIZATION_PERIOD;
-import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_PROPERTIES;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_RUN_DURATION_NANOS;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_YIELD_DURATION;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.IT_IS_NOT_A_VALID_SCHEDULING_STRATEGY;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.PENALIZATION_PERIOD_KEY;
-import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.PROCESSOR_PROPS_KEY;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.RUN_DURATION_NANOS_KEY;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.isSchedulingStrategy;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CLASS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.DEFAULT_PROPERTIES;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.MAX_CONCURRENT_TASKS_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.NAME_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROCESSORS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROPERTIES_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SCHEDULING_PERIOD_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SCHEDULING_STRATEGY_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.YIELD_PERIOD_KEY;
@@ -73,7 +73,7 @@ public class ProcessorSchemaV1 extends BaseSchema implements ConvertableSchema<P
yieldPeriod = getOptionalKeyAsType(map, YIELD_PERIOD_KEY, String.class, PROCESSORS_KEY, DEFAULT_YIELD_DURATION);
runDurationNanos = getOptionalKeyAsType(map, RUN_DURATION_NANOS_KEY, Number.class, PROCESSORS_KEY, DEFAULT_RUN_DURATION_NANOS);
autoTerminatedRelationshipsList = getOptionalKeyAsType(map, AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY, List.class, PROCESSORS_KEY, DEFAULT_AUTO_TERMINATED_RELATIONSHIPS_LIST);
- properties = getOptionalKeyAsType(map, PROCESSOR_PROPS_KEY, Map.class, PROCESSORS_KEY, DEFAULT_PROPERTIES);
+ properties = getOptionalKeyAsType(map, PROPERTIES_KEY, Map.class, PROCESSORS_KEY, DEFAULT_PROPERTIES);
}
@Override
@@ -88,7 +88,7 @@ public class ProcessorSchemaV1 extends BaseSchema implements ConvertableSchema<P
map.put(YIELD_PERIOD_KEY, yieldPeriod);
map.put(RUN_DURATION_NANOS_KEY, runDurationNanos);
map.put(AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY, autoTerminatedRelationshipsList);
- map.put(PROCESSOR_PROPS_KEY, new HashMap<>(properties));
+ map.put(PROPERTIES_KEY, new HashMap<>(properties));
return new ProcessorSchema(map);
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/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
new file mode 100644
index 0000000..20a9da2
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v2/ConfigSchemaV2.java
@@ -0,0 +1,242 @@
+/*
+ * 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.v2;
+
+import org.apache.nifi.minifi.commons.schema.ComponentStatusRepositorySchema;
+import org.apache.nifi.minifi.commons.schema.ConfigSchema;
+import org.apache.nifi.minifi.commons.schema.ConnectionSchema;
+import org.apache.nifi.minifi.commons.schema.ContentRepositorySchema;
+import org.apache.nifi.minifi.commons.schema.CorePropertiesSchema;
+import org.apache.nifi.minifi.commons.schema.FlowControllerSchema;
+import org.apache.nifi.minifi.commons.schema.FlowFileRepositorySchema;
+import org.apache.nifi.minifi.commons.schema.FunnelSchema;
+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.RemoteProcessGroupSchema;
+import org.apache.nifi.minifi.commons.schema.SecurityPropertiesSchema;
+import org.apache.nifi.minifi.commons.schema.common.BaseSchema;
+import org.apache.nifi.minifi.commons.schema.common.ConvertableSchema;
+import org.apache.nifi.minifi.commons.schema.common.StringUtil;
+import org.apache.nifi.minifi.commons.schema.common.WritableSchema;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.COMPONENT_STATUS_REPO_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CONTENT_REPO_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CORE_PROPS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.FLOWFILE_REPO_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.FLOW_CONTROLLER_PROPS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROVENANCE_REPORTING_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROVENANCE_REPO_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SECURITY_PROPS_KEY;
+
+public class ConfigSchemaV2 extends BaseSchema implements WritableSchema, ConvertableSchema<ConfigSchema> {
+ public static final int CONFIG_VERSION = 2;
+ 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_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): ";
+ public static final String CONNECTION_WITH_ID = "Connection with id ";
+ public static final String HAS_INVALID_SOURCE_ID = " has invalid source id ";
+ public static final String HAS_INVALID_DESTINATION_ID = " has invalid destination id ";
+ public static final String FOUND_THE_FOLLOWING_DUPLICATE_PROCESSOR_IDS = "Found the following duplicate processor ids: ";
+ public static final String FOUND_THE_FOLLOWING_DUPLICATE_CONNECTION_IDS = "Found the following duplicate connection ids: ";
+ public static final String FOUND_THE_FOLLOWING_DUPLICATE_FUNNEL_IDS = "Found the following duplicate funnel ids: ";
+ public static final String FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_PROCESS_GROUP_NAMES = "Found the following duplicate remote process group names: ";
+ public static String TOP_LEVEL_NAME = "top level";
+ private FlowControllerSchema flowControllerProperties;
+ private CorePropertiesSchema coreProperties;
+ private FlowFileRepositorySchema flowfileRepositoryProperties;
+ private ContentRepositorySchema contentRepositoryProperties;
+ private ComponentStatusRepositorySchema componentStatusRepositoryProperties;
+ private SecurityPropertiesSchema securityProperties;
+ private ProcessGroupSchemaV2 processGroupSchema;
+ private ProvenanceReportingSchema provenanceReportingProperties;
+
+ private ProvenanceRepositorySchema provenanceRepositorySchema;
+
+ public ConfigSchemaV2(Map map) {
+ this(map, Collections.emptyList());
+ }
+
+ public ConfigSchemaV2(Map map, List<String> validationIssues) {
+ validationIssues.stream().forEach(this::addValidationIssue);
+ flowControllerProperties = getMapAsType(map, FLOW_CONTROLLER_PROPS_KEY, FlowControllerSchema.class, TOP_LEVEL_NAME, true);
+
+ coreProperties = getMapAsType(map, CORE_PROPS_KEY, CorePropertiesSchema.class, TOP_LEVEL_NAME, false);
+ flowfileRepositoryProperties = getMapAsType(map, FLOWFILE_REPO_KEY, FlowFileRepositorySchema.class, TOP_LEVEL_NAME, false);
+ contentRepositoryProperties = getMapAsType(map, CONTENT_REPO_KEY, ContentRepositorySchema.class, TOP_LEVEL_NAME, false);
+ provenanceRepositorySchema = getMapAsType(map, PROVENANCE_REPO_KEY, ProvenanceRepositorySchema.class, TOP_LEVEL_NAME, false);
+ componentStatusRepositoryProperties = getMapAsType(map, COMPONENT_STATUS_REPO_KEY, ComponentStatusRepositorySchema.class, TOP_LEVEL_NAME, false);
+ securityProperties = getMapAsType(map, SECURITY_PROPS_KEY, SecurityPropertiesSchema.class, TOP_LEVEL_NAME, false);
+
+ processGroupSchema = new ProcessGroupSchemaV2(map, TOP_LEVEL_NAME);
+
+ provenanceReportingProperties = getMapAsType(map, PROVENANCE_REPORTING_KEY, ProvenanceReportingSchema.class, TOP_LEVEL_NAME, false, false);
+
+ addIssuesIfNotNull(flowControllerProperties);
+ addIssuesIfNotNull(coreProperties);
+ addIssuesIfNotNull(flowfileRepositoryProperties);
+ addIssuesIfNotNull(contentRepositoryProperties);
+ addIssuesIfNotNull(componentStatusRepositoryProperties);
+ addIssuesIfNotNull(securityProperties);
+ addIssuesIfNotNull(processGroupSchema);
+ addIssuesIfNotNull(provenanceReportingProperties);
+ addIssuesIfNotNull(provenanceRepositorySchema);
+
+ List<ProcessGroupSchemaV2> allProcessGroups = getAllProcessGroups(processGroupSchema);
+ List<ConnectionSchema> allConnectionSchemas = allProcessGroups.stream().flatMap(p -> p.getConnections().stream()).collect(Collectors.toList());
+ List<RemoteProcessGroupSchema> allRemoteProcessGroups = allProcessGroups.stream().flatMap(p -> p.getRemoteProcessGroups().stream()).collect(Collectors.toList());
+
+ List<String> allProcessorIds = allProcessGroups.stream().flatMap(p -> p.getProcessors().stream()).map(ProcessorSchema::getId).collect(Collectors.toList());
+ List<String> allFunnelIds = allProcessGroups.stream().flatMap(p -> p.getFunnels().stream()).map(FunnelSchema::getId).collect(Collectors.toList());
+ 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());
+ 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());
+
+ checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_PROCESSOR_IDS, allProcessorIds);
+ checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_FUNNEL_IDS, allFunnelIds);
+ 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_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
+ OverlapResults<String> overlapResults = findOverlap(new HashSet<>(allProcessorIds), new HashSet<>(allRemoteInputPortIds), new HashSet<>(allInputPortIds), new HashSet<>(allOutputPortIds),
+ new HashSet<>(allFunnelIds));
+ if (overlapResults.duplicates.size() > 0) {
+ addValidationIssue(FOUND_THE_FOLLOWING_DUPLICATE_IDS + overlapResults.duplicates.stream().sorted().collect(Collectors.joining(", ")));
+ }
+
+ allConnectionSchemas.forEach(c -> {
+ String destinationId = c.getDestinationId();
+ if (!StringUtil.isNullOrEmpty(destinationId) && !overlapResults.seen.contains(destinationId)) {
+ addValidationIssue(CONNECTION_WITH_ID + c.getId() + HAS_INVALID_DESTINATION_ID + destinationId);
+ }
+ String sourceId = c.getSourceId();
+ if (!StringUtil.isNullOrEmpty(sourceId) && !overlapResults.seen.contains(sourceId)) {
+ addValidationIssue(CONNECTION_WITH_ID + c.getId() + HAS_INVALID_SOURCE_ID + sourceId);
+ }
+ });
+ }
+
+ protected static <T> OverlapResults<T> findOverlap(Collection<T>... collections) {
+ Set<T> seen = new HashSet<>();
+ return new OverlapResults<>(seen, Arrays.stream(collections).flatMap(c -> c.stream()).sequential().filter(s -> !seen.add(s)).collect(Collectors.toSet()));
+ }
+
+ public static List<ProcessGroupSchemaV2> getAllProcessGroups(ProcessGroupSchemaV2 processGroupSchema) {
+ List<ProcessGroupSchemaV2> result = new ArrayList<>();
+ addProcessGroups(processGroupSchema, result);
+ return result;
+ }
+
+ private static void addProcessGroups(ProcessGroupSchemaV2 processGroupSchema, List<ProcessGroupSchemaV2> result) {
+ result.add(processGroupSchema);
+ processGroupSchema.getProcessGroupSchemas().forEach(p -> addProcessGroups(p, result));
+ }
+
+ public Map<String, Object> toMap() {
+ Map<String, Object> result = mapSupplier.get();
+ result.put(VERSION, getVersion());
+ putIfNotNull(result, FLOW_CONTROLLER_PROPS_KEY, flowControllerProperties);
+ putIfNotNull(result, CORE_PROPS_KEY, coreProperties);
+ putIfNotNull(result, FLOWFILE_REPO_KEY, flowfileRepositoryProperties);
+ putIfNotNull(result, CONTENT_REPO_KEY, contentRepositoryProperties);
+ putIfNotNull(result, PROVENANCE_REPO_KEY, provenanceRepositorySchema);
+ putIfNotNull(result, COMPONENT_STATUS_REPO_KEY, componentStatusRepositoryProperties);
+ putIfNotNull(result, SECURITY_PROPS_KEY, securityProperties);
+ result.putAll(processGroupSchema.toMap());
+ putIfNotNull(result, PROVENANCE_REPORTING_KEY, provenanceReportingProperties);
+ return result;
+ }
+
+ public FlowControllerSchema getFlowControllerProperties() {
+ return flowControllerProperties;
+ }
+
+ public CorePropertiesSchema getCoreProperties() {
+ return coreProperties;
+ }
+
+ public FlowFileRepositorySchema getFlowfileRepositoryProperties() {
+ return flowfileRepositoryProperties;
+ }
+
+ public ContentRepositorySchema getContentRepositoryProperties() {
+ return contentRepositoryProperties;
+ }
+
+ public SecurityPropertiesSchema getSecurityProperties() {
+ return securityProperties;
+ }
+
+ public ProcessGroupSchemaV2 getProcessGroupSchema() {
+ return processGroupSchema;
+ }
+
+ public ProvenanceReportingSchema getProvenanceReportingProperties() {
+ return provenanceReportingProperties;
+ }
+
+ public ComponentStatusRepositorySchema getComponentStatusRepositoryProperties() {
+ return componentStatusRepositoryProperties;
+ }
+
+ public ProvenanceRepositorySchema getProvenanceRepositorySchema() {
+ return provenanceRepositorySchema;
+ }
+
+ @Override
+ public int getVersion() {
+ return CONFIG_VERSION;
+ }
+
+ @Override
+ public ConfigSchema convert() {
+ Map<String, Object> map = this.toMap();
+ List<String> validationIssues = getValidationIssues();
+ return new ConfigSchema(map, validationIssues);
+ }
+
+ private static class OverlapResults<T> {
+ private final Set<T> seen;
+ private final Set<T> duplicates;
+
+ private OverlapResults(Set<T> seen, Set<T> duplicates) {
+ this.seen = seen;
+ this.duplicates = duplicates;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v2/ProcessGroupSchemaV2.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v2/ProcessGroupSchemaV2.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v2/ProcessGroupSchemaV2.java
new file mode 100644
index 0000000..a9f0449
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v2/ProcessGroupSchemaV2.java
@@ -0,0 +1,186 @@
+/*
+ *
+ * * 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.v2;
+
+import org.apache.nifi.minifi.commons.schema.ConfigSchema;
+import org.apache.nifi.minifi.commons.schema.ConnectionSchema;
+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.RemoteProcessGroupSchema;
+import org.apache.nifi.minifi.commons.schema.common.BaseSchemaWithIdAndName;
+import org.apache.nifi.minifi.commons.schema.common.ConvertableSchema;
+import org.apache.nifi.minifi.commons.schema.common.StringUtil;
+import org.apache.nifi.minifi.commons.schema.common.WritableSchema;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CONNECTIONS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CONTROLLER_SERVICES_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.DEFAULT_PROPERTIES;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.FUNNELS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.ID_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.INPUT_PORTS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.NAME_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.OUTPUT_PORTS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROCESSORS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.REMOTE_PROCESS_GROUPS_KEY;
+
+public class ProcessGroupSchemaV2 extends BaseSchemaWithIdAndName implements WritableSchema, ConvertableSchema<ProcessGroupSchema> {
+
+ public static final String PROCESS_GROUPS_KEY = "Process Groups";
+ public static final String ID_DEFAULT = "Root-Group";
+
+ private String comment;
+ private List<ProcessorSchema> processors;
+ private List<FunnelSchema> funnels;
+ private List<ConnectionSchema> connections;
+ private List<RemoteProcessGroupSchema> remoteProcessGroups;
+ private List<ProcessGroupSchemaV2> processGroupSchemas;
+ private List<PortSchema> inputPortSchemas;
+ private List<PortSchema> outputPortSchemas;
+
+ public ProcessGroupSchemaV2(Map map, String wrapperName) {
+ super(map, wrapperName);
+
+ processors = getOptionalKeyAsList(map, PROCESSORS_KEY, ProcessorSchema::new, wrapperName);
+ funnels = getOptionalKeyAsList(map, FUNNELS_KEY, FunnelSchema::new, wrapperName);
+ remoteProcessGroups = getOptionalKeyAsList(map, REMOTE_PROCESS_GROUPS_KEY, RemoteProcessGroupSchema::new, wrapperName);
+ connections = getOptionalKeyAsList(map, CONNECTIONS_KEY, ConnectionSchema::new, wrapperName);
+ inputPortSchemas = getOptionalKeyAsList(map, INPUT_PORTS_KEY, m -> new PortSchema(m, "InputPort(id: {id}, name: {name})"), wrapperName);
+ outputPortSchemas = getOptionalKeyAsList(map, OUTPUT_PORTS_KEY, m -> new PortSchema(m, "OutputPort(id: {id}, name: {name})"), wrapperName);
+ processGroupSchemas = getOptionalKeyAsList(map, PROCESS_GROUPS_KEY, m -> new ProcessGroupSchemaV2(m, "ProcessGroup(id: {id}, name: {name})"), wrapperName);
+
+ if (ConfigSchema.TOP_LEVEL_NAME.equals(wrapperName)) {
+ if (inputPortSchemas.size() > 0) {
+ addValidationIssue(INPUT_PORTS_KEY, wrapperName, "must be empty in root group as external input/output ports are currently unsupported");
+ }
+ if (outputPortSchemas.size() > 0) {
+ addValidationIssue(OUTPUT_PORTS_KEY, wrapperName, "must be empty in root group as external input/output ports are currently unsupported");
+ }
+ } else if (ID_DEFAULT.equals(getId())) {
+ addValidationIssue(ID_KEY, wrapperName, "must be set to a value not " + ID_DEFAULT + " if not in root group");
+ }
+
+ Set<String> portIds = getPortIds();
+ connections.stream().filter(c -> portIds.contains(c.getSourceId())).forEachOrdered(c -> c.setNeedsSourceRelationships(false));
+
+
+ Set<String> funnelIds = new HashSet<>(funnels.stream().map(FunnelSchema::getId).collect(Collectors.toList()));
+ connections.stream().filter(c -> funnelIds.contains(c.getSourceId())).forEachOrdered(c -> c.setNeedsSourceRelationships(false));
+
+ addIssuesIfNotNull(processors);
+ addIssuesIfNotNull(remoteProcessGroups);
+ addIssuesIfNotNull(processGroupSchemas);
+ addIssuesIfNotNull(funnels);
+ addIssuesIfNotNull(connections);
+ }
+
+ public Map<String, Object> toMap() {
+ Map<String, Object> result = mapSupplier.get();
+ String id = getId();
+ if (!ID_DEFAULT.equals(id)) {
+ result.put(ID_KEY, id);
+ }
+ StringUtil.doIfNotNullOrEmpty(getName(), name -> result.put(NAME_KEY, name));
+ putListIfNotNull(result, PROCESSORS_KEY, processors);
+ putListIfNotNull(result, PROCESS_GROUPS_KEY, processGroupSchemas);
+ putListIfNotNull(result, INPUT_PORTS_KEY, inputPortSchemas);
+ putListIfNotNull(result, OUTPUT_PORTS_KEY, outputPortSchemas);
+ putListIfNotNull(result, FUNNELS_KEY, funnels);
+ putListIfNotNull(result, CONNECTIONS_KEY, connections);
+ putListIfNotNull(result, REMOTE_PROCESS_GROUPS_KEY, remoteProcessGroups);
+ return result;
+ }
+
+ public List<ProcessorSchema> getProcessors() {
+ return processors;
+ }
+
+ public List<FunnelSchema> getFunnels() {
+ return funnels;
+ }
+
+ public List<ConnectionSchema> getConnections() {
+ return connections;
+ }
+
+ public List<RemoteProcessGroupSchema> getRemoteProcessGroups() {
+ return remoteProcessGroups;
+ }
+
+ public List<ProcessGroupSchemaV2> getProcessGroupSchemas() {
+ return processGroupSchemas;
+ }
+
+ public Set<String> getPortIds() {
+ Set<String> result = new HashSet<>();
+ inputPortSchemas.stream().map(PortSchema::getId).forEachOrdered(result::add);
+ outputPortSchemas.stream().map(PortSchema::getId).forEachOrdered(result::add);
+ processGroupSchemas.stream().flatMap(p -> p.getPortIds().stream()).forEachOrdered(result::add);
+ return result;
+ }
+
+ public String getComment() {
+ return comment;
+ }
+
+ public void setComment(String comment) {
+ this.comment = comment;
+ }
+
+ @Override
+ protected String getId(Map map, String wrapperName) {
+ return getOptionalKeyAsType(map, ID_KEY, String.class, wrapperName, ID_DEFAULT);
+ }
+
+ @Override
+ public ProcessGroupSchema convert() {
+ Map<String, Object> map = this.toMap();
+ map.put(CONTROLLER_SERVICES_KEY, DEFAULT_PROPERTIES);
+ return new ProcessGroupSchema(map, getWrapperName());
+ }
+
+ @Override
+ public int getVersion() {
+ return ConfigSchema.CONFIG_VERSION;
+ }
+
+ public List<PortSchema> getOutputPortSchemas() {
+ return outputPortSchemas;
+ }
+
+ public List<PortSchema> getInputPortSchemas() {
+ return inputPortSchemas;
+ }
+
+ @Override
+ protected boolean isValidId(String value) {
+ if (ID_DEFAULT.equals(value)) {
+ return true;
+ }
+ return super.isValidId(value);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoaderTest.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoaderTest.java b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoaderTest.java
index 928bc03..045492e 100644
--- a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoaderTest.java
+++ b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoaderTest.java
@@ -65,6 +65,14 @@ public class SchemaLoaderTest {
}
@Test
+ public void testMinimalConfigV3Version() throws IOException, SchemaLoaderException {
+ Map<String, Object> yamlAsMap = SchemaLoader.loadYamlAsMap(SchemaLoaderTest.class.getClassLoader().getResourceAsStream("config-minimal-v3.yml"));
+ yamlAsMap.put(ConfigSchema.VERSION, ConfigSchema.CONFIG_VERSION);
+ ConfigSchema configSchema = SchemaLoader.loadConfigSchemaFromYaml(yamlAsMap);
+ validateMinimalConfigVersion1Parse(configSchema);
+ }
+
+ @Test
public void testUnsupportedVersion() throws IOException, SchemaLoaderException {
Map<String, Object> yamlAsMap = SchemaLoader.loadYamlAsMap(SchemaLoaderTest.class.getClassLoader().getResourceAsStream("config-minimal-v2.yml"));
yamlAsMap.put(ConfigSchema.VERSION, "9999999");
@@ -72,7 +80,7 @@ public class SchemaLoaderTest {
SchemaLoader.loadConfigSchemaFromYaml(yamlAsMap);
fail();
} catch (SchemaLoaderException e) {
- assertEquals("YAML configuration version 9999999 not supported. Supported versions: 1, 2", e.getMessage());
+ assertEquals("YAML configuration version 9999999 not supported. Supported versions: 1, 2, 3", e.getMessage());
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1Test.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1Test.java b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1Test.java
index 3668c0e..f23145e 100644
--- a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1Test.java
+++ b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1Test.java
@@ -30,18 +30,18 @@ import java.util.List;
import java.util.Map;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY;
-import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.CLASS_KEY;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_AUTO_TERMINATED_RELATIONSHIPS_LIST;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_MAX_CONCURRENT_TASKS;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_PENALIZATION_PERIOD;
-import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_PROPERTIES;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_RUN_DURATION_NANOS;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_YIELD_DURATION;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.PENALIZATION_PERIOD_KEY;
-import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.PROCESSOR_PROPS_KEY;
import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.RUN_DURATION_NANOS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CLASS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.DEFAULT_PROPERTIES;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.MAX_CONCURRENT_TASKS_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.NAME_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROPERTIES_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SCHEDULING_PERIOD_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SCHEDULING_STRATEGY_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.YIELD_PERIOD_KEY;
@@ -109,7 +109,7 @@ public class ProcessorSchemaV1Test {
map.put(YIELD_PERIOD_KEY, testYieldPeriod);
map.put(RUN_DURATION_NANOS_KEY, testRunDurationNanos);
map.put(AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY, testAutoTerminatedRelationships);
- map.put(PROCESSOR_PROPS_KEY, testProperties);
+ map.put(PROPERTIES_KEY, testProperties);
return map;
}
@@ -235,7 +235,7 @@ public class ProcessorSchemaV1Test {
@Test
public void testNoProperties() {
Map<String, Object> map = createMap();
- map.remove(PROCESSOR_PROPS_KEY);
+ map.remove(PROPERTIES_KEY);
assertEquals(DEFAULT_PROPERTIES, createSchema(map, 0).convert().getProperties());
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-commons/minifi-commons-schema/src/test/resources/config-minimal-v3.yml
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/test/resources/config-minimal-v3.yml b/minifi-commons/minifi-commons-schema/src/test/resources/config-minimal-v3.yml
new file mode 100644
index 0000000..9b4c397
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/test/resources/config-minimal-v3.yml
@@ -0,0 +1,38 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the \"License\"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an \"AS IS\" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# This is a minimal V3 config.
+MiNiFi Config Version: 3
+Flow Controller:
+ name: MiNiFi Flow
+
+# When running the Flow (not just doing the transform) these processors will be invalid due to not having the necesary properties/auto-terminated relationships
+Processors:
+ - id: d4b7c284-882c-39e2-88bb-65e8abd5f4c8
+ class: org.apache.nifi.processors.standard.TailFile
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 10 sec
+ - id: a256e6b3-36af-3c38-8564-789c399b516c
+ class: org.apache.nifi.processors.standard.PutFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 0 sec
+
+Connections:
+ - id: 90015098-3cd2-3fb0-9696-3f7d28e17f72
+ source id: d4b7c284-882c-39e2-88bb-65e8abd5f4c8
+ source relationship names:
+ - success
+ destination id: a256e6b3-36af-3c38-8564-789c399b516c
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-docs/src/main/assembly/dependencies.xml
----------------------------------------------------------------------
diff --git a/minifi-docs/src/main/assembly/dependencies.xml b/minifi-docs/src/main/assembly/dependencies.xml
index 1ea6053..014b895 100644
--- a/minifi-docs/src/main/assembly/dependencies.xml
+++ b/minifi-docs/src/main/assembly/dependencies.xml
@@ -28,6 +28,13 @@
<filtered>true</filtered>
</file>
<file>
+ <source>${project.build.directory}/markdown/minifi-java-agent-quick-start.md</source>
+ <outputDirectory>./</outputDirectory>
+ <destName>System_Admin_Guide.md</destName>
+ <fileMode>0644</fileMode>
+ <filtered>true</filtered>
+ </file>
+ <file>
<source>./LICENSE</source>
<outputDirectory>./</outputDirectory>
<destName>LICENSE</destName>
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/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 d65f8ed..d95cd84 100644
--- a/minifi-docs/src/main/markdown/System_Admin_Guide.md
+++ b/minifi-docs/src/main/markdown/System_Admin_Guide.md
@@ -37,7 +37,7 @@ After a new configuration has been pulled/received the Ingestors use a Different
After a new config is determined to be new, the MiNiFi agent will attempt to restart. The bootstrap first saves the old config into a swap file. The bootstrap monitors the agent as it restarts and if it fails it will roll back to the old config. If it succeeds then the swap file will be deleted and the agent will start processing using the new config.
-Note: Data left in connections when the agent attempts to restart will either be mapped to a connection with the same ID in the new config, or orphaned and deleted.
+**Note:** Data left in connections when the agent attempts to restart will either be mapped to a connection with the same ID in the new config, or orphaned and deleted.
The configuration for Warm-Redeploy is done in the bootstrap.conf and primarily revolve around the Config Change Ingestors. The configuration in the bootstrap.conf is done using the "nifi.minifi.notifier.ingestors" key followed by the full path name of the desired Ingestor implementation to run. Use a comma separated list to define more than one Ingestor implementation. For example:
@@ -69,7 +69,7 @@ class name: org.apache.nifi.minifi.bootstrap.configuration.ingestors.RestChangeI
This Config Change Ingestor sets up a light-weight Jetty HTTP(S) REST service in order to listen to HTTP(S) requests. A potential new configuration is sent via a POST request with the BODY being the potential new config.
-NOTE: The encoding is expected to be Unicode and the exact version specified by the BOM mark ('UTF-8','UTF-16BE' or 'UTF-16LE'). If there is no BOM mark, then UTF-8 is used.
+**Note:** The encoding is expected to be Unicode and the exact version specified by the BOM mark ('UTF-8','UTF-16BE' or 'UTF-16LE'). If there is no BOM mark, then UTF-8 is used.
Here is an example post request using 'curl' hitting the local machine on pot 8338 and it is executed with the config file "config.yml" in the directory the command is run from:
@@ -330,7 +330,7 @@ Alternatively, the MiNiFi Toolkit Converter can aid in creating a config.yml fro
tool can be downloaded from http://nifi.apache.org/minifi/download.html under the `MiNiFi Toolkit Binaries` section. Information on the toolkit's usage is
available at https://nifi.apache.org/minifi/minifi-toolkit.html.
-NOTE: Note that values for periods of time and data sizes must include the unit of measure,
+**Note:** Values for periods of time and data sizes must include the unit of measure,
for example "10 sec" or "10 MB", not simply "10".
## Versioning
@@ -344,8 +344,11 @@ parses and upconverts to the current version without issue.
1. Use ids instead of names for processors, connections.
2. Allow multiple source relationships for connections.
-3. Support process groups, input ports, output ports
-4. Change Id Key for RPGs from "Remote Processing Groups" to the proper "Remote Process Groups" (not "ing")
+3. Added support for process groups, and internal input ports an output ports.
+4. Change Id Key for RPGs from "Remote Processing Groups" to the proper "Remote Process Groups" (not "ing").
+
+### Version 2 -> Version 3 changes
+1. Added support for Controller Services.
## Flow Controller
@@ -418,7 +421,7 @@ always sync | If set to _true_, any change to the reposito
-------------------------------- | -------------
provenance rollover time | The amount of time to wait before rolling over the latest data provenance information so that it is available to be accessed by components. The default value is 1 min.
-## *Component Status Repository*
+## Component Status Repository
The Component Status Repository contains the information for the Component Status History tool in the User Interface. These
properties govern how that tool works.
@@ -434,7 +437,7 @@ of 576.
buffer size | Specifies the buffer size for the Component Status Repository. The default value is 1440.
snapshot frequency | This value indicates how often to present a snapshot of the components' status history. The default value is 1 min.
-## *Security Properties*
+## Security Properties
These properties pertain to various security features in NiFi. Many of these properties are covered in more detail in the
Security Configuration section of this Administrator's Guide.
@@ -450,6 +453,8 @@ truststore type | The truststore type. It is blank by default.
truststore password | The truststore password. It is blank by default.
ssl protocol | The protocol to use when communicating via https. Necessary to transfer provenance securely.
+**Note:** A StandardSSLContextService will be made automatically with the ID "SSL-Context-Service" if "ssl protocol" is configured.
+
#### Sensitive Properties Subsection
Some properties for processors are marked as _sensitive_ and should be encrypted. These following properties will be used to encrypt the properties while in use by MiNiFi. This will currently *not* be used to encrypt properties in the config file.
@@ -462,7 +467,7 @@ provider | The sensitive property provider. The default value is BC.
## Processors
-The current implementation of MiNiFi supports multiple processors. the "Processors" subsection is a list of these processors. Each processor must specify these properties. They are the basic configuration general to all processor implementations. Make sure that all relationships for a processor are accounted for in the auto-terminated relationship list or are used in a connection.
+The current implementation of MiNiFi supports multiple processors. The "Processors" subsection is a list of these processors. Each processor must specify these properties. They are the basic configuration general to all processor implementations. Make sure that all relationships for a processor are accounted for in the auto-terminated relationship list or are used in a connection.
*Property* | *Description*
----------------------------------- | -------------
@@ -488,6 +493,35 @@ Within the Processor Configuration section, there is the `Properties` subsection
State File: ./conf/state/tail-file
Initial Start Position: Beginning of File
+
+### Controller Services
+
+The current implementation of MiNiFi supports Controller Services. The "Controller Services" subsection is a list of these services. Each Controller Service must specify the following properties. They are the basic configuration general to all Controller Service implementations.
+
+*Property* | *Description*
+------ | -----------
+name | The name of what this Controller Service will do. This is not used for any underlying implementation but solely for the users of this configuration and MiNiFi agent.
+id | The id of this Controller Service. This must be a valid UUID. To reference this Controller Service in the properties of another component, this ID is used.
+type | The fully qualified java class name of the processor to run. For example for the standard StandardSSLContextService processor would be: org.apache.nifi.ssl.StandardSSLContextService
+
+**Note:** If the "Security Properties" is configured with an "ssl protocol" then a StandardSSLContextService will be made automatically with the ID "SSL-Context-Service".
+
+
+#### Controller Service Properties
+
+Within the Controller Service Configuration section, there is the `Properties` subsection. The keys and values in this section are the property names and values for the service. For example the StandardSSLContextService would have a section like this:
+
+ Properties:
+ Keystore Filename: /tmp/localhost/keystore.jks
+ Keystore Password: keystorePassword
+ Keystore Type: JKS
+ SSL Protocol: TLS
+ Truststore Filename: /tmp/truststore.jks
+ Truststore Password: truststorePassword
+ Truststore Type: JKS
+ key-password: keyPassword
+
+
## Process Groups
Process groups can be nested from the top level. They can contain other process groups as well and can be used to logically group related operations.
@@ -588,7 +622,7 @@ use compression | Indicates whether or not to compress the events when bein
timeout | How long MiNiFi should wait before timing out the connection.
batch size | Specifies how many records to send in a single batch, at most. This should be significantly above the expected amount of records generated between scheduling. If it is not, then there is the potential for the Provenance reporting to lag behind event generation and never catch up.
-
+**Note:** In order to send via HTTPS, the "Security Properties" must be fully configured. A StandardSSLContextService will be made automatically with the ID "SSL-Context-Service" and used by the Provenance Reporting.
# Example Config File
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-docs/src/main/markdown/minifi-java-agent-quick-start.md
----------------------------------------------------------------------
diff --git a/minifi-docs/src/main/markdown/minifi-java-agent-quick-start.md b/minifi-docs/src/main/markdown/minifi-java-agent-quick-start.md
index 63ee343..3fd7425 100644
--- a/minifi-docs/src/main/markdown/minifi-java-agent-quick-start.md
+++ b/minifi-docs/src/main/markdown/minifi-java-agent-quick-start.md
@@ -102,7 +102,7 @@ You can use the MiNiFi Toolkit, located in your MiNiFi installation directory, a
1. Launch NiFi
2. Create a dataflow.
3. Convert your dataflow into a template.
-4. Download your template as an .xml file. For more information on working with templates, see the [Templates](http://docs.hortonworks.com/HDPDocuments/HDF2/HDF-2.0.0/bk_user-guide/content/templates.html) section in the *User Guide*.
+4. Download your template as an .xml file. For more information on working with templates, see the [Templates](https://nifi.apache.org/docs/nifi-docs/html/user-guide.html#templates) section in the *User Guide*.
5. From the MiNiFi Toolkit, run the following command to turn your .xml file into a .yml file:
```
config.sh transform input_file output_file
@@ -180,13 +180,28 @@ MiNiFi is able to use following processors out of the box:
* UnpackContent
* ValidateXml
+MiNiFi is able to use the StandardSSLContextService out of the box.
+
If you want to create a dataflow with a processor not shipped with MiNiFi, you can do so.
1. Set up your dataflow as described above.
2. Copy the desired NAR file into the MiNiFi lib directory.
3. Restart your MiNiFi instance.
-**Note:** Currently only the StandardSSLContextService is supported as a controller service. It is created automatically if the the "Security Properties" section is set and can be referenced in the processor configuration using the ID "SSL-Context-Service".
-
+**Note:** The following processors are also a part of the default distribution but require adding a NAR for a Controller Service not packaged by default. The processors are grouped by the NAR that is required.
+* nifi-dbcp-service-nar
+ * ConvertJSONToSQL
+ * PutSQL
+ * GenerateTableFetch
+ * ListDatabaseTable
+ * QueryDatabaseTable
+ * ExecuteSQL
+* nifi-distributed-cache-services-nar
+ * DetectDuplicate
+ * FetchDistributedMapCache
+ * PutDistributedMapCache
+* nifi-http-context-map-nar
+ * HandleHttpRequest
+ * HandleHttpResponse
# Securing your Dataflow
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-nar-bundles/minifi-framework-bundle/minifi-framework-nar/pom.xml
----------------------------------------------------------------------
diff --git a/minifi-nar-bundles/minifi-framework-bundle/minifi-framework-nar/pom.xml b/minifi-nar-bundles/minifi-framework-bundle/minifi-framework-nar/pom.xml
index cbdce76..8a161a8 100644
--- a/minifi-nar-bundles/minifi-framework-bundle/minifi-framework-nar/pom.xml
+++ b/minifi-nar-bundles/minifi-framework-bundle/minifi-framework-nar/pom.xml
@@ -112,10 +112,12 @@ limitations under the License.
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
+ <scope>provided</scope>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
+ <scope>provided</scope>
</dependency>
</dependencies>
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-nar-bundles/minifi-provenance-reporting-bundle/minifi-provenance-reporting-nar/pom.xml
----------------------------------------------------------------------
diff --git a/minifi-nar-bundles/minifi-provenance-reporting-bundle/minifi-provenance-reporting-nar/pom.xml b/minifi-nar-bundles/minifi-provenance-reporting-bundle/minifi-provenance-reporting-nar/pom.xml
index f187920..a70fd37 100644
--- a/minifi-nar-bundles/minifi-provenance-reporting-bundle/minifi-provenance-reporting-nar/pom.xml
+++ b/minifi-nar-bundles/minifi-provenance-reporting-bundle/minifi-provenance-reporting-nar/pom.xml
@@ -30,11 +30,10 @@
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-site-to-site-reporting-task</artifactId>
- <version>1.0.0</version>
</dependency>
<dependency>
- <groupId>org.apache.nifi.minifi</groupId>
- <artifactId>minifi-standard-services-api-nar</artifactId>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-standard-services-api-nar</artifactId>
<type>nar</type>
</dependency>
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-nar-bundles/minifi-ssl-context-service-nar/pom.xml
----------------------------------------------------------------------
diff --git a/minifi-nar-bundles/minifi-ssl-context-service-nar/pom.xml b/minifi-nar-bundles/minifi-ssl-context-service-nar/pom.xml
index 475195a..73a07b6 100644
--- a/minifi-nar-bundles/minifi-ssl-context-service-nar/pom.xml
+++ b/minifi-nar-bundles/minifi-ssl-context-service-nar/pom.xml
@@ -28,8 +28,8 @@
</properties>
<dependencies>
<dependency>
- <groupId>org.apache.nifi.minifi</groupId>
- <artifactId>minifi-standard-services-api-nar</artifactId>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-standard-services-api-nar</artifactId>
<type>nar</type>
<exclusions>
<exclusion>
@@ -71,10 +71,12 @@
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
+ <scope>provided</scope>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
+ <scope>provided</scope>
</dependency>
</dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-nar-bundles/minifi-ssl-context-service-nar/src/main/resources/META-INF/NOTICE
----------------------------------------------------------------------
diff --git a/minifi-nar-bundles/minifi-ssl-context-service-nar/src/main/resources/META-INF/NOTICE b/minifi-nar-bundles/minifi-ssl-context-service-nar/src/main/resources/META-INF/NOTICE
index b7d1e8b..41dd8dd 100644
--- a/minifi-nar-bundles/minifi-ssl-context-service-nar/src/main/resources/META-INF/NOTICE
+++ b/minifi-nar-bundles/minifi-ssl-context-service-nar/src/main/resources/META-INF/NOTICE
@@ -3,22 +3,3 @@ Copyright 2014-2016 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
-
-******************
-Apache Software License v2
-******************
-
-The following binary components are provided under the Apache Software License v2
-
- (ASLv2) Apache Commons IO
- The following NOTICE information applies:
- Apache Commons IO
- Copyright 2002-2012 The Apache Software Foundation
-
- (ASLv2) Apache Commons Lang
- The following NOTICE information applies:
- Apache Commons Lang
- Copyright 2001-2015 The Apache Software Foundation
-
- This product includes software from the Spring Framework,
- under the Apache License 2.0 (see: StringUtils.containsWhitespace())
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/cbb2bdd8/minifi-nar-bundles/minifi-standard-nar/pom.xml
----------------------------------------------------------------------
diff --git a/minifi-nar-bundles/minifi-standard-nar/pom.xml b/minifi-nar-bundles/minifi-standard-nar/pom.xml
index 695ac7c..2d08698 100644
--- a/minifi-nar-bundles/minifi-standard-nar/pom.xml
+++ b/minifi-nar-bundles/minifi-standard-nar/pom.xml
@@ -32,8 +32,8 @@ limitations under the License.
</properties>
<dependencies>
<dependency>
- <groupId>org.apache.nifi.minifi</groupId>
- <artifactId>minifi-standard-services-api-nar</artifactId>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-standard-services-api-nar</artifactId>
<type>nar</type>
</dependency>
<dependency>
@@ -70,7 +70,6 @@ limitations under the License.
<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bcpg-jdk15on</artifactId>
- <version>1.54</version>
<scope>provided</scope>
</dependency>
<dependency>