You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jp...@apache.org on 2016/06/14 19:27:43 UTC
[1/3] nifi-minifi git commit: MINIFI-30 Added changes to the config
transformer to support multiple processors, multiple RPGs,
multiple Input Ports in RPGS, and greater validation on the config yaml
Repository: nifi-minifi
Updated Branches:
refs/heads/master fb5548198 -> 6f528bbc1
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-docs/Properties_Guide.md
----------------------------------------------------------------------
diff --git a/minifi-docs/Properties_Guide.md b/minifi-docs/Properties_Guide.md
index 1e2b09c..b67b478 100644
--- a/minifi-docs/Properties_Guide.md
+++ b/minifi-docs/Properties_Guide.md
@@ -50,6 +50,7 @@ flow controller graceful shutdown period | Indicates the shutdown period. The de
flow service write delay interval | When many changes are made to the flow.xml, this property specifies how long to wait before writing out the changes, so as to batch the changes into a single write. The default value is 500 ms.
administrative yield duration | If a component allows an unexpected exception to escape, it is considered a bug. As a result, the framework will pause (or administratively yield) the component for this amount of time. This is done so that the component does not use up massive amounts of system resources, since it is known to have problems in the existing state. The default value is 30 sec.
bored yield duration | When a component has no work to do (i.e., is "bored"), this is the amount of time it will wait before checking to see if it has new data to work on. This way, it does not use up CPU resources by checking for new work too often. When setting this property, be aware that it could add extra latency for components that do not constantly have work to do, as once they go into this "bored" state, they will wait this amount of time before checking for more work. The default value is 10 millis.
+max concurrent threads | The maximum number of threads any processor can have running at one time.
## FlowFile Repository
@@ -142,9 +143,9 @@ key | This is the password used to encrypt any sensitive property values
algorithm | The algorithm used to encrypt sensitive properties. The default value is `PBEWITHMD5AND256BITAES-CBC-OPENSSL`.
provider | The sensitive property provider. The default value is BC.
-## Processor Configuration
+## Processors
-The current implementation of MiNiFi supports one source processor. These properties are the basic configuration general to all processor implementations.
+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*
----------------------------------- | -------------
@@ -168,21 +169,24 @@ Within the Processor Configuration section, there is the `Properties` subsection
State File: ./conf/state/tail-file
Initial Start Position: Beginning of File
-## Connection Properties
+## Connections
-There is only one connection in this initial version of MiNiFi, between the source processor and the Remote Processing Group. These properties allow for customization of the of the connection.
+There can be multiple connections in this version of MiNiFi. The "Connections" subsection is a list of connections. Each connection must specify these properties.
*Property* | *Description*
-------------------- | -------------
-name | The name of what this connection will do. This is not used for any underlying implementation but solely for the users of this configuration and MiNiFi agent.
+name | The name of what this connection will do. This is used for the id of the connection so it must be unique.
+source name | The name of what of the processor that is the source for this connection.
+source relationship name | The name of the processors relationship to route to this connection
+destination name | The name of the component to receive this connection.
max work queue size | This property is the max number of FlowFiles that can be in the queue before back pressure is applied. When back pressure is applied the source processor will no longer be scheduled to run.
max work queue data size | This property specifies the maximum amount of data (in size) that should be queued up before applying back pressure. When back pressure is applied the source processor will no longer be scheduled to run.
flowfile expiration | Indicates how long FlowFiles are allowed to exist in the connection before be expired (automatically removed from the flow).
queue prioritizer class | This configuration option specifies the fully qualified java class path of a queue prioritizer to use. If no special prioritizer is desired then it should be left blank. An example value of this property is: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
-## Remote Processing Group
+## Remote Processing Groups
-MiNiFi is currently designed only to send back data to core NiFi instances using the Site to Site protocol. These properties configure the Remote Processing Group that links to that core instance.
+MiNiFi can be used to send data using the Site to Site protocol (via a Remote Processing Group) or a Processor. These properties configure the Remote Processing Groups that use Site-To-Site to send data to a core instance.
*Property* | *Description*
------------ | -------------
@@ -193,7 +197,7 @@ timeout | How long MiNiFi should wait before timing out the connection.
yield period | When communication with this Remote Processing Group fails, it will not be scheduled again for this amount of time.
-#### Input Port Subsection
+#### Input 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.
@@ -225,7 +229,7 @@ batch size | Specifies how many records to send in a single batch, at
# Example Config File
-Below is an example config YAML file to tail the minifi-app.log send it and provenance data back to a secure instance of NiFi.
+Below are two example config YAML files. The first tails the minifi-app.log, send the tailed log and provenance data back to a secure instance of NiFi. The second uses a series of processors to tail the app log, routes off only lines that contain "WriteAheadFlowFileRepository" and puts it as a file in the "./" directory.
``` yaml
@@ -250,6 +254,9 @@ FlowFile Repository:
out period: 5 sec
out threads: 4
+Provenance Repository:
+ provenance rollover time: 1 min
+
Content Repository:
content claim max appendable size: 10 MB
content claim max flow files: 100
@@ -273,49 +280,206 @@ Security Properties:
algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
provider: BC
-Processor Configuration:
- name: TailFile
- class: org.apache.nifi.processors.standard.TailFile
- max concurrent tasks: 1
+Processors:
+ - name: TailFile
+ class: org.apache.nifi.processors.standard.TailFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 1 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ Properties:
+ File to Tail: logs/minifi-app.log
+ Rolling Filename Pattern: minifi-app*
+ Initial Start Position: Beginning of File
+
+Connections:
+ - name: TailToS2S
+ source name: TailFile
+ source relationship name: success
+ destination name: 8644cbcc-a45c-40e0-964d-5e536e2ada61
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
+
+Remote Processing Groups:
+ - name: NiFi Flow
+ comment:
+ url: https://localhost:8090/nifi
+ timeout: 30 secs
+ yield period: 10 sec
+ Input Ports:
+ - id: 8644cbcc-a45c-40e0-964d-5e536e2ada61
+ name: tailed log
+ comments:
+ max concurrent tasks: 1
+ use compression: false
+
+Provenance Reporting:
+ comment:
scheduling strategy: TIMER_DRIVEN
- scheduling period: 1 sec
- penalization period: 30 sec
- yield period: 1 sec
- run duration nanos: 0
- auto-terminated relationships list:
- Properties:
- File to Tail: logs/minifi-app.log
- Rolling Filename Pattern: minifi-app*
- Initial Start Position: Beginning of File
+ scheduling period: 30 sec
+ destination url: https://localhost:8090/
+ port name: provenance
+ originating url: http://${hostname(true)}:8081/nifi
+ use compression: true
+ timeout: 30 secs
+ batch size: 1000
+```
-Connection Properties:
- name:
- max work queue size: 0
- max work queue data size: 0 MB
- flowfile expiration: 0 sec
- queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
-Remote Processing Group:
- name: NiFi Flow
+``` yaml
+Flow Controller:
+ name: MiNiFi Flow
comment:
- url: https://localhost:8090/nifi
- timeout: 30 secs
- yield period: 10 sec
- Input Port:
- id: 8644cbcc-a45c-40e0-964d-5e536e2ada61
- name: tailed log
- comments:
- max concurrent tasks: 1
- use compression: false
+
+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
+
+Component Status Repository:
+ buffer size: 1440
+ snapshot frequency: 1 min
+
+Security Properties:
+ keystore: /tmp/ssl/localhost-ks.jks
+ keystore type: JKS
+ keystore password: localtest
+ key password: localtest
+ truststore: /tmp/ssl/localhost-ts.jks
+ truststore type: JKS
+ truststore password: localtest
+ ssl protocol: TLS
+ Sensitive Props:
+ key:
+ algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
+ provider: BC
+
+Processors:
+ - name: TailAppLog
+ class: org.apache.nifi.processors.standard.TailFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 10 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ Properties:
+ File to Tail: logs/minifi-app.log
+ Rolling Filename Pattern: minifi-app*
+ Initial Start Position: Beginning of File
+ - name: SplitIntoSingleLines
+ class: org.apache.nifi.processors.standard.SplitText
+ 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:
+ - failure
+ - original
+ Properties:
+ Line Split Count: 1
+ Header Line Count: 0
+ Remove Trailing Newlines: true
+ - name: RouteErrors
+ class: org.apache.nifi.processors.standard.RouteText
+ 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:
+ - unmatched
+ - original
+ Properties:
+ Routing Strategy: Route to 'matched' if line matches all conditions
+ Matching Strategy: Contains
+ Character Set: UTF-8
+ Ignore Leading/Trailing Whitespace: true
+ Ignore Case: true
+ Grouping Regular Expression:
+ WALFFR: WriteAheadFlowFileRepository
+ - name: PutFile
+ class: org.apache.nifi.processors.standard.PutFile
+ 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:
+ - failure
+ - success
+ Properties:
+ Directory: ./
+ Conflict Resolution Strategy: replace
+ Create Missing Directories: true
+ Maximum File Count:
+ Last Modified Time:
+ Permissions:
+ Owner:
+ Group:
+
+Connections:
+ - name: TailToSplit
+ source name: TailAppLog
+ source relationship name: success
+ destination name: SplitIntoSingleLines
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
+ - name: SplitToRoute
+ source name: SplitIntoSingleLines
+ source relationship name: splits
+ destination name: RouteErrors
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
+ - name: RouteToS2S
+ source name: RouteErrors
+ source relationship name: matched
+ destination name: PutFile
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
Provenance Reporting:
comment:
scheduling strategy: TIMER_DRIVEN
scheduling period: 30 sec
- destination url: https://localhost:8090/
+ destination url: https://localhost:8080/
port name: provenance
originating url: http://${hostname(true)}:8081/nifi
use compression: true
timeout: 30 secs
batch size: 1000
-```
+```
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-resources/src/main/resources/conf/config.yml
----------------------------------------------------------------------
diff --git a/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-resources/src/main/resources/conf/config.yml b/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-resources/src/main/resources/conf/config.yml
index 5043900..89b1011 100644
--- a/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-resources/src/main/resources/conf/config.yml
+++ b/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-resources/src/main/resources/conf/config.yml
@@ -22,6 +22,7 @@ Core Properties:
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
@@ -34,7 +35,7 @@ FlowFile Repository:
out period: 5 sec
out threads: 4
-Content Repository:
+Content Repository:
content claim max appendable size: 10 MB
content claim max flow files: 100
always sync: false
@@ -60,45 +61,10 @@ Security Properties:
algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
provider: BC
-Processor Configuration:
- name:
- class:
- max concurrent tasks:
- scheduling strategy:
- scheduling period:
- penalization period:
- yield period:
- run duration nanos:
- auto-terminated relationships list:
- Properties:
+Processors:
-Connection Properties:
- name:
- max work queue size: 0
- max work queue data size: 0 MB
- flowfile expiration: 0 sec
- queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
+Connections:
-Remote Processing Group:
- name:
- comment:
- url:
- timeout:
- yield period:
- Input Port:
- id:
- name:
- comments:
- max concurrent tasks:
- use compression:
+Remote Processing Groups:
-Provenance Reporting:
- comment:
- scheduling strategy:
- scheduling period:
- destination url:
- port name:
- originating url:
- use compression:
- timeout:
- batch size:
\ No newline at end of file
+Provenance Reporting:
\ No newline at end of file
[2/3] nifi-minifi git commit: MINIFI-30 Added changes to the config
transformer to support multiple processors, multiple RPGs,
multiple Input Ports in RPGS, and greater validation on the config yaml
Posted by jp...@apache.org.
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ProvenanceReportingSchema.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ProvenanceReportingSchema.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ProvenanceReportingSchema.java
new file mode 100644
index 0000000..8e98ffe
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ProvenanceReportingSchema.java
@@ -0,0 +1,112 @@
+/*
+ * 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.bootstrap.util.schema;
+
+import org.apache.nifi.minifi.bootstrap.util.schema.common.BaseSchema;
+import org.apache.nifi.scheduling.SchedulingStrategy;
+
+import java.util.Map;
+
+import static org.apache.nifi.minifi.bootstrap.util.schema.RemoteProcessingGroupSchema.TIMEOUT_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.COMMENT_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.PROVENANCE_REPORTING_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.SCHEDULING_PERIOD_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.SCHEDULING_STRATEGY_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.USE_COMPRESSION_KEY;
+
+/**
+ *
+ */
+public class ProvenanceReportingSchema extends BaseSchema {
+ public static final String DESTINATION_URL_KEY = "destination url";
+ public static final String PORT_NAME_KEY = "port name";
+ public static final String ORIGINATING_URL_KEY = "originating url";
+ public static final String BATCH_SIZE_KEY = "batch size";
+
+ private String comment;
+ private String schedulingStrategy;
+ private String schedulingPeriod;
+ private String destinationUrl;
+ private String portName;
+ private String originatingUrl = "http://${hostname(true)}:8080/nifi";
+ private Boolean useCompression = true;
+ private String timeout = "30 secs";
+ private Number batchSize = 1000;
+
+ public ProvenanceReportingSchema() {
+ }
+
+ public ProvenanceReportingSchema(Map map) {
+ comment = getOptionalKeyAsType(map, COMMENT_KEY, String.class, PROVENANCE_REPORTING_KEY, null);
+
+ schedulingStrategy = getRequiredKeyAsType(map, SCHEDULING_STRATEGY_KEY, String.class, PROVENANCE_REPORTING_KEY);
+ try {
+ SchedulingStrategy.valueOf(schedulingStrategy);
+ } catch (IllegalArgumentException e) {
+ addValidationIssue(SCHEDULING_STRATEGY_KEY, PROVENANCE_REPORTING_KEY, "it is not a valid scheduling strategy");
+ }
+
+ schedulingPeriod = getRequiredKeyAsType(map, SCHEDULING_PERIOD_KEY, String.class, PROVENANCE_REPORTING_KEY);
+
+ destinationUrl = getRequiredKeyAsType(map, DESTINATION_URL_KEY, String.class, PROVENANCE_REPORTING_KEY);
+ portName = getRequiredKeyAsType(map, PORT_NAME_KEY, String.class, PROVENANCE_REPORTING_KEY);
+
+ originatingUrl = getOptionalKeyAsType(map, ORIGINATING_URL_KEY, String.class, PROVENANCE_REPORTING_KEY, "http://${hostname(true)}:8080/nifi");
+
+ useCompression = getOptionalKeyAsType(map, USE_COMPRESSION_KEY, Boolean.class, PROVENANCE_REPORTING_KEY, true);
+
+ timeout = getOptionalKeyAsType(map, TIMEOUT_KEY, String.class, PROVENANCE_REPORTING_KEY, "30 secs");
+
+ batchSize = getOptionalKeyAsType(map, BATCH_SIZE_KEY, Number.class, PROVENANCE_REPORTING_KEY, 1000);
+ }
+
+ public String getComment() {
+ return comment;
+ }
+
+ public String getSchedulingStrategy() {
+ return schedulingStrategy;
+ }
+
+ public String getSchedulingPeriod() {
+ return schedulingPeriod;
+ }
+
+ public String getDestinationUrl() {
+ return destinationUrl;
+ }
+
+ public String getPortName() {
+ return portName;
+ }
+
+ public String getOriginatingUrl() {
+ return originatingUrl;
+ }
+
+ public boolean getUseCompression() {
+ return useCompression;
+ }
+
+ public String getTimeout() {
+ return timeout;
+ }
+
+ public Number getBatchSize() {
+ return batchSize;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ProvenanceRepositorySchema.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ProvenanceRepositorySchema.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ProvenanceRepositorySchema.java
new file mode 100644
index 0000000..4a3b404
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ProvenanceRepositorySchema.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.bootstrap.util.schema;
+
+import org.apache.nifi.minifi.bootstrap.util.schema.common.BaseSchema;
+
+import java.util.Map;
+
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.PROVENANCE_REPO_KEY;
+
+public class ProvenanceRepositorySchema extends BaseSchema {
+
+
+ public static final String PROVENANCE_REPO_ROLLOVER_TIME_KEY = "provenance rollover time";
+
+ private String provenanceRepoRolloverTime = "1 min";
+
+ public ProvenanceRepositorySchema(){
+ }
+
+ public ProvenanceRepositorySchema(Map map) {
+ provenanceRepoRolloverTime = getOptionalKeyAsType(map, PROVENANCE_REPO_ROLLOVER_TIME_KEY, String.class, PROVENANCE_REPO_KEY, "1 min");
+ }
+
+ public String getProvenanceRepoRolloverTimeKey() {
+ return provenanceRepoRolloverTime;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/RemoteInputPortSchema.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/RemoteInputPortSchema.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/RemoteInputPortSchema.java
new file mode 100644
index 0000000..cfcb611
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/RemoteInputPortSchema.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.minifi.bootstrap.util.schema;
+
+import org.apache.nifi.minifi.bootstrap.util.schema.common.BaseSchema;
+
+import java.util.Map;
+
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.COMMENT_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.ID_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.INPUT_PORTS_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.MAX_CONCURRENT_TASKS_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.NAME_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.USE_COMPRESSION_KEY;
+
+/**
+ *
+ */
+public class RemoteInputPortSchema extends BaseSchema {
+
+ private String id;
+ private String name;
+ private String comment = "";
+ private Number maxConcurrentTasks = 1;
+ private Boolean useCompression = true;
+
+ public RemoteInputPortSchema() {
+ }
+
+ public RemoteInputPortSchema(Map map) {
+ id = getRequiredKeyAsType(map, ID_KEY, String.class, INPUT_PORTS_KEY);
+ name = getRequiredKeyAsType(map, NAME_KEY, String.class, INPUT_PORTS_KEY);
+
+ comment = getOptionalKeyAsType(map, COMMENT_KEY, String.class, INPUT_PORTS_KEY, "");
+ maxConcurrentTasks = getOptionalKeyAsType(map, MAX_CONCURRENT_TASKS_KEY, Number.class, INPUT_PORTS_KEY, 1);
+ useCompression = getOptionalKeyAsType(map, USE_COMPRESSION_KEY, Boolean.class, INPUT_PORTS_KEY, true);
+ }
+
+ public String getId() {
+ return id;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ 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/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/RemoteProcessingGroupSchema.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/RemoteProcessingGroupSchema.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/RemoteProcessingGroupSchema.java
new file mode 100644
index 0000000..e69f023
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/RemoteProcessingGroupSchema.java
@@ -0,0 +1,88 @@
+/*
+ * 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.bootstrap.util.schema;
+
+import org.apache.nifi.minifi.bootstrap.util.schema.common.BaseSchema;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.COMMENT_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.INPUT_PORTS_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.NAME_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.REMOTE_PROCESSING_GROUPS_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.YIELD_PERIOD_KEY;
+
+/**
+ *
+ */
+public class RemoteProcessingGroupSchema extends BaseSchema {
+ public static final String URL_KEY = "url";
+ public static final String TIMEOUT_KEY = "timeout";
+
+ private String name;
+ private String comment = "";
+ private String url;
+ private String timeout = "30 secs";
+ private String yieldPeriod = "10 sec";
+ private List<RemoteInputPortSchema> inputPorts;
+
+ public RemoteProcessingGroupSchema() {
+ }
+
+ public RemoteProcessingGroupSchema(Map map) {
+ name = getRequiredKeyAsType(map, NAME_KEY, String.class, REMOTE_PROCESSING_GROUPS_KEY);
+ url = getRequiredKeyAsType(map, URL_KEY, String.class, REMOTE_PROCESSING_GROUPS_KEY);
+
+ comment = getOptionalKeyAsType(map, COMMENT_KEY, String.class, REMOTE_PROCESSING_GROUPS_KEY, "");
+ timeout = getOptionalKeyAsType(map, TIMEOUT_KEY, String.class, REMOTE_PROCESSING_GROUPS_KEY, "30 secs");
+ yieldPeriod = getOptionalKeyAsType(map, YIELD_PERIOD_KEY, String.class, REMOTE_PROCESSING_GROUPS_KEY, "10 sec");
+
+ inputPorts = getRequiredKeyAsType(map, INPUT_PORTS_KEY, List.class, REMOTE_PROCESSING_GROUPS_KEY);
+ if (inputPorts != null) {
+ transformListToType(inputPorts, "input port", RemoteInputPortSchema.class, INPUT_PORTS_KEY);
+
+ for (RemoteInputPortSchema remoteInputPortSchema: inputPorts) {
+ addIssuesIfNotNull(remoteInputPortSchema);
+ }
+ }
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public String getComment() {
+ return comment;
+ }
+
+ public String getUrl() {
+ return url;
+ }
+
+ public String getTimeout() {
+ return timeout;
+ }
+
+ public String getYieldPeriod() {
+ return yieldPeriod;
+ }
+
+ public List<RemoteInputPortSchema> getInputPorts() {
+ return inputPorts;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/SecurityPropertiesSchema.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/SecurityPropertiesSchema.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/SecurityPropertiesSchema.java
new file mode 100644
index 0000000..96a9261
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/SecurityPropertiesSchema.java
@@ -0,0 +1,164 @@
+/*
+ * 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.bootstrap.util.schema;
+
+import org.apache.nifi.minifi.bootstrap.util.schema.common.BaseSchema;
+
+import java.util.Map;
+
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.SECURITY_PROPS_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.SENSITIVE_PROPS_KEY;
+
+/**
+ *
+ */
+public class SecurityPropertiesSchema extends BaseSchema {
+
+ public static final String KEYSTORE_KEY = "keystore";
+ public static final String KEYSTORE_TYPE_KEY = "keystore type";
+ public static final String KEYSTORE_PASSWORD_KEY = "keystore password";
+ public static final String KEY_PASSWORD_KEY = "key password";
+ public static final String TRUSTSTORE_KEY = "truststore";
+ public static final String TRUSTSTORE_TYPE_KEY = "truststore type";
+ public static final String TRUSTSTORE_PASSWORD_KEY = "truststore password";
+ public static final String SSL_PROTOCOL_KEY = "ssl protocol";
+
+ private String keystore;
+ private String keystoreType;
+ private String keystorePassword;
+ private String keyPassword;
+ private String truststore;
+ private String truststoreType;
+ private String truststorePassword;
+ private String sslProtocol;
+ private SensitivePropsSchema sensitiveProps;
+
+ public SecurityPropertiesSchema() {
+ sensitiveProps = new SensitivePropsSchema();
+ }
+
+ public SecurityPropertiesSchema(Map map) {
+ keystore = getOptionalKeyAsType(map, KEYSTORE_KEY, String.class, SECURITY_PROPS_KEY, null);
+
+ keystoreType = getOptionalKeyAsType(map, KEYSTORE_TYPE_KEY, String.class, SECURITY_PROPS_KEY, null);
+ if (keystoreType != null) {
+ if (validateStoreType(keystoreType)) {
+ addValidationIssue(KEYSTORE_TYPE_KEY, SECURITY_PROPS_KEY, "it is not a supported type (must be either PKCS12 or JKS format)");
+ }
+ }
+
+ keystorePassword = getOptionalKeyAsType(map, KEYSTORE_PASSWORD_KEY, String.class, SECURITY_PROPS_KEY, null);
+
+ keyPassword = getOptionalKeyAsType(map, KEY_PASSWORD_KEY, String.class, SECURITY_PROPS_KEY, null);
+
+ truststore = getOptionalKeyAsType(map, TRUSTSTORE_KEY, String.class, SECURITY_PROPS_KEY, null);
+
+ truststoreType = getOptionalKeyAsType(map, TRUSTSTORE_TYPE_KEY, String.class, SECURITY_PROPS_KEY, null);
+ if (truststoreType != null) {
+ if (validateStoreType(truststoreType)) {
+ addValidationIssue(TRUSTSTORE_TYPE_KEY, SECURITY_PROPS_KEY, "it is not a supported type (must be either PKCS12 or JKS format)");
+ }
+ }
+
+ truststorePassword = getOptionalKeyAsType(map, TRUSTSTORE_PASSWORD_KEY, String.class, SECURITY_PROPS_KEY, null);
+
+ sslProtocol = getOptionalKeyAsType(map, SSL_PROTOCOL_KEY, String.class, SECURITY_PROPS_KEY, null);
+ if (sslProtocol != null) {
+ switch (sslProtocol) {
+ case "SSL":
+ break;
+ case "SSLv2Hello":
+ break;
+ case "SSLv3":
+ break;
+ case "TLS":
+ break;
+ case "TLSv1":
+ break;
+ case "TLSv1.1":
+ break;
+ case "TLSv1.2":
+ break;
+ default:
+ addValidationIssue(SSL_PROTOCOL_KEY, SECURITY_PROPS_KEY, "it is not an allowable value of SSL protocol");
+ break;
+ }
+ }
+
+ if (sslProtocol != null) {
+ if (keystore == null) {
+ validationIssues.add("When the '" + SSL_PROTOCOL_KEY + "' key of '" + SECURITY_PROPS_KEY + "' is set, the '" + KEYSTORE_KEY + "' must also be set");
+ } else if (keystoreType == null || keystorePassword == null || keyPassword == null) {
+ validationIssues.add("When the '" + KEYSTORE_KEY + "' key of '" + SECURITY_PROPS_KEY + "' is set, the '" + KEYSTORE_TYPE_KEY + "', '" + KEYSTORE_PASSWORD_KEY +
+ "' and '" + KEY_PASSWORD_KEY + "' all must also be set");
+ }
+
+ if (truststore != null && (truststoreType == null || truststorePassword == null)) {
+ validationIssues.add("When the '" + TRUSTSTORE_KEY + "' key of '" + SECURITY_PROPS_KEY + "' is set, the '" + TRUSTSTORE_TYPE_KEY + "' and '" +
+ TRUSTSTORE_PASSWORD_KEY + "' must also be set");
+ }
+ }
+
+ sensitiveProps = getMapAsType(map, SENSITIVE_PROPS_KEY, SensitivePropsSchema.class, SECURITY_PROPS_KEY, false);
+
+ addIssuesIfNotNull(sensitiveProps);
+ }
+
+ private boolean validateStoreType(String store) {
+ return !store.isEmpty() && !(store.equalsIgnoreCase("JKS") || store.equalsIgnoreCase("PKCS12"));
+ }
+
+ public boolean useSSL() {
+ return sslProtocol != null && !(sslProtocol.isEmpty());
+ }
+
+ public String getKeystore() {
+ return keystore;
+ }
+
+ public String getKeystoreType() {
+ return keystoreType;
+ }
+
+ public String getKeystorePassword() {
+ return keystorePassword;
+ }
+
+ public String getKeyPassword() {
+ return keyPassword;
+ }
+
+ public String getTruststore() {
+ return truststore;
+ }
+
+ public String getTruststoreType() {
+ return truststoreType;
+ }
+
+ public String getTruststorePassword() {
+ return truststorePassword;
+ }
+
+ public String getSslProtocol() {
+ return sslProtocol;
+ }
+
+ public SensitivePropsSchema getSensitiveProps() {
+ return sensitiveProps;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/SensitivePropsSchema.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/SensitivePropsSchema.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/SensitivePropsSchema.java
new file mode 100644
index 0000000..36cd07a
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/SensitivePropsSchema.java
@@ -0,0 +1,59 @@
+/*
+ * 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.bootstrap.util.schema;
+
+import org.apache.nifi.minifi.bootstrap.util.schema.common.BaseSchema;
+
+import java.util.Map;
+
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.SENSITIVE_PROPS_KEY;
+
+/**
+ *
+ */
+public class SensitivePropsSchema extends BaseSchema {
+ public static final String SENSITIVE_PROPS_KEY_KEY = "key";
+ public static final String SENSITIVE_PROPS_ALGORITHM_KEY = "algorithm";
+ public static final String SENSITIVE_PROPS_PROVIDER_KEY = "provider";
+
+ private String key = "";
+ private String algorithm = "PBEWITHMD5AND256BITAES-CBC-OPENSSL";
+ private String provider = "BC";
+
+ public SensitivePropsSchema() {
+ }
+
+ public SensitivePropsSchema(Map map) {
+ key = getOptionalKeyAsType(map, SENSITIVE_PROPS_KEY_KEY, String.class, SENSITIVE_PROPS_KEY, "");
+
+ algorithm = getOptionalKeyAsType(map, SENSITIVE_PROPS_ALGORITHM_KEY, String.class, SENSITIVE_PROPS_KEY, "PBEWITHMD5AND256BITAES-CBC-OPENSSL");
+
+ provider = getOptionalKeyAsType(map, SENSITIVE_PROPS_PROVIDER_KEY, String.class, SENSITIVE_PROPS_KEY, "BC");
+ }
+
+ public String getKey() {
+ return key;
+ }
+
+ public String getAlgorithm() {
+ return algorithm;
+ }
+
+ public String getProvider() {
+ return provider;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/SwapSchema.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/SwapSchema.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/SwapSchema.java
new file mode 100644
index 0000000..ec6563f
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/SwapSchema.java
@@ -0,0 +1,75 @@
+/*
+ * 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.bootstrap.util.schema;
+
+import org.apache.nifi.minifi.bootstrap.util.schema.common.BaseSchema;
+
+import java.util.Map;
+
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.SWAP_PROPS_KEY;
+
+/**
+ *
+ */
+public class SwapSchema extends BaseSchema {
+ public static final String THRESHOLD_KEY = "threshold";
+ public static final String IN_PERIOD_KEY = "in period";
+ public static final String IN_THREADS_KEY = "in threads";
+ public static final String OUT_PERIOD_KEY = "out period";
+ public static final String OUT_THREADS_KEY = "out threads";
+
+ private Number threshold = 20000;
+ private String inPeriod = "5 sec";
+ private Number inThreads = 1;
+ private String outPeriod = "5 sec";
+ private Number outThreads = 4;
+
+ public SwapSchema() {
+ }
+
+ public SwapSchema(Map map) {
+ threshold = getOptionalKeyAsType(map, THRESHOLD_KEY, Number.class, SWAP_PROPS_KEY, 20000);
+
+ inPeriod = getOptionalKeyAsType(map, IN_PERIOD_KEY, String.class, SWAP_PROPS_KEY, "5 sec");
+
+ inThreads = getOptionalKeyAsType(map, IN_THREADS_KEY, Number.class, SWAP_PROPS_KEY, 1);
+
+ outPeriod = getOptionalKeyAsType(map, OUT_PERIOD_KEY, String.class, SWAP_PROPS_KEY, "5 sec");
+
+ outThreads = getOptionalKeyAsType(map, OUT_THREADS_KEY, Number.class, SWAP_PROPS_KEY, 4);
+ }
+
+ public Number getThreshold() {
+ return threshold;
+ }
+
+ public String getInPeriod() {
+ return inPeriod;
+ }
+
+ public Number getInThreads() {
+ return inThreads;
+ }
+
+ public String getOutPeriod() {
+ return outPeriod;
+ }
+
+ public Number getOutThreads() {
+ return outThreads;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/common/BaseSchema.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/common/BaseSchema.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/common/BaseSchema.java
new file mode 100644
index 0000000..a29be6d
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/common/BaseSchema.java
@@ -0,0 +1,134 @@
+/*
+ * 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.bootstrap.util.schema.common;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+public class BaseSchema {
+
+ /******* Validation Issue helper methods *******/
+ public List<String> validationIssues = new LinkedList<>();
+
+ public boolean isValid() {
+ return validationIssues.isEmpty();
+ }
+
+ public List<String> getValidationIssues() {
+ return validationIssues;
+ }
+
+ public String getValidationIssuesAsString() {
+ StringBuilder stringBuilder = new StringBuilder();
+ boolean first = true;
+ for (String validationIssue : validationIssues) {
+ if (!first) {
+ stringBuilder.append(", ");
+ }
+ stringBuilder.append("[");
+ stringBuilder.append(validationIssue);
+ stringBuilder.append("]");
+ first = false;
+ }
+ return stringBuilder.toString();
+ }
+
+ public void addValidationIssue(String keyName, String wrapperName, String reason) {
+ validationIssues.add("'" + keyName + "' in section '" + wrapperName + "' because " + reason);
+ }
+
+ public void addIssuesIfNotNull(BaseSchema baseSchema) {
+ if (baseSchema != null) {
+ validationIssues.addAll(baseSchema.getValidationIssues());
+ }
+ }
+
+ /******* Value Access/Interpretation helper methods *******/
+ public <T> T getOptionalKeyAsType(Map valueMap, String key, Class targetClass, String wrapperName, T defaultValue) {
+ return getKeyAsType(valueMap, key, targetClass, wrapperName, false, defaultValue);
+ }
+
+ public <T> T getRequiredKeyAsType(Map valueMap, String key, Class targetClass, String wrapperName) {
+ return getKeyAsType(valueMap, key, targetClass, wrapperName, true, null);
+ }
+
+ <T> T getKeyAsType(Map valueMap, String key, Class targetClass, String wrapperName, boolean required, T defaultValue) {
+ Object value = valueMap.get(key);
+ if (value == null) {
+ if (defaultValue != null) {
+ return defaultValue;
+ } else if(required) {
+ addValidationIssue(key, wrapperName, "it was not found and it is required");
+ }
+ } else {
+ if (targetClass.isInstance(value)) {
+ return (T) value;
+ } else {
+ addValidationIssue(key, wrapperName, "it is found but could not be parsed as a " + targetClass.getSimpleName());
+ }
+ }
+ return null;
+ }
+
+
+ public <T> T getMapAsType(Map valueMap, String key, Class targetClass, String wrapperName, boolean required) {
+ Object obj = valueMap.get(key);
+ return interpretValueAsType(obj, key, targetClass, wrapperName, required);
+ }
+
+ public void transformListToType(List list, String simpleListType, Class targetClass, String wrapperName){
+ for (int i = 0; i < list.size(); i++) {
+ Object obj = interpretValueAsType(list.get(i), simpleListType + " number " + i, targetClass, wrapperName, false);
+ if (obj != null) {
+ list.set(i, obj);
+ }
+ }
+ }
+
+ private <T> T interpretValueAsType(Object obj, String key, Class targetClass, String wrapperName, boolean required) {
+ if (obj == null) {
+ if (required){
+ addValidationIssue(key, wrapperName, "it is a required property but was not found");
+ } else {
+ try {
+ return (T) targetClass.newInstance();
+ } catch (InstantiationException | IllegalAccessException e) {
+ addValidationIssue(key, wrapperName, "it is optional and when attempting to create it the following exception was thrown:" + e.getMessage());
+ }
+ }
+ } else if (obj instanceof Map) {
+ Constructor<?> constructor;
+ try {
+ constructor = targetClass.getConstructor(Map.class);
+ return (T) constructor.newInstance((Map) obj);
+ } catch (NoSuchMethodException | IllegalAccessException | InstantiationException | InvocationTargetException e) {
+ addValidationIssue(key, wrapperName, "it is found as a map and when attempting to interpret it the following exception was thrown:" + e.getMessage());
+ }
+ } else {
+ try {
+ return (T) obj;
+ } catch (ClassCastException e) {
+ addValidationIssue(key, wrapperName, "it is found but could not be parsed as a map");
+ }
+ }
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/common/CommonPropertyKeys.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/common/CommonPropertyKeys.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/common/CommonPropertyKeys.java
new file mode 100644
index 0000000..fc0a681
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/common/CommonPropertyKeys.java
@@ -0,0 +1,49 @@
+/*
+ * 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.bootstrap.util.schema.common;
+
+public class CommonPropertyKeys {
+ public static final String CORE_PROPS_KEY = "Core Properties";
+ public static final String FLOWFILE_REPO_KEY = "FlowFile Repository";
+ public static final String SWAP_PROPS_KEY = "Swap";
+ public static final String FLOW_CONTROLLER_PROPS_KEY = "Flow Controller";
+ public static final String CONTENT_REPO_KEY = "Content Repository";
+ public static final String COMPONENT_STATUS_REPO_KEY = "Component Status Repository";
+ public static final String SECURITY_PROPS_KEY = "Security Properties";
+ public static final String SENSITIVE_PROPS_KEY = "Sensitive Props";
+ public static final String PROCESSORS_KEY = "Processors";
+ public static final String CONNECTIONS_KEY = "Connections";
+ public static final String PROVENANCE_REPORTING_KEY = "Provenance Reporting";
+ public static final String REMOTE_PROCESSING_GROUPS_KEY = "Remote Processing Groups";
+ public static final String INPUT_PORTS_KEY = "Input Ports";
+ public static final String PROVENANCE_REPO_KEY = "Provenance Repository";
+
+
+ public static final String NAME_KEY = "name";
+ public static final String COMMENT_KEY = "comment";
+ public static final String ALWAYS_SYNC_KEY = "always sync";
+ public static final String YIELD_PERIOD_KEY = "yield period";
+ public static final String MAX_CONCURRENT_THREADS_KEY = "max concurrent threads";
+ public static final String MAX_CONCURRENT_TASKS_KEY = "max concurrent tasks";
+ public static final String ID_KEY = "id";
+ 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";
+
+ private CommonPropertyKeys() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/TestConfigTransformer.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/TestConfigTransformer.java b/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/TestConfigTransformer.java
index d0a7d71..cb373bf 100644
--- a/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/TestConfigTransformer.java
+++ b/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/TestConfigTransformer.java
@@ -17,12 +17,13 @@
package org.apache.nifi.minifi.bootstrap.util;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.File;
import java.io.FileInputStream;
-import org.apache.nifi.minifi.bootstrap.configuration.ConfigurationChangeException;
+import org.apache.nifi.minifi.bootstrap.exception.InvalidConfigurationException;
import org.junit.Assert;
import org.junit.Test;
@@ -30,7 +31,6 @@ public class TestConfigTransformer {
@Test
public void doesTransformFile() throws Exception {
-
ConfigTransformer.transformConfigFile("./src/test/resources/config.yml", "./target/");
File nifiPropertiesFile = new File("./target/nifi.properties");
@@ -66,7 +66,6 @@ public class TestConfigTransformer {
@Test
public void doesTransformOnDefaultFile() throws Exception {
-
ConfigTransformer.transformConfigFile("./src/test/resources/default.yml", "./target/");
File nifiPropertiesFile = new File("./target/nifi.properties");
@@ -82,19 +81,122 @@ public class TestConfigTransformer {
flowXml.deleteOnExit();
}
- @Test(expected = IllegalArgumentException.class)
- public void handleTransformInvalidFile() throws Exception {
+ @Test
+ public void doesTransformOnMultipleProcessors() throws Exception {
+ ConfigTransformer.transformConfigFile("./src/test/resources/config-multiple-processors.yml", "./target/");
+ File nifiPropertiesFile = new File("./target/nifi.properties");
+
+ assertTrue(nifiPropertiesFile.exists());
+ assertTrue(nifiPropertiesFile.canRead());
+
+ nifiPropertiesFile.deleteOnExit();
+
+ File flowXml = new File("./target/flow.xml.gz");
+ assertTrue(flowXml.exists());
+ assertTrue(flowXml.canRead());
+
+ flowXml.deleteOnExit();
+ }
+
+ @Test
+ public void doesTransformOnMultipleRemoteProcessingGroups() throws Exception {
+ ConfigTransformer.transformConfigFile("./src/test/resources/config-multiple-RPGs.yml", "./target/");
+ File nifiPropertiesFile = new File("./target/nifi.properties");
+
+ assertTrue(nifiPropertiesFile.exists());
+ assertTrue(nifiPropertiesFile.canRead());
- ConfigTransformer.transformConfigFile("./src/test/resources/config-invalid.yml", "./target/");
+ nifiPropertiesFile.deleteOnExit();
+
+ File flowXml = new File("./target/flow.xml.gz");
+ assertTrue(flowXml.exists());
+ assertTrue(flowXml.canRead());
- Assert.fail("Invalid configuration file was not detected.");
+ flowXml.deleteOnExit();
}
- @Test(expected = ConfigurationChangeException.class)
+ @Test
+ public void doesTransformOnMultipleInputPorts() throws Exception {
+ ConfigTransformer.transformConfigFile("./src/test/resources/config-multiple-input-ports.yml", "./target/");
+ File nifiPropertiesFile = new File("./target/nifi.properties");
+
+ assertTrue(nifiPropertiesFile.exists());
+ assertTrue(nifiPropertiesFile.canRead());
+
+ nifiPropertiesFile.deleteOnExit();
+
+ File flowXml = new File("./target/flow.xml.gz");
+ assertTrue(flowXml.exists());
+ assertTrue(flowXml.canRead());
+
+ flowXml.deleteOnExit();
+ }
+
+ @Test
+ public void doesTransformOnMinimal() throws Exception {
+ ConfigTransformer.transformConfigFile("./src/test/resources/config-minimal.yml", "./target/");
+ File nifiPropertiesFile = new File("./target/nifi.properties");
+
+ assertTrue(nifiPropertiesFile.exists());
+ assertTrue(nifiPropertiesFile.canRead());
+
+ nifiPropertiesFile.deleteOnExit();
+
+ File flowXml = new File("./target/flow.xml.gz");
+ assertTrue(flowXml.exists());
+ assertTrue(flowXml.canRead());
+
+ flowXml.deleteOnExit();
+ }
+
+ @Test
+ public void handleTransformInvalidFile() throws Exception {
+ try {
+ ConfigTransformer.transformConfigFile("./src/test/resources/config-invalid.yml", "./target/");
+ Assert.fail("Invalid configuration file was not detected.");
+ } catch (InvalidConfigurationException e){
+ assertEquals("Provided YAML configuration is not a Map", e.getMessage());
+ }
+ }
+
+ @Test
+ public void handleTransformMalformedField() throws Exception {
+ try {
+ ConfigTransformer.transformConfigFile("./src/test/resources/config-malformed-field.yml", "./target/");
+ Assert.fail("Invalid configuration file was not detected.");
+ } catch (InvalidConfigurationException e){
+ assertEquals("Failed to transform config file due to:['threshold' in section 'Swap' because it is found but could not be parsed as a Number]", e.getMessage());
+ }
+ }
+
+ @Test
public void handleTransformEmptyFile() throws Exception {
+ try {
+ ConfigTransformer.transformConfigFile("./src/test/resources/config-empty.yml", "./target/");
+ Assert.fail("Invalid configuration file was not detected.");
+ } catch (InvalidConfigurationException e){
+ assertEquals("Provided YAML configuration is not a Map", e.getMessage());
+ }
+ }
- ConfigTransformer.transformConfigFile("./src/test/resources/config-empty.yml", "./target/");
+ @Test
+ public void handleTransformFileMissingRequiredField() throws Exception {
+ try {
+ ConfigTransformer.transformConfigFile("./src/test/resources/config-missing-required-field.yml", "./target/");
+ Assert.fail("Invalid configuration file was not detected.");
+ } catch (InvalidConfigurationException e){
+ assertEquals("Failed to transform config file due to:['class' in section 'Processors' because it was not found and it is required]", e.getMessage());
+ }
+ }
- Assert.fail("Invalid configuration file was not detected.");
+ @Test
+ public void handleTransformFileMultipleProblems() throws Exception {
+ try {
+ ConfigTransformer.transformConfigFile("./src/test/resources/config-multiple-problems.yml", "./target/");
+ Assert.fail("Invalid configuration file was not detected.");
+ } catch (InvalidConfigurationException e){
+ assertEquals("Failed to transform config file due to:['scheduling strategy' in section 'Provenance Reporting' because it is not a valid scheduling strategy], ['class' in section " +
+ "'Processors' because it was not found and it is required], ['source name' in section 'Connections' because it was not found and it is required]", e.getMessage());
+ }
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/test/resources/config-empty.yml
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/resources/config-empty.yml b/minifi-bootstrap/src/test/resources/config-empty.yml
index fbbbeb9..8141ad7 100644
--- a/minifi-bootstrap/src/test/resources/config-empty.yml
+++ b/minifi-bootstrap/src/test/resources/config-empty.yml
@@ -11,8 +11,4 @@
# 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.
-
-Flow Controller:
- name: MiNiFi Flow
- comment:
\ No newline at end of file
+# limitations under the License.
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/test/resources/config-malformed-field.yml
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/resources/config-malformed-field.yml b/minifi-bootstrap/src/test/resources/config-malformed-field.yml
new file mode 100644
index 0000000..ecce9a8
--- /dev/null
+++ b/minifi-bootstrap/src/test/resources/config-malformed-field.yml
@@ -0,0 +1,109 @@
+# 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.
+
+Flow Controller:
+ name: MiNiFi Flow
+ 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 should be a number
+ threshold: this is not a number
+ 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
+
+Component Status Repository:
+ buffer size: 1440
+ snapshot frequency: 1 min
+
+Security Properties:
+ keystore: /tmp/ssl/localhost-ks.jks
+ keystore type: JKS
+ keystore password: localtest
+ key password: localtest
+ truststore: /tmp/ssl/localhost-ts.jks
+ truststore type: JKS
+ truststore password: localtest
+ ssl protocol: TLS
+ Sensitive Props:
+ key:
+ algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
+ provider: BC
+
+Processors:
+ - name: TailFile
+ class: org.apache.nifi.processors.standard.TailFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 1 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ Properties:
+ File to Tail: logs/minifi-app.log
+ Rolling Filename Pattern: minifi-app*
+ Initial Start Position: Beginning of File
+
+Connections:
+ - name: TailToS2S
+ source name: TailFile
+ source relationship name: success
+ destination name: 8644cbcc-a45c-40e0-964d-5e536e2ada61
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
+
+Remote Processing Groups:
+ - name: NiFi Flow
+ comment:
+ url: https://localhost:8090/nifi
+ timeout: 30 secs
+ yield period: 10 sec
+ Input Ports:
+ - id: 8644cbcc-a45c-40e0-964d-5e536e2ada61
+ name: tailed log
+ comments:
+ max concurrent tasks: 1
+ use compression: false
+
+Provenance Reporting:
+ comment:
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 30 sec
+ destination url: https://localhost:8090/
+ port name: provenance
+ originating url: http://${hostname(true)}:8081/nifi
+ use compression: true
+ timeout: 30 secs
+ batch size: 1000
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/test/resources/config-minimal.yml
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/resources/config-minimal.yml b/minifi-bootstrap/src/test/resources/config-minimal.yml
new file mode 100644
index 0000000..573fef6
--- /dev/null
+++ b/minifi-bootstrap/src/test/resources/config-minimal.yml
@@ -0,0 +1,35 @@
+# 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.
+
+Flow Controller:
+ name: MiNiFi Flow
+
+# When creating the Flow (not doing the transform) these processors will be invalid due to not having the necesary properties/auto-terminated relationships
+Processors:
+ - name: TailAppLog
+ class: org.apache.nifi.processors.standard.TailFile
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 10 sec
+ - name: PutFile
+ class: org.apache.nifi.processors.standard.PutFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 0 sec
+
+Connections:
+ - name: TailToSplit
+ source name: RouteErrors
+ source relationship name: success
+ destination name: PutFile
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/test/resources/config-missing-required-field.yml
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/resources/config-missing-required-field.yml b/minifi-bootstrap/src/test/resources/config-missing-required-field.yml
new file mode 100644
index 0000000..9339d8c
--- /dev/null
+++ b/minifi-bootstrap/src/test/resources/config-missing-required-field.yml
@@ -0,0 +1,109 @@
+# 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.
+
+Flow Controller:
+ name: MiNiFi Flow
+ 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
+
+Component Status Repository:
+ buffer size: 1440
+ snapshot frequency: 1 min
+
+Security Properties:
+ keystore: /tmp/ssl/localhost-ks.jks
+ keystore type: JKS
+ keystore password: localtest
+ key password: localtest
+ truststore: /tmp/ssl/localhost-ts.jks
+ truststore type: JKS
+ truststore password: localtest
+ ssl protocol: TLS
+ Sensitive Props:
+ key:
+ algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
+ provider: BC
+
+Processors:
+ - name: TailFile
+ # 'class' is a required field
+ # class: org.apache.nifi.processors.standard.TailFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 1 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ Properties:
+ File to Tail: logs/minifi-app.log
+ Rolling Filename Pattern: minifi-app*
+ Initial Start Position: Beginning of File
+
+Connections:
+ - name: TailToS2S
+ source name: TailFile
+ source relationship name: success
+ destination name: 8644cbcc-a45c-40e0-964d-5e536e2ada61
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
+
+Remote Processing Groups:
+ - name: NiFi Flow
+ comment:
+ url: https://localhost:8090/nifi
+ timeout: 30 secs
+ yield period: 10 sec
+ Input Ports:
+ - id: 8644cbcc-a45c-40e0-964d-5e536e2ada61
+ name: tailed log
+ comments:
+ max concurrent tasks: 1
+ use compression: false
+
+Provenance Reporting:
+ comment:
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 30 sec
+ destination url: https://localhost:8090/
+ port name: provenance
+ originating url: http://${hostname(true)}:8081/nifi
+ use compression: true
+ timeout: 30 secs
+ batch size: 1000
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/test/resources/config-multiple-RPGs.yml
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/resources/config-multiple-RPGs.yml b/minifi-bootstrap/src/test/resources/config-multiple-RPGs.yml
new file mode 100644
index 0000000..48ce749
--- /dev/null
+++ b/minifi-bootstrap/src/test/resources/config-multiple-RPGs.yml
@@ -0,0 +1,127 @@
+# 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.
+
+Flow Controller:
+ name: MiNiFi Flow
+ 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
+
+Component Status Repository:
+ buffer size: 1440
+ snapshot frequency: 1 min
+
+Security Properties:
+ keystore: /tmp/ssl/localhost-ks.jks
+ keystore type: JKS
+ keystore password: localtest
+ key password: localtest
+ truststore: /tmp/ssl/localhost-ts.jks
+ truststore type: JKS
+ truststore password: localtest
+ ssl protocol: TLS
+ Sensitive Props:
+ key:
+ algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
+ provider: BC
+
+Processors:
+ - name: TailAppLog
+ class: org.apache.nifi.processors.standard.TailFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 10 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ Properties:
+ File to Tail: logs/minifi-app.log
+ Rolling Filename Pattern: minifi-app*
+ Initial Start Position: Beginning of File
+
+Connections:
+ - name: TailToS2S
+ source name: TailFile
+ source relationship name: success
+ destination name: dgs809qe-9qw7-q985-jl1u-y789ohi87g78
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
+ - name: TailToS2S-2
+ source name: TailFile
+ source relationship name: success
+ destination name: 8644cbcc-a45c-40e0-964d-5e536e2ada61
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
+
+Remote Processing Groups:
+ - name: NiFi Flow
+ comment:
+ url: https://localhost:8090/nifi
+ timeout: 30 secs
+ yield period: 10 sec
+ Input Ports:
+ - id: 8644cbcc-a45c-40e0-964d-5e536e2ada61
+ name: tailed log
+ comments:
+ max concurrent tasks: 1
+ use compression: false
+ - name: NiFi Flow 2
+ comment:
+ url: https://localhost:8089/nifi
+ timeout: 30 secs
+ yield period: 10 sec
+ Input Ports:
+ - id: dgs809qe-9qw7-q985-jl1u-y789ohi87g78
+ name: tailed log
+ comments:
+ max concurrent tasks: 1
+ use compression: false
+
+Provenance Reporting:
+ comment:
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 30 sec
+ destination url: https://localhost:8080/
+ port name: provenance
+ originating url: http://${hostname(true)}:8081/nifi
+ use compression: true
+ timeout: 30 secs
+ batch size: 1000
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/test/resources/config-multiple-input-ports.yml
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/resources/config-multiple-input-ports.yml b/minifi-bootstrap/src/test/resources/config-multiple-input-ports.yml
new file mode 100644
index 0000000..9498eae
--- /dev/null
+++ b/minifi-bootstrap/src/test/resources/config-multiple-input-ports.yml
@@ -0,0 +1,121 @@
+# 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.
+
+Flow Controller:
+ name: MiNiFi Flow
+ 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
+
+Component Status Repository:
+ buffer size: 1440
+ snapshot frequency: 1 min
+
+Security Properties:
+ keystore: /tmp/ssl/localhost-ks.jks
+ keystore type: JKS
+ keystore password: localtest
+ key password: localtest
+ truststore: /tmp/ssl/localhost-ts.jks
+ truststore type: JKS
+ truststore password: localtest
+ ssl protocol: TLS
+ Sensitive Props:
+ key:
+ algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
+ provider: BC
+
+Processors:
+ - name: TailAppLog
+ class: org.apache.nifi.processors.standard.TailFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 10 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ Properties:
+ File to Tail: logs/minifi-app.log
+ Rolling Filename Pattern: minifi-app*
+ Initial Start Position: Beginning of File
+
+Connections:
+ - name: TailToS2S
+ source name: TailFile
+ source relationship name: success
+ destination name: dgs809qe-9qw7-q985-jl1u-y789ohi87g78
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
+ - name: TailToS2S-2
+ source name: TailFile
+ source relationship name: success
+ destination name: 8644cbcc-a45c-40e0-964d-5e536e2ada61
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
+
+Remote Processing Groups:
+ - name: NiFi Flow
+ comment:
+ url: https://localhost:8090/nifi
+ timeout: 30 secs
+ yield period: 10 sec
+ Input Ports:
+ - id: 8644cbcc-a45c-40e0-964d-5e536e2ada61
+ name: tailed log
+ comments:
+ max concurrent tasks: 1
+ use compression: false
+ - id: dgs809qe-9qw7-q985-jl1u-y789ohi87g78
+ name: tailed log2
+ comments:
+ max concurrent tasks: 1
+ use compression: false
+
+Provenance Reporting:
+ comment:
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 30 sec
+ destination url: https://localhost:8080/
+ port name: provenance
+ originating url: http://${hostname(true)}:8081/nifi
+ use compression: true
+ timeout: 30 secs
+ batch size: 1000
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/test/resources/config-multiple-problems.yml
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/resources/config-multiple-problems.yml b/minifi-bootstrap/src/test/resources/config-multiple-problems.yml
new file mode 100644
index 0000000..c30bfb9
--- /dev/null
+++ b/minifi-bootstrap/src/test/resources/config-multiple-problems.yml
@@ -0,0 +1,111 @@
+# 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.
+
+Flow Controller:
+ name: MiNiFi Flow
+ 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
+
+Component Status Repository:
+ buffer size: 1440
+ snapshot frequency: 1 min
+
+Security Properties:
+ keystore: /tmp/ssl/localhost-ks.jks
+ keystore type: JKS
+ keystore password: localtest
+ key password: localtest
+ truststore: /tmp/ssl/localhost-ts.jks
+ truststore type: JKS
+ truststore password: localtest
+ ssl protocol: TLS
+ Sensitive Props:
+ key:
+ algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
+ provider: BC
+
+Processors:
+ - name: TailFile
+ # 'class' is a required field
+ # class: org.apache.nifi.processors.standard.TailFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 1 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ Properties:
+ File to Tail: logs/minifi-app.log
+ Rolling Filename Pattern: minifi-app*
+ Initial Start Position: Beginning of File
+
+Connections:
+ - name: TailToS2S
+ # Source name is needed
+ source name:
+ source relationship name: success
+ destination name: 8644cbcc-a45c-40e0-964d-5e536e2ada61
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ queue prioritizer class:
+
+Remote Processing Groups:
+ - name: NiFi Flow
+ comment:
+ url: https://localhost:8090/nifi
+ timeout: 30 secs
+ yield period: 10 sec
+ Input Ports:
+ - id: 8644cbcc-a45c-40e0-964d-5e536e2ada61
+ name: tailed log
+ comments:
+ max concurrent tasks: 1
+ use compression: false
+
+Provenance Reporting:
+ comment:
+ # Not a valid strategy
+ scheduling strategy: ALL THE TIEMS!
+ scheduling period: 30 sec
+ destination url: https://localhost:8090/
+ port name: provenance
+ originating url: http://${hostname(true)}:8081/nifi
+ use compression: true
+ timeout: 30 secs
+ batch size: 1000
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/test/resources/config-multiple-processors.yml
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/resources/config-multiple-processors.yml b/minifi-bootstrap/src/test/resources/config-multiple-processors.yml
new file mode 100644
index 0000000..5a3cf92
--- /dev/null
+++ b/minifi-bootstrap/src/test/resources/config-multiple-processors.yml
@@ -0,0 +1,165 @@
+# 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.
+
+Flow Controller:
+ name: MiNiFi Flow
+ 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
+
+Component Status Repository:
+ buffer size: 1440
+ snapshot frequency: 1 min
+
+Security Properties:
+ keystore: /tmp/ssl/localhost-ks.jks
+ keystore type: JKS
+ keystore password: localtest
+ key password: localtest
+ truststore: /tmp/ssl/localhost-ts.jks
+ truststore type: JKS
+ truststore password: localtest
+ ssl protocol: TLS
+ Sensitive Props:
+ key:
+ algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
+ provider: BC
+
+Processors:
+ - name: TailAppLog
+ class: org.apache.nifi.processors.standard.TailFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 10 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ Properties:
+ File to Tail: logs/minifi-app.log
+ Rolling Filename Pattern: minifi-app*
+ Initial Start Position: Beginning of File
+ - name: SplitIntoSingleLines
+ class: org.apache.nifi.processors.standard.SplitText
+ 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:
+ - failure
+ - original
+ Properties:
+ Line Split Count: 1
+ Header Line Count: 0
+ Remove Trailing Newlines: true
+ - name: RouteErrors
+ class: org.apache.nifi.processors.standard.RouteText
+ 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:
+ - unmatched
+ - original
+ Properties:
+ Routing Strategy: Route to 'matched' if line matches all conditions
+ Matching Strategy: Contains
+ Character Set: UTF-8
+ Ignore Leading/Trailing Whitespace: true
+ Ignore Case: true
+ Grouping Regular Expression:
+ WALFFR: WriteAheadFlowFileRepository
+ - name: PutFile
+ class: org.apache.nifi.processors.standard.PutFile
+ 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:
+ - failure
+ - success
+ Properties:
+ Directory: ./
+ Conflict Resolution Strategy: replace
+ Create Missing Directories: true
+ Maximum File Count:
+ Last Modified Time:
+ Permissions:
+ Owner:
+ Group:
+
+Connections:
+ - name: TailToSplit
+ source name: TailAppLog
+ source relationship name: success
+ destination name: SplitIntoSingleLines
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
+ - name: SplitToRoute
+ source name: SplitIntoSingleLines
+ source relationship name: splits
+ destination name: RouteErrors
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
+ - name: RouteToS2S
+ source name: RouteErrors
+ source relationship name: matched
+ destination name: PutFile
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
+
+Provenance Reporting:
+ comment:
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 30 sec
+ destination url: https://localhost:8080/
+ port name: provenance
+ originating url: http://${hostname(true)}:8081/nifi
+ use compression: true
+ timeout: 30 secs
+ batch size: 1000
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/test/resources/config.yml
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/resources/config.yml b/minifi-bootstrap/src/test/resources/config.yml
index bb4dec4..2af6b9b 100644
--- a/minifi-bootstrap/src/test/resources/config.yml
+++ b/minifi-bootstrap/src/test/resources/config.yml
@@ -22,6 +22,7 @@ Core Properties:
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
@@ -60,40 +61,43 @@ Security Properties:
algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
provider: BC
-Processor Configuration:
- name: TailFile
- class: org.apache.nifi.processors.standard.TailFile
- max concurrent tasks: 1
- scheduling strategy: TIMER_DRIVEN
- scheduling period: 1 sec
- penalization period: 30 sec
- yield period: 1 sec
- run duration nanos: 0
- auto-terminated relationships list:
- Properties:
- File to Tail: logs/minifi-app.log
- Rolling Filename Pattern: minifi-app*
- Initial Start Position: Beginning of File
+Processors:
+ - name: TailFile
+ class: org.apache.nifi.processors.standard.TailFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 1 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ Properties:
+ File to Tail: logs/minifi-app.log
+ Rolling Filename Pattern: minifi-app*
+ Initial Start Position: Beginning of File
-Connection Properties:
- name:
- max work queue size: 0
- max work queue data size: 0 MB
- flowfile expiration: 0 sec
- queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
+Connections:
+ - name: TailToS2S
+ source name: TailFile
+ source relationship name: success
+ destination name: 8644cbcc-a45c-40e0-964d-5e536e2ada61
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
-Remote Processing Group:
- name: NiFi Flow
- comment:
- url: https://localhost:8090/nifi
- timeout: 30 secs
- yield period: 10 sec
- Input Port:
- id: 8644cbcc-a45c-40e0-964d-5e536e2ada61
- name: tailed log
- comments:
- max concurrent tasks: 1
- use compression: false
+Remote Processing Groups:
+ - name: NiFi Flow
+ comment:
+ url: https://localhost:8090/nifi
+ timeout: 30 secs
+ yield period: 10 sec
+ Input Ports:
+ - id: 8644cbcc-a45c-40e0-964d-5e536e2ada61
+ name: tailed log
+ comments:
+ max concurrent tasks: 1
+ use compression: false
Provenance Reporting:
comment:
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/test/resources/default.yml
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/resources/default.yml b/minifi-bootstrap/src/test/resources/default.yml
index 5043900..89b1011 100644
--- a/minifi-bootstrap/src/test/resources/default.yml
+++ b/minifi-bootstrap/src/test/resources/default.yml
@@ -22,6 +22,7 @@ Core Properties:
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
@@ -34,7 +35,7 @@ FlowFile Repository:
out period: 5 sec
out threads: 4
-Content Repository:
+Content Repository:
content claim max appendable size: 10 MB
content claim max flow files: 100
always sync: false
@@ -60,45 +61,10 @@ Security Properties:
algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
provider: BC
-Processor Configuration:
- name:
- class:
- max concurrent tasks:
- scheduling strategy:
- scheduling period:
- penalization period:
- yield period:
- run duration nanos:
- auto-terminated relationships list:
- Properties:
+Processors:
-Connection Properties:
- name:
- max work queue size: 0
- max work queue data size: 0 MB
- flowfile expiration: 0 sec
- queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
+Connections:
-Remote Processing Group:
- name:
- comment:
- url:
- timeout:
- yield period:
- Input Port:
- id:
- name:
- comments:
- max concurrent tasks:
- use compression:
+Remote Processing Groups:
-Provenance Reporting:
- comment:
- scheduling strategy:
- scheduling period:
- destination url:
- port name:
- originating url:
- use compression:
- timeout:
- batch size:
\ No newline at end of file
+Provenance Reporting:
\ No newline at end of file
[3/3] nifi-minifi git commit: MINIFI-30 Added changes to the config
transformer to support multiple processors, multiple RPGs,
multiple Input Ports in RPGS, and greater validation on the config yaml
Posted by jp...@apache.org.
MINIFI-30 Added changes to the config transformer to support multiple processors, multiple RPGs, multiple Input Ports in RPGS, and greater validation on the config yaml
This closes #17
Project: http://git-wip-us.apache.org/repos/asf/nifi-minifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi-minifi/commit/6f528bbc
Tree: http://git-wip-us.apache.org/repos/asf/nifi-minifi/tree/6f528bbc
Diff: http://git-wip-us.apache.org/repos/asf/nifi-minifi/diff/6f528bbc
Branch: refs/heads/master
Commit: 6f528bbc1454ee66f2fa26d7c01c8de28290f649
Parents: fb55481
Author: Joseph Percivall <jo...@yahoo.com>
Authored: Tue Apr 26 15:11:00 2016 -0400
Committer: Joseph Percivall <jo...@yahoo.com>
Committed: Tue Jun 14 15:27:10 2016 -0400
----------------------------------------------------------------------
.../InvalidConfigurationException.java | 38 ++
.../bootstrap/util/ConfigTransformer.java | 484 ++++++++-----------
.../schema/ComponentStatusRepositorySchema.java | 50 ++
.../bootstrap/util/schema/ConfigSchema.java | 157 ++++++
.../bootstrap/util/schema/ConnectionSchema.java | 93 ++++
.../util/schema/ContentRepositorySchema.java | 57 +++
.../util/schema/CorePropertiesSchema.java | 73 +++
.../util/schema/FlowControllerSchema.java | 51 ++
.../util/schema/FlowFileRepositorySchema.java | 68 +++
.../bootstrap/util/schema/ProcessorSchema.java | 127 +++++
.../util/schema/ProvenanceReportingSchema.java | 112 +++++
.../util/schema/ProvenanceRepositorySchema.java | 43 ++
.../util/schema/RemoteInputPortSchema.java | 72 +++
.../schema/RemoteProcessingGroupSchema.java | 88 ++++
.../util/schema/SecurityPropertiesSchema.java | 164 +++++++
.../util/schema/SensitivePropsSchema.java | 59 +++
.../bootstrap/util/schema/SwapSchema.java | 75 +++
.../util/schema/common/BaseSchema.java | 134 +++++
.../util/schema/common/CommonPropertyKeys.java | 49 ++
.../bootstrap/util/TestConfigTransformer.java | 122 ++++-
.../src/test/resources/config-empty.yml | 6 +-
.../test/resources/config-malformed-field.yml | 109 +++++
.../src/test/resources/config-minimal.yml | 35 ++
.../resources/config-missing-required-field.yml | 109 +++++
.../src/test/resources/config-multiple-RPGs.yml | 127 +++++
.../resources/config-multiple-input-ports.yml | 121 +++++
.../test/resources/config-multiple-problems.yml | 111 +++++
.../resources/config-multiple-processors.yml | 165 +++++++
minifi-bootstrap/src/test/resources/config.yml | 68 +--
minifi-bootstrap/src/test/resources/default.yml | 46 +-
minifi-docs/Properties_Guide.md | 246 ++++++++--
.../src/main/resources/conf/config.yml | 46 +-
32 files changed, 2865 insertions(+), 440 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/exception/InvalidConfigurationException.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/exception/InvalidConfigurationException.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/exception/InvalidConfigurationException.java
new file mode 100644
index 0000000..bdc51d4
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/exception/InvalidConfigurationException.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.minifi.bootstrap.exception;
+
+public class InvalidConfigurationException extends Exception {
+
+ private static final long serialVersionUID = 1L;
+
+ public InvalidConfigurationException() {
+ super();
+ }
+
+ public InvalidConfigurationException(final String message) {
+ super(message);
+ }
+
+ public InvalidConfigurationException(final Throwable t) {
+ super(t);
+ }
+
+ public InvalidConfigurationException(final String message, final Throwable t) {
+ super(message, t);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/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 e7383ad..956f4dc 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
@@ -20,6 +20,22 @@ package org.apache.nifi.minifi.bootstrap.util;
import org.apache.nifi.controller.FlowSerializationException;
import org.apache.nifi.minifi.bootstrap.configuration.ConfigurationChangeException;
+import org.apache.nifi.minifi.bootstrap.exception.InvalidConfigurationException;
+import org.apache.nifi.minifi.bootstrap.util.schema.ComponentStatusRepositorySchema;
+import org.apache.nifi.minifi.bootstrap.util.schema.ConfigSchema;
+import org.apache.nifi.minifi.bootstrap.util.schema.ConnectionSchema;
+import org.apache.nifi.minifi.bootstrap.util.schema.ContentRepositorySchema;
+import org.apache.nifi.minifi.bootstrap.util.schema.CorePropertiesSchema;
+import org.apache.nifi.minifi.bootstrap.util.schema.FlowControllerSchema;
+import org.apache.nifi.minifi.bootstrap.util.schema.FlowFileRepositorySchema;
+import org.apache.nifi.minifi.bootstrap.util.schema.ProcessorSchema;
+import org.apache.nifi.minifi.bootstrap.util.schema.ProvenanceReportingSchema;
+import org.apache.nifi.minifi.bootstrap.util.schema.ProvenanceRepositorySchema;
+import org.apache.nifi.minifi.bootstrap.util.schema.RemoteInputPortSchema;
+import org.apache.nifi.minifi.bootstrap.util.schema.RemoteProcessingGroupSchema;
+import org.apache.nifi.minifi.bootstrap.util.schema.SecurityPropertiesSchema;
+import org.apache.nifi.minifi.bootstrap.util.schema.SensitivePropsSchema;
+import org.apache.nifi.minifi.bootstrap.util.schema.SwapSchema;
import org.apache.nifi.stream.io.ByteArrayOutputStream;
import org.w3c.dom.DOMException;
import org.w3c.dom.Document;
@@ -50,96 +66,17 @@ import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Collection;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.zip.GZIPOutputStream;
public final class ConfigTransformer {
- // Underlying version NIFI POC will be using
+ // Underlying version of NIFI will be using
public static final String NIFI_VERSION = "0.6.0";
- public static final String NAME_KEY = "name";
- public static final String COMMENT_KEY = "comment";
- public static final String ALWAYS_SYNC_KEY = "always sync";
- public static final String YIELD_PERIOD_KEY = "yield period";
- public static final String MAX_CONCURRENT_TASKS_KEY = "max concurrent tasks";
- public static final String ID_KEY = "id";
-
- public static final String FLOW_CONTROLLER_PROPS_KEY = "Flow Controller";
-
- public static final String CORE_PROPS_KEY = "Core Properties";
- public static final String FLOW_CONTROLLER_SHUTDOWN_PERIOD_KEY = "flow controller graceful shutdown period";
- public static final String FLOW_SERVICE_WRITE_DELAY_INTERVAL_KEY = "flow service write delay interval";
- public static final String ADMINISTRATIVE_YIELD_DURATION_KEY = "administrative yield duration";
- public static final String BORED_YIELD_DURATION_KEY = "bored yield duration";
-
- public static final String FLOWFILE_REPO_KEY = "FlowFile Repository";
- public static final String PARTITIONS_KEY = "partitions";
- public static final String CHECKPOINT_INTERVAL_KEY = "checkpoint interval";
- public static final String THRESHOLD_KEY = "queue swap threshold";
- public static final String SWAP_PROPS_KEY = "Swap";
- public static final String IN_PERIOD_KEY = "in period";
- public static final String IN_THREADS_KEY = "in threads";
- public static final String OUT_PERIOD_KEY = "out period";
- public static final String OUT_THREADS_KEY = "out threads";
-
-
- public static final String PROVENANCE_REPO_KEY = "Provenance Repository";
- public static final String PROVENANCE_REPO_ROLLOVER_TIME_KEY = "provenance rollover time";
-
- public static final String CONTENT_REPO_KEY = "Content Repository";
- public static final String CONTENT_CLAIM_MAX_APPENDABLE_SIZE_KEY = "content claim max appendable size";
- public static final String CONTENT_CLAIM_MAX_FLOW_FILES_KEY = "content claim max flow files";
-
- public static final String COMPONENT_STATUS_REPO_KEY = "Component Status Repository";
- public static final String BUFFER_SIZE_KEY = "buffer size";
- public static final String SNAPSHOT_FREQUENCY_KEY = "snapshot frequency";
-
- public static final String SECURITY_PROPS_KEY = "Security Properties";
- public static final String KEYSTORE_KEY = "keystore";
- public static final String KEYSTORE_TYPE_KEY = "keystore type";
- public static final String KEYSTORE_PASSWORD_KEY = "keystore password";
- public static final String KEY_PASSWORD_KEY = "key password";
- public static final String TRUSTSTORE_KEY = "truststore";
- public static final String TRUSTSTORE_TYPE_KEY = "truststore type";
- public static final String TRUSTSTORE_PASSWORD_KEY = "truststore password";
- public static final String SENSITIVE_PROPS_KEY = "Sensitive Props";
- public static final String SENSITIVE_PROPS_KEY__KEY = "key";
- public static final String SENSITIVE_PROPS_ALGORITHM_KEY = "algorithm";
- public static final String SENSITIVE_PROPS_PROVIDER_KEY = "provider";
-
- public static final String PROCESSOR_CONFIG_KEY = "Processor Configuration";
- public static final String CLASS_KEY = "class";
- public static final String SCHEDULING_PERIOD_KEY = "scheduling period";
- public static final String PENALIZATION_PERIOD_KEY = "penalization period";
- public static final String SCHEDULING_STRATEGY_KEY = "scheduling strategy";
- 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 CONNECTION_PROPS_KEY = "Connection Properties";
- public static final String MAX_WORK_QUEUE_SIZE_KEY = "max work queue size";
- public static final String MAX_WORK_QUEUE_DATA_SIZE_KEY = "max work queue data size";
- public static final String FLOWFILE_EXPIRATION__KEY = "flowfile expiration";
- public static final String QUEUE_PRIORITIZER_CLASS_KEY = "queue prioritizer class";
-
- public static final String REMOTE_PROCESSING_GROUP_KEY = "Remote Processing Group";
- public static final String URL_KEY = "url";
- public static final String TIMEOUT_KEY = "timeout";
-
- public static final String INPUT_PORT_KEY = "Input Port";
- public static final String USE_COMPRESSION_KEY = "use compression";
-
- public static final String PROVENANCE_REPORTING_KEY = "Provenance Reporting";
- public static final String DESTINATION_URL_KEY = "destination url";
- public static final String PORT_NAME_KEY = "port name";
- public static final String ORIGINATING_URL_KEY = "originating url";
- public static final String BATCH_SIZE_KEY = "batch size";
-
- public static final String SSL_PROTOCOL_KEY = "ssl protocol";
-
// Final util classes should have private constructor
- private ConfigTransformer() {}
+ private ConfigTransformer() {
+ }
public static void transformConfigFile(String sourceFile, String destPath) throws Exception {
File ymlConfigFile = new File(sourceFile);
@@ -157,20 +94,24 @@ public final class ConfigTransformer {
// Verify the parsed object is a Map structure
if (loadedObject instanceof Map) {
- final Map<String, Object> result = (Map<String, Object>) loadedObject;
+ final Map<String, Object> loadedMap = (Map<String, Object>) loadedObject;
+ ConfigSchema configSchema = new ConfigSchema(loadedMap);
+ if (!configSchema.isValid()) {
+ throw new InvalidConfigurationException("Failed to transform config file due to:" + configSchema.getValidationIssuesAsString());
+ }
// Create nifi.properties and flow.xml.gz in memory
ByteArrayOutputStream nifiPropertiesOutputStream = new ByteArrayOutputStream();
- writeNiFiProperties(result, nifiPropertiesOutputStream);
+ writeNiFiProperties(configSchema, nifiPropertiesOutputStream);
- DOMSource flowXml = createFlowXml(result);
+ DOMSource flowXml = createFlowXml(configSchema);
// Write nifi.properties and flow.xml.gz
writeNiFiPropertiesFile(nifiPropertiesOutputStream, destPath);
writeFlowXmlFile(flowXml, destPath);
} else {
- throw new IllegalArgumentException("Provided YAML configuration is not a Map.");
+ throw new InvalidConfigurationException("Provided YAML configuration is not a Map");
}
} finally {
if (sourceStream != null) {
@@ -180,12 +121,11 @@ public final class ConfigTransformer {
}
private static void writeNiFiPropertiesFile(ByteArrayOutputStream nifiPropertiesOutputStream, String destPath) throws IOException {
- try {
- final Path nifiPropertiesPath = Paths.get(destPath, "nifi.properties");
- FileOutputStream nifiProperties = new FileOutputStream(new File(nifiPropertiesPath.toString()));
- nifiProperties.write(nifiPropertiesOutputStream.getUnderlyingBuffer());
+ final Path nifiPropertiesPath = Paths.get(destPath, "nifi.properties");
+ try (FileOutputStream nifiProperties = new FileOutputStream(nifiPropertiesPath.toString())) {
+ nifiPropertiesOutputStream.writeTo(nifiProperties);
} finally {
- if (nifiPropertiesOutputStream != null){
+ if (nifiPropertiesOutputStream != null) {
nifiPropertiesOutputStream.flush();
nifiPropertiesOutputStream.close();
}
@@ -210,32 +150,32 @@ public final class ConfigTransformer {
outStream.close();
}
- private static void writeNiFiProperties(Map<String, Object> topLevelYaml, OutputStream outputStream) throws FileNotFoundException, UnsupportedEncodingException, ConfigurationChangeException {
+ private static void writeNiFiProperties(ConfigSchema configSchema, OutputStream outputStream) throws FileNotFoundException, UnsupportedEncodingException, ConfigurationChangeException {
PrintWriter writer = null;
try {
writer = new PrintWriter(outputStream, true);
- Map<String, Object> coreProperties = (Map<String, Object>) topLevelYaml.get(CORE_PROPS_KEY);
- Map<String, Object> flowfileRepo = (Map<String, Object>) topLevelYaml.get(FLOWFILE_REPO_KEY);
- Map<String, Object> swapProperties = (Map<String, Object>) flowfileRepo.get(SWAP_PROPS_KEY);
- Map<String, Object> contentRepo = (Map<String, Object>) topLevelYaml.get(CONTENT_REPO_KEY);
- Map<String, Object> provenanceRepo = (Map<String, Object>) topLevelYaml.get(PROVENANCE_REPO_KEY);
- Map<String, Object> componentStatusRepo = (Map<String, Object>) topLevelYaml.get(COMPONENT_STATUS_REPO_KEY);
- Map<String, Object> securityProperties = (Map<String, Object>) topLevelYaml.get(SECURITY_PROPS_KEY);
- Map<String, Object> sensitiveProperties = (Map<String, Object>) securityProperties.get(SENSITIVE_PROPS_KEY);
+ CorePropertiesSchema coreProperties = configSchema.getCoreProperties();
+ FlowFileRepositorySchema flowfileRepoSchema = configSchema.getFlowfileRepositoryProperties();
+ SwapSchema swapProperties = flowfileRepoSchema.getSwapProperties();
+ ContentRepositorySchema contentRepoProperties = configSchema.getContentRepositoryProperties();
+ ComponentStatusRepositorySchema componentStatusRepoProperties = configSchema.getComponentStatusRepositoryProperties();
+ SecurityPropertiesSchema securityProperties = configSchema.getSecurityProperties();
+ SensitivePropsSchema sensitiveProperties = securityProperties.getSensitiveProps();
+ ProvenanceRepositorySchema provenanceRepositorySchema = configSchema.getProvenanceRepositorySchema();
writer.print(PROPERTIES_FILE_APACHE_2_0_LICENSE);
writer.println("# Core Properties #");
writer.println();
- writer.println("nifi.version="+NIFI_VERSION);
+ writer.println("nifi.version=" + NIFI_VERSION);
writer.println("nifi.flow.configuration.file=./conf/flow.xml.gz");
writer.println("nifi.flow.configuration.archive.dir=./conf/archive/");
writer.println("nifi.flowcontroller.autoResumeState=true");
- writer.println("nifi.flowcontroller.graceful.shutdown.period=" + getValueString(coreProperties, FLOW_CONTROLLER_SHUTDOWN_PERIOD_KEY));
- writer.println("nifi.flowservice.writedelay.interval=" + getValueString(coreProperties, FLOW_SERVICE_WRITE_DELAY_INTERVAL_KEY));
- writer.println("nifi.administrative.yield.duration=" + getValueString(coreProperties, ADMINISTRATIVE_YIELD_DURATION_KEY));
+ writer.println("nifi.flowcontroller.graceful.shutdown.period=" + coreProperties.getFlowControllerGracefulShutdownPeriod());
+ writer.println("nifi.flowservice.writedelay.interval=" + coreProperties.getFlowServiceWriteDelayInterval());
+ writer.println("nifi.administrative.yield.duration=" + coreProperties.getAdministrativeYieldDuration());
writer.println("# If a component has no work to do (is \"bored\"), how long should we wait before checking again for work?");
- writer.println("nifi.bored.yield.duration=" + getValueString(coreProperties, BORED_YIELD_DURATION_KEY));
+ writer.println("nifi.bored.yield.duration=" + coreProperties.getBoredYieldDuration());
writer.println();
writer.println("nifi.authority.provider.configuration.file=./conf/authority-providers.xml");
writer.println("nifi.login.identity.provider.configuration.file=./conf/login-identity-providers.xml");
@@ -260,38 +200,38 @@ public final class ConfigTransformer {
writer.println("# FlowFile Repository");
writer.println("nifi.flowfile.repository.implementation=org.apache.nifi.controller.repository.WriteAheadFlowFileRepository");
writer.println("nifi.flowfile.repository.directory=./flowfile_repository");
- writer.println("nifi.flowfile.repository.partitions=" + getValueString(flowfileRepo, PARTITIONS_KEY));
- writer.println("nifi.flowfile.repository.checkpoint.interval=" + getValueString(flowfileRepo,CHECKPOINT_INTERVAL_KEY));
- writer.println("nifi.flowfile.repository.always.sync=" + getValueString(flowfileRepo,ALWAYS_SYNC_KEY));
+ writer.println("nifi.flowfile.repository.partitions=" + flowfileRepoSchema.getPartitions());
+ writer.println("nifi.flowfile.repository.checkpoint.interval=" + flowfileRepoSchema.getCheckpointInterval());
+ writer.println("nifi.flowfile.repository.always.sync=" + flowfileRepoSchema.getAlwaysSync());
writer.println();
writer.println("nifi.swap.manager.implementation=org.apache.nifi.controller.FileSystemSwapManager");
- writer.println("nifi.queue.swap.threshold=" + getValueString(swapProperties, THRESHOLD_KEY));
- writer.println("nifi.swap.in.period=" + getValueString(swapProperties, IN_PERIOD_KEY));
- writer.println("nifi.swap.in.threads=" + getValueString(swapProperties, IN_THREADS_KEY));
- writer.println("nifi.swap.out.period=" + getValueString(swapProperties, OUT_PERIOD_KEY));
- writer.println("nifi.swap.out.threads=" + getValueString(swapProperties, OUT_THREADS_KEY));
+ writer.println("nifi.queue.swap.threshold=" + swapProperties.getThreshold());
+ writer.println("nifi.swap.in.period=" + swapProperties.getInPeriod());
+ writer.println("nifi.swap.in.threads=" + swapProperties.getInThreads());
+ writer.println("nifi.swap.out.period=" + swapProperties.getOutPeriod());
+ writer.println("nifi.swap.out.threads=" + swapProperties.getOutThreads());
writer.println();
writer.println("# Content Repository");
writer.println("nifi.content.repository.implementation=org.apache.nifi.controller.repository.FileSystemRepository");
- writer.println("nifi.content.claim.max.appendable.size=" + getValueString(contentRepo, CONTENT_CLAIM_MAX_APPENDABLE_SIZE_KEY));
- writer.println("nifi.content.claim.max.flow.files=" + getValueString(contentRepo, CONTENT_CLAIM_MAX_FLOW_FILES_KEY));
+ writer.println("nifi.content.claim.max.appendable.size=" + contentRepoProperties.getContentClaimMaxAppendableSize());
+ writer.println("nifi.content.claim.max.flow.files=" + contentRepoProperties.getContentClaimMaxFlowFiles());
writer.println("nifi.content.repository.archive.max.retention.period=");
writer.println("nifi.content.repository.archive.max.usage.percentage=");
writer.println("nifi.content.repository.archive.enabled=false");
writer.println("nifi.content.repository.directory.default=./content_repository");
- writer.println("nifi.content.repository.always.sync=" + getValueString(contentRepo, ALWAYS_SYNC_KEY));
+ writer.println("nifi.content.repository.always.sync=" + contentRepoProperties.getAlwaysSync());
writer.println();
writer.println("# Provenance Repository Properties");
writer.println("nifi.provenance.repository.implementation=org.apache.nifi.provenance.MiNiFiPersistentProvenanceRepository");
- writer.println("nifi.provenance.repository.rollover.time=" + getValueString(provenanceRepo, PROVENANCE_REPO_ROLLOVER_TIME_KEY));
+ writer.println("nifi.provenance.repository.rollover.time=" + provenanceRepositorySchema.getProvenanceRepoRolloverTimeKey());
writer.println();
writer.println("# Volatile Provenance Respository Properties");
- writer.println("nifi.provenance.repository.buffer.size=100000");
+ writer.println("nifi.provenance.repository.buffer.size=10000");
writer.println();
writer.println("# Component Status Repository");
writer.println("nifi.components.status.repository.implementation=org.apache.nifi.controller.status.history.VolatileComponentStatusRepository");
- writer.println("nifi.components.status.repository.buffer.size=" + getValueString(componentStatusRepo, BUFFER_SIZE_KEY));
- writer.println("nifi.components.status.snapshot.frequency=" + getValueString(componentStatusRepo, SNAPSHOT_FREQUENCY_KEY));
+ writer.println("nifi.components.status.repository.buffer.size=" + componentStatusRepoProperties.getBufferSize());
+ writer.println("nifi.components.status.snapshot.frequency=" + componentStatusRepoProperties.getSnapshotFrequency());
writer.println();
writer.println("# web properties #");
writer.println("nifi.web.war.directory=./lib");
@@ -303,17 +243,17 @@ public final class ConfigTransformer {
writer.println("nifi.web.jetty.threads=200");
writer.println();
writer.println("# security properties #");
- writer.println("nifi.sensitive.props.key=" + getValueString(sensitiveProperties, SENSITIVE_PROPS_KEY__KEY));
- writer.println("nifi.sensitive.props.algorithm=" + getValueString(sensitiveProperties, SENSITIVE_PROPS_ALGORITHM_KEY));
- writer.println("nifi.sensitive.props.provider=" + getValueString(sensitiveProperties, SENSITIVE_PROPS_PROVIDER_KEY));
+ writer.println("nifi.sensitive.props.key=" + sensitiveProperties.getKey());
+ writer.println("nifi.sensitive.props.algorithm=" + sensitiveProperties.getAlgorithm());
+ writer.println("nifi.sensitive.props.provider=" + sensitiveProperties.getProvider());
writer.println();
- writer.println("nifi.security.keystore=" + getValueString(securityProperties, KEYSTORE_KEY));
- writer.println("nifi.security.keystoreType=" + getValueString(securityProperties, KEYSTORE_TYPE_KEY));
- writer.println("nifi.security.keystorePasswd=" + getValueString(securityProperties, KEYSTORE_PASSWORD_KEY));
- writer.println("nifi.security.keyPasswd=" + getValueString(securityProperties, KEY_PASSWORD_KEY));
- writer.println("nifi.security.truststore=" + getValueString(securityProperties, TRUSTSTORE_KEY));
- writer.println("nifi.security.truststoreType=" + getValueString(securityProperties, TRUSTSTORE_TYPE_KEY));
- writer.println("nifi.security.truststorePasswd=" + getValueString(securityProperties, TRUSTSTORE_PASSWORD_KEY));
+ writer.println("nifi.security.keystore=" + securityProperties.getKeystore());
+ writer.println("nifi.security.keystoreType=" + securityProperties.getKeystoreType());
+ writer.println("nifi.security.keystorePasswd=" + securityProperties.getKeystorePassword());
+ writer.println("nifi.security.keyPasswd=" + securityProperties.getKeyPassword());
+ writer.println("nifi.security.truststore=" + securityProperties.getTruststore());
+ writer.println("nifi.security.truststoreType=" + securityProperties.getTruststoreType());
+ writer.println("nifi.security.truststorePasswd=" + securityProperties.getTruststorePassword());
writer.println("nifi.security.needClientAuth=");
writer.println("nifi.security.user.credential.cache.duration=24 hours");
writer.println("nifi.security.user.authority.provider=file-provider");
@@ -333,13 +273,14 @@ public final class ConfigTransformer {
} catch (NullPointerException e) {
throw new ConfigurationChangeException("Failed to parse the config YAML while creating the nifi.properties", e);
} finally {
- if (writer != null){
+ if (writer != null) {
writer.flush();
writer.close();
}
}
}
- private static DOMSource createFlowXml(Map<String, Object> topLevelYaml) throws IOException, ConfigurationChangeException {
+
+ private static DOMSource createFlowXml(ConfigSchema configSchema) throws IOException, ConfigurationChangeException {
try {
// create a new, empty document
final DocumentBuilderFactory docFactory = DocumentBuilderFactory.newInstance();
@@ -351,50 +292,34 @@ public final class ConfigTransformer {
// populate document with controller state
final Element rootNode = doc.createElement("flowController");
doc.appendChild(rootNode);
- Map<String, Object> processorConfig = (Map<String, Object>) topLevelYaml.get(PROCESSOR_CONFIG_KEY);
- addTextElement(rootNode, "maxTimerDrivenThreadCount", getValueString(processorConfig, MAX_CONCURRENT_TASKS_KEY, "1"));
- addTextElement(rootNode, "maxEventDrivenThreadCount", getValueString(processorConfig, MAX_CONCURRENT_TASKS_KEY, "1"));
- addProcessGroup(rootNode, topLevelYaml, "rootGroup");
-
- Map<String, Object> securityProps = (Map<String, Object>) topLevelYaml.get(SECURITY_PROPS_KEY);
- if (securityProps != null) {
- String sslAlgorithm = (String) securityProps.get(SSL_PROTOCOL_KEY);
- if (sslAlgorithm != null && !(sslAlgorithm.isEmpty())) {
- final Element controllerServicesNode = doc.createElement("controllerServices");
- rootNode.appendChild(controllerServicesNode);
- addSSLControllerService(controllerServicesNode, securityProps);
- }
+ CorePropertiesSchema coreProperties = configSchema.getCoreProperties();
+ addTextElement(rootNode, "maxTimerDrivenThreadCount", String.valueOf(coreProperties.getMaxConcurrentThreads()));
+ addTextElement(rootNode, "maxEventDrivenThreadCount", String.valueOf(coreProperties.getMaxConcurrentThreads()));
+ addProcessGroup(rootNode, configSchema, "rootGroup");
+
+ SecurityPropertiesSchema securityProperties = configSchema.getSecurityProperties();
+ if (securityProperties.useSSL()) {
+ final Element controllerServicesNode = doc.createElement("controllerServices");
+ rootNode.appendChild(controllerServicesNode);
+ addSSLControllerService(controllerServicesNode, securityProperties);
}
- Map<String, Object> provenanceProperties = (Map<String, Object>) topLevelYaml.get(PROVENANCE_REPORTING_KEY);
- if (provenanceProperties.get(SCHEDULING_STRATEGY_KEY) != null) {
+ ProvenanceReportingSchema provenanceProperties = configSchema.getProvenanceReportingProperties();
+ if (provenanceProperties != null) {
final Element reportingTasksNode = doc.createElement("reportingTasks");
rootNode.appendChild(reportingTasksNode);
- addProvenanceReportingTask(reportingTasksNode, topLevelYaml);
+ addProvenanceReportingTask(reportingTasksNode, configSchema);
}
return new DOMSource(doc);
} catch (final ParserConfigurationException | DOMException | TransformerFactoryConfigurationError | IllegalArgumentException e) {
throw new FlowSerializationException(e);
- } catch (Exception e){
+ } catch (Exception e) {
throw new ConfigurationChangeException("Failed to parse the config YAML while writing the top level of the flow xml", e);
}
}
- private static <K> String getValueString(Map<K,Object> map, K key){
- Object value = map.get(key);
- return value == null ? "" : value.toString();
- }
-
- private static <K> String getValueString(Map<K,Object> map, K key, String theDefault){
- Object value = null;
- if (map != null){
- value = map.get(key);
- }
- return value == null ? theDefault : value.toString();
- }
-
- private static void addSSLControllerService(final Element element, Map<String, Object> securityProperties) throws ConfigurationChangeException {
+ private static void addSSLControllerService(final Element element, SecurityPropertiesSchema securityProperties) throws ConfigurationChangeException {
try {
final Element serviceElement = element.getOwnerDocument().createElement("controllerService");
addTextElement(serviceElement, "id", "SSL-Context-Service");
@@ -405,126 +330,133 @@ public final class ConfigTransformer {
addTextElement(serviceElement, "enabled", "true");
Map<String, Object> attributes = new HashMap<>();
- attributes.put("Keystore Filename", securityProperties.get(KEYSTORE_KEY));
- attributes.put("Keystore Type", securityProperties.get(KEYSTORE_TYPE_KEY));
- attributes.put("Keystore Password", securityProperties.get(KEYSTORE_PASSWORD_KEY));
- attributes.put("Truststore Filename", securityProperties.get(TRUSTSTORE_KEY));
- attributes.put("Truststore Type", securityProperties.get(TRUSTSTORE_TYPE_KEY));
- attributes.put("Truststore Password", securityProperties.get(TRUSTSTORE_PASSWORD_KEY));
- attributes.put("SSL Protocol", securityProperties.get(SSL_PROTOCOL_KEY));
+ attributes.put("Keystore Filename", securityProperties.getKeystore());
+ attributes.put("Keystore Type", securityProperties.getKeystoreType());
+ attributes.put("Keystore Password", securityProperties.getKeyPassword());
+ attributes.put("Truststore Filename", securityProperties.getTruststore());
+ attributes.put("Truststore Type", securityProperties.getTruststoreType());
+ attributes.put("Truststore Password", securityProperties.getTruststorePassword());
+ attributes.put("SSL Protocol", securityProperties.getSslProtocol());
addConfiguration(serviceElement, attributes);
element.appendChild(serviceElement);
- } catch (Exception e){
+ } catch (Exception e) {
throw new ConfigurationChangeException("Failed to parse the config YAML while trying to create an SSL Controller Service", e);
}
}
- private static void addProcessGroup(final Element parentElement, Map<String, Object> topLevelYaml, final String elementName) throws ConfigurationChangeException {
+ private static void addProcessGroup(final Element parentElement, ConfigSchema configSchema, final String elementName) throws ConfigurationChangeException {
try {
- Map<String, Object> flowControllerProperties = (Map<String, Object>) topLevelYaml.get(FLOW_CONTROLLER_PROPS_KEY);
+ FlowControllerSchema flowControllerProperties = configSchema.getFlowControllerProperties();
final Document doc = parentElement.getOwnerDocument();
final Element element = doc.createElement(elementName);
parentElement.appendChild(element);
addTextElement(element, "id", "Root-Group");
- addTextElement(element, "name", getValueString(flowControllerProperties, NAME_KEY));
+ addTextElement(element, "name", flowControllerProperties.getName());
addPosition(element);
- addTextElement(element, "comment", getValueString(flowControllerProperties, COMMENT_KEY));
+ addTextElement(element, "comment", flowControllerProperties.getComment());
- Map<String, Object> processorConfig = (Map<String, Object>) topLevelYaml.get(PROCESSOR_CONFIG_KEY);
- addProcessor(element, processorConfig);
+ List<ProcessorSchema> processors = configSchema.getProcessors();
+ if (processors != null) {
+ for (ProcessorSchema processorConfig : processors) {
+ addProcessor(element, processorConfig);
+ }
+ }
- Map<String, Object> remoteProcessingGroup = (Map<String, Object>) topLevelYaml.get(REMOTE_PROCESSING_GROUP_KEY);
- addRemoteProcessGroup(element, remoteProcessingGroup);
+ List<RemoteProcessingGroupSchema> remoteProcessingGroups = configSchema.getRemoteProcessingGroups();
+ if (remoteProcessingGroups != null) {
+ for (RemoteProcessingGroupSchema remoteProcessingGroupSchema: remoteProcessingGroups) {
+ addRemoteProcessGroup(element, remoteProcessingGroupSchema);
+ }
+ }
- addConnection(element, topLevelYaml);
- } catch (ConfigurationChangeException e){
+ List<ConnectionSchema> connections = configSchema.getConnections();
+ if (connections != null) {
+ for (ConnectionSchema connectionConfig : connections) {
+ addConnection(element, connectionConfig, configSchema);
+ }
+ }
+ } catch (ConfigurationChangeException e) {
throw e;
- } catch (Exception e){
+ } catch (Exception e) {
throw new ConfigurationChangeException("Failed to parse the config YAML while trying to creating the root Process Group", e);
}
}
- private static void addProcessor(final Element parentElement, Map<String, Object> processorConfig) throws ConfigurationChangeException {
-
+ private static void addProcessor(final Element parentElement, ProcessorSchema processorConfig) throws ConfigurationChangeException {
try {
- if (processorConfig.get(CLASS_KEY) == null) {
- // Only add a processor if it has a class
- return;
- }
-
final Document doc = parentElement.getOwnerDocument();
final Element element = doc.createElement("processor");
parentElement.appendChild(element);
- addTextElement(element, "id", "Processor");
- addTextElement(element, "name", getValueString(processorConfig, NAME_KEY));
+ addTextElement(element, "id", processorConfig.getName());
+ addTextElement(element, "name", processorConfig.getName());
addPosition(element);
addStyle(element);
- addTextElement(element, "comment", getValueString(processorConfig, COMMENT_KEY));
- addTextElement(element, "class", getValueString(processorConfig, CLASS_KEY));
- addTextElement(element, "maxConcurrentTasks", getValueString(processorConfig, MAX_CONCURRENT_TASKS_KEY));
- addTextElement(element, "schedulingPeriod", getValueString(processorConfig, SCHEDULING_PERIOD_KEY));
- addTextElement(element, "penalizationPeriod", getValueString(processorConfig, PENALIZATION_PERIOD_KEY));
- addTextElement(element, "yieldPeriod", getValueString(processorConfig, YIELD_PERIOD_KEY));
+ addTextElement(element, "comment", "");
+ addTextElement(element, "class", processorConfig.getProcessorClass());
+ addTextElement(element, "maxConcurrentTasks", String.valueOf(processorConfig.getMaxConcurrentTasks()));
+ addTextElement(element, "schedulingPeriod", processorConfig.getSchedulingPeriod());
+ addTextElement(element, "penalizationPeriod", processorConfig.getPenalizationPeriod());
+ addTextElement(element, "yieldPeriod", processorConfig.getYieldPeriod());
addTextElement(element, "bulletinLevel", "WARN");
addTextElement(element, "lossTolerant", "false");
addTextElement(element, "scheduledState", "RUNNING");
- addTextElement(element, "schedulingStrategy", getValueString(processorConfig, SCHEDULING_STRATEGY_KEY));
- addTextElement(element, "runDurationNanos", getValueString(processorConfig, RUN_DURATION_NANOS_KEY));
+ addTextElement(element, "schedulingStrategy", processorConfig.getSchedulingStrategy());
+ addTextElement(element, "runDurationNanos", String.valueOf(processorConfig.getRunDurationNanos()));
- addConfiguration(element, (Map<String, Object>) processorConfig.get(PROCESSOR_PROPS_KEY));
+ addConfiguration(element, processorConfig.getProperties());
- Collection<String> autoTerminatedRelationships = (Collection<String>) processorConfig.get(AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY);
+ Collection<String> autoTerminatedRelationships = processorConfig.getAutoTerminatedRelationshipsList();
if (autoTerminatedRelationships != null) {
for (String rel : autoTerminatedRelationships) {
addTextElement(element, "autoTerminatedRelationship", rel);
}
}
- } catch (Exception e){
- throw new ConfigurationChangeException("Failed to parse the config YAML while trying to add the Processor", e);
+ } catch (Exception e) {
+ throw new ConfigurationChangeException("Failed to parse the config YAML while trying to add a Processor", e);
}
}
- private static void addProvenanceReportingTask(final Element element, Map<String, Object> topLevelYaml) throws ConfigurationChangeException {
+ private static void addProvenanceReportingTask(final Element element, ConfigSchema configSchema) throws ConfigurationChangeException {
try {
- Map<String, Object> provenanceProperties = (Map<String, Object>) topLevelYaml.get(PROVENANCE_REPORTING_KEY);
+ ProvenanceReportingSchema provenanceProperties = configSchema.getProvenanceReportingProperties();
final Element taskElement = element.getOwnerDocument().createElement("reportingTask");
addTextElement(taskElement, "id", "Provenance-Reporting");
addTextElement(taskElement, "name", "Site-To-Site-Provenance-Reporting");
- addTextElement(taskElement, "comment", getValueString(provenanceProperties, COMMENT_KEY));
+ addTextElement(taskElement, "comment", provenanceProperties.getComment());
addTextElement(taskElement, "class", "org.apache.nifi.minifi.provenance.reporting.ProvenanceReportingTask");
- addTextElement(taskElement, "schedulingPeriod", getValueString(provenanceProperties, SCHEDULING_PERIOD_KEY));
+ addTextElement(taskElement, "schedulingPeriod", provenanceProperties.getSchedulingPeriod());
addTextElement(taskElement, "scheduledState", "RUNNING");
- addTextElement(taskElement, "schedulingStrategy", getValueString(provenanceProperties, SCHEDULING_STRATEGY_KEY));
+ addTextElement(taskElement, "schedulingStrategy", provenanceProperties.getSchedulingStrategy());
Map<String, Object> attributes = new HashMap<>();
- attributes.put("Destination URL", provenanceProperties.get(DESTINATION_URL_KEY));
- attributes.put("Input Port Name", provenanceProperties.get(PORT_NAME_KEY));
- attributes.put("MiNiFi URL", provenanceProperties.get(ORIGINATING_URL_KEY));
- attributes.put("Compress Events", provenanceProperties.get(USE_COMPRESSION_KEY));
- attributes.put("Batch Size", provenanceProperties.get(BATCH_SIZE_KEY));
-
- Map<String, Object> securityProps = (Map<String, Object>) topLevelYaml.get(SECURITY_PROPS_KEY);
- String sslAlgorithm = (String) securityProps.get(SSL_PROTOCOL_KEY);
- if (sslAlgorithm != null && !(sslAlgorithm.isEmpty())) {
+ attributes.put("Destination URL", provenanceProperties.getDestinationUrl());
+ attributes.put("Input Port Name", provenanceProperties.getPortName());
+ attributes.put("MiNiFi URL", provenanceProperties.getOriginatingUrl());
+ attributes.put("Compress Events", provenanceProperties.getUseCompression());
+ attributes.put("Batch Size", provenanceProperties.getBatchSize());
+ attributes.put("Communications Timeout", provenanceProperties.getTimeout());
+
+ SecurityPropertiesSchema securityProps = configSchema.getSecurityProperties();
+ if (securityProps.useSSL()) {
attributes.put("SSL Context Service", "SSL-Context-Service");
}
addConfiguration(taskElement, attributes);
element.appendChild(taskElement);
- } catch (Exception e){
+ } catch (Exception e) {
throw new ConfigurationChangeException("Failed to parse the config YAML while trying to add the Provenance Reporting Task", e);
}
}
private static void addConfiguration(final Element element, Map<String, Object> elementConfig) {
final Document doc = element.getOwnerDocument();
- if (elementConfig == null){
+ if (elementConfig == null) {
return;
}
for (final Map.Entry<String, Object> entry : elementConfig.entrySet()) {
@@ -544,76 +476,57 @@ public final class ConfigTransformer {
parentElement.appendChild(element);
}
- private static void addRemoteProcessGroup(final Element parentElement, Map<String, Object> remoteProcessingGroup) throws ConfigurationChangeException {
+ private static void addRemoteProcessGroup(final Element parentElement, RemoteProcessingGroupSchema remoteProcessingGroupProperties) throws ConfigurationChangeException {
try {
- if (remoteProcessingGroup.get(URL_KEY) == null) {
- // Only add an an RPG if it has a URL
- return;
- }
-
final Document doc = parentElement.getOwnerDocument();
final Element element = doc.createElement("remoteProcessGroup");
parentElement.appendChild(element);
- addTextElement(element, "id", "Remote-Process-Group");
- addTextElement(element, "name", getValueString(remoteProcessingGroup, NAME_KEY));
+ addTextElement(element, "id", remoteProcessingGroupProperties.getName());
+ addTextElement(element, "name", remoteProcessingGroupProperties.getName());
addPosition(element);
- addTextElement(element, "comment", getValueString(remoteProcessingGroup, COMMENT_KEY));
- addTextElement(element, "url", getValueString(remoteProcessingGroup, URL_KEY));
- addTextElement(element, "timeout", getValueString(remoteProcessingGroup, TIMEOUT_KEY));
- addTextElement(element, "yieldPeriod", getValueString(remoteProcessingGroup, YIELD_PERIOD_KEY));
+ addTextElement(element, "comment", remoteProcessingGroupProperties.getComment());
+ addTextElement(element, "url", remoteProcessingGroupProperties.getUrl());
+ addTextElement(element, "timeout", remoteProcessingGroupProperties.getTimeout());
+ addTextElement(element, "yieldPeriod", remoteProcessingGroupProperties.getYieldPeriod());
addTextElement(element, "transmitting", "true");
- Map<String, Object> inputPort = (Map<String, Object>) remoteProcessingGroup.get(INPUT_PORT_KEY);
- addRemoteGroupPort(element, inputPort, "inputPort");
+ List<RemoteInputPortSchema> remoteInputPorts = remoteProcessingGroupProperties.getInputPorts();
+ for(RemoteInputPortSchema remoteInputPortSchema: remoteInputPorts) {
+ addRemoteGroupPort(element, remoteInputPortSchema);
+ }
parentElement.appendChild(element);
- } catch (Exception e){
+ } catch (Exception e) {
throw new ConfigurationChangeException("Failed to parse the config YAML while trying to add the Remote Process Group", e);
}
}
- private static void addRemoteGroupPort(final Element parentElement, Map<String, Object> inputPort, final String elementName) throws ConfigurationChangeException {
-
+ private static void addRemoteGroupPort(final Element parentElement, RemoteInputPortSchema inputPort) throws ConfigurationChangeException {
try {
- if (inputPort.get(ID_KEY) == null) {
- // Only add an input port if it has an ID
- return;
- }
-
final Document doc = parentElement.getOwnerDocument();
- final Element element = doc.createElement(elementName);
+ final Element element = doc.createElement("inputPort");
parentElement.appendChild(element);
- addTextElement(element, "id", getValueString(inputPort, ID_KEY));
- addTextElement(element, "name", getValueString(inputPort, NAME_KEY));
+ addTextElement(element, "id", inputPort.getId());
+ addTextElement(element, "name", inputPort.getName());
addPosition(element);
- addTextElement(element, "comments", getValueString(inputPort, COMMENT_KEY));
+ addTextElement(element, "comments", inputPort.getComment());
addTextElement(element, "scheduledState", "RUNNING");
- addTextElement(element, "maxConcurrentTasks", getValueString(inputPort, MAX_CONCURRENT_TASKS_KEY));
- addTextElement(element, "useCompression", getValueString(inputPort, USE_COMPRESSION_KEY));
+ addTextElement(element, "maxConcurrentTasks", String.valueOf(inputPort.getMax_concurrent_tasks()));
+ addTextElement(element, "useCompression", String.valueOf(inputPort.getUseCompression()));
parentElement.appendChild(element);
- } catch (Exception e){
+ } catch (Exception e) {
throw new ConfigurationChangeException("Failed to parse the config YAML while trying to add the input port of the Remote Process Group", e);
}
}
- private static void addConnection(final Element parentElement, Map<String, Object> topLevelYaml) throws ConfigurationChangeException {
+ private static void addConnection(final Element parentElement, ConnectionSchema connectionProperties, ConfigSchema configSchema) throws ConfigurationChangeException {
try {
- Map<String, Object> connectionProperties = (Map<String, Object>) topLevelYaml.get(CONNECTION_PROPS_KEY);
- Map<String, Object> remoteProcessingGroup = (Map<String, Object>) topLevelYaml.get(REMOTE_PROCESSING_GROUP_KEY);
- Map<String, Object> inputPort = (Map<String, Object>) remoteProcessingGroup.get(INPUT_PORT_KEY);
- Map<String, Object> processorConfig = (Map<String, Object>) topLevelYaml.get(PROCESSOR_CONFIG_KEY);
-
- if (inputPort.get(ID_KEY) == null || processorConfig.get(CLASS_KEY) == null) {
- // Only add the connection if the input port and processor config are created
- return;
- }
-
final Document doc = parentElement.getOwnerDocument();
final Element element = doc.createElement("connection");
parentElement.appendChild(element);
- addTextElement(element, "id", "Connection");
- addTextElement(element, "name", getValueString(connectionProperties, NAME_KEY));
+ addTextElement(element, "id", connectionProperties.getName());
+ addTextElement(element, "name", connectionProperties.getName());
final Element bendPointsElement = doc.createElement("bendPoints");
element.appendChild(bendPointsElement);
@@ -621,28 +534,55 @@ public final class ConfigTransformer {
addTextElement(element, "labelIndex", "1");
addTextElement(element, "zIndex", "0");
- addTextElement(element, "sourceId", "Processor");
+ addTextElement(element, "sourceId", connectionProperties.getSourceName());
addTextElement(element, "sourceGroupId", "Root-Group");
addTextElement(element, "sourceType", "PROCESSOR");
- addTextElement(element, "destinationId", getValueString(inputPort, ID_KEY));
- addTextElement(element, "destinationGroupId", "Remote-Process-Group");
- addTextElement(element, "destinationType", "REMOTE_INPUT_PORT");
+ addTextElement(element, "destinationId", connectionProperties.getDestinationName());
- addTextElement(element, "relationship", "success");
+ if (isInputPortId(connectionProperties.getDestinationName(), configSchema)) {
+ addTextElement(element, "destinationGroupId", "Remote-Process-Group");
+ addTextElement(element, "destinationType", "REMOTE_INPUT_PORT");
+ } else {
+ addTextElement(element, "destinationGroupId", "Root-Group");
+ addTextElement(element, "destinationType", "PROCESSOR");
+ }
+
+ addTextElement(element, "relationship", connectionProperties.getSourceRelationshipName());
- addTextElement(element, "maxWorkQueueSize", getValueString(connectionProperties, MAX_WORK_QUEUE_SIZE_KEY));
- addTextElement(element, "maxWorkQueueDataSize", getValueString(connectionProperties, MAX_WORK_QUEUE_DATA_SIZE_KEY));
+ addTextElement(element, "maxWorkQueueSize", String.valueOf(connectionProperties.getMaxWorkQueueSize()));
+ addTextElement(element, "maxWorkQueueDataSize", connectionProperties.getMaxWorkQueueDataSize());
- addTextElement(element, "flowFileExpiration", getValueString(connectionProperties, FLOWFILE_EXPIRATION__KEY));
- addTextElement(element, "queuePrioritizerClass", getValueString(connectionProperties, QUEUE_PRIORITIZER_CLASS_KEY));
+ addTextElement(element, "flowFileExpiration", connectionProperties.getFlowfileExpiration());
+ addTextElement(element, "queuePrioritizerClass", connectionProperties.getQueuePrioritizerClass());
parentElement.appendChild(element);
- } catch (Exception e){
+ } catch (Exception e) {
throw new ConfigurationChangeException("Failed to parse the config YAML while trying to add the connection from the Processor to the input port of the Remote Process Group", e);
}
}
+ private static boolean isInputPortId(String id, ConfigSchema configSchema) {
+ boolean isInputPortId = false;
+ try {
+ List<RemoteProcessingGroupSchema> remoteProcessingGroups = configSchema.getRemoteProcessingGroups();
+ if (remoteProcessingGroups != null) {
+ for (RemoteProcessingGroupSchema remoteProcessingGroupSchema: remoteProcessingGroups) {
+ List<RemoteInputPortSchema> remoteInputPorts = remoteProcessingGroupSchema.getInputPorts();
+ for (RemoteInputPortSchema remoteInputPortSchema: remoteInputPorts) {
+ if (remoteInputPortSchema != null && id.equals(remoteInputPortSchema.getId())) {
+ isInputPortId = true;
+ break;
+ }
+ }
+ }
+ }
+ } catch (Exception e) {
+ // If an exception was thrown then it isn't the InputPort
+ }
+ return isInputPortId;
+ }
+
private static void addPosition(final Element parentElement) {
final Element element = parentElement.getOwnerDocument().createElement("position");
element.setAttribute("x", String.valueOf("0"));
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ComponentStatusRepositorySchema.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ComponentStatusRepositorySchema.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ComponentStatusRepositorySchema.java
new file mode 100644
index 0000000..5b6ac2e
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ComponentStatusRepositorySchema.java
@@ -0,0 +1,50 @@
+/*
+ * 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.bootstrap.util.schema;
+
+import org.apache.nifi.minifi.bootstrap.util.schema.common.BaseSchema;
+
+import java.util.Map;
+
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.COMPONENT_STATUS_REPO_KEY;
+
+/**
+ *
+ */
+public class ComponentStatusRepositorySchema extends BaseSchema {
+ public static final String BUFFER_SIZE_KEY = "buffer size";
+ public static final String SNAPSHOT_FREQUENCY_KEY = "snapshot frequency";
+
+ private Number bufferSize = 1440;
+ private String snapshotFrequency = "1 min";
+
+ public ComponentStatusRepositorySchema() {
+ }
+
+ public ComponentStatusRepositorySchema(Map map) {
+ bufferSize = getOptionalKeyAsType(map, BUFFER_SIZE_KEY, Number.class, COMPONENT_STATUS_REPO_KEY, 1440);
+ snapshotFrequency = getOptionalKeyAsType(map, SNAPSHOT_FREQUENCY_KEY, String.class, COMPONENT_STATUS_REPO_KEY, "1 min");
+ }
+
+ public Number getBufferSize() {
+ return bufferSize;
+ }
+
+ public String getSnapshotFrequency() {
+ return snapshotFrequency;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ConfigSchema.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ConfigSchema.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ConfigSchema.java
new file mode 100644
index 0000000..6d3d7ca
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ConfigSchema.java
@@ -0,0 +1,157 @@
+/*
+ * 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.bootstrap.util.schema;
+
+import org.apache.nifi.minifi.bootstrap.util.schema.common.BaseSchema;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.COMPONENT_STATUS_REPO_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.CONNECTIONS_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.CONTENT_REPO_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.CORE_PROPS_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.FLOWFILE_REPO_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.FLOW_CONTROLLER_PROPS_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.PROCESSORS_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.PROVENANCE_REPORTING_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.PROVENANCE_REPO_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.REMOTE_PROCESSING_GROUPS_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.SECURITY_PROPS_KEY;
+
+/**
+ *
+ */
+public class ConfigSchema extends BaseSchema {
+ 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 List<ProcessorSchema> processors;
+ private List<ConnectionSchema> connections;
+ private List<RemoteProcessingGroupSchema> remoteProcessingGroups;
+ private ProvenanceReportingSchema provenanceReportingProperties;
+
+ private ProvenanceRepositorySchema provenanceRepositorySchema;
+
+ public ConfigSchema() {
+ }
+
+ public ConfigSchema(Map map) {
+ Object obj ;
+ 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);
+
+ processors = getOptionalKeyAsType(map, PROCESSORS_KEY, List.class, TOP_LEVEL_NAME, null);
+ if (processors != null) {
+ transformListToType(processors, "processor", ProcessorSchema.class, PROCESSORS_KEY);
+ }
+
+ connections = getOptionalKeyAsType(map, CONNECTIONS_KEY, List.class, TOP_LEVEL_NAME, null);
+ if (connections != null) {
+ transformListToType(connections, "connection", ConnectionSchema.class, CONNECTIONS_KEY);
+ }
+
+ remoteProcessingGroups = getOptionalKeyAsType(map, REMOTE_PROCESSING_GROUPS_KEY, List.class, TOP_LEVEL_NAME, null);
+ if (remoteProcessingGroups != null) {
+ transformListToType(remoteProcessingGroups, "remote processing group", RemoteProcessingGroupSchema.class, REMOTE_PROCESSING_GROUPS_KEY);
+ }
+
+ provenanceReportingProperties = getMapAsType(map, PROVENANCE_REPORTING_KEY, ProvenanceReportingSchema.class, TOP_LEVEL_NAME, false);
+
+ addIssuesIfNotNull(flowControllerProperties);
+ addIssuesIfNotNull(coreProperties);
+ addIssuesIfNotNull(flowfileRepositoryProperties);
+ addIssuesIfNotNull(contentRepositoryProperties);
+ addIssuesIfNotNull(componentStatusRepositoryProperties);
+ addIssuesIfNotNull(securityProperties);
+ addIssuesIfNotNull(provenanceReportingProperties);
+ addIssuesIfNotNull(provenanceRepositorySchema);
+
+ if (processors != null) {
+ for (ProcessorSchema processorSchema : processors) {
+ addIssuesIfNotNull(processorSchema);
+ }
+ }
+
+ if (connections != null) {
+ for (ConnectionSchema connectionSchema : connections) {
+ addIssuesIfNotNull(connectionSchema);
+ }
+ }
+
+ if (remoteProcessingGroups != null) {
+ for (RemoteProcessingGroupSchema remoteProcessingGroupSchema : remoteProcessingGroups) {
+ addIssuesIfNotNull(remoteProcessingGroupSchema);
+ }
+ }
+ }
+
+ 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 List<ProcessorSchema> getProcessors() {
+ return processors;
+ }
+
+ public List<ConnectionSchema> getConnections() {
+ return connections;
+ }
+
+ public List<RemoteProcessingGroupSchema> getRemoteProcessingGroups() {
+ return remoteProcessingGroups;
+ }
+
+ public ProvenanceReportingSchema getProvenanceReportingProperties() {
+ return provenanceReportingProperties;
+ }
+
+ public ComponentStatusRepositorySchema getComponentStatusRepositoryProperties() {
+ return componentStatusRepositoryProperties;
+ }
+
+ public ProvenanceRepositorySchema getProvenanceRepositorySchema() {
+ return provenanceRepositorySchema;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ConnectionSchema.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ConnectionSchema.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ConnectionSchema.java
new file mode 100644
index 0000000..f7024f8
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ConnectionSchema.java
@@ -0,0 +1,93 @@
+/*
+ * 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.bootstrap.util.schema;
+
+import org.apache.nifi.minifi.bootstrap.util.schema.common.BaseSchema;
+
+import java.util.Map;
+
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.CONNECTIONS_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.NAME_KEY;
+
+/**
+ *
+ */
+public class ConnectionSchema extends BaseSchema {
+ public static final String SOURCE_NAME_KEY = "source name";
+ public static final String SOURCE_RELATIONSHIP_NAME_KEY = "source relationship name";
+ public static final String DESTINATION_NAME_KEY = "destination name";
+ public static final String MAX_WORK_QUEUE_SIZE_KEY = "max work queue size";
+ public static final String MAX_WORK_QUEUE_DATA_SIZE_KEY = "max work queue data size";
+ public static final String FLOWFILE_EXPIRATION__KEY = "flowfile expiration";
+ public static final String QUEUE_PRIORITIZER_CLASS_KEY = "queue prioritizer class";
+
+ private String name;
+ private String sourceName;
+ private String sourceRelationshipName;
+ private String destinationName;
+ private Number maxWorkQueueSize = 0;
+ private String maxWorkQueueDataSize = "0 MB";
+ private String flowfileExpiration = "0 sec";
+ private String queuePrioritizerClass = "";
+
+ public ConnectionSchema() {
+ }
+
+ public ConnectionSchema(Map map) {
+ name = getRequiredKeyAsType(map, NAME_KEY, String.class, CONNECTIONS_KEY);
+ sourceName = getRequiredKeyAsType(map, SOURCE_NAME_KEY, String.class, CONNECTIONS_KEY);
+ sourceRelationshipName = getRequiredKeyAsType(map, SOURCE_RELATIONSHIP_NAME_KEY, String.class, CONNECTIONS_KEY);
+ destinationName = getRequiredKeyAsType(map, DESTINATION_NAME_KEY, String.class, CONNECTIONS_KEY);
+
+ maxWorkQueueSize = getOptionalKeyAsType(map, MAX_WORK_QUEUE_SIZE_KEY, Number.class, CONNECTIONS_KEY, 0);
+ maxWorkQueueDataSize = getOptionalKeyAsType(map, MAX_WORK_QUEUE_DATA_SIZE_KEY, String.class, CONNECTIONS_KEY, "0 MB");
+ flowfileExpiration = getOptionalKeyAsType(map, FLOWFILE_EXPIRATION__KEY, String.class, CONNECTIONS_KEY, "0 sec");
+ queuePrioritizerClass = getOptionalKeyAsType(map, QUEUE_PRIORITIZER_CLASS_KEY, String.class, CONNECTIONS_KEY, "");
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public String getSourceName() {
+ return sourceName;
+ }
+
+ public String getSourceRelationshipName() {
+ return sourceRelationshipName;
+ }
+
+ public String getDestinationName() {
+ return destinationName;
+ }
+
+ public Number getMaxWorkQueueSize() {
+ return maxWorkQueueSize;
+ }
+
+ public String getMaxWorkQueueDataSize() {
+ return maxWorkQueueDataSize;
+ }
+
+ public String getFlowfileExpiration() {
+ return flowfileExpiration;
+ }
+
+ public String getQueuePrioritizerClass() {
+ return queuePrioritizerClass;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ContentRepositorySchema.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ContentRepositorySchema.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ContentRepositorySchema.java
new file mode 100644
index 0000000..1a4d07d
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ContentRepositorySchema.java
@@ -0,0 +1,57 @@
+/*
+ * 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.bootstrap.util.schema;
+
+import org.apache.nifi.minifi.bootstrap.util.schema.common.BaseSchema;
+
+import java.util.Map;
+
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.ALWAYS_SYNC_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.CONTENT_REPO_KEY;
+
+/**
+ *
+ */
+public class ContentRepositorySchema extends BaseSchema {
+ public static final String CONTENT_CLAIM_MAX_APPENDABLE_SIZE_KEY = "content claim max appendable size";
+ public static final String CONTENT_CLAIM_MAX_FLOW_FILES_KEY = "content claim max flow files";
+
+ private String contentClaimMaxAppendableSize = "10 MB";
+ private Number contentClaimMaxFlowFiles = 100;
+ private Boolean alwaysSync = false;
+
+ public ContentRepositorySchema() {
+ }
+
+ public ContentRepositorySchema(Map map) {
+ contentClaimMaxAppendableSize = getOptionalKeyAsType(map, CONTENT_CLAIM_MAX_APPENDABLE_SIZE_KEY, String.class, CONTENT_REPO_KEY, "10 MB");
+ contentClaimMaxFlowFiles = getOptionalKeyAsType(map, CONTENT_CLAIM_MAX_FLOW_FILES_KEY, Number.class, CONTENT_REPO_KEY, 100);
+ alwaysSync = getOptionalKeyAsType(map, ALWAYS_SYNC_KEY, Boolean.class, CONTENT_REPO_KEY, false);
+ }
+
+ public String getContentClaimMaxAppendableSize() {
+ return contentClaimMaxAppendableSize;
+ }
+
+ public Number getContentClaimMaxFlowFiles() {
+ return contentClaimMaxFlowFiles;
+ }
+
+ public boolean getAlwaysSync() {
+ return alwaysSync;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/CorePropertiesSchema.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/CorePropertiesSchema.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/CorePropertiesSchema.java
new file mode 100644
index 0000000..d4bae0b
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/CorePropertiesSchema.java
@@ -0,0 +1,73 @@
+/*
+ * 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.bootstrap.util.schema;
+
+import org.apache.nifi.minifi.bootstrap.util.schema.common.BaseSchema;
+
+import java.util.Map;
+
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.CORE_PROPS_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.MAX_CONCURRENT_THREADS_KEY;
+
+/**
+ *
+ */
+public class CorePropertiesSchema extends BaseSchema {
+
+ public static final String FLOW_CONTROLLER_SHUTDOWN_PERIOD_KEY = "flow controller graceful shutdown period";
+ public static final String FLOW_SERVICE_WRITE_DELAY_INTERVAL_KEY = "flow service write delay interval";
+ public static final String ADMINISTRATIVE_YIELD_DURATION_KEY = "administrative yield duration";
+ public static final String BORED_YIELD_DURATION_KEY = "bored yield duration";
+
+
+ private String flowControllerGracefulShutdownPeriod = "10 sec";
+ private String flowServiceWriteDelayInterval = "500 ms";
+ private String administrativeYieldDuration = "30 sec";
+ private String boredYieldDuration = "10 millis";
+ private Number maxConcurrentThreads = 1;
+
+ public CorePropertiesSchema() {
+ }
+
+ public CorePropertiesSchema(Map map) {
+ flowControllerGracefulShutdownPeriod = getOptionalKeyAsType(map, FLOW_CONTROLLER_SHUTDOWN_PERIOD_KEY, String.class, CORE_PROPS_KEY, "10 sec");
+ flowServiceWriteDelayInterval = getOptionalKeyAsType(map, FLOW_SERVICE_WRITE_DELAY_INTERVAL_KEY, String.class, CORE_PROPS_KEY, "500 ms");
+ administrativeYieldDuration = getOptionalKeyAsType(map, ADMINISTRATIVE_YIELD_DURATION_KEY, String.class, CORE_PROPS_KEY, "30 sec");
+ boredYieldDuration = getOptionalKeyAsType(map, BORED_YIELD_DURATION_KEY, String.class, CORE_PROPS_KEY, "10 millis");
+ maxConcurrentThreads = getOptionalKeyAsType(map, MAX_CONCURRENT_THREADS_KEY, Number.class, CORE_PROPS_KEY, 1);
+ }
+
+ public String getFlowControllerGracefulShutdownPeriod() {
+ return flowControllerGracefulShutdownPeriod;
+ }
+
+ public String getFlowServiceWriteDelayInterval() {
+ return flowServiceWriteDelayInterval;
+ }
+
+ public String getAdministrativeYieldDuration() {
+ return administrativeYieldDuration;
+ }
+
+ public String getBoredYieldDuration() {
+ return boredYieldDuration;
+ }
+
+ public Number getMaxConcurrentThreads() {
+ return maxConcurrentThreads;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/FlowControllerSchema.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/FlowControllerSchema.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/FlowControllerSchema.java
new file mode 100644
index 0000000..1d7500a
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/FlowControllerSchema.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.minifi.bootstrap.util.schema;
+
+import org.apache.nifi.minifi.bootstrap.util.schema.common.BaseSchema;
+
+import java.util.Map;
+
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.COMMENT_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.FLOW_CONTROLLER_PROPS_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.NAME_KEY;
+
+/**
+ *
+ */
+public class FlowControllerSchema extends BaseSchema {
+
+ private String name;
+ private String comment = "";
+
+ public FlowControllerSchema() {
+ }
+
+ public FlowControllerSchema(Map map) {
+ name = getRequiredKeyAsType(map, NAME_KEY, String.class, FLOW_CONTROLLER_PROPS_KEY);
+
+ comment = getOptionalKeyAsType(map, COMMENT_KEY, String.class, FLOW_CONTROLLER_PROPS_KEY, "");
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public String getComment() {
+ return comment;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/FlowFileRepositorySchema.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/FlowFileRepositorySchema.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/FlowFileRepositorySchema.java
new file mode 100644
index 0000000..10a8c55
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/FlowFileRepositorySchema.java
@@ -0,0 +1,68 @@
+/*
+ * 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.bootstrap.util.schema;
+
+import org.apache.nifi.minifi.bootstrap.util.schema.common.BaseSchema;
+
+import java.util.Map;
+
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.ALWAYS_SYNC_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.FLOWFILE_REPO_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.SWAP_PROPS_KEY;
+
+/**
+ *
+ */
+public class FlowFileRepositorySchema extends BaseSchema {
+
+ public static final String PARTITIONS_KEY = "partitions";
+ public static final String CHECKPOINT_INTERVAL_KEY = "checkpoint interval";
+
+ private Number partitions = 256;
+ private String checkpointInterval = "2 mins";
+ private Boolean alwaysSync = false;
+ private SwapSchema swapProperties;
+
+ public FlowFileRepositorySchema() {
+ swapProperties = new SwapSchema();
+ }
+
+ public FlowFileRepositorySchema(Map map) {
+ partitions = getOptionalKeyAsType(map, PARTITIONS_KEY, Number.class, FLOWFILE_REPO_KEY, 256);
+ checkpointInterval = getOptionalKeyAsType(map, CHECKPOINT_INTERVAL_KEY, String.class, FLOWFILE_REPO_KEY, "2 mins");
+ alwaysSync = getOptionalKeyAsType(map, ALWAYS_SYNC_KEY, Boolean.class, FLOWFILE_REPO_KEY, false);
+
+ swapProperties = getMapAsType(map, SWAP_PROPS_KEY, SwapSchema.class, FLOWFILE_REPO_KEY, false);
+ addIssuesIfNotNull(swapProperties);
+ }
+
+ public Number getPartitions() {
+ return partitions;
+ }
+
+ public String getCheckpointInterval() {
+ return checkpointInterval;
+ }
+
+ public boolean getAlwaysSync() {
+ return alwaysSync;
+ }
+
+ public SwapSchema getSwapProperties() {
+ return swapProperties;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/6f528bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ProcessorSchema.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ProcessorSchema.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ProcessorSchema.java
new file mode 100644
index 0000000..bb86c8c
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/schema/ProcessorSchema.java
@@ -0,0 +1,127 @@
+/*
+ * 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.bootstrap.util.schema;
+
+import org.apache.nifi.minifi.bootstrap.util.schema.common.BaseSchema;
+import org.apache.nifi.scheduling.SchedulingStrategy;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.MAX_CONCURRENT_TASKS_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.NAME_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.PROCESSORS_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.SCHEDULING_PERIOD_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.SCHEDULING_STRATEGY_KEY;
+import static org.apache.nifi.minifi.bootstrap.util.schema.common.CommonPropertyKeys.YIELD_PERIOD_KEY;
+
+/**
+ *
+ */
+public class ProcessorSchema extends BaseSchema {
+ 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";
+
+
+ private String name;
+ private String processorClass;
+ private Number maxConcurrentTasks = 1;
+ private String schedulingStrategy;
+ private String schedulingPeriod;
+ private String penalizationPeriod = "30 sec";
+ private String yieldPeriod = "1 sec";
+ private Number runDurationNanos = 0;
+ private List<String> autoTerminatedRelationshipsList = Collections.emptyList();
+ private Map<String, Object> properties = Collections.emptyMap();
+
+ public ProcessorSchema() {
+ }
+
+ public ProcessorSchema(Map map) {
+ name = getRequiredKeyAsType(map, NAME_KEY, String.class, PROCESSORS_KEY);
+ processorClass = getRequiredKeyAsType(map, CLASS_KEY, String.class, PROCESSORS_KEY);
+
+ maxConcurrentTasks = getOptionalKeyAsType(map, MAX_CONCURRENT_TASKS_KEY, Number.class, PROCESSORS_KEY, 1);
+
+ schedulingStrategy = getRequiredKeyAsType(map, SCHEDULING_STRATEGY_KEY, String.class, PROCESSORS_KEY);
+ try {
+ SchedulingStrategy.valueOf(schedulingStrategy);
+ } catch (IllegalArgumentException e) {
+ addValidationIssue(SCHEDULING_STRATEGY_KEY, PROCESSORS_KEY, "it is not a valid scheduling strategy");
+ }
+
+ schedulingPeriod = getRequiredKeyAsType(map, SCHEDULING_PERIOD_KEY, String.class, PROCESSORS_KEY);
+
+ penalizationPeriod = getOptionalKeyAsType(map, PENALIZATION_PERIOD_KEY, String.class, PROCESSORS_KEY, "30 sec");
+
+ yieldPeriod = getOptionalKeyAsType(map, YIELD_PERIOD_KEY, String.class, PROCESSORS_KEY, "1 sec");
+
+ runDurationNanos = getOptionalKeyAsType(map, RUN_DURATION_NANOS_KEY, Number.class, PROCESSORS_KEY, 0);
+
+ autoTerminatedRelationshipsList = getOptionalKeyAsType(map, AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY, List.class, PROCESSORS_KEY, null);
+
+ properties = getOptionalKeyAsType(map, PROCESSOR_PROPS_KEY, Map.class, PROCESSORS_KEY, null);
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getProcessorClass() {
+ return processorClass;
+ }
+
+ public Number getMaxConcurrentTasks() {
+ return maxConcurrentTasks;
+ }
+
+ public String getSchedulingStrategy() {
+ return schedulingStrategy;
+ }
+
+ public String getSchedulingPeriod() {
+ return schedulingPeriod;
+ }
+
+ public String getPenalizationPeriod() {
+ return penalizationPeriod;
+ }
+
+ public String getYieldPeriod() {
+ return yieldPeriod;
+ }
+
+ public Number getRunDurationNanos() {
+ return runDurationNanos;
+ }
+
+ public List<String> getAutoTerminatedRelationshipsList() {
+ return autoTerminatedRelationshipsList;
+ }
+
+ public Map<String, Object> getProperties() {
+ return properties;
+ }
+}