Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'develop' of http://git-wip-us.apache.org/repos/asf/incubator-nifi into develop

This commit is contained in:
Mark Payne 2015-05-01 10:30:55 -04:00
parent 7db27c01b2 393efc836e
commit 1a1cfe2344
14 changed files with 1023 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

111
nifi/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
View File

@ -16,9 +16,6 @@
*/
package org.apache.nifi.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
@ -27,10 +24,15 @@ import java.net.InetSocketAddress;
import java.nio.file.InvalidPathException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class NiFiProperties extends Properties {
private static final long serialVersionUID = 2119177359005492702L;
@ -50,6 +52,7 @@ public class NiFiProperties extends Properties {
public static final String AUTO_RESUME_STATE = "nifi.flowcontroller.autoResumeState";
public static final String FLOW_CONTROLLER_GRACEFUL_SHUTDOWN_PERIOD = "nifi.flowcontroller.graceful.shutdown.period";
public static final String NAR_LIBRARY_DIRECTORY = "nifi.nar.library.directory";
public static final String NAR_LIBRARY_DIRECTORY_PREFIX = "nifi.nar.library.directory.";
public static final String NAR_WORKING_DIRECTORY = "nifi.nar.working.directory";
public static final String COMPONENT_DOCS_DIRECTORY = "nifi.documentation.working.directory";
public static final String SENSITIVE_PROPS_KEY = "nifi.sensitive.props.key";
@ -230,21 +233,27 @@ public class NiFiProperties extends Properties {
* obtained.
*
* @return the NiFiProperties object to use
* @throws RuntimeException if unable to load properties file
* @throws RuntimeException
* if unable to load properties file
*/
public static synchronized NiFiProperties getInstance() {
if (null == instance) {
final NiFiProperties suspectInstance = new NiFiProperties();
final String nfPropertiesFilePath = System.getProperty(NiFiProperties.PROPERTIES_FILE_PATH);
final String nfPropertiesFilePath = System
.getProperty(NiFiProperties.PROPERTIES_FILE_PATH);
if (null == nfPropertiesFilePath || nfPropertiesFilePath.trim().length() == 0) {
throw new RuntimeException("Requires a system property called \'" + NiFiProperties.PROPERTIES_FILE_PATH + "\' and this is not set or has no value");
throw new RuntimeException("Requires a system property called \'"
+ NiFiProperties.PROPERTIES_FILE_PATH
+ "\' and this is not set or has no value");
}
final File propertiesFile = new File(nfPropertiesFilePath);
if (!propertiesFile.exists()) {
throw new RuntimeException("Properties file doesn't exist \'" + propertiesFile.getAbsolutePath() + "\'");
throw new RuntimeException("Properties file doesn't exist \'"
+ propertiesFile.getAbsolutePath() + "\'");
}
if (!propertiesFile.canRead()) {
throw new RuntimeException("Properties file exists but cannot be read \'" + propertiesFile.getAbsolutePath() + "\'");
throw new RuntimeException("Properties file exists but cannot be read \'"
+ propertiesFile.getAbsolutePath() + "\'");
}
InputStream inStream = null;
try {
@ -252,7 +261,8 @@ public class NiFiProperties extends Properties {
suspectInstance.load(inStream);
} catch (final Exception ex) {
LOG.error("Cannot load properties file due to " + ex.getLocalizedMessage());
throw new RuntimeException("Cannot load properties file due to " + ex.getLocalizedMessage(), ex);
throw new RuntimeException("Cannot load properties file due to "
+ ex.getLocalizedMessage(), ex);
} finally {
if (null != inStream) {
try {
@ -374,7 +384,7 @@ public class NiFiProperties extends Properties {
if ("false".equalsIgnoreCase(secureVal)) {
return false;
}else{
} else {
return true;
}
@ -406,7 +416,8 @@ public class NiFiProperties extends Properties {
* @return Whether to auto start the processors or not
*/
public boolean getAutoResumeState() {
final String rawAutoResumeState = getProperty(AUTO_RESUME_STATE, DEFAULT_AUTO_RESUME_STATE.toString());
final String rawAutoResumeState = getProperty(AUTO_RESUME_STATE,
DEFAULT_AUTO_RESUME_STATE.toString());
return Boolean.parseBoolean(rawAutoResumeState);
}
@ -417,7 +428,8 @@ public class NiFiProperties extends Properties {
* @return the number of partitions
*/
public int getFlowFileRepositoryPartitions() {
final String rawProperty = getProperty(FLOWFILE_REPOSITORY_PARTITIONS, DEFAULT_FLOWFILE_REPO_PARTITIONS);
final String rawProperty = getProperty(FLOWFILE_REPOSITORY_PARTITIONS,
DEFAULT_FLOWFILE_REPO_PARTITIONS);
return Integer.parseInt(rawProperty);
}
@ -428,7 +440,8 @@ public class NiFiProperties extends Properties {
* @return the number of milliseconds between checkpoint events
*/
public String getFlowFileRepositoryCheckpointInterval() {
return getProperty(FLOWFILE_REPOSITORY_CHECKPOINT_INTERVAL, DEFAULT_FLOWFILE_CHECKPOINT_INTERVAL);
return getProperty(FLOWFILE_REPOSITORY_CHECKPOINT_INTERVAL,
DEFAULT_FLOWFILE_CHECKPOINT_INTERVAL);
}
/**
@ -470,7 +483,8 @@ public class NiFiProperties extends Properties {
}
public String getUserCredentialCacheDuration() {
return getProperty(SECURITY_USER_CREDENTIAL_CACHE_DURATION, DEFAULT_USER_CREDENTIAL_CACHE_DURATION);
return getProperty(SECURITY_USER_CREDENTIAL_CACHE_DURATION,
DEFAULT_USER_CREDENTIAL_CACHE_DURATION);
}
public boolean getSupportNewAccountRequests() {
@ -525,8 +539,28 @@ public class NiFiProperties extends Properties {
return new File(getNarWorkingDirectory(), "extensions");
}
public File getNarLibraryDirectory() {
return new File(getProperty(NAR_LIBRARY_DIRECTORY, DEFAULT_NAR_LIBRARY_DIR));
public List<Path> getNarLibraryDirectories() {
List<Path> narLibraryPaths = new ArrayList<>();
// go through each property
for (String propertyName : stringPropertyNames()) {
// determine if the property is a nar library path
if (StringUtils.startsWith(propertyName, NAR_LIBRARY_DIRECTORY_PREFIX)
|| NAR_LIBRARY_DIRECTORY.equals(propertyName)) {
// attempt to resolve the path specified
String narLib = getProperty(propertyName);
if (!StringUtils.isBlank(narLib)) {
narLibraryPaths.add(Paths.get(narLib));
}
}
}
if (narLibraryPaths.isEmpty()) {
narLibraryPaths.add(Paths.get(DEFAULT_NAR_LIBRARY_DIR));
}
return narLibraryPaths;
}
// getters for ui properties //
@ -559,7 +593,8 @@ public class NiFiProperties extends Properties {
// getters for cluster protocol properties //
public String getClusterProtocolHeartbeatInterval() {
return getProperty(CLUSTER_PROTOCOL_HEARTBEAT_INTERVAL, DEFAULT_CLUSTER_PROTOCOL_HEARTBEAT_INTERVAL);
return getProperty(CLUSTER_PROTOCOL_HEARTBEAT_INTERVAL,
DEFAULT_CLUSTER_PROTOCOL_HEARTBEAT_INTERVAL);
}
public String getNodeHeartbeatInterval() {
@ -571,7 +606,8 @@ public class NiFiProperties extends Properties {
}
public String getClusterProtocolConnectionHandshakeTimeout() {
return getProperty(CLUSTER_PROTOCOL_CONNECTION_HANDSHAKE_TIMEOUT, DEFAULT_CLUSTER_PROTOCOL_CONNECTION_HANDSHAKE_TIMEOUT);
return getProperty(CLUSTER_PROTOCOL_CONNECTION_HANDSHAKE_TIMEOUT,
DEFAULT_CLUSTER_PROTOCOL_CONNECTION_HANDSHAKE_TIMEOUT);
}
public boolean getClusterProtocolUseMulticast() {
@ -593,7 +629,8 @@ public class NiFiProperties extends Properties {
}
public File getPersistentStateDirectory() {
final String dirName = getProperty(PERSISTENT_STATE_DIRECTORY, DEFAULT_PERSISTENT_STATE_DIRECTORY);
final String dirName = getProperty(PERSISTENT_STATE_DIRECTORY,
DEFAULT_PERSISTENT_STATE_DIRECTORY);
final File file = new File(dirName);
if (!file.exists()) {
file.mkdirs();
@ -603,14 +640,16 @@ public class NiFiProperties extends Properties {
public int getClusterProtocolMulticastServiceLocatorAttempts() {
try {
return Integer.parseInt(getProperty(CLUSTER_PROTOCOL_MULTICAST_SERVICE_LOCATOR_ATTEMPTS));
return Integer
.parseInt(getProperty(CLUSTER_PROTOCOL_MULTICAST_SERVICE_LOCATOR_ATTEMPTS));
} catch (NumberFormatException nfe) {
return DEFAULT_CLUSTER_PROTOCOL_MULTICAST_SERVICE_LOCATOR_ATTEMPTS;
}
}
public String getClusterProtocolMulticastServiceLocatorAttemptsDelay() {
return getProperty(CLUSTER_PROTOCOL_MULTICAST_SERVICE_LOCATOR_ATTEMPTS_DELAY, DEFAULT_CLUSTER_PROTOCOL_MULTICAST_SERVICE_LOCATOR_ATTEMPTS_DELAY);
return getProperty(CLUSTER_PROTOCOL_MULTICAST_SERVICE_LOCATOR_ATTEMPTS_DELAY,
DEFAULT_CLUSTER_PROTOCOL_MULTICAST_SERVICE_LOCATOR_ATTEMPTS_DELAY);
}
// getters for cluster node properties //
@ -653,7 +692,8 @@ public class NiFiProperties extends Properties {
if (StringUtils.isBlank(socketAddress)) {
socketAddress = "localhost";
}
int socketPort = Integer.parseInt(getProperty(CLUSTER_NODE_UNICAST_MANAGER_PROTOCOL_PORT));
int socketPort = Integer
.parseInt(getProperty(CLUSTER_NODE_UNICAST_MANAGER_PROTOCOL_PORT));
return InetSocketAddress.createUnresolved(socketAddress, socketPort);
} catch (Exception ex) {
throw new RuntimeException("Invalid unicast manager address/port due to: " + ex, ex);
@ -704,11 +744,13 @@ public class NiFiProperties extends Properties {
}
public String getClusterManagerNodeApiConnectionTimeout() {
return getProperty(CLUSTER_MANAGER_NODE_API_CONNECTION_TIMEOUT, DEFAULT_CLUSTER_MANAGER_NODE_API_CONNECTION_TIMEOUT);
return getProperty(CLUSTER_MANAGER_NODE_API_CONNECTION_TIMEOUT,
DEFAULT_CLUSTER_MANAGER_NODE_API_CONNECTION_TIMEOUT);
}
public String getClusterManagerNodeApiReadTimeout() {
return getProperty(CLUSTER_MANAGER_NODE_API_READ_TIMEOUT, DEFAULT_CLUSTER_MANAGER_NODE_API_READ_TIMEOUT);
return getProperty(CLUSTER_MANAGER_NODE_API_READ_TIMEOUT,
DEFAULT_CLUSTER_MANAGER_NODE_API_READ_TIMEOUT);
}
public int getClusterManagerNodeApiRequestThreads() {
@ -720,7 +762,8 @@ public class NiFiProperties extends Properties {
}
public String getClusterManagerFlowRetrievalDelay() {
return getProperty(CLUSTER_MANAGER_FLOW_RETRIEVAL_DELAY, DEFAULT_CLUSTER_MANAGER_FLOW_RETRIEVAL_DELAY);
return getProperty(CLUSTER_MANAGER_FLOW_RETRIEVAL_DELAY,
DEFAULT_CLUSTER_MANAGER_FLOW_RETRIEVAL_DELAY);
}
public int getClusterManagerProtocolThreads() {
@ -732,7 +775,8 @@ public class NiFiProperties extends Properties {
}
public String getClusterManagerSafeModeDuration() {
return getProperty(CLUSTER_MANAGER_SAFEMODE_DURATION, DEFAULT_CLUSTER_MANAGER_SAFEMODE_DURATION);
return getProperty(CLUSTER_MANAGER_SAFEMODE_DURATION,
DEFAULT_CLUSTER_MANAGER_SAFEMODE_DURATION);
}
public String getClusterProtocolManagerToNodeApiScheme() {
@ -780,7 +824,8 @@ public class NiFiProperties extends Properties {
* configured. No directories will be created as a result of this operation.
*
* @return database repository path
* @throws InvalidPathException If the configured path is invalid
* @throws InvalidPathException
* If the configured path is invalid
*/
public Path getDatabaseRepositoryPath() {
return Paths.get(getProperty(REPOSITORY_DATABASE_DIRECTORY));
@ -791,7 +836,8 @@ public class NiFiProperties extends Properties {
* configured. No directories will be created as a result of this operation.
*
* @return database repository path
* @throws InvalidPathException If the configured path is invalid
* @throws InvalidPathException
* If the configured path is invalid
*/
public Path getFlowFileRepositoryPath() {
return Paths.get(getProperty(FLOWFILE_REPOSITORY_DIRECTORY));
@ -804,7 +850,8 @@ public class NiFiProperties extends Properties {
* operation.
*
* @return file repositories paths
* @throws InvalidPathException If any of the configured paths are invalid
* @throws InvalidPathException
* If any of the configured paths are invalid
*/
public Map<String, Path> getContentRepositoryPaths() {
final Map<String, Path> contentRepositoryPaths = new HashMap<>();
@ -814,7 +861,8 @@ public class NiFiProperties extends Properties {
// determine if the property is a file repository path
if (StringUtils.startsWith(propertyName, REPOSITORY_CONTENT_PREFIX)) {
// get the repository key
final String key = StringUtils.substringAfter(propertyName, REPOSITORY_CONTENT_PREFIX);
final String key = StringUtils.substringAfter(propertyName,
REPOSITORY_CONTENT_PREFIX);
// attempt to resolve the path specified
contentRepositoryPaths.put(key, Paths.get(getProperty(propertyName)));
@ -839,7 +887,8 @@ public class NiFiProperties extends Properties {
// determine if the property is a file repository path
if (StringUtils.startsWith(propertyName, PROVENANCE_REPO_DIRECTORY_PREFIX)) {
// get the repository key
final String key = StringUtils.substringAfter(propertyName, PROVENANCE_REPO_DIRECTORY_PREFIX);
final String key = StringUtils.substringAfter(propertyName,
PROVENANCE_REPO_DIRECTORY_PREFIX);
// attempt to resolve the path specified
provenanceRepositoryPaths.put(key, Paths.get(getProperty(propertyName)));

95
nifi/nifi-commons/nifi-properties/src/test/java/org/apache/nifi/util/NiFiPropertiesTest.java Normal file
View File

@ -0,0 +1,95 @@
package org.apache.nifi.util;
import static org.junit.Assert.assertEquals;
import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.nio.file.Path;
import java.util.List;
import org.junit.Test;
public class NiFiPropertiesTest {
@Test
public void testProperties() {
NiFiProperties properties = loadSpecifiedProperties("/NiFiProperties/conf/nifi.properties");
assertEquals("UI Banner Text", properties.getBannerText());
List<Path> directories = properties.getNarLibraryDirectories();
assertEquals(new File("./target/resources/NiFiProperties/lib/").getPath(),
directories.get(0).toString());
assertEquals(new File("./target/resources/NiFiProperties/lib2/").getPath(), directories
.get(1).toString());
}
@Test
public void testMissingProperties() {
NiFiProperties properties = loadSpecifiedProperties("/NiFiProperties/conf/nifi.missing.properties");
List<Path> directories = properties.getNarLibraryDirectories();
assertEquals(1, directories.size());
assertEquals(new File(NiFiProperties.DEFAULT_NAR_LIBRARY_DIR).getPath(), directories.get(0)
.toString());
}
@Test
public void testBlankProperties() {
NiFiProperties properties = loadSpecifiedProperties("/NiFiProperties/conf/nifi.blank.properties");
List<Path> directories = properties.getNarLibraryDirectories();
assertEquals(1, directories.size());
assertEquals(new File(NiFiProperties.DEFAULT_NAR_LIBRARY_DIR).getPath(), directories.get(0)
.toString());
}
private NiFiProperties loadSpecifiedProperties(String propertiesFile) {
String file = NiFiPropertiesTest.class.getResource(propertiesFile).getFile();
System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, file);
NiFiProperties properties = NiFiProperties.getInstance();
// clear out existing properties
for (String prop : properties.stringPropertyNames()) {
properties.remove(prop);
}
InputStream inStream = null;
try {
inStream = new BufferedInputStream(new FileInputStream(file));
properties.load(inStream);
} catch (final Exception ex) {
throw new RuntimeException("Cannot load properties file due to "
+ ex.getLocalizedMessage(), ex);
} finally {
if (null != inStream) {
try {
inStream.close();
} catch (final Exception ex) {
/**
* do nothing *
*/
}
}
}
return properties;
}
}

128
nifi/nifi-commons/nifi-properties/src/test/resources/NiFiProperties/conf/nifi.blank.properties Normal file
View File

@ -0,0 +1,128 @@
# 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.
# Core Properties #
nifi.version=nifi-test 3.0.0
nifi.flow.configuration.file=./target/flow.xml.gz
nifi.flow.configuration.archive.dir=./target/archive/
nifi.flowcontroller.autoResumeState=true
nifi.flowcontroller.graceful.shutdown.period=10 sec
nifi.flowservice.writedelay.interval=2 sec
nifi.administrative.yield.duration=30 sec
nifi.reporting.task.configuration.file=./target/reporting-tasks.xml
nifi.controller.service.configuration.file=./target/controller-services.xml
nifi.templates.directory=./target/templates
nifi.ui.banner.text=UI Banner Text
nifi.ui.autorefresh.interval=30 sec
nifi.nar.library.directory=
nifi.custom.nar.library.directory.alt=
nifi.nar.working.directory=./target/work/nar/
# H2 Settings
nifi.database.directory=./target/database_repository
nifi.h2.url.append=;LOCK_TIMEOUT=25000;WRITE_DELAY=0;AUTO_SERVER=FALSE
# FlowFile Repository
nifi.flowfile.repository.directory=./target/test-repo
nifi.flowfile.repository.partitions=1
nifi.flowfile.repository.checkpoint.interval=2 mins
nifi.queue.swap.threshold=20000
nifi.swap.storage.directory=./target/test-repo/swap
nifi.swap.in.period=5 sec
nifi.swap.in.threads=1
nifi.swap.out.period=5 sec
nifi.swap.out.threads=4
# Content Repository
nifi.content.claim.max.appendable.size=10 MB
nifi.content.claim.max.flow.files=100
nifi.content.repository.directory.default=./target/content_repository
# Provenance Repository Properties
nifi.provenance.repository.storage.directory=./target/provenance_repository
nifi.provenance.repository.max.storage.time=24 hours
nifi.provenance.repository.max.storage.size=1 GB
nifi.provenance.repository.rollover.time=5 mins
nifi.provenance.repository.rollover.size=100 MB
# Site to Site properties
nifi.remote.input.socket.port=9990
nifi.remote.input.secure=true
# web properties #
nifi.web.war.directory=./target/lib
nifi.web.http.host=
nifi.web.http.port=8080
nifi.web.https.host=
nifi.web.https.port=
nifi.web.jetty.working.directory=./target/work/jetty
# security properties #
nifi.sensitive.props.key=key
nifi.sensitive.props.algorithm=PBEWITHMD5AND256BITAES-CBC-OPENSSL
nifi.sensitive.props.provider=BC
nifi.security.keystore=
nifi.security.keystoreType=
nifi.security.keystorePasswd=
nifi.security.keyPasswd=
nifi.security.truststore=
nifi.security.truststoreType=
nifi.security.truststorePasswd=
nifi.security.needClientAuth=
nifi.security.authorizedUsers.file=./target/conf/authorized-users.xml
nifi.security.user.credential.cache.duration=24 hours
nifi.security.user.authority.provider=nifi.authorization.FileAuthorizationProvider
nifi.security.support.new.account.requests=
nifi.security.default.user.roles=
# cluster common properties (cluster manager and nodes must have same values) #
nifi.cluster.protocol.heartbeat.interval=5 sec
nifi.cluster.protocol.is.secure=false
nifi.cluster.protocol.socket.timeout=30 sec
nifi.cluster.protocol.connection.handshake.timeout=45 sec
# if multicast is used, then nifi.cluster.protocol.multicast.xxx properties must be configured #
nifi.cluster.protocol.use.multicast=false
nifi.cluster.protocol.multicast.address=
nifi.cluster.protocol.multicast.port=
nifi.cluster.protocol.multicast.service.broadcast.delay=500 ms
nifi.cluster.protocol.multicast.service.locator.attempts=3
nifi.cluster.protocol.multicast.service.locator.attempts.delay=1 sec
# cluster node properties (only configure for cluster nodes) #
nifi.cluster.is.node=false
nifi.cluster.node.address=
nifi.cluster.node.protocol.port=
nifi.cluster.node.protocol.threads=2
# if multicast is not used, nifi.cluster.node.unicast.xxx must have same values as nifi.cluster.manager.xxx #
nifi.cluster.node.unicast.manager.address=
nifi.cluster.node.unicast.manager.protocol.port=
nifi.cluster.node.unicast.manager.authority.provider.port=
# cluster manager properties (only configure for cluster manager) #
nifi.cluster.is.manager=false
nifi.cluster.manager.address=
nifi.cluster.manager.protocol.port=
nifi.cluster.manager.authority.provider.port=
nifi.cluster.manager.authority.provider.threads=10
nifi.cluster.manager.node.firewall.file=
nifi.cluster.manager.node.event.history.size=10
nifi.cluster.manager.node.api.connection.timeout=30 sec
nifi.cluster.manager.node.api.read.timeout=30 sec
nifi.cluster.manager.node.api.request.threads=10
nifi.cluster.manager.flow.retrieval.delay=5 sec
nifi.cluster.manager.protocol.threads=10
nifi.cluster.manager.safemode.duration=0 sec

126
nifi/nifi-commons/nifi-properties/src/test/resources/NiFiProperties/conf/nifi.missing.properties Normal file
View File

@ -0,0 +1,126 @@
# 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.
# Core Properties #
nifi.version=nifi-test 3.0.0
nifi.flow.configuration.file=./target/flow.xml.gz
nifi.flow.configuration.archive.dir=./target/archive/
nifi.flowcontroller.autoResumeState=true
nifi.flowcontroller.graceful.shutdown.period=10 sec
nifi.flowservice.writedelay.interval=2 sec
nifi.administrative.yield.duration=30 sec
nifi.reporting.task.configuration.file=./target/reporting-tasks.xml
nifi.controller.service.configuration.file=./target/controller-services.xml
nifi.templates.directory=./target/templates
nifi.ui.banner.text=UI Banner Text
nifi.ui.autorefresh.interval=30 sec
nifi.nar.working.directory=./target/work/nar/
# H2 Settings
nifi.database.directory=./target/database_repository
nifi.h2.url.append=;LOCK_TIMEOUT=25000;WRITE_DELAY=0;AUTO_SERVER=FALSE
# FlowFile Repository
nifi.flowfile.repository.directory=./target/test-repo
nifi.flowfile.repository.partitions=1
nifi.flowfile.repository.checkpoint.interval=2 mins
nifi.queue.swap.threshold=20000
nifi.swap.storage.directory=./target/test-repo/swap
nifi.swap.in.period=5 sec
nifi.swap.in.threads=1
nifi.swap.out.period=5 sec
nifi.swap.out.threads=4
# Content Repository
nifi.content.claim.max.appendable.size=10 MB
nifi.content.claim.max.flow.files=100
nifi.content.repository.directory.default=./target/content_repository
# Provenance Repository Properties
nifi.provenance.repository.storage.directory=./target/provenance_repository
nifi.provenance.repository.max.storage.time=24 hours
nifi.provenance.repository.max.storage.size=1 GB
nifi.provenance.repository.rollover.time=5 mins
nifi.provenance.repository.rollover.size=100 MB
# Site to Site properties
nifi.remote.input.socket.port=9990
nifi.remote.input.secure=true
# web properties #
nifi.web.war.directory=./target/lib
nifi.web.http.host=
nifi.web.http.port=8080
nifi.web.https.host=
nifi.web.https.port=
nifi.web.jetty.working.directory=./target/work/jetty
# security properties #
nifi.sensitive.props.key=key
nifi.sensitive.props.algorithm=PBEWITHMD5AND256BITAES-CBC-OPENSSL
nifi.sensitive.props.provider=BC
nifi.security.keystore=
nifi.security.keystoreType=
nifi.security.keystorePasswd=
nifi.security.keyPasswd=
nifi.security.truststore=
nifi.security.truststoreType=
nifi.security.truststorePasswd=
nifi.security.needClientAuth=
nifi.security.authorizedUsers.file=./target/conf/authorized-users.xml
nifi.security.user.credential.cache.duration=24 hours
nifi.security.user.authority.provider=nifi.authorization.FileAuthorizationProvider
nifi.security.support.new.account.requests=
nifi.security.default.user.roles=
# cluster common properties (cluster manager and nodes must have same values) #
nifi.cluster.protocol.heartbeat.interval=5 sec
nifi.cluster.protocol.is.secure=false
nifi.cluster.protocol.socket.timeout=30 sec
nifi.cluster.protocol.connection.handshake.timeout=45 sec
# if multicast is used, then nifi.cluster.protocol.multicast.xxx properties must be configured #
nifi.cluster.protocol.use.multicast=false
nifi.cluster.protocol.multicast.address=
nifi.cluster.protocol.multicast.port=
nifi.cluster.protocol.multicast.service.broadcast.delay=500 ms
nifi.cluster.protocol.multicast.service.locator.attempts=3
nifi.cluster.protocol.multicast.service.locator.attempts.delay=1 sec
# cluster node properties (only configure for cluster nodes) #
nifi.cluster.is.node=false
nifi.cluster.node.address=
nifi.cluster.node.protocol.port=
nifi.cluster.node.protocol.threads=2
# if multicast is not used, nifi.cluster.node.unicast.xxx must have same values as nifi.cluster.manager.xxx #
nifi.cluster.node.unicast.manager.address=
nifi.cluster.node.unicast.manager.protocol.port=
nifi.cluster.node.unicast.manager.authority.provider.port=
# cluster manager properties (only configure for cluster manager) #
nifi.cluster.is.manager=false
nifi.cluster.manager.address=
nifi.cluster.manager.protocol.port=
nifi.cluster.manager.authority.provider.port=
nifi.cluster.manager.authority.provider.threads=10
nifi.cluster.manager.node.firewall.file=
nifi.cluster.manager.node.event.history.size=10
nifi.cluster.manager.node.api.connection.timeout=30 sec
nifi.cluster.manager.node.api.read.timeout=30 sec
nifi.cluster.manager.node.api.request.threads=10
nifi.cluster.manager.flow.retrieval.delay=5 sec
nifi.cluster.manager.protocol.threads=10
nifi.cluster.manager.safemode.duration=0 sec

128
nifi/nifi-commons/nifi-properties/src/test/resources/NiFiProperties/conf/nifi.properties Normal file
View File

@ -0,0 +1,128 @@
# 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.
# Core Properties #
nifi.version=nifi-test 3.0.0
nifi.flow.configuration.file=./target/flow.xml.gz
nifi.flow.configuration.archive.dir=./target/archive/
nifi.flowcontroller.autoResumeState=true
nifi.flowcontroller.graceful.shutdown.period=10 sec
nifi.flowservice.writedelay.interval=2 sec
nifi.administrative.yield.duration=30 sec
nifi.reporting.task.configuration.file=./target/reporting-tasks.xml
nifi.controller.service.configuration.file=./target/controller-services.xml
nifi.templates.directory=./target/templates
nifi.ui.banner.text=UI Banner Text
nifi.ui.autorefresh.interval=30 sec
nifi.nar.library.directory=./target/resources/NiFiProperties/lib/
nifi.nar.library.directory.alt=./target/resources/NiFiProperties/lib2/
nifi.nar.working.directory=./target/work/nar/
# H2 Settings
nifi.database.directory=./target/database_repository
nifi.h2.url.append=;LOCK_TIMEOUT=25000;WRITE_DELAY=0;AUTO_SERVER=FALSE
# FlowFile Repository
nifi.flowfile.repository.directory=./target/test-repo
nifi.flowfile.repository.partitions=1
nifi.flowfile.repository.checkpoint.interval=2 mins
nifi.queue.swap.threshold=20000
nifi.swap.storage.directory=./target/test-repo/swap
nifi.swap.in.period=5 sec
nifi.swap.in.threads=1
nifi.swap.out.period=5 sec
nifi.swap.out.threads=4
# Content Repository
nifi.content.claim.max.appendable.size=10 MB
nifi.content.claim.max.flow.files=100
nifi.content.repository.directory.default=./target/content_repository
# Provenance Repository Properties
nifi.provenance.repository.storage.directory=./target/provenance_repository
nifi.provenance.repository.max.storage.time=24 hours
nifi.provenance.repository.max.storage.size=1 GB
nifi.provenance.repository.rollover.time=5 mins
nifi.provenance.repository.rollover.size=100 MB
# Site to Site properties
nifi.remote.input.socket.port=9990
nifi.remote.input.secure=true
# web properties #
nifi.web.war.directory=./target/lib
nifi.web.http.host=
nifi.web.http.port=8080
nifi.web.https.host=
nifi.web.https.port=
nifi.web.jetty.working.directory=./target/work/jetty
# security properties #
nifi.sensitive.props.key=key
nifi.sensitive.props.algorithm=PBEWITHMD5AND256BITAES-CBC-OPENSSL
nifi.sensitive.props.provider=BC
nifi.security.keystore=
nifi.security.keystoreType=
nifi.security.keystorePasswd=
nifi.security.keyPasswd=
nifi.security.truststore=
nifi.security.truststoreType=
nifi.security.truststorePasswd=
nifi.security.needClientAuth=
nifi.security.authorizedUsers.file=./target/conf/authorized-users.xml
nifi.security.user.credential.cache.duration=24 hours
nifi.security.user.authority.provider=nifi.authorization.FileAuthorizationProvider
nifi.security.support.new.account.requests=
nifi.security.default.user.roles=
# cluster common properties (cluster manager and nodes must have same values) #
nifi.cluster.protocol.heartbeat.interval=5 sec
nifi.cluster.protocol.is.secure=false
nifi.cluster.protocol.socket.timeout=30 sec
nifi.cluster.protocol.connection.handshake.timeout=45 sec
# if multicast is used, then nifi.cluster.protocol.multicast.xxx properties must be configured #
nifi.cluster.protocol.use.multicast=false
nifi.cluster.protocol.multicast.address=
nifi.cluster.protocol.multicast.port=
nifi.cluster.protocol.multicast.service.broadcast.delay=500 ms
nifi.cluster.protocol.multicast.service.locator.attempts=3
nifi.cluster.protocol.multicast.service.locator.attempts.delay=1 sec
# cluster node properties (only configure for cluster nodes) #
nifi.cluster.is.node=false
nifi.cluster.node.address=
nifi.cluster.node.protocol.port=
nifi.cluster.node.protocol.threads=2
# if multicast is not used, nifi.cluster.node.unicast.xxx must have same values as nifi.cluster.manager.xxx #
nifi.cluster.node.unicast.manager.address=
nifi.cluster.node.unicast.manager.protocol.port=
nifi.cluster.node.unicast.manager.authority.provider.port=
# cluster manager properties (only configure for cluster manager) #
nifi.cluster.is.manager=false
nifi.cluster.manager.address=
nifi.cluster.manager.protocol.port=
nifi.cluster.manager.authority.provider.port=
nifi.cluster.manager.authority.provider.threads=10
nifi.cluster.manager.node.firewall.file=
nifi.cluster.manager.node.event.history.size=10
nifi.cluster.manager.node.api.connection.timeout=30 sec
nifi.cluster.manager.node.api.read.timeout=30 sec
nifi.cluster.manager.node.api.request.threads=10
nifi.cluster.manager.flow.retrieval.delay=5 sec
nifi.cluster.manager.protocol.threads=10
nifi.cluster.manager.safemode.duration=0 sec

61
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/main/java/org/apache/nifi/documentation/html/HtmlDocumentationWriter.java
View File

@ -20,6 +20,7 @@ import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import javax.xml.stream.FactoryConfigurationError;
import javax.xml.stream.XMLOutputFactory;
@ -36,6 +37,7 @@ import org.apache.nifi.components.ConfigurableComponent;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.documentation.DocumentationWriter;
import org.apache.nifi.nar.ExtensionManager;
/**
* Generates HTML documentation for a ConfigurableComponent. This class is used
@ -148,9 +150,7 @@ public class HtmlDocumentationWriter implements DocumentationWriter {
xmlStreamWriter.writeCharacters(", ");
}
final String link = "../" + linkedComponent.getCanonicalName() + "/index.html";
writeLink(xmlStreamWriter, linkedComponent.getSimpleName(), link);
writeLinkForComponent(xmlStreamWriter, linkedComponent);
++index;
}
@ -434,12 +434,24 @@ public class HtmlDocumentationWriter implements DocumentationWriter {
}
xmlStreamWriter.writeEndElement();
} else if (property.getControllerServiceDefinition() != null) {
Class<? extends ControllerService> controllerServiceClass = property
.getControllerServiceDefinition();
Class<? extends ControllerService> controllerServiceClass = property.getControllerServiceDefinition();
writeSimpleElement(xmlStreamWriter, "strong", "Controller Service: ");
writeSimpleElement(xmlStreamWriter, "strong", "Controller Service API: ");
xmlStreamWriter.writeEmptyElement("br");
xmlStreamWriter.writeCharacters(controllerServiceClass.getSimpleName());
final List<Class<? extends ControllerService>> implementations = lookupControllerServiceImpls(controllerServiceClass);
xmlStreamWriter.writeEmptyElement("br");
if (implementations.size() > 0) {
final String title = implementations.size() > 1 ? "Implementations: " : "Implementation:";
writeSimpleElement(xmlStreamWriter, "strong", title);
for (int i = 0; i < implementations.size(); i++) {
xmlStreamWriter.writeEmptyElement("br");
writeLinkForComponent(xmlStreamWriter, implementations.get(i));
}
} else {
xmlStreamWriter.writeCharacters("No implementations found.");
}
}
}
@ -519,4 +531,41 @@ public class HtmlDocumentationWriter implements DocumentationWriter {
xmlStreamWriter.writeCharacters(text);
xmlStreamWriter.writeEndElement();
}
/**
* Writes a link to another configurable component
*
* @param xmlStreamWriter the xml stream writer
* @param clazz the configurable component to link to
* @throws XMLStreamException thrown if there is a problem writing the XML
*/
protected void writeLinkForComponent(final XMLStreamWriter xmlStreamWriter, final Class<?> clazz) throws XMLStreamException {
writeLink(xmlStreamWriter, clazz.getSimpleName(), "../" + clazz.getCanonicalName() + "/index.html");
}
/**
* Uses the {@link ExtensionManager} to discover any {@link ControllerService} implementations that implement a specific
* ControllerService API.
*
* @param parent the controller service API
* @return a list of controller services that implement the controller service API
*/
private List<Class<? extends ControllerService>> lookupControllerServiceImpls(
final Class<? extends ControllerService> parent) {
final List<Class<? extends ControllerService>> implementations = new ArrayList<>();
// first get all ControllerService implementations
final Set<Class> controllerServices = ExtensionManager.getExtensions(ControllerService.class);
// then iterate over all controller services looking for any that is a child of the parent
// ControllerService API that was passed in as a parameter
for (final Class<? extends ControllerService> controllerServiceClass : controllerServices) {
if (parent.isAssignableFrom(controllerServiceClass)) {
implementations.add(controllerServiceClass);
}
}
return implementations;
}
}

6
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/test/java/org/apache/nifi/documentation/example/FullyDocumentedControllerService.java
View File

@ -28,7 +28,7 @@ import org.apache.nifi.processor.util.StandardValidators;
@CapabilityDescription("A documented controller service that can help you do things")
@Tags({"one", "two", "three"})
public class FullyDocumentedControllerService extends AbstractControllerService {
public class FullyDocumentedControllerService extends AbstractControllerService implements SampleService{
public static final PropertyDescriptor KEYSTORE = new PropertyDescriptor.Builder().name("Keystore Filename")
.description("The fully-qualified filename of the Keystore").defaultValue(null)
@ -55,4 +55,8 @@ public class FullyDocumentedControllerService extends AbstractControllerService
return properties;
}
@Override
public void doSomething() {
// TODO Auto-generated method stub
}
}

2
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/test/java/org/apache/nifi/documentation/html/ProcessorDocumentationWriterTest.java
View File

@ -59,7 +59,7 @@ public class ProcessorDocumentationWriterTest {
assertContains(results, FullyDocumentedProcessor.REL_SUCCESS.getDescription());
assertContains(results, FullyDocumentedProcessor.REL_FAILURE.getName());
assertContains(results, FullyDocumentedProcessor.REL_FAILURE.getDescription());
assertContains(results, "Controller Service: ");
assertContains(results, "Controller Service API: ");
assertContains(results, "SampleService");
assertNotContains(results, "iconSecure.png");

129
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/main/java/org/apache/nifi/nar/NarUnpacker.java
View File

@ -25,6 +25,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
@ -41,7 +42,6 @@ import java.util.jar.Manifest;
import org.apache.nifi.util.FileUtils;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -61,22 +61,34 @@ public final class NarUnpacker {
};
public static ExtensionMapping unpackNars(final NiFiProperties props) {
final File narLibraryDir = props.getNarLibraryDirectory();
final List<Path> narLibraryDirs = props.getNarLibraryDirectories();
final File frameworkWorkingDir = props.getFrameworkWorkingDirectory();
final File extensionsWorkingDir = props.getExtensionsWorkingDirectory();
final File docsWorkingDir = props.getComponentDocumentationWorkingDirectory();
try {
File unpackedFramework = null;
final Set<File> unpackedExtensions = new HashSet<>();
final List<File> narFiles = new ArrayList<>();
// make sure the nar directories are there and accessible
FileUtils.ensureDirectoryExistAndCanAccess(narLibraryDir);
FileUtils.ensureDirectoryExistAndCanAccess(frameworkWorkingDir);
FileUtils.ensureDirectoryExistAndCanAccess(extensionsWorkingDir);
FileUtils.ensureDirectoryExistAndCanAccess(docsWorkingDir);
File unpackedFramework = null;
final Set<File> unpackedExtensions = new HashSet<>();
final File[] narFiles = narLibraryDir.listFiles(NAR_FILTER);
if (narFiles != null) {
for (Path narLibraryDir : narLibraryDirs) {
File narDir = narLibraryDir.toFile();
FileUtils.ensureDirectoryExistAndCanAccess(narDir);
File[] dirFiles = narDir.listFiles(NAR_FILTER);
if (dirFiles != null) {
List<File> fileList = Arrays.asList(dirFiles);
narFiles.addAll(fileList);
}
}
if (!narFiles.isEmpty()) {
for (File narFile : narFiles) {
logger.debug("Expanding NAR file: " + narFile.getAbsolutePath());
@ -91,7 +103,8 @@ public final class NarUnpacker {
// determine if this is the framework
if (NarClassLoaders.FRAMEWORK_NAR_ID.equals(narId)) {
if (unpackedFramework != null) {
throw new IllegalStateException("Multiple framework NARs discovered. Only one framework is permitted.");
throw new IllegalStateException(
"Multiple framework NARs discovered. Only one framework is permitted.");
}
unpackedFramework = unpackNar(narFile, frameworkWorkingDir);
@ -108,7 +121,8 @@ public final class NarUnpacker {
throw new IllegalStateException("Framework NAR cannot be read.");
}
// Determine if any nars no longer exist and delete their working directories. This happens
// Determine if any nars no longer exist and delete their
// working directories. This happens
// if a new version of a nar is dropped into the lib dir.
// ensure no old framework are present
final File[] frameworkWorkingDirContents = frameworkWorkingDir.listFiles();
@ -131,7 +145,8 @@ public final class NarUnpacker {
}
}
// attempt to delete any docs files that exist so that any components that have been removed
// attempt to delete any docs files that exist so that any
// components that have been removed
// will no longer have entries in the docs folder
final File[] docsFiles = docsWorkingDir.listFiles();
if (docsFiles != null) {
@ -144,7 +159,8 @@ public final class NarUnpacker {
mapExtensions(extensionsWorkingDir, docsWorkingDir, extensionMapping);
return extensionMapping;
} catch (IOException e) {
logger.warn("Unable to load NAR library bundles due to " + e + " Will proceed without loading any further Nar bundles");
logger.warn("Unable to load NAR library bundles due to " + e
+ " Will proceed without loading any further Nar bundles");
if (logger.isDebugEnabled()) {
logger.warn("", e);
}
@ -153,7 +169,8 @@ public final class NarUnpacker {
return null;
}
private static void mapExtensions(final File workingDirectory, final File docsDirectory, final ExtensionMapping mapping) throws IOException {
private static void mapExtensions(final File workingDirectory, final File docsDirectory,
final ExtensionMapping mapping) throws IOException {
final File[] directoryContents = workingDirectory.listFiles();
if (directoryContents != null) {
for (final File file : directoryContents) {
@ -169,19 +186,24 @@ public final class NarUnpacker {
/**
* Unpacks the specified nar into the specified base working directory.
*
* @param nar the nar to unpack
* @param baseWorkingDirectory the directory to unpack to
* @param nar
* the nar to unpack
* @param baseWorkingDirectory
* the directory to unpack to
* @return the directory to the unpacked NAR
* @throws IOException if unable to explode nar
* @throws IOException
* if unable to explode nar
*/
private static File unpackNar(final File nar, final File baseWorkingDirectory) throws IOException {
private static File unpackNar(final File nar, final File baseWorkingDirectory)
throws IOException {
final File narWorkingDirectory = new File(baseWorkingDirectory, nar.getName() + "-unpacked");
// if the working directory doesn't exist, unpack the nar
if (!narWorkingDirectory.exists()) {
unpack(nar, narWorkingDirectory, calculateMd5sum(nar));
} else {
// the working directory does exist. Run MD5 sum against the nar file and check if the nar has changed since it was deployed.
// the working directory does exist. Run MD5 sum against the nar
// file and check if the nar has changed since it was deployed.
final byte[] narMd5 = calculateMd5sum(nar);
final File workingHashFile = new File(narWorkingDirectory, HASH_FILENAME);
if (!workingHashFile.exists()) {
@ -190,7 +212,8 @@ public final class NarUnpacker {
} else {
final byte[] hashFileContents = Files.readAllBytes(workingHashFile.toPath());
if (!Arrays.equals(hashFileContents, narMd5)) {
logger.info("Contents of nar {} have changed. Reloading.", new Object[]{nar.getAbsolutePath()});
logger.info("Contents of nar {} have changed. Reloading.",
new Object[] { nar.getAbsolutePath() });
FileUtils.deleteFile(narWorkingDirectory, true);
unpack(nar, narWorkingDirectory, narMd5);
}
@ -204,11 +227,13 @@ public final class NarUnpacker {
* Unpacks the NAR to the specified directory. Creates a checksum file that
* used to determine if future expansion is necessary.
*
* @param workingDirectory the root directory to which the NAR should be
* unpacked.
* @throws IOException if the NAR could not be unpacked.
* @param workingDirectory
* the root directory to which the NAR should be unpacked.
* @throws IOException
* if the NAR could not be unpacked.
*/
private static void unpack(final File nar, final File workingDirectory, final byte[] hash) throws IOException {
private static void unpack(final File nar, final File workingDirectory, final byte[] hash)
throws IOException {
try (JarFile jarFile = new JarFile(nar)) {
Enumeration<JarEntry> jarEntries = jarFile.entries();
@ -230,7 +255,8 @@ public final class NarUnpacker {
}
}
private static void unpackDocumentation(final File jar, final File docsDirectory, final ExtensionMapping extensionMapping) throws IOException {
private static void unpackDocumentation(final File jar, final File docsDirectory,
final ExtensionMapping extensionMapping) throws IOException {
// determine the components that may have documentation
determineDocumentedNiFiComponents(jar, extensionMapping);
@ -240,7 +266,8 @@ public final class NarUnpacker {
final String entryName = "docs/" + componentName;
// go through each entry in this jar
for (final Enumeration<JarEntry> jarEnumeration = jarFile.entries(); jarEnumeration.hasMoreElements();) {
for (final Enumeration<JarEntry> jarEnumeration = jarFile.entries(); jarEnumeration
.hasMoreElements();) {
final JarEntry jarEntry = jarEnumeration.nextElement();
// if this entry is documentation for this component
@ -252,8 +279,10 @@ public final class NarUnpacker {
final File componentDocsDirectory = new File(docsDirectory, name);
// ensure the documentation directory can be created
if (!componentDocsDirectory.exists() && !componentDocsDirectory.mkdirs()) {
logger.warn("Unable to create docs directory " + componentDocsDirectory.getAbsolutePath());
if (!componentDocsDirectory.exists()
&& !componentDocsDirectory.mkdirs()) {
logger.warn("Unable to create docs directory "
+ componentDocsDirectory.getAbsolutePath());
break;
}
} else {
@ -268,19 +297,27 @@ public final class NarUnpacker {
}
}
private static void determineDocumentedNiFiComponents(final File jar, final ExtensionMapping extensionMapping) throws IOException {
private static void determineDocumentedNiFiComponents(final File jar,
final ExtensionMapping extensionMapping) throws IOException {
try (final JarFile jarFile = new JarFile(jar)) {
final JarEntry processorEntry = jarFile.getJarEntry("META-INF/services/org.apache.nifi.processor.Processor");
final JarEntry reportingTaskEntry = jarFile.getJarEntry("META-INF/services/org.apache.nifi.reporting.ReportingTask");
final JarEntry controllerServiceEntry = jarFile.getJarEntry("META-INF/services/org.apache.nifi.controller.ControllerService");
final JarEntry processorEntry = jarFile
.getJarEntry("META-INF/services/org.apache.nifi.processor.Processor");
final JarEntry reportingTaskEntry = jarFile
.getJarEntry("META-INF/services/org.apache.nifi.reporting.ReportingTask");
final JarEntry controllerServiceEntry = jarFile
.getJarEntry("META-INF/services/org.apache.nifi.controller.ControllerService");
extensionMapping.addAllProcessors(determineDocumentedNiFiComponents(jarFile, processorEntry));
extensionMapping.addAllReportingTasks(determineDocumentedNiFiComponents(jarFile, reportingTaskEntry));
extensionMapping.addAllControllerServices(determineDocumentedNiFiComponents(jarFile, controllerServiceEntry));
extensionMapping.addAllProcessors(determineDocumentedNiFiComponents(jarFile,
processorEntry));
extensionMapping.addAllReportingTasks(determineDocumentedNiFiComponents(jarFile,
reportingTaskEntry));
extensionMapping.addAllControllerServices(determineDocumentedNiFiComponents(jarFile,
controllerServiceEntry));
}
}
private static List<String> determineDocumentedNiFiComponents(final JarFile jarFile, final JarEntry jarEntry) throws IOException {
private static List<String> determineDocumentedNiFiComponents(final JarFile jarFile,
final JarEntry jarEntry) throws IOException {
final List<String> componentNames = new ArrayList<>();
if (jarEntry == null) {
@ -288,13 +325,15 @@ public final class NarUnpacker {
}
try (final InputStream entryInputStream = jarFile.getInputStream(jarEntry);
final BufferedReader reader = new BufferedReader(new InputStreamReader(entryInputStream))) {
final BufferedReader reader = new BufferedReader(new InputStreamReader(
entryInputStream))) {
String line;
while ((line = reader.readLine()) != null) {
final String trimmedLine = line.trim();
if (!trimmedLine.isEmpty() && !trimmedLine.startsWith("#")) {
final int indexOfPound = trimmedLine.indexOf("#");
final String effectiveLine = (indexOfPound > 0) ? trimmedLine.substring(0, indexOfPound) : trimmedLine;
final String effectiveLine = (indexOfPound > 0) ? trimmedLine.substring(0,
indexOfPound) : trimmedLine;
componentNames.add(effectiveLine);
}
}
@ -307,12 +346,16 @@ public final class NarUnpacker {
* Creates the specified file, whose contents will come from the
* <tt>InputStream</tt>.
*
* @param inputStream the contents of the file to create.
* @param file the file to create.
* @throws IOException if the file could not be created.
* @param inputStream
* the contents of the file to create.
* @param file
* the file to create.
* @throws IOException
* if the file could not be created.
*/
private static void makeFile(final InputStream inputStream, final File file) throws IOException {
try (final InputStream in = inputStream; final FileOutputStream fos = new FileOutputStream(file)) {
try (final InputStream in = inputStream;
final FileOutputStream fos = new FileOutputStream(file)) {
byte[] bytes = new byte[65536];
int numRead;
while ((numRead = in.read(bytes)) != -1) {
@ -324,9 +367,11 @@ public final class NarUnpacker {
/**
* Calculates an md5 sum of the specified file.
*
* @param file to calculate the md5sum of
* @param file
* to calculate the md5sum of
* @return the md5sum bytes
* @throws IOException if cannot read file
* @throws IOException
* if cannot read file
*/
private static byte[] calculateMd5sum(final File file) throws IOException {
try (final FileInputStream inputStream = new FileInputStream(file)) {

185
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/test/java/org/apache/nifi/nar/NarUnpackerTest.java Normal file
View File

@ -0,0 +1,185 @@
package org.apache.nifi.nar;
import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.FileVisitResult;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import org.apache.nifi.util.NiFiProperties;
import org.junit.BeforeClass;
import org.junit.Test;
public class NarUnpackerTest {
@BeforeClass
public static void copyResources() throws IOException {
final Path sourcePath = Paths.get("./src/test/resources");
final Path targetPath = Paths.get("./target");
Files.walkFileTree(sourcePath, new SimpleFileVisitor<Path>() {
@Override
public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs)
throws IOException {
Path relativeSource = sourcePath.relativize(dir);
Path target = targetPath.resolve(relativeSource);
Files.createDirectories(target);
return FileVisitResult.CONTINUE;
}
@Override
public FileVisitResult visitFile(Path file, BasicFileAttributes attrs)
throws IOException {
Path relativeSource = sourcePath.relativize(file);
Path target = targetPath.resolve(relativeSource);
Files.copy(file, target, REPLACE_EXISTING);
return FileVisitResult.CONTINUE;
}
});
}
@Test
public void testUnpackNars() {
NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties");
assertEquals("./target/NarUnpacker/lib/",
properties.getProperty("nifi.nar.library.directory"));
assertEquals("./target/NarUnpacker/lib2/",
properties.getProperty("nifi.nar.library.directory.alt"));
final ExtensionMapping extensionMapping = NarUnpacker.unpackNars(properties);
assertEquals(2, extensionMapping.getAllExtensionNames().size());
assertTrue(extensionMapping.getAllExtensionNames().contains(
"org.apache.nifi.processors.dummy.one"));
assertTrue(extensionMapping.getAllExtensionNames().contains(
"org.apache.nifi.processors.dummy.two"));
final File extensionsWorkingDir = properties.getExtensionsWorkingDirectory();
File[] extensionFiles = extensionsWorkingDir.listFiles();
assertEquals(2, extensionFiles.length);
assertEquals("dummy-one.nar-unpacked", extensionFiles[0].getName());
assertEquals("dummy-two.nar-unpacked", extensionFiles[1].getName());
}
@Test
public void testUnpackNarsFromEmptyDir() throws IOException {
NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties");
final File emptyDir = new File("./target/empty/dir");
emptyDir.delete();
emptyDir.deleteOnExit();
assertTrue(emptyDir.mkdirs());
properties.setProperty("nifi.nar.library.directory.alt", emptyDir.toString());
final ExtensionMapping extensionMapping = NarUnpacker.unpackNars(properties);
assertEquals(1, extensionMapping.getAllExtensionNames().size());
assertTrue(extensionMapping.getAllExtensionNames().contains(
"org.apache.nifi.processors.dummy.one"));
final File extensionsWorkingDir = properties.getExtensionsWorkingDirectory();
File[] extensionFiles = extensionsWorkingDir.listFiles();
assertEquals(1, extensionFiles.length);
assertEquals("dummy-one.nar-unpacked", extensionFiles[0].getName());
}
@Test
public void testUnpackNarsFromNonExistantDir() {
final File nonExistantDir = new File("./target/this/dir/should/not/exist/");
nonExistantDir.delete();
nonExistantDir.deleteOnExit();
NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties");
properties.setProperty("nifi.nar.library.directory.alt", nonExistantDir.toString());
final ExtensionMapping extensionMapping = NarUnpacker.unpackNars(properties);
assertTrue(extensionMapping.getAllExtensionNames().contains(
"org.apache.nifi.processors.dummy.one"));
assertEquals(1, extensionMapping.getAllExtensionNames().size());
final File extensionsWorkingDir = properties.getExtensionsWorkingDirectory();
File[] extensionFiles = extensionsWorkingDir.listFiles();
assertEquals(1, extensionFiles.length);
assertEquals("dummy-one.nar-unpacked", extensionFiles[0].getName());
}
@Test
public void testUnpackNarsFromNonDir() throws IOException {
final File nonDir = new File("./target/file.txt");
nonDir.createNewFile();
nonDir.deleteOnExit();
NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties");
properties.setProperty("nifi.nar.library.directory.alt", nonDir.toString());
final ExtensionMapping extensionMapping = NarUnpacker.unpackNars(properties);
assertNull(extensionMapping);
}
private NiFiProperties loadSpecifiedProperties(String propertiesFile) {
String file = NarUnpackerTest.class.getResource(propertiesFile).getFile();
System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, file);
NiFiProperties properties = NiFiProperties.getInstance();
// clear out existing properties
for (String prop : properties.stringPropertyNames()) {
properties.remove(prop);
}
InputStream inStream = null;
try {
inStream = new BufferedInputStream(new FileInputStream(file));
properties.load(inStream);
} catch (final Exception ex) {
throw new RuntimeException("Cannot load properties file due to "
+ ex.getLocalizedMessage(), ex);
} finally {
if (null != inStream) {
try {
inStream.close();
} catch (final Exception ex) {
/**
* do nothing *
*/
}
}
}
return properties;
}
}

129
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/test/resources/NarUnpacker/conf/nifi.properties Normal file
View File

@ -0,0 +1,129 @@
# 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.
# Core Properties #
nifi.version=nifi-test 3.0.0
nifi.flow.configuration.file=./target/flow.xml.gz
nifi.flow.configuration.archive.dir=./target/archive/
nifi.flowcontroller.autoResumeState=true
nifi.flowcontroller.graceful.shutdown.period=10 sec
nifi.flowservice.writedelay.interval=2 sec
nifi.administrative.yield.duration=30 sec
nifi.reporting.task.configuration.file=./target/reporting-tasks.xml
nifi.controller.service.configuration.file=./target/controller-services.xml
nifi.templates.directory=./target/templates
nifi.ui.banner.text=UI Banner Text
nifi.ui.autorefresh.interval=30 sec
nifi.nar.library.directory=./target/NarUnpacker/lib/
nifi.nar.library.directory.alt=./target/NarUnpacker/lib2/
nifi.nar.working.directory=./target/work/nar/
# H2 Settings
nifi.database.directory=./target/database_repository
nifi.h2.url.append=;LOCK_TIMEOUT=25000;WRITE_DELAY=0;AUTO_SERVER=FALSE
# FlowFile Repository
nifi.flowfile.repository.directory=./target/test-repo
nifi.flowfile.repository.partitions=1
nifi.flowfile.repository.checkpoint.interval=2 mins
nifi.queue.swap.threshold=20000
nifi.swap.storage.directory=./target/test-repo/swap
nifi.swap.in.period=5 sec
nifi.swap.in.threads=1
nifi.swap.out.period=5 sec
nifi.swap.out.threads=4
# Content Repository
nifi.content.claim.max.appendable.size=10 MB
nifi.content.claim.max.flow.files=100
nifi.content.repository.directory.default=./target/content_repository
# Provenance Repository Properties
nifi.provenance.repository.storage.directory=./target/provenance_repository
nifi.provenance.repository.max.storage.time=24 hours
nifi.provenance.repository.max.storage.size=1 GB
nifi.provenance.repository.rollover.time=5 mins
nifi.provenance.repository.rollover.size=100 MB
# Site to Site properties
nifi.remote.input.socket.port=9990
nifi.remote.input.secure=true
# web properties #
nifi.web.war.directory=./target/lib
nifi.web.http.host=
nifi.web.http.port=8080
nifi.web.https.host=
nifi.web.https.port=
nifi.web.jetty.working.directory=./target/work/jetty
# security properties #
nifi.sensitive.props.key=key
nifi.sensitive.props.algorithm=PBEWITHMD5AND256BITAES-CBC-OPENSSL
nifi.sensitive.props.provider=BC
nifi.security.keystore=
nifi.security.keystoreType=
nifi.security.keystorePasswd=
nifi.security.keyPasswd=
nifi.security.truststore=
nifi.security.truststoreType=
nifi.security.truststorePasswd=
nifi.security.needClientAuth=
nifi.security.authorizedUsers.file=./target/conf/authorized-users.xml
nifi.security.user.credential.cache.duration=24 hours
nifi.security.user.authority.provider=nifi.authorization.FileAuthorizationProvider
nifi.security.support.new.account.requests=
nifi.security.default.user.roles=
# cluster common properties (cluster manager and nodes must have same values) #
nifi.cluster.protocol.heartbeat.interval=5 sec
nifi.cluster.protocol.is.secure=false
nifi.cluster.protocol.socket.timeout=30 sec
nifi.cluster.protocol.connection.handshake.timeout=45 sec
# if multicast is used, then nifi.cluster.protocol.multicast.xxx properties must be configured #
nifi.cluster.protocol.use.multicast=false
nifi.cluster.protocol.multicast.address=
nifi.cluster.protocol.multicast.port=
nifi.cluster.protocol.multicast.service.broadcast.delay=500 ms
nifi.cluster.protocol.multicast.service.locator.attempts=3
nifi.cluster.protocol.multicast.service.locator.attempts.delay=1 sec
# cluster node properties (only configure for cluster nodes) #
nifi.cluster.is.node=false
nifi.cluster.node.address=
nifi.cluster.node.protocol.port=
nifi.cluster.node.protocol.threads=2
# if multicast is not used, nifi.cluster.node.unicast.xxx must have same values as nifi.cluster.manager.xxx #
nifi.cluster.node.unicast.manager.address=
nifi.cluster.node.unicast.manager.protocol.port=
nifi.cluster.node.unicast.manager.authority.provider.port=
# cluster manager properties (only configure for cluster manager) #
nifi.cluster.is.manager=false
nifi.cluster.manager.address=
nifi.cluster.manager.protocol.port=
nifi.cluster.manager.authority.provider.port=
nifi.cluster.manager.authority.provider.threads=10
nifi.cluster.manager.node.firewall.file=
nifi.cluster.manager.node.event.history.size=10
nifi.cluster.manager.node.api.connection.timeout=30 sec
nifi.cluster.manager.node.api.read.timeout=30 sec
nifi.cluster.manager.node.api.request.threads=10
nifi.cluster.manager.flow.retrieval.delay=5 sec
nifi.cluster.manager.protocol.threads=10
nifi.cluster.manager.safemode.duration=0 sec

BIN
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/test/resources/NarUnpacker/lib/dummy-one.nar Normal file
View File

Binary file not shown.

BIN
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/test/resources/NarUnpacker/lib/nifi-framework-nar.nar Normal file
View File

Binary file not shown.

BIN
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/test/resources/NarUnpacker/lib2/dummy-two.nar Normal file
View File

Binary file not shown.