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/10/31 17:39:01 UTC
[2/3] nifi-minifi git commit: MINIFI-117 - Maintainable Configuration
Versioning
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/1bbeedf6/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1.java
new file mode 100644
index 0000000..2ebf28a
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1.java
@@ -0,0 +1,263 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. See the NOTICE file distributed with
+ * * this work for additional information regarding copyright ownership.
+ * * The ASF licenses this file to You under the Apache License, Version 2.0
+ * * (the "License"); you may not use this file except in compliance with
+ * * the License. You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing, software
+ * * distributed under the License is distributed on an "AS IS" BASIS,
+ * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * * See the License for the specific language governing permissions and
+ * * limitations under the License.
+ *
+ */
+
+package org.apache.nifi.minifi.commons.schema.v1;
+
+import org.apache.nifi.minifi.commons.schema.ComponentStatusRepositorySchema;
+import org.apache.nifi.minifi.commons.schema.ConfigSchema;
+import org.apache.nifi.minifi.commons.schema.ConnectionSchema;
+import org.apache.nifi.minifi.commons.schema.ContentRepositorySchema;
+import org.apache.nifi.minifi.commons.schema.CorePropertiesSchema;
+import org.apache.nifi.minifi.commons.schema.FlowControllerSchema;
+import org.apache.nifi.minifi.commons.schema.FlowFileRepositorySchema;
+import org.apache.nifi.minifi.commons.schema.ProcessorSchema;
+import org.apache.nifi.minifi.commons.schema.ProvenanceReportingSchema;
+import org.apache.nifi.minifi.commons.schema.ProvenanceRepositorySchema;
+import org.apache.nifi.minifi.commons.schema.RemoteInputPortSchema;
+import org.apache.nifi.minifi.commons.schema.RemoteProcessingGroupSchema;
+import org.apache.nifi.minifi.commons.schema.SecurityPropertiesSchema;
+import org.apache.nifi.minifi.commons.schema.common.BaseSchema;
+import org.apache.nifi.minifi.commons.schema.common.ConvertableSchema;
+import org.apache.nifi.minifi.commons.schema.common.StringUtil;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.nifi.minifi.commons.schema.ConfigSchema.TOP_LEVEL_NAME;
+import static org.apache.nifi.minifi.commons.schema.ConfigSchema.VERSION;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.COMPONENT_STATUS_REPO_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CONNECTIONS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CONTENT_REPO_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CORE_PROPS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.FLOWFILE_REPO_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.FLOW_CONTROLLER_PROPS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROCESSORS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROVENANCE_REPORTING_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROVENANCE_REPO_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.REMOTE_PROCESSING_GROUPS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SECURITY_PROPS_KEY;
+
+public class ConfigSchemaV1 extends BaseSchema implements ConvertableSchema<ConfigSchema> {
+ public static final String FOUND_THE_FOLLOWING_DUPLICATE_PROCESSOR_NAMES = "Found the following duplicate processor names: ";
+ public static final String FOUND_THE_FOLLOWING_DUPLICATE_CONNECTION_NAMES = "Found the following duplicate connection names: ";
+ public static final String FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_PROCESSING_GROUP_NAMES = "Found the following duplicate remote processing group names: ";
+ public static final String CANNOT_LOOK_UP_PROCESSOR_ID_FROM_PROCESSOR_NAME_DUE_TO_DUPLICATE_PROCESSOR_NAMES = "Cannot look up Processor id from Processor name due to duplicate Processor names: ";
+ public static final int CONFIG_VERSION = 1;
+ public static final String CONNECTION_WITH_NAME = "Connection with name ";
+ public static final String HAS_INVALID_DESTINATION_NAME = " has invalid destination name ";
+ public static final String HAS_INVALID_SOURCE_NAME = " has invalid source name ";
+ private FlowControllerSchema flowControllerProperties;
+ private CorePropertiesSchema coreProperties;
+ private FlowFileRepositorySchema flowfileRepositoryProperties;
+ private ContentRepositorySchema contentRepositoryProperties;
+ private ComponentStatusRepositorySchema componentStatusRepositoryProperties;
+ private SecurityPropertiesSchema securityProperties;
+ private List<ProcessorSchemaV1> processors;
+ private List<ConnectionSchemaV1> connections;
+ private List<RemoteProcessingGroupSchema> remoteProcessingGroups;
+ private ProvenanceReportingSchema provenanceReportingProperties;
+
+ private ProvenanceRepositorySchema provenanceRepositorySchema;
+
+ public ConfigSchemaV1(Map map) {
+ 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 = convertListToType(getOptionalKeyAsType(map, PROCESSORS_KEY, List.class, TOP_LEVEL_NAME, new ArrayList<>()), PROCESSORS_KEY, ProcessorSchemaV1.class, TOP_LEVEL_NAME);
+
+ remoteProcessingGroups = convertListToType(getOptionalKeyAsType(map, REMOTE_PROCESSING_GROUPS_KEY, List.class, TOP_LEVEL_NAME, new ArrayList<>()), "remote processing group",
+ RemoteProcessingGroupSchema.class, REMOTE_PROCESSING_GROUPS_KEY);
+
+ connections = convertListToType(getOptionalKeyAsType(map, CONNECTIONS_KEY, List.class, TOP_LEVEL_NAME, new ArrayList<>()), CONNECTIONS_KEY, ConnectionSchemaV1.class, TOP_LEVEL_NAME);
+
+ provenanceReportingProperties = getMapAsType(map, PROVENANCE_REPORTING_KEY, ProvenanceReportingSchema.class, TOP_LEVEL_NAME, false, false);
+
+ addIssuesIfNotNull(flowControllerProperties);
+ addIssuesIfNotNull(coreProperties);
+ addIssuesIfNotNull(flowfileRepositoryProperties);
+ addIssuesIfNotNull(contentRepositoryProperties);
+ addIssuesIfNotNull(componentStatusRepositoryProperties);
+ addIssuesIfNotNull(securityProperties);
+ addIssuesIfNotNull(provenanceReportingProperties);
+ addIssuesIfNotNull(provenanceRepositorySchema);
+ addIssuesIfNotNull(processors);
+ addIssuesIfNotNull(connections);
+ addIssuesIfNotNull(remoteProcessingGroups);
+
+ List<String> processorNames = processors.stream().map(ProcessorSchemaV1::getName).collect(Collectors.toList());
+
+ checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_PROCESSOR_NAMES, processorNames);
+ checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_CONNECTION_NAMES, connections.stream().map(ConnectionSchemaV1::getName).collect(Collectors.toList()));
+ checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_PROCESSING_GROUP_NAMES, remoteProcessingGroups.stream().map(RemoteProcessingGroupSchema::getName)
+ .collect(Collectors.toList()));
+
+ Set<String> connectableNames = new HashSet<>(processorNames);
+ connectableNames.addAll(remoteProcessingGroups.stream().flatMap(r -> r.getInputPorts().stream()).map(RemoteInputPortSchema::getId).collect(Collectors.toList()));
+ connections.forEach(c -> {
+ String destinationName = c.getDestinationName();
+ if (!StringUtil.isNullOrEmpty(destinationName) && !connectableNames.contains(destinationName)) {
+ addValidationIssue(CONNECTION_WITH_NAME + c.getName() + HAS_INVALID_DESTINATION_NAME + destinationName);
+ }
+ String sourceName = c.getSourceName();
+ if (!StringUtil.isNullOrEmpty(sourceName) && !connectableNames.contains(sourceName)) {
+ addValidationIssue(CONNECTION_WITH_NAME + c.getName() + HAS_INVALID_SOURCE_NAME + sourceName);
+ }
+ });
+ }
+
+ protected List<ProcessorSchema> getProcessorSchemas() {
+ Map<String, Integer> idMap = new HashMap<>();
+ List<ProcessorSchema> processorSchemas = new ArrayList<>(processors.size());
+
+ for (ProcessorSchemaV1 processor : processors) {
+ ProcessorSchema processorSchema = processor.convert();
+ processorSchema.setId(getUniqueId(idMap, processorSchema.getName()));
+ processorSchemas.add(processorSchema);
+ }
+
+ return processorSchemas;
+ }
+
+ protected List<ConnectionSchema> getConnectionSchemas(List<ProcessorSchema> processors, List<String> validationIssues) {
+ Map<String, Integer> idMap = new HashMap<>();
+
+ Map<String, String> processorNameToIdMap = new HashMap<>();
+
+ // We can't look up id by name for names that appear more than once
+ Set<String> duplicateProcessorNames = new HashSet<>();
+
+ if (processors != null) {
+ processors.stream().forEachOrdered(p -> processorNameToIdMap.put(p.getName(), p.getId()));
+
+ Set<String> processorNames = new HashSet<>();
+ processors.stream().map(ProcessorSchema::getName).forEachOrdered(n -> {
+ if (!processorNames.add(n)) {
+ duplicateProcessorNames.add(n);
+ }
+ });
+ }
+
+ Set<String> remoteInputPortIds = new HashSet<>();
+ if (remoteProcessingGroups != null) {
+ remoteInputPortIds.addAll(remoteProcessingGroups.stream().filter(r -> r.getInputPorts() != null)
+ .flatMap(r -> r.getInputPorts().stream()).map(RemoteInputPortSchema::getId).collect(Collectors.toSet()));
+ }
+
+ Set<String> problematicDuplicateNames = new HashSet<>();
+
+ List<ConnectionSchema> connectionSchemas = new ArrayList<>(connections.size());
+ for (ConnectionSchemaV1 connection : connections) {
+ ConnectionSchema convert = connection.convert();
+ convert.setId(getUniqueId(idMap, convert.getName()));
+
+ String sourceName = connection.getSourceName();
+ if (remoteInputPortIds.contains(sourceName)) {
+ convert.setSourceId(sourceName);
+ } else {
+ if (duplicateProcessorNames.contains(sourceName)) {
+ problematicDuplicateNames.add(sourceName);
+ }
+ String sourceId = processorNameToIdMap.get(sourceName);
+ if (!StringUtil.isNullOrEmpty(sourceId)) {
+ convert.setSourceId(sourceId);
+ }
+ }
+
+ String destinationName = connection.getDestinationName();
+ if (remoteInputPortIds.contains(destinationName)) {
+ convert.setDestinationId(destinationName);
+ } else {
+ if (duplicateProcessorNames.contains(destinationName)) {
+ problematicDuplicateNames.add(destinationName);
+ }
+ String destinationId = processorNameToIdMap.get(destinationName);
+ if (!StringUtil.isNullOrEmpty(destinationId)) {
+ convert.setDestinationId(destinationId);
+ }
+ }
+ connectionSchemas.add(convert);
+ }
+
+ if (problematicDuplicateNames.size() > 0) {
+ validationIssues.add(CANNOT_LOOK_UP_PROCESSOR_ID_FROM_PROCESSOR_NAME_DUE_TO_DUPLICATE_PROCESSOR_NAMES
+ + problematicDuplicateNames.stream().collect(Collectors.joining(", ")));
+ }
+ return connectionSchemas;
+ }
+
+ @Override
+ public ConfigSchema convert() {
+ Map<String, Object> map = new HashMap<>();
+ map.put(VERSION, getVersion());
+ putIfNotNull(map, FLOW_CONTROLLER_PROPS_KEY, flowControllerProperties);
+ putIfNotNull(map, CORE_PROPS_KEY, coreProperties);
+ putIfNotNull(map, FLOWFILE_REPO_KEY, flowfileRepositoryProperties);
+ putIfNotNull(map, CONTENT_REPO_KEY, contentRepositoryProperties);
+ putIfNotNull(map, PROVENANCE_REPO_KEY, provenanceRepositorySchema);
+ putIfNotNull(map, COMPONENT_STATUS_REPO_KEY, componentStatusRepositoryProperties);
+ putIfNotNull(map, SECURITY_PROPS_KEY, securityProperties);
+ List<ProcessorSchema> processorSchemas = getProcessorSchemas();
+ putListIfNotNull(map, PROCESSORS_KEY, processorSchemas);
+ List<String> validationIssues = getValidationIssues();
+ putListIfNotNull(map, CONNECTIONS_KEY, getConnectionSchemas(processorSchemas, validationIssues));
+ putListIfNotNull(map, REMOTE_PROCESSING_GROUPS_KEY, remoteProcessingGroups);
+ putIfNotNull(map, PROVENANCE_REPORTING_KEY, provenanceReportingProperties);
+ return new ConfigSchema(map, validationIssues);
+ }
+
+ /**
+ * Will replace all characters not in [A-Za-z0-9_] with _
+ * <p>
+ * This has potential for collisions so it will also append numbers as necessary to prevent that
+ *
+ * @param ids id map of already incremented numbers
+ * @param name the name
+ * @return a unique filesystem-friendly id
+ */
+ public static String getUniqueId(Map<String, Integer> ids, String name) {
+ String baseId = StringUtil.isNullOrEmpty(name) ? EMPTY_NAME : ID_REPLACE_PATTERN.matcher(name).replaceAll("_");
+ String id = baseId;
+ Integer idNum = ids.get(baseId);
+ while (ids.containsKey(id)) {
+ id = baseId + "_" + idNum++;
+ }
+ // Using != on a string comparison here is intentional. The two will be reference equal iff the body of the while loop was never executed.
+ if (id != baseId) {
+ ids.put(baseId, idNum);
+ }
+ ids.put(id, 2);
+ return id;
+ }
+
+ @Override
+ public int getVersion() {
+ return CONFIG_VERSION;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/1bbeedf6/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ConnectionSchemaV1.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ConnectionSchemaV1.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ConnectionSchemaV1.java
new file mode 100644
index 0000000..cdf9290
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ConnectionSchemaV1.java
@@ -0,0 +1,104 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. See the NOTICE file distributed with
+ * * this work for additional information regarding copyright ownership.
+ * * The ASF licenses this file to You under the Apache License, Version 2.0
+ * * (the "License"); you may not use this file except in compliance with
+ * * the License. You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing, software
+ * * distributed under the License is distributed on an "AS IS" BASIS,
+ * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * * See the License for the specific language governing permissions and
+ * * limitations under the License.
+ *
+ */
+
+package org.apache.nifi.minifi.commons.schema.v1;
+
+import org.apache.nifi.minifi.commons.schema.ConnectionSchema;
+import org.apache.nifi.minifi.commons.schema.common.BaseSchema;
+import org.apache.nifi.minifi.commons.schema.common.ConvertableSchema;
+import org.apache.nifi.minifi.commons.schema.common.StringUtil;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.nifi.minifi.commons.schema.ConnectionSchema.DEFAULT_FLOWFILE_EXPIRATION;
+import static org.apache.nifi.minifi.commons.schema.ConnectionSchema.DEFAULT_MAX_QUEUE_DATA_SIZE;
+import static org.apache.nifi.minifi.commons.schema.ConnectionSchema.DEFAULT_MAX_WORK_QUEUE_SIZE;
+import static org.apache.nifi.minifi.commons.schema.ConnectionSchema.FLOWFILE_EXPIRATION__KEY;
+import static org.apache.nifi.minifi.commons.schema.ConnectionSchema.MAX_WORK_QUEUE_DATA_SIZE_KEY;
+import static org.apache.nifi.minifi.commons.schema.ConnectionSchema.MAX_WORK_QUEUE_SIZE_KEY;
+import static org.apache.nifi.minifi.commons.schema.ConnectionSchema.QUEUE_PRIORITIZER_CLASS_KEY;
+import static org.apache.nifi.minifi.commons.schema.ConnectionSchema.SOURCE_RELATIONSHIP_NAMES_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CONNECTIONS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.NAME_KEY;
+
+public class ConnectionSchemaV1 extends BaseSchema implements ConvertableSchema<ConnectionSchema> {
+ public static final String SOURCE_RELATIONSHIP_NAME_KEY = "source relationship name";
+ public static final String DESTINATION_NAME_KEY = "destination name";
+ public static final String SOURCE_NAME_KEY = "source name";
+
+ private String name;
+
+ private String sourceRelationshipName;
+ private String destinationName;
+
+ private String sourceName;
+
+ private Number maxWorkQueueSize = DEFAULT_MAX_WORK_QUEUE_SIZE;
+ private String maxWorkQueueDataSize = DEFAULT_MAX_QUEUE_DATA_SIZE;
+ private String flowfileExpiration = DEFAULT_FLOWFILE_EXPIRATION;
+ private String queuePrioritizerClass;
+
+ public ConnectionSchemaV1(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, DEFAULT_MAX_WORK_QUEUE_SIZE);
+ maxWorkQueueDataSize = getOptionalKeyAsType(map, MAX_WORK_QUEUE_DATA_SIZE_KEY, String.class, CONNECTIONS_KEY, DEFAULT_MAX_QUEUE_DATA_SIZE);
+ flowfileExpiration = getOptionalKeyAsType(map, FLOWFILE_EXPIRATION__KEY, String.class, CONNECTIONS_KEY, DEFAULT_FLOWFILE_EXPIRATION);
+ queuePrioritizerClass = getOptionalKeyAsType(map, QUEUE_PRIORITIZER_CLASS_KEY, String.class, CONNECTIONS_KEY, "");
+ }
+
+ @Override
+ public ConnectionSchema convert() {
+ Map<String, Object> map = new HashMap<>();
+ map.put(NAME_KEY, name);
+ if (StringUtil.isNullOrEmpty(sourceRelationshipName)) {
+ map.put(SOURCE_RELATIONSHIP_NAMES_KEY, new ArrayList<>());
+ } else {
+ map.put(SOURCE_RELATIONSHIP_NAMES_KEY, new ArrayList<>(Arrays.asList(sourceRelationshipName)));
+ }
+ map.put(MAX_WORK_QUEUE_SIZE_KEY, maxWorkQueueSize);
+ map.put(MAX_WORK_QUEUE_DATA_SIZE_KEY, maxWorkQueueDataSize);
+ map.put(FLOWFILE_EXPIRATION__KEY, flowfileExpiration);
+ map.put(QUEUE_PRIORITIZER_CLASS_KEY, queuePrioritizerClass);
+ return new ConnectionSchema(map);
+ }
+
+ public String getSourceName() {
+ return sourceName;
+ }
+
+ public String getDestinationName() {
+ return destinationName;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ @Override
+ public int getVersion() {
+ return ConfigSchemaV1.CONFIG_VERSION;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/1bbeedf6/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1.java
new file mode 100644
index 0000000..d64c98b
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1.java
@@ -0,0 +1,103 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. See the NOTICE file distributed with
+ * * this work for additional information regarding copyright ownership.
+ * * The ASF licenses this file to You under the Apache License, Version 2.0
+ * * (the "License"); you may not use this file except in compliance with
+ * * the License. You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing, software
+ * * distributed under the License is distributed on an "AS IS" BASIS,
+ * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * * See the License for the specific language governing permissions and
+ * * limitations under the License.
+ *
+ */
+
+package org.apache.nifi.minifi.commons.schema.v1;
+
+import org.apache.nifi.minifi.commons.schema.ProcessorSchema;
+import org.apache.nifi.minifi.commons.schema.common.BaseSchema;
+import org.apache.nifi.minifi.commons.schema.common.ConvertableSchema;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.CLASS_KEY;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_AUTO_TERMINATED_RELATIONSHIPS_LIST;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_MAX_CONCURRENT_TASKS;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_PENALIZATION_PERIOD;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_PROPERTIES;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_RUN_DURATION_NANOS;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_YIELD_DURATION;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.IT_IS_NOT_A_VALID_SCHEDULING_STRATEGY;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.PENALIZATION_PERIOD_KEY;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.PROCESSOR_PROPS_KEY;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.RUN_DURATION_NANOS_KEY;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.isSchedulingStrategy;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.MAX_CONCURRENT_TASKS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.NAME_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROCESSORS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SCHEDULING_PERIOD_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SCHEDULING_STRATEGY_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.YIELD_PERIOD_KEY;
+
+public class ProcessorSchemaV1 extends BaseSchema implements ConvertableSchema<ProcessorSchema> {
+ private String name;
+ private String processorClass;
+ private String schedulingStrategy;
+ private String schedulingPeriod;
+ private Number maxConcurrentTasks = DEFAULT_MAX_CONCURRENT_TASKS;
+ private String penalizationPeriod = DEFAULT_PENALIZATION_PERIOD;
+ private String yieldPeriod = DEFAULT_YIELD_DURATION;
+ private Number runDurationNanos = DEFAULT_RUN_DURATION_NANOS;
+ private List<String> autoTerminatedRelationshipsList = DEFAULT_AUTO_TERMINATED_RELATIONSHIPS_LIST;
+ private Map<String, Object> properties = DEFAULT_PROPERTIES;
+
+ public ProcessorSchemaV1(Map map) {
+ name = getRequiredKeyAsType(map, NAME_KEY, String.class, PROCESSORS_KEY);
+ processorClass = getRequiredKeyAsType(map, CLASS_KEY, String.class, PROCESSORS_KEY);
+ schedulingStrategy = getRequiredKeyAsType(map, SCHEDULING_STRATEGY_KEY, String.class, PROCESSORS_KEY);
+ if (schedulingStrategy != null && !isSchedulingStrategy(schedulingStrategy)) {
+ addValidationIssue(SCHEDULING_STRATEGY_KEY, PROCESSORS_KEY, IT_IS_NOT_A_VALID_SCHEDULING_STRATEGY);
+ }
+ schedulingPeriod = getRequiredKeyAsType(map, SCHEDULING_PERIOD_KEY, String.class, PROCESSORS_KEY);
+
+ maxConcurrentTasks = getOptionalKeyAsType(map, MAX_CONCURRENT_TASKS_KEY, Number.class, PROCESSORS_KEY, DEFAULT_MAX_CONCURRENT_TASKS);
+ penalizationPeriod = getOptionalKeyAsType(map, PENALIZATION_PERIOD_KEY, String.class, PROCESSORS_KEY, DEFAULT_PENALIZATION_PERIOD);
+ yieldPeriod = getOptionalKeyAsType(map, YIELD_PERIOD_KEY, String.class, PROCESSORS_KEY, DEFAULT_YIELD_DURATION);
+ runDurationNanos = getOptionalKeyAsType(map, RUN_DURATION_NANOS_KEY, Number.class, PROCESSORS_KEY, DEFAULT_RUN_DURATION_NANOS);
+ autoTerminatedRelationshipsList = getOptionalKeyAsType(map, AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY, List.class, PROCESSORS_KEY, DEFAULT_AUTO_TERMINATED_RELATIONSHIPS_LIST);
+ properties = getOptionalKeyAsType(map, PROCESSOR_PROPS_KEY, Map.class, PROCESSORS_KEY, DEFAULT_PROPERTIES);
+ }
+
+ @Override
+ public ProcessorSchema convert() {
+ Map<String, Object> map = new HashMap<>();
+ map.put(NAME_KEY, name);
+ map.put(CLASS_KEY, processorClass);
+ map.put(MAX_CONCURRENT_TASKS_KEY, maxConcurrentTasks);
+ map.put(SCHEDULING_STRATEGY_KEY, schedulingStrategy);
+ map.put(SCHEDULING_PERIOD_KEY, schedulingPeriod);
+ map.put(PENALIZATION_PERIOD_KEY, penalizationPeriod);
+ map.put(YIELD_PERIOD_KEY, yieldPeriod);
+ map.put(RUN_DURATION_NANOS_KEY, runDurationNanos);
+ map.put(AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY, autoTerminatedRelationshipsList);
+ map.put(PROCESSOR_PROPS_KEY, new HashMap<>(properties));
+ return new ProcessorSchema(map);
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ @Override
+ public int getVersion() {
+ return ConfigSchemaV1.CONFIG_VERSION;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/1bbeedf6/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/ConfigSchemaTest.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/ConfigSchemaTest.java b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/ConfigSchemaTest.java
index 2713d15..afca6e0 100644
--- a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/ConfigSchemaTest.java
+++ b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/ConfigSchemaTest.java
@@ -17,7 +17,6 @@
package org.apache.nifi.minifi.commons.schema;
-import org.apache.nifi.minifi.commons.schema.common.BaseSchema;
import org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys;
import org.apache.nifi.minifi.commons.schema.exception.SchemaLoaderException;
import org.apache.nifi.minifi.commons.schema.serialization.SchemaLoader;
@@ -26,75 +25,53 @@ import org.junit.Test;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.ID_KEY;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.fail;
public class ConfigSchemaTest {
@Test
- public void testGetUniqueIdEmptySet() {
- String testId = "testId";
- assertEquals(testId + "___", ConfigSchema.getUniqueId(new HashMap<>(), testId + "/ $"));
- }
-
- @Test
- public void testConnectionGeneratedIds() {
- List<Map<String, Object>> listWithKeyValues = getListWithKeyValues(CommonPropertyKeys.NAME_KEY, "test", "test", "test_2");
-
- // These ids should be honored even though they're last
- listWithKeyValues.addAll(getListWithKeyValues(CommonPropertyKeys.ID_KEY, "test", "test_2"));
-
- ConfigSchema configSchema = new ConfigSchema(Collections.singletonMap(CommonPropertyKeys.CONNECTIONS_KEY, listWithKeyValues));
- assertMessageDoesNotExist(configSchema, ConfigSchema.FOUND_THE_FOLLOWING_DUPLICATE_CONNECTION_IDS);
- List<ConnectionSchema> connections = configSchema.getConnections();
- assertEquals(5, connections.size());
-
- // Generated unique ids
- assertEquals("test_3", connections.get(0).getId());
- assertEquals("test_4", connections.get(1).getId());
- assertEquals("test_2_2", connections.get(2).getId());
-
- // Specified ids
- assertEquals("test", connections.get(3).getId());
- assertEquals("test_2", connections.get(4).getId());
+ public void testValid() throws IOException, SchemaLoaderException {
+ Map<String, Object> yamlAsMap = SchemaLoader.loadYamlAsMap(ConfigSchemaTest.class.getClassLoader().getResourceAsStream("config-minimal-v2.yml"));
+ ConfigSchema configSchema = new ConfigSchema(yamlAsMap);
+ List<String> validationIssues = configSchema.getValidationIssues();
+ assertEquals(0, validationIssues.size());
}
@Test
- public void testGetUniqueIdConflicts() {
- Map<String, Integer> ids = new HashMap<>();
- assertEquals("test_id", ConfigSchema.getUniqueId(ids, "test/id"));
- assertEquals("test_id_2", ConfigSchema.getUniqueId(ids, "test$id"));
- assertEquals("test_id_3", ConfigSchema.getUniqueId(ids, "test$id"));
- assertEquals("test_id_4", ConfigSchema.getUniqueId(ids, "test$id"));
- assertEquals("test_id_5", ConfigSchema.getUniqueId(ids, "test$id"));
- assertEquals("test_id_2_2", ConfigSchema.getUniqueId(ids, "test_id_2"));
+ public void testValidationIssuesFromOlder() throws IOException, SchemaLoaderException {
+ Map<String, Object> yamlAsMap = SchemaLoader.loadYamlAsMap(ConfigSchemaTest.class.getClassLoader().getResourceAsStream("config-minimal.yml"));
+ ConfigSchema configSchema = new ConfigSchema(yamlAsMap);
+ List<String> validationIssues = configSchema.getValidationIssues();
+ assertNotEquals(0, validationIssues.size());
}
@Test
public void testProcessorDuplicateValidationNegativeCase() {
- ConfigSchema configSchema = new ConfigSchema(Collections.singletonMap(CommonPropertyKeys.PROCESSORS_KEY, getListWithKeyValues(CommonPropertyKeys.ID_KEY, "testId1", "testId2")));
+ ConfigSchema configSchema = new ConfigSchema(Collections.singletonMap(CommonPropertyKeys.PROCESSORS_KEY, getListWithKeyValues(ID_KEY, "testId1", "testId2")));
assertMessageDoesNotExist(configSchema, ConfigSchema.FOUND_THE_FOLLOWING_DUPLICATE_PROCESSOR_IDS);
}
@Test
public void testProcessorDuplicateValidationPositiveCase() {
- ConfigSchema configSchema = new ConfigSchema(Collections.singletonMap(CommonPropertyKeys.PROCESSORS_KEY, getListWithKeyValues(CommonPropertyKeys.ID_KEY, "testId1", "testId1")));
+ ConfigSchema configSchema = new ConfigSchema(Collections.singletonMap(CommonPropertyKeys.PROCESSORS_KEY, getListWithKeyValues(ID_KEY, "testId1", "testId1")));
assertMessageDoesExist(configSchema, ConfigSchema.FOUND_THE_FOLLOWING_DUPLICATE_PROCESSOR_IDS);
}
@Test
public void testConnectionDuplicateValidationNegativeCase() {
- ConfigSchema configSchema = new ConfigSchema(Collections.singletonMap(CommonPropertyKeys.CONNECTIONS_KEY, getListWithKeyValues(CommonPropertyKeys.ID_KEY, "testId1", "testId2")));
+ ConfigSchema configSchema = new ConfigSchema(Collections.singletonMap(CommonPropertyKeys.CONNECTIONS_KEY, getListWithKeyValues(ID_KEY, "testId1", "testId2")));
assertMessageDoesNotExist(configSchema, ConfigSchema.FOUND_THE_FOLLOWING_DUPLICATE_CONNECTION_IDS);
}
@Test
public void testConnectionDuplicateValidationPositiveCase() {
- ConfigSchema configSchema = new ConfigSchema(Collections.singletonMap(CommonPropertyKeys.CONNECTIONS_KEY, getListWithKeyValues(CommonPropertyKeys.ID_KEY, "testId1", "testId1")));
+ ConfigSchema configSchema = new ConfigSchema(Collections.singletonMap(CommonPropertyKeys.CONNECTIONS_KEY, getListWithKeyValues(ID_KEY, "testId1", "testId1")));
assertMessageDoesExist(configSchema, ConfigSchema.FOUND_THE_FOLLOWING_DUPLICATE_CONNECTION_IDS);
}
@@ -111,8 +88,8 @@ public class ConfigSchemaTest {
}
@Test
- public void testInvalidSourceAndDestinationNames() throws IOException, SchemaLoaderException {
- Map<String, Object> yamlAsMap = SchemaLoader.loadYamlAsMap(ConfigSchemaTest.class.getClassLoader().getResourceAsStream("config-minimal.yml"));
+ public void testInvalidSourceAndDestinationIds() throws IOException, SchemaLoaderException {
+ Map<String, Object> yamlAsMap = SchemaLoader.loadYamlAsMap(ConfigSchemaTest.class.getClassLoader().getResourceAsStream("config-minimal-v2.yml"));
List<Map<String, Object>> connections = (List<Map<String, Object>>) yamlAsMap.get(CommonPropertyKeys.CONNECTIONS_KEY);
assertEquals(1, connections.size());
@@ -120,15 +97,14 @@ public class ConfigSchemaTest {
String fakeDestination = "fakeDestination";
Map<String, Object> connection = connections.get(0);
- connection.put(ConnectionSchema.SOURCE_NAME_KEY, fakeSource);
- connection.put(ConnectionSchema.DESTINATION_NAME_KEY, fakeDestination);
+ connection.put(ConnectionSchema.SOURCE_ID_KEY, fakeSource);
+ connection.put(ConnectionSchema.DESTINATION_ID_KEY, fakeDestination);
ConfigSchema configSchema = new ConfigSchema(yamlAsMap);
List<String> validationIssues = configSchema.getValidationIssues();
- assertEquals(3, validationIssues.size());
- assertEquals(ConfigSchema.CONNECTIONS_REFER_TO_PROCESSOR_NAMES_THAT_DONT_EXIST + fakeDestination + ", " + fakeSource, validationIssues.get(0));
- assertEquals(BaseSchema.getIssueText(ConnectionSchema.SOURCE_ID_KEY, CommonPropertyKeys.CONNECTIONS_KEY, BaseSchema.IT_WAS_NOT_FOUND_AND_IT_IS_REQUIRED), validationIssues.get(1));
- assertEquals(BaseSchema.getIssueText(ConnectionSchema.DESTINATION_ID_KEY, CommonPropertyKeys.CONNECTIONS_KEY, BaseSchema.IT_WAS_NOT_FOUND_AND_IT_IS_REQUIRED), validationIssues.get(2));
+ assertEquals(2, validationIssues.size());
+ assertEquals(ConfigSchema.CONNECTION_WITH_ID + connection.get(ID_KEY) + ConfigSchema.HAS_INVALID_DESTINATION_ID + fakeDestination, validationIssues.get(0));
+ assertEquals(ConfigSchema.CONNECTION_WITH_ID + connection.get(ID_KEY) + ConfigSchema.HAS_INVALID_SOURCE_ID + fakeSource, validationIssues.get(1));
}
public static List<Map<String, Object>> getListWithNames(String... names) {
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/1bbeedf6/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/ConnectionSchemaTest.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/ConnectionSchemaTest.java b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/ConnectionSchemaTest.java
index 0d37df9..61f9e96 100644
--- a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/ConnectionSchemaTest.java
+++ b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/ConnectionSchemaTest.java
@@ -30,8 +30,6 @@ import java.util.List;
import java.util.Map;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNull;
public class ConnectionSchemaTest {
private String testId;
@@ -41,8 +39,6 @@ public class ConnectionSchemaTest {
private String testSourceRelationShip2;
private List<String> testSourceRelationships;
private String testDestinationId;
- private String testSourceName;
- private String testDestinationName;
private int testMaxWorkQueueSize;
private String testMaxWorkQueueDataSize;
private String testFlowfileExpiration;
@@ -57,8 +53,6 @@ public class ConnectionSchemaTest {
testSourceRelationShip2 = "testSourceRelationShip2";
testSourceRelationships = Arrays.asList(testSourceRelationShip1, testSourceRelationShip2);
testDestinationId = "testDestinationId";
- testSourceName = "testSourceName";
- testDestinationName = "testDestinationName";
testMaxWorkQueueSize = 55;
testMaxWorkQueueDataSize = "testMaxWorkQueueDataSize";
testFlowfileExpiration = "testFlowfileExpiration";
@@ -71,7 +65,7 @@ public class ConnectionSchemaTest {
private ConnectionSchema createSchema(Map<String, Object> map, int expectedValidationIssues) {
ConnectionSchema connectionSchema = new ConnectionSchema(map);
- assertEquals(expectedValidationIssues, connectionSchema.getValidationIssues().size());
+ assertEquals(connectionSchema.getValidationIssues().toString(), expectedValidationIssues, connectionSchema.getValidationIssues().size());
return connectionSchema;
}
@@ -82,8 +76,6 @@ public class ConnectionSchemaTest {
map.put(ConnectionSchema.SOURCE_ID_KEY, testSourceId);
map.put(ConnectionSchema.SOURCE_RELATIONSHIP_NAMES_KEY, testSourceRelationships);
map.put(ConnectionSchema.DESTINATION_ID_KEY, testDestinationId);
- map.put(ConnectionSchema.SOURCE_NAME_KEY, testSourceName);
- map.put(ConnectionSchema.DESTINATION_NAME_KEY, testDestinationName);
map.put(ConnectionSchema.MAX_WORK_QUEUE_SIZE_KEY, testMaxWorkQueueSize);
map.put(ConnectionSchema.MAX_WORK_QUEUE_DATA_SIZE_KEY, testMaxWorkQueueDataSize);
map.put(ConnectionSchema.FLOWFILE_EXPIRATION__KEY, testFlowfileExpiration);
@@ -140,64 +132,13 @@ public class ConnectionSchemaTest {
}
@Test
- public void testNoSourceNameWithId() {
- ConnectionSchema schema = createSchema(0);
- assertNull(schema.getSourceName());
- assertFalse(schema.toMap().containsKey(ConnectionSchema.SOURCE_NAME_KEY));
- }
-
- @Test
- public void testSourceNameNoId() {
- Map<String, Object> map = createMap();
- map.remove(ConnectionSchema.SOURCE_ID_KEY);
- ConnectionSchema schema = createSchema(map, 1);
- assertEquals("", schema.getSourceId());
- assertEquals(testSourceName, schema.getSourceName());
- Map<String, Object> outputMap = schema.toMap();
- assertEquals(schema.getSourceId(), outputMap.get(ConnectionSchema.SOURCE_ID_KEY));
- assertFalse(schema.toMap().containsKey(ConnectionSchema.SOURCE_NAME_KEY));
- }
-
- @Test
- public void testNoSourceIdOrSourceName() {
- Map<String, Object> map = createMap();
- map.remove(ConnectionSchema.SOURCE_ID_KEY);
- map.remove(ConnectionSchema.SOURCE_NAME_KEY);
- ConnectionSchema schema = createSchema(map, 2);
- assertEquals("", schema.getSourceId());
- assertNull(schema.getSourceName());
- Map<String, Object> outputMap = schema.toMap();
- assertEquals(schema.getSourceId(), outputMap.get(ConnectionSchema.SOURCE_ID_KEY));
- assertFalse(schema.toMap().containsKey(ConnectionSchema.SOURCE_NAME_KEY));
- }
-
- @Test
- public void testSourceRelationShipNames() {
+ public void testSourceRelationshipNames() {
ConnectionSchema schema = createSchema(0);
assertEquals(testSourceRelationships, schema.getSourceRelationshipNames());
assertEquals(schema.getSourceRelationshipNames(), schema.toMap().get(ConnectionSchema.SOURCE_RELATIONSHIP_NAMES_KEY));
}
@Test
- public void testSourceRelationshipName() {
- Map<String, Object> map = createMap();
- map.remove(ConnectionSchema.SOURCE_RELATIONSHIP_NAMES_KEY);
- map.put(ConnectionSchema.SOURCE_RELATIONSHIP_NAME_KEY, testSourceRelationShip1);
- ConnectionSchema schema = createSchema(map, 0);
- assertEquals(new ArrayList<>(Arrays.asList(testSourceRelationShip1)), schema.getSourceRelationshipNames());
- assertEquals(schema.getSourceRelationshipNames(), schema.toMap().get(ConnectionSchema.SOURCE_RELATIONSHIP_NAMES_KEY));
- }
-
- @Test
- public void testSourceRelationshipNameAndSourceRelationshipNames() {
- Map<String, Object> map = createMap();
- map.put(ConnectionSchema.SOURCE_RELATIONSHIP_NAME_KEY, testSourceRelationShip1);
- ConnectionSchema schema = createSchema(map, 1);
- assertEquals(testSourceRelationships, schema.getSourceRelationshipNames());
- assertEquals(schema.getSourceRelationshipNames(), schema.toMap().get(ConnectionSchema.SOURCE_RELATIONSHIP_NAMES_KEY));
- }
-
- @Test
public void testNoSourceRelationshipNames() {
Map<String, Object> map = createMap();
map.remove(ConnectionSchema.SOURCE_RELATIONSHIP_NAMES_KEY);
@@ -223,35 +164,6 @@ public class ConnectionSchemaTest {
}
@Test
- public void testDestinationNameWithId() {
- ConnectionSchema schema = createSchema(0);
- assertNull(schema.getDestinationName());
- assertFalse(schema.toMap().containsKey(ConnectionSchema.DESTINATION_NAME_KEY));
- }
-
- @Test
- public void testDestinationNameNoId() {
- Map<String, Object> map = createMap();
- map.remove(ConnectionSchema.DESTINATION_ID_KEY);
- ConnectionSchema schema = createSchema(map, 1);
- assertEquals(testDestinationName, schema.getDestinationName());
- assertFalse(schema.toMap().containsKey(ConnectionSchema.DESTINATION_NAME_KEY));
- }
-
- @Test
- public void testNoDestinationNameNoId() {
- Map<String, Object> map = createMap();
- map.remove(ConnectionSchema.DESTINATION_ID_KEY);
- map.remove(ConnectionSchema.DESTINATION_NAME_KEY);
- ConnectionSchema schema = createSchema(map, 2);
- assertEquals("", schema.getDestinationId());
- assertNull(schema.getDestinationName());
- Map<String, Object> outputMap = schema.toMap();
- assertEquals(schema.getDestinationId(), outputMap.get(ConnectionSchema.DESTINATION_ID_KEY));
- assertFalse(outputMap.containsKey(ConnectionSchema.DESTINATION_NAME_KEY));
- }
-
- @Test
public void testMaxWorkQueueSize() {
ConnectionSchema schema = createSchema(0);
assertEquals(testMaxWorkQueueSize, schema.getMaxWorkQueueSize());
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/1bbeedf6/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoaderTest.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoaderTest.java b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoaderTest.java
index c1b63d0..5b602ac 100644
--- a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoaderTest.java
+++ b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoaderTest.java
@@ -21,6 +21,7 @@ import org.apache.nifi.minifi.commons.schema.ConfigSchema;
import org.apache.nifi.minifi.commons.schema.ConnectionSchema;
import org.apache.nifi.minifi.commons.schema.ProcessorSchema;
import org.apache.nifi.minifi.commons.schema.exception.SchemaLoaderException;
+import org.apache.nifi.minifi.commons.schema.v1.ConfigSchemaV1;
import org.junit.Test;
import java.io.IOException;
@@ -30,6 +31,7 @@ import java.util.Map;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
public class SchemaLoaderTest {
@Test
@@ -49,19 +51,31 @@ public class SchemaLoaderTest {
@Test
public void testMinimalConfigV1Version() throws IOException, SchemaLoaderException {
Map<String, Object> yamlAsMap = SchemaLoader.loadYamlAsMap(SchemaLoaderTest.class.getClassLoader().getResourceAsStream("config-minimal.yml"));
- yamlAsMap.put(ConfigSchema.VERSION, "1");
+ yamlAsMap.put(ConfigSchema.VERSION, ConfigSchemaV1.CONFIG_VERSION);
ConfigSchema configSchema = SchemaLoader.loadConfigSchemaFromYaml(yamlAsMap);
validateMinimalConfigVersion1Parse(configSchema);
}
@Test
public void testMinimalConfigV2Version() throws IOException, SchemaLoaderException {
- Map<String, Object> yamlAsMap = SchemaLoader.loadYamlAsMap(SchemaLoaderTest.class.getClassLoader().getResourceAsStream("config-minimal.yml"));
+ Map<String, Object> yamlAsMap = SchemaLoader.loadYamlAsMap(SchemaLoaderTest.class.getClassLoader().getResourceAsStream("config-minimal-v2.yml"));
yamlAsMap.put(ConfigSchema.VERSION, ConfigSchema.CONFIG_VERSION);
ConfigSchema configSchema = SchemaLoader.loadConfigSchemaFromYaml(yamlAsMap);
validateMinimalConfigVersion1Parse(configSchema);
}
+ @Test
+ public void testUnsupportedVersion() throws IOException, SchemaLoaderException {
+ Map<String, Object> yamlAsMap = SchemaLoader.loadYamlAsMap(SchemaLoaderTest.class.getClassLoader().getResourceAsStream("config-minimal-v2.yml"));
+ yamlAsMap.put(ConfigSchema.VERSION, "9999999");
+ try {
+ SchemaLoader.loadConfigSchemaFromYaml(yamlAsMap);
+ fail();
+ } catch (SchemaLoaderException e) {
+ assertEquals("YAML configuration version 9999999 not supported. Supported versions: 1, 2", e.getMessage());
+ }
+ }
+
private void validateMinimalConfigVersion1Parse(ConfigSchema configSchema) {
assertTrue(configSchema instanceof ConfigSchema);
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/1bbeedf6/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1Test.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1Test.java b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1Test.java
new file mode 100644
index 0000000..56a8103
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1Test.java
@@ -0,0 +1,94 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. See the NOTICE file distributed with
+ * * this work for additional information regarding copyright ownership.
+ * * The ASF licenses this file to You under the Apache License, Version 2.0
+ * * (the "License"); you may not use this file except in compliance with
+ * * the License. You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing, software
+ * * distributed under the License is distributed on an "AS IS" BASIS,
+ * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * * See the License for the specific language governing permissions and
+ * * limitations under the License.
+ *
+ */
+
+package org.apache.nifi.minifi.commons.schema.v1;
+
+import org.apache.nifi.minifi.commons.schema.ConfigSchema;
+import org.apache.nifi.minifi.commons.schema.ConfigSchemaTest;
+import org.apache.nifi.minifi.commons.schema.ConnectionSchema;
+import org.apache.nifi.minifi.commons.schema.common.BaseSchema;
+import org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys;
+import org.apache.nifi.minifi.commons.schema.exception.SchemaLoaderException;
+import org.apache.nifi.minifi.commons.schema.serialization.SchemaLoader;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.NAME_KEY;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
+public class ConfigSchemaV1Test {
+ @Test
+ public void testValid() throws IOException, SchemaLoaderException {
+ Map<String, Object> yamlAsMap = SchemaLoader.loadYamlAsMap(ConfigSchemaTest.class.getClassLoader().getResourceAsStream("config-minimal.yml"));
+ ConfigSchema configSchema = new ConfigSchemaV1(yamlAsMap).convert();
+ List<String> validationIssues = configSchema.getValidationIssues();
+ assertEquals(0, validationIssues.size());
+ }
+ @Test
+ public void testValidationIssuesFromNewer() throws IOException, SchemaLoaderException {
+ Map<String, Object> yamlAsMap = SchemaLoader.loadYamlAsMap(ConfigSchemaTest.class.getClassLoader().getResourceAsStream("config-minimal-v2.yml"));
+ ConfigSchema configSchema = new ConfigSchemaV1(yamlAsMap).convert();
+ List<String> validationIssues = configSchema.getValidationIssues();
+ assertNotEquals(0, validationIssues.size());
+ }
+
+ @Test
+ public void testInvalidSourceAndDestinationNames() throws IOException, SchemaLoaderException {
+ Map<String, Object> yamlAsMap = SchemaLoader.loadYamlAsMap(ConfigSchemaTest.class.getClassLoader().getResourceAsStream("config-minimal.yml"));
+ List<Map<String, Object>> connections = (List<Map<String, Object>>) yamlAsMap.get(CommonPropertyKeys.CONNECTIONS_KEY);
+ assertEquals(1, connections.size());
+
+ String fakeSource = "fakeSource";
+ String fakeDestination = "fakeDestination";
+
+ Map<String, Object> connection = connections.get(0);
+ connection.put(ConnectionSchemaV1.SOURCE_NAME_KEY, fakeSource);
+ connection.put(ConnectionSchemaV1.DESTINATION_NAME_KEY, fakeDestination);
+
+ ConfigSchema configSchema = new ConfigSchemaV1(yamlAsMap).convert();
+ List<String> validationIssues = configSchema.getValidationIssues();
+ assertEquals(4, validationIssues.size());
+ assertEquals(BaseSchema.getIssueText(ConnectionSchema.DESTINATION_ID_KEY, CommonPropertyKeys.CONNECTIONS_KEY, BaseSchema.IT_WAS_NOT_FOUND_AND_IT_IS_REQUIRED), validationIssues.get(0));
+ assertEquals(BaseSchema.getIssueText(ConnectionSchema.SOURCE_ID_KEY, CommonPropertyKeys.CONNECTIONS_KEY, BaseSchema.IT_WAS_NOT_FOUND_AND_IT_IS_REQUIRED), validationIssues.get(1));
+ assertEquals(ConfigSchemaV1.CONNECTION_WITH_NAME + connection.get(NAME_KEY) + ConfigSchemaV1.HAS_INVALID_DESTINATION_NAME + fakeDestination, validationIssues.get(2));
+ assertEquals(ConfigSchemaV1.CONNECTION_WITH_NAME + connection.get(NAME_KEY) + ConfigSchemaV1.HAS_INVALID_SOURCE_NAME + fakeSource, validationIssues.get(3));
+ }
+
+ @Test
+ public void testGetUniqueIdConflicts() {
+ Map<String, Integer> ids = new HashMap<>();
+ assertEquals("test_id", ConfigSchemaV1.getUniqueId(ids, "test/id"));
+ assertEquals("test_id_2", ConfigSchemaV1.getUniqueId(ids, "test$id"));
+ assertEquals("test_id_3", ConfigSchemaV1.getUniqueId(ids, "test$id"));
+ assertEquals("test_id_4", ConfigSchemaV1.getUniqueId(ids, "test$id"));
+ assertEquals("test_id_5", ConfigSchemaV1.getUniqueId(ids, "test$id"));
+ assertEquals("test_id_2_2", ConfigSchemaV1.getUniqueId(ids, "test_id_2"));
+ }
+
+ @Test
+ public void testGetUniqueIdEmptySet() {
+ String testId = "testId";
+ assertEquals(testId + "___", ConfigSchemaV1.getUniqueId(new HashMap<>(), testId + "/ $"));
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/1bbeedf6/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConnectionSchemaV1Test.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConnectionSchemaV1Test.java b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConnectionSchemaV1Test.java
new file mode 100644
index 0000000..e9acf4a
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConnectionSchemaV1Test.java
@@ -0,0 +1,192 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. See the NOTICE file distributed with
+ * * this work for additional information regarding copyright ownership.
+ * * The ASF licenses this file to You under the Apache License, Version 2.0
+ * * (the "License"); you may not use this file except in compliance with
+ * * the License. You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing, software
+ * * distributed under the License is distributed on an "AS IS" BASIS,
+ * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * * See the License for the specific language governing permissions and
+ * * limitations under the License.
+ *
+ */
+
+package org.apache.nifi.minifi.commons.schema.v1;
+
+import org.apache.nifi.minifi.commons.schema.ConfigSchema;
+import org.apache.nifi.minifi.commons.schema.ConnectionSchema;
+import org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.nifi.minifi.commons.schema.ConfigSchemaTest.assertMessageDoesNotExist;
+import static org.apache.nifi.minifi.commons.schema.ConfigSchemaTest.getListWithKeyValues;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+public class ConnectionSchemaV1Test {
+ private String testName;
+ private String testSourceRelationship;
+ private String testSourceName;
+ private String testDestinationName;
+ private int testMaxWorkQueueSize;
+ private String testMaxWorkQueueDataSize;
+ private String testFlowfileExpiration;
+ private String testQueuePrioritizerClass;
+
+ @Before
+ public void setup() {
+ testName = "testName";
+ testSourceRelationship = "testSourceRelationship";
+ testSourceName = "testSourceName";
+ testDestinationName = "testDestinationName";
+ testMaxWorkQueueSize = 55;
+ testMaxWorkQueueDataSize = "testMaxWorkQueueDataSize";
+ testFlowfileExpiration = "testFlowfileExpiration";
+ testQueuePrioritizerClass = "testQueuePrioritizerClass";
+ }
+
+ private ConnectionSchemaV1 createSchema(int expectedValidationIssues) {
+ return createSchema(createMap(), expectedValidationIssues);
+ }
+
+ private ConnectionSchemaV1 createSchema(Map<String, Object> map, int expectedValidationIssues) {
+ ConnectionSchemaV1 connectionSchema = new ConnectionSchemaV1(map);
+ assertEquals(expectedValidationIssues, connectionSchema.getValidationIssues().size());
+ return connectionSchema;
+ }
+
+ private Map<String, Object> createMap() {
+ Map<String, Object> map = new HashMap<>();
+ map.put(CommonPropertyKeys.NAME_KEY, testName);
+ map.put(ConnectionSchemaV1.SOURCE_RELATIONSHIP_NAME_KEY, testSourceRelationship);
+ map.put(ConnectionSchemaV1.SOURCE_NAME_KEY, testSourceName);
+ map.put(ConnectionSchemaV1.DESTINATION_NAME_KEY, testDestinationName);
+ map.put(ConnectionSchema.MAX_WORK_QUEUE_SIZE_KEY, testMaxWorkQueueSize);
+ map.put(ConnectionSchema.MAX_WORK_QUEUE_DATA_SIZE_KEY, testMaxWorkQueueDataSize);
+ map.put(ConnectionSchema.FLOWFILE_EXPIRATION__KEY, testFlowfileExpiration);
+ map.put(ConnectionSchema.QUEUE_PRIORITIZER_CLASS_KEY, testQueuePrioritizerClass);
+ return map;
+ }
+
+ @Test
+ public void testName() {
+ ConnectionSchemaV1 schema = createSchema(0);
+ assertEquals(testName, schema.getName());
+ assertEquals(schema.getName(), schema.convert().getName());
+ }
+
+ @Test
+ public void testNoName() {
+ Map<String, Object> map = createMap();
+ map.remove(CommonPropertyKeys.NAME_KEY);
+ ConnectionSchemaV1 schema = createSchema(map, 1);
+ assertNull(schema.getName());
+ assertEquals("", schema.convert().getName());
+ }
+
+ @Test
+ public void testSourceRelationShipName() {
+ ConnectionSchemaV1 schema = createSchema(0);
+ List<String> sourceRelationshipNames = schema.convert().getSourceRelationshipNames();
+ assertEquals(1, sourceRelationshipNames.size());
+ assertEquals(testSourceRelationship, sourceRelationshipNames.get(0));
+ }
+
+ @Test
+ public void testNoSourceRelationshipName() {
+ Map<String, Object> map = createMap();
+ map.remove(ConnectionSchemaV1.SOURCE_RELATIONSHIP_NAME_KEY);
+ ConnectionSchemaV1 schema = createSchema(map, 1);
+ List<String> sourceRelationshipNames = schema.convert().getSourceRelationshipNames();
+ assertEquals(0, sourceRelationshipNames.size());
+ }
+
+ @Test
+ public void testDestinationName() {
+ assertEquals(testDestinationName, createSchema(0).getDestinationName());
+ }
+
+ @Test
+ public void testNoDestinationName() {
+ Map<String, Object> map = createMap();
+ map.remove(ConnectionSchemaV1.DESTINATION_NAME_KEY);
+ assertNull(createSchema(map, 1).getDestinationName());
+ }
+
+ @Test
+ public void testMaxWorkQueueSize() {
+ assertEquals(testMaxWorkQueueSize, createSchema(0).convert().getMaxWorkQueueSize());
+ }
+
+ @Test
+ public void testNoMaxWorkQueueSize() {
+ Map<String, Object> map = createMap();
+ map.remove(ConnectionSchema.MAX_WORK_QUEUE_SIZE_KEY);
+ assertEquals(ConnectionSchema.DEFAULT_MAX_WORK_QUEUE_SIZE, createSchema(map, 0).convert().getMaxWorkQueueSize());
+ }
+
+ @Test
+ public void testMaxWorkQueueDataSize() {
+ assertEquals(testMaxWorkQueueDataSize, createSchema(0).convert().getMaxWorkQueueDataSize());
+ }
+
+ @Test
+ public void testNoMaxWorkQueueDataSize() {
+ Map<String, Object> map = createMap();
+ map.remove(ConnectionSchema.MAX_WORK_QUEUE_DATA_SIZE_KEY);
+ assertEquals(ConnectionSchema.DEFAULT_MAX_QUEUE_DATA_SIZE, createSchema(map, 0).convert().getMaxWorkQueueDataSize());
+ }
+
+ @Test
+ public void testFlowFileExpiration() {
+ assertEquals(testFlowfileExpiration, createSchema(0).convert().getFlowfileExpiration());
+ }
+
+ @Test
+ public void testNoFlowFileExpiration() {
+ Map<String, Object> map = createMap();
+ map.remove(ConnectionSchema.FLOWFILE_EXPIRATION__KEY);
+ assertEquals(ConnectionSchema.DEFAULT_FLOWFILE_EXPIRATION, createSchema(map, 0).convert().getFlowfileExpiration());
+ }
+
+ @Test
+ public void testQueuePrioritizer() {
+ assertEquals(testQueuePrioritizerClass, createSchema(0).convert().getQueuePrioritizerClass());
+ }
+
+ @Test
+ public void testNoQueuePrioritizer() {
+ Map<String, Object> map = createMap();
+ map.remove(ConnectionSchema.QUEUE_PRIORITIZER_CLASS_KEY);
+ assertEquals("", createSchema(map, 0).convert().getQueuePrioritizerClass());
+ }
+
+ @Test
+ public void testConnectionGeneratedIds() {
+ List<Map<String, Object>> listWithKeyValues = getListWithKeyValues(CommonPropertyKeys.NAME_KEY, "test", "test", "test_2", "test", "test_2");
+
+ ConfigSchema configSchema = new ConfigSchemaV1(Collections.singletonMap(CommonPropertyKeys.CONNECTIONS_KEY, listWithKeyValues)).convert();
+ assertMessageDoesNotExist(configSchema, ConfigSchema.FOUND_THE_FOLLOWING_DUPLICATE_CONNECTION_IDS);
+ List<ConnectionSchema> connections = configSchema.getConnections();
+ assertEquals(5, connections.size());
+
+ // Generated unique ids
+ assertEquals("test", connections.get(0).getId());
+ assertEquals("test_2", connections.get(1).getId());
+ assertEquals("test_2_2", connections.get(2).getId());
+ assertEquals("test_3", connections.get(3).getId());
+ assertEquals("test_2_3", connections.get(4).getId());
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/1bbeedf6/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1Test.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1Test.java b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1Test.java
new file mode 100644
index 0000000..3668c0e
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ProcessorSchemaV1Test.java
@@ -0,0 +1,241 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. See the NOTICE file distributed with
+ * * this work for additional information regarding copyright ownership.
+ * * The ASF licenses this file to You under the Apache License, Version 2.0
+ * * (the "License"); you may not use this file except in compliance with
+ * * the License. You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing, software
+ * * distributed under the License is distributed on an "AS IS" BASIS,
+ * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * * See the License for the specific language governing permissions and
+ * * limitations under the License.
+ *
+ */
+
+package org.apache.nifi.minifi.commons.schema.v1;
+
+import org.apache.nifi.scheduling.SchedulingStrategy;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.CLASS_KEY;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_AUTO_TERMINATED_RELATIONSHIPS_LIST;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_MAX_CONCURRENT_TASKS;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_PENALIZATION_PERIOD;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_PROPERTIES;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_RUN_DURATION_NANOS;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.DEFAULT_YIELD_DURATION;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.PENALIZATION_PERIOD_KEY;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.PROCESSOR_PROPS_KEY;
+import static org.apache.nifi.minifi.commons.schema.ProcessorSchema.RUN_DURATION_NANOS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.MAX_CONCURRENT_TASKS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.NAME_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SCHEDULING_PERIOD_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SCHEDULING_STRATEGY_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.YIELD_PERIOD_KEY;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+public class ProcessorSchemaV1Test {
+ private String testName;
+ private String testProcessorClass;
+ private String testSchedulingStrategy;
+ private String testSchedulingPeriod;
+ private int testMaxConcurrentTasks;
+ private String testPenalizationPeriod;
+ private String testYieldPeriod;
+ private int testRunDurationNanos;
+ private String testAutoTerminatedRelationship1;
+ private String testAutoTerminatedRelationship2;
+ private List<String> testAutoTerminatedRelationships;
+ private String testKey1;
+ private String testValue1;
+ private String testKey2;
+ private String testValue2;
+ private Map<String, Object> testProperties;
+
+ @Before
+ public void setup() {
+ testName = "testName";
+ testProcessorClass = "testProcessorClass";
+ testSchedulingStrategy = SchedulingStrategy.PRIMARY_NODE_ONLY.toString();
+ testSchedulingPeriod = "testSchedulingPeriod";
+ testMaxConcurrentTasks = 55;
+ testPenalizationPeriod = "testPenalizationPeriod";
+ testYieldPeriod = "testYieldPeriod";
+ testRunDurationNanos = 125;
+ testAutoTerminatedRelationship1 = "testAutoTerminatedRelationship1";
+ testAutoTerminatedRelationship2 = "testAutoTerminatedRelationship2";
+ testAutoTerminatedRelationships = new ArrayList<>(Arrays.asList(testAutoTerminatedRelationship1, testAutoTerminatedRelationship2));
+ testKey1 = "testKey1";
+ testValue1 = "testValue1";
+ testKey2 = "testKey2";
+ testValue2 = "testValue2";
+ testProperties = new HashMap<>();
+ testProperties.put(testKey1, testValue1);
+ testProperties.put(testKey2, testValue2);
+ }
+
+ private ProcessorSchemaV1 createSchema(int expectedValidationIssues) {
+ return createSchema(createMap(), expectedValidationIssues);
+ }
+
+ private ProcessorSchemaV1 createSchema(Map<String, Object> map, int expectedValidationIssues) {
+ ProcessorSchemaV1 processorSchemaV1 = new ProcessorSchemaV1(map);
+ assertEquals(expectedValidationIssues, processorSchemaV1.getValidationIssues().size());
+ return processorSchemaV1;
+ }
+
+ private Map<String, Object> createMap() {
+ Map<String, Object> map = new HashMap<>();
+ map.put(NAME_KEY, testName);
+ map.put(CLASS_KEY, testProcessorClass);
+ map.put(SCHEDULING_STRATEGY_KEY, testSchedulingStrategy);
+ map.put(SCHEDULING_PERIOD_KEY, testSchedulingPeriod);
+ map.put(MAX_CONCURRENT_TASKS_KEY, testMaxConcurrentTasks);
+ map.put(PENALIZATION_PERIOD_KEY, testPenalizationPeriod);
+ map.put(YIELD_PERIOD_KEY, testYieldPeriod);
+ map.put(RUN_DURATION_NANOS_KEY, testRunDurationNanos);
+ map.put(AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY, testAutoTerminatedRelationships);
+ map.put(PROCESSOR_PROPS_KEY, testProperties);
+ return map;
+ }
+
+ @Test
+ public void testName() {
+ assertEquals(testName, createSchema(0).convert().getName());
+ }
+
+ @Test
+ public void testNoName() {
+ Map<String, Object> map = createMap();
+ map.remove(NAME_KEY);
+ assertNull(createSchema(map, 1).getName());
+ }
+
+ @Test
+ public void testProcessorClass() {
+ assertEquals(testProcessorClass, createSchema(0).convert().getProcessorClass());
+ }
+
+ @Test
+ public void testNoProcessorClass() {
+ Map<String, Object> map = createMap();
+ map.remove(CLASS_KEY);
+ assertNull(createSchema(map, 1).convert().getProcessorClass());
+ }
+
+ @Test
+ public void testSchedulingStrategy() {
+ assertEquals(testSchedulingStrategy, createSchema(0).convert().getSchedulingStrategy());
+ }
+
+ @Test
+ public void testNoSchedulingStrategy() {
+ Map<String, Object> map = createMap();
+ map.remove(SCHEDULING_STRATEGY_KEY);
+ assertNull(createSchema(map, 1).convert().getSchedulingStrategy());
+ }
+
+ @Test
+ public void testInvalidSchedulingStrategy() {
+ testSchedulingStrategy = "fake strategy";
+ assertEquals(testSchedulingStrategy, createSchema(1).convert().getSchedulingStrategy());
+ }
+
+ @Test
+ public void testSchedulingPeriod() {
+ assertEquals(testSchedulingPeriod, createSchema(0).convert().getSchedulingPeriod());
+ }
+
+ @Test
+ public void testNoSchedulingPeriod() {
+ Map<String, Object> map = createMap();
+ map.remove(SCHEDULING_PERIOD_KEY);
+ assertNull(createSchema(map, 1).convert().getSchedulingPeriod());
+ }
+
+ @Test
+ public void testMaxConcurrentTasks() {
+ assertEquals(testMaxConcurrentTasks, createSchema(0).convert().getMaxConcurrentTasks());
+ }
+
+ @Test
+ public void testNoMaxConcurrentTasks() {
+ Map<String, Object> map = createMap();
+ map.remove(MAX_CONCURRENT_TASKS_KEY);
+ assertEquals(DEFAULT_MAX_CONCURRENT_TASKS, createSchema(map, 0).convert().getMaxConcurrentTasks());
+ }
+
+ @Test
+ public void testPenalizationPeriod() {
+ assertEquals(testPenalizationPeriod, createSchema(0).convert().getPenalizationPeriod());
+ }
+
+ @Test
+ public void testNoPenalizationPeriod() {
+ Map<String, Object> map = createMap();
+ map.remove(PENALIZATION_PERIOD_KEY);
+ assertEquals(DEFAULT_PENALIZATION_PERIOD, createSchema(map, 0).convert().getPenalizationPeriod());
+ }
+
+ @Test
+ public void testYieldPeriod() {
+ assertEquals(testYieldPeriod, createSchema(0).convert().getYieldPeriod());
+ }
+
+ @Test
+ public void testNoYieldPeriod() {
+ Map<String, Object> map = createMap();
+ map.remove(YIELD_PERIOD_KEY);
+ assertEquals(DEFAULT_YIELD_DURATION, createSchema(map, 0).convert().getYieldPeriod());
+ }
+
+ @Test
+ public void testRunDurationNanos() {
+ assertEquals(testRunDurationNanos, createSchema(0).convert().getRunDurationNanos());
+ }
+
+ @Test
+ public void testNoRunDurationNanos() {
+ Map<String, Object> map = createMap();
+ map.remove(RUN_DURATION_NANOS_KEY);
+ assertEquals(DEFAULT_RUN_DURATION_NANOS, createSchema(map, 0).convert().getRunDurationNanos());
+ }
+
+ @Test
+ public void testAutoTerminatedRelationships() {
+ assertEquals(testAutoTerminatedRelationships, createSchema(0).convert().getAutoTerminatedRelationshipsList());
+ }
+
+ @Test
+ public void testNoAutoTerminatedRelationships() {
+ Map<String, Object> map = createMap();
+ map.remove(AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY);
+ assertEquals(DEFAULT_AUTO_TERMINATED_RELATIONSHIPS_LIST, createSchema(map, 0).convert().getAutoTerminatedRelationshipsList());
+ }
+
+ @Test
+ public void testProperties() {
+ assertEquals(testProperties, createSchema(0).convert().getProperties());
+ }
+
+ @Test
+ public void testNoProperties() {
+ Map<String, Object> map = createMap();
+ map.remove(PROCESSOR_PROPS_KEY);
+ assertEquals(DEFAULT_PROPERTIES, createSchema(map, 0).convert().getProperties());
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/1bbeedf6/minifi-commons/minifi-commons-schema/src/test/resources/config-minimal-v2.yml
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/test/resources/config-minimal-v2.yml b/minifi-commons/minifi-commons-schema/src/test/resources/config-minimal-v2.yml
new file mode 100644
index 0000000..2472a32
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/test/resources/config-minimal-v2.yml
@@ -0,0 +1,38 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the \"License\"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an \"AS IS\" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# This is a minimal V2 config.
+MiNiFi Config Version: 2
+Flow Controller:
+ name: MiNiFi Flow
+
+# When running the Flow (not just doing the transform) these processors will be invalid due to not having the necesary properties/auto-terminated relationships
+Processors:
+ - id: bcd
+ class: org.apache.nifi.processors.standard.TailFile
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 10 sec
+ - id: cde
+ class: org.apache.nifi.processors.standard.PutFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 0 sec
+
+Connections:
+ - id: abc
+ source id: bcd
+ source relationship names:
+ - success
+ destination id: cde
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/1bbeedf6/minifi-docs/src/main/markdown/System_Admin_Guide.md
----------------------------------------------------------------------
diff --git a/minifi-docs/src/main/markdown/System_Admin_Guide.md b/minifi-docs/src/main/markdown/System_Admin_Guide.md
index 3f3d414..38c9588 100644
--- a/minifi-docs/src/main/markdown/System_Admin_Guide.md
+++ b/minifi-docs/src/main/markdown/System_Admin_Guide.md
@@ -237,6 +237,18 @@ available at https://nifi.apache.org/minifi/minifi-toolkit.html.
NOTE: Note that values for periods of time and data sizes must include the unit of measure,
for example "10 sec" or "10 MB", not simply "10".
+## Versioning
+
+The "MiNiFi Config Version" property is used to indicate to the configuration parser which version of the config file it is looking at. If the property is empty or missing, version 1 is assumed.
+
+The MiNiFi Toolkit Converter is capable of parsing previous versions (possibly subject to a future deprecation policy) and writing out the current version. It can also validate that a given config file
+parses and upconverts to the current version without issue.
+
+### Version 1 -> Version 2 changes
+
+1. Use ids instead of names for processors, connections.
+2. Allow multiple source relationships for connections.
+
## Flow Controller
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/1bbeedf6/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 89b1011..44d8d4e 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
@@ -13,58 +13,48 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+MiNiFi Config Version: 2
Flow Controller:
- name: MiNiFi Flow
- comment:
-
+ 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
-
+ 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
-
+ 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
-
+ content claim max appendable size: 10 MB
+ content claim max flow files: 100
+ always sync: false
Provenance Repository:
- provenance rollover time: 1 min
-
+ provenance rollover time: 1 min
Component Status Repository:
- buffer size: 1440
- snapshot frequency: 1 min
-
+ buffer size: 1440
+ snapshot frequency: 1 min
Security Properties:
- keystore:
- keystore type:
- keystore password:
- key password:
- truststore:
- truststore type:
- truststore password:
- ssl protocol:
- Sensitive Props:
- key:
- algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
- provider: BC
-
-Processors:
-
-Connections:
-
-Remote Processing Groups:
-
-Provenance Reporting:
\ No newline at end of file
+ keystore: ''
+ keystore type: ''
+ keystore password: ''
+ key password: ''
+ truststore: ''
+ truststore type: ''
+ truststore password: ''
+ ssl protocol: ''
+ Sensitive Props:
+ key: ''
+ algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
+ provider: BC
+Processors: []
+Connections: []
+Remote Processing Groups: []
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/1bbeedf6/minifi-toolkit/minifi-toolkit-assembly/src/main/resources/config.sh
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-assembly/src/main/resources/config.sh b/minifi-toolkit/minifi-toolkit-assembly/src/main/resources/config.sh
index ba29fdc..12963cb 100755
--- a/minifi-toolkit/minifi-toolkit-assembly/src/main/resources/config.sh
+++ b/minifi-toolkit/minifi-toolkit-assembly/src/main/resources/config.sh
@@ -93,7 +93,7 @@ init() {
detectOS
# Locate the Java VM to execute
- locateJava "$1"
+ locateJava
}
run() {
@@ -118,19 +118,5 @@ run() {
return $?
}
-
-case "$1" in
- transform|validate)
- init "$1"
- run "$@"
- ;;
- *)
- echo ""
- echo "MiNiFi toolkit Usage, valid commands include: {transform|validate}"
- echo "transform: Transform template xml into MiNiFi config YAML"
- echo " config.sh transform INPUT_FILE OUTPUT_FILE"
- echo "validate: Validate config YAML"
- echo " config.sh validate INPUT_FILE"
- echo ""
- ;;
-esac
\ No newline at end of file
+init
+run "$@"
\ No newline at end of file