ConfigTransformer { protected static void addConnection(final Element parentElement, ConnectionSchema connectionProperties, ParentGroupIdResolver parentGroupIdResolver) throws ConfigurationChangeException { try { final Document doc = parentElement.getOwnerDocument(); final Element element = doc.createElement("connection"); parentElement.appendChild(element); addTextElement(element, "id", connectionProperties.getId()); addTextElement(element, "name", connectionProperties.getName()); final Element bendPointsElement = doc.createElement("bendPoints"); element.appendChild(bendPointsElement); addTextElement(element, "labelIndex", "1"); addTextElement(element, "zIndex", "0"); addConnectionSourceOrDestination(element, "source", connectionProperties.getSourceId(), parentGroupIdResolver); addConnectionSourceOrDestination(element, "destination", connectionProperties.getDestinationId(), parentGroupIdResolver); List<String> sourceRelationshipNames = connectionProperties.getSourceRelationshipNames(); if (sourceRelationshipNames.isEmpty()) { addTextElement(element, "relationship", null); } else { for (String relationshipName : sourceRelationshipNames) { addTextElement(element, "relationship", relationshipName); } } addTextElement(element, "maxWorkQueueSize", String.valueOf(connectionProperties.getMaxWorkQueueSize())); addTextElement(element, "maxWorkQueueDataSize", connectionProperties.getMaxWorkQueueDataSize()); addTextElement(element, "flowFileExpiration", connectionProperties.getFlowfileExpiration()); addTextElementIfNotNullOrEmpty(element, "queuePrioritizerClass", connectionProperties.getQueuePrioritizerClass()); parentElement.appendChild(element); } 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  ConfigTransformer(); static void transformConfigFile(InputStream sourceStream, String destPath, Properties bootstrapProperties); static final String NIFI_VERSION; static final String ROOT_GROUP; static final String NIFI_VERSION_KEY; static final Logger logger; static final String PROPERTIES_FILE_APACHE_2_0_LICENSE; }
@Test public void testNullQueuePrioritizerNotWritten() throws ConfigurationChangeException, XPathExpressionException { ConfigTransformer.addConnection(config, new ConnectionSchema(Collections.emptyMap()), new ParentGroupIdResolver(new ProcessGroupSchema(Collections.emptyMap(), ConfigSchema.TOP_LEVEL_NAME))); XPath xpath = xPathFactory.newXPath(); String expression = "connection/queuePrioritizerClass"; assertNull(xpath.evaluate(expression, config, XPathConstants.NODE)); }
@Test public void testEmptyQueuePrioritizerNotWritten() throws ConfigurationChangeException, XPathExpressionException { Map<String, Object> map = new HashMap<>(); map.put(ConnectionSchema.QUEUE_PRIORITIZER_CLASS_KEY, ""); ConfigTransformer.addConnection(config, new ConnectionSchema(map), new ParentGroupIdResolver(new ProcessGroupSchema(Collections.emptyMap(), ConfigSchema.TOP_LEVEL_NAME))); XPath xpath = xPathFactory.newXPath(); String expression = "connection/queuePrioritizerClass"; assertNull(xpath.evaluate(expression, config, XPathConstants.NODE)); }
@Test public void testQueuePrioritizerWritten() throws ConfigurationChangeException, XPathExpressionException { Map<String, Object> map = new HashMap<>(); map.put(ConnectionSchema.QUEUE_PRIORITIZER_CLASS_KEY, "org.apache.nifi.prioritizer.FirstInFirstOutPrioritizer"); ConfigTransformer.addConnection(config, new ConnectionSchema(map), new ParentGroupIdResolver(new ProcessGroupSchema(Collections.emptyMap(), ConfigSchema.TOP_LEVEL_NAME))); XPath xpath = xPathFactory.newXPath(); String expression = "connection/queuePrioritizerClass/text()"; assertEquals("org.apache.nifi.prioritizer.FirstInFirstOutPrioritizer", xpath.evaluate(expression, config, XPathConstants.STRING)); }