http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/main/java/org/apache/nifi/nar/NarThreadContextClassLoader.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/main/java/org/apache/nifi/nar/NarThreadContextClassLoader.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/main/java/org/apache/nifi/nar/NarThreadContextClassLoader.java index 381b54b..827abdd 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/main/java/org/apache/nifi/nar/NarThreadContextClassLoader.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/main/java/org/apache/nifi/nar/NarThreadContextClassLoader.java @@ -34,11 +34,14 @@ import org.apache.nifi.reporting.ReportingTask; import java.io.IOException; import java.io.InputStream; +import java.lang.reflect.Constructor; +import java.lang.reflect.InvocationTargetException; import java.net.URL; import java.net.URLClassLoader; import java.util.ArrayList; import java.util.Enumeration; import java.util.List; +import org.apache.nifi.util.NiFiProperties; /** * THREAD SAFE @@ -165,7 +168,22 @@ public class NarThreadContextClassLoader extends URLClassLoader { } } - public static <T> T createInstance(final String implementationClassName, final Class<T> typeDefinition) throws InstantiationException, IllegalAccessException, ClassNotFoundException { + /** + * Constructs an instance of the given type using either default no args + * constructor or a constructor which takes a NiFiProperties object + * (preferred). + * + * @param <T> type + * @param implementationClassName class + * @param typeDefinition def + * @param nifiProperties props + * @return constructed instance + * @throws InstantiationException ex + * @throws IllegalAccessException ex + * @throws ClassNotFoundException ex + */ + public static <T> T createInstance(final String implementationClassName, final Class<T> typeDefinition, final NiFiProperties nifiProperties) + throws InstantiationException, IllegalAccessException, ClassNotFoundException { final ClassLoader originalClassLoader = Thread.currentThread().getContextClassLoader(); Thread.currentThread().setContextClassLoader(NarThreadContextClassLoader.getInstance()); try { @@ -181,7 +199,16 @@ public class NarThreadContextClassLoader extends URLClassLoader { Thread.currentThread().setContextClassLoader(detectedClassLoaderForType); final Class<?> desiredClass = rawClass.asSubclass(typeDefinition); - return typeDefinition.cast(desiredClass.newInstance()); + if (nifiProperties == null) { + return typeDefinition.cast(desiredClass.newInstance()); + } + Constructor<?> constructor = null; + try { + constructor = desiredClass.getConstructor(NiFiProperties.class); + return typeDefinition.cast(constructor.newInstance(nifiProperties)); + } catch (final NoSuchMethodException | InvocationTargetException ex) { + return typeDefinition.cast(desiredClass.newInstance()); + } } finally { Thread.currentThread().setContextClassLoader(originalClassLoader); }
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/test/java/org/apache/nifi/nar/NarUnpackerTest.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/test/java/org/apache/nifi/nar/NarUnpackerTest.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/test/java/org/apache/nifi/nar/NarUnpackerTest.java index 882c8c6..0fc8f4d 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/test/java/org/apache/nifi/nar/NarUnpackerTest.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/test/java/org/apache/nifi/nar/NarUnpackerTest.java @@ -21,11 +21,8 @@ import org.junit.Assert; import org.junit.BeforeClass; import org.junit.Test; -import java.io.BufferedInputStream; import java.io.File; -import java.io.FileInputStream; import java.io.IOException; -import java.io.InputStream; import java.net.URISyntaxException; import java.nio.file.FileVisitResult; import java.nio.file.Files; @@ -37,6 +34,9 @@ import java.util.HashSet; import java.util.Set; import static java.nio.file.StandardCopyOption.REPLACE_EXISTING; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; @@ -81,7 +81,7 @@ public class NarUnpackerTest { @Test public void testUnpackNars() { - NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties"); + NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties", Collections.EMPTY_MAP); assertEquals("./target/NarUnpacker/lib/", properties.getProperty("nifi.nar.library.directory")); @@ -112,14 +112,14 @@ public class NarUnpackerTest { @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 Map<String, String> others = new HashMap<>(); + others.put("nifi.nar.library.directory.alt", emptyDir.toString()); + NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties", others); final ExtensionMapping extensionMapping = NarUnpacker.unpackNars(properties); @@ -141,8 +141,9 @@ public class NarUnpackerTest { nonExistantDir.delete(); nonExistantDir.deleteOnExit(); - NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties"); - properties.setProperty("nifi.nar.library.directory.alt", nonExistantDir.toString()); + final Map<String, String> others = new HashMap<>(); + others.put("nifi.nar.library.directory.alt", nonExistantDir.toString()); + NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties", others); final ExtensionMapping extensionMapping = NarUnpacker.unpackNars(properties); @@ -165,15 +166,16 @@ public class NarUnpackerTest { nonDir.createNewFile(); nonDir.deleteOnExit(); - NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties"); - properties.setProperty("nifi.nar.library.directory.alt", nonDir.toString()); + final Map<String, String> others = new HashMap<>(); + others.put("nifi.nar.library.directory.alt", nonDir.toString()); + NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties", others); final ExtensionMapping extensionMapping = NarUnpacker.unpackNars(properties); assertNull(extensionMapping); } - private NiFiProperties loadSpecifiedProperties(String propertiesFile) { + private NiFiProperties loadSpecifiedProperties(final String propertiesFile, final Map<String, String> others) { String filePath; try { filePath = NarUnpackerTest.class.getResource(propertiesFile).toURI().getPath(); @@ -181,34 +183,6 @@ public class NarUnpackerTest { throw new RuntimeException("Cannot load properties file due to " + ex.getLocalizedMessage(), ex); } - System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, filePath); - - 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(filePath)); - 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; + return NiFiProperties.createBasicNiFiProperties(filePath, others); } -} \ No newline at end of file +} http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-runtime/src/main/java/org/apache/nifi/NiFi.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-runtime/src/main/java/org/apache/nifi/NiFi.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-runtime/src/main/java/org/apache/nifi/NiFi.java index a426c40..6d0fa97 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-runtime/src/main/java/org/apache/nifi/NiFi.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-runtime/src/main/java/org/apache/nifi/NiFi.java @@ -224,7 +224,7 @@ public class NiFi { public static void main(String[] args) { logger.info("Launching NiFi..."); try { - new NiFi(NiFiProperties.getInstance()); + new NiFi(NiFiProperties.createBasicNiFiProperties(null, null)); } catch (final Throwable t) { logger.error("Failure to launch NiFi due to " + t, t); } http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslServerSocketFactory.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslServerSocketFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslServerSocketFactory.java index 59ea312..458157c 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslServerSocketFactory.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslServerSocketFactory.java @@ -36,8 +36,8 @@ public class SslServerSocketFactory extends SSLServerSocketFactory { private SSLServerSocketFactory sslServerSocketFactory; - public SslServerSocketFactory() { - final SSLContext sslCtx = SslContextFactory.createSslContext(NiFiProperties.getInstance()); + public SslServerSocketFactory(final NiFiProperties nifiProperties) { + final SSLContext sslCtx = SslContextFactory.createSslContext(nifiProperties); if (sslCtx == null) { try { sslServerSocketFactory = SSLContext.getDefault().getServerSocketFactory(); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslSocketFactory.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslSocketFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslSocketFactory.java index da0e7fb..fa6de56 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslSocketFactory.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslSocketFactory.java @@ -36,8 +36,8 @@ public class SslSocketFactory extends SSLSocketFactory { private final SSLSocketFactory sslSocketFactory; - public SslSocketFactory() { - final SSLContext sslCtx = SslContextFactory.createSslContext(NiFiProperties.getInstance()); + public SslSocketFactory(final NiFiProperties nifiProperties) { + final SSLContext sslCtx = SslContextFactory.createSslContext(nifiProperties); if (sslCtx == null) { try { sslSocketFactory = SSLContext.getDefault().getSocketFactory(); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/HttpRemoteSiteListener.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/HttpRemoteSiteListener.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/HttpRemoteSiteListener.java index b335f48..7a001ab 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/HttpRemoteSiteListener.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/HttpRemoteSiteListener.java @@ -51,7 +51,7 @@ public class HttpRemoteSiteListener implements RemoteSiteListener { private ProcessGroup rootGroup; private ScheduledFuture<?> transactionMaintenanceTask; - private HttpRemoteSiteListener() { + private HttpRemoteSiteListener(final NiFiProperties nifiProperties) { super(); taskExecutor = Executors.newScheduledThreadPool(1, new ThreadFactory() { private final ThreadFactory defaultFactory = Executors.defaultThreadFactory(); @@ -65,10 +65,9 @@ public class HttpRemoteSiteListener implements RemoteSiteListener { } }); - NiFiProperties properties = NiFiProperties.getInstance(); int txTtlSec; try { - final String snapshotFrequency = properties.getProperty(SITE_TO_SITE_HTTP_TRANSACTION_TTL, DEFAULT_SITE_TO_SITE_HTTP_TRANSACTION_TTL); + final String snapshotFrequency = nifiProperties.getProperty(SITE_TO_SITE_HTTP_TRANSACTION_TTL, DEFAULT_SITE_TO_SITE_HTTP_TRANSACTION_TTL); txTtlSec = (int) FormatUtils.getTimeDuration(snapshotFrequency, TimeUnit.SECONDS); } catch (final Exception e) { txTtlSec = (int) FormatUtils.getTimeDuration(DEFAULT_SITE_TO_SITE_HTTP_TRANSACTION_TTL, TimeUnit.SECONDS); @@ -78,11 +77,11 @@ public class HttpRemoteSiteListener implements RemoteSiteListener { transactionTtlSec = txTtlSec; } - public static HttpRemoteSiteListener getInstance() { + public static HttpRemoteSiteListener getInstance(final NiFiProperties nifiProperties) { if (instance == null) { synchronized (HttpRemoteSiteListener.class) { if (instance == null) { - instance = new HttpRemoteSiteListener(); + instance = new HttpRemoteSiteListener(nifiProperties); } } } @@ -90,6 +89,7 @@ public class HttpRemoteSiteListener implements RemoteSiteListener { } private class TransactionWrapper { + private final FlowFileTransaction transaction; private final HandshakeProperties handshakenProperties; private long lastCommunicationAt; @@ -129,7 +129,7 @@ public class HttpRemoteSiteListener implements RemoteSiteListener { try { Set<String> transactionIds = transactions.keySet().stream().collect(Collectors.toSet()); transactionIds.stream().filter(tid -> !isTransactionActive(tid)) - .forEach(tid -> cancelTransaction(tid)); + .forEach(tid -> cancelTransaction(tid)); } catch (Exception e) { // Swallow exception so that this thread can keep working. logger.error("An exception occurred while maintaining transactions", e); @@ -146,7 +146,7 @@ public class HttpRemoteSiteListener implements RemoteSiteListener { } else { logger.debug("Cancel a transaction. transactionId={}", transactionId); FlowFileTransaction t = wrapper.transaction; - if(t != null && t.getSession() != null){ + if (t != null && t.getSession() != null) { logger.info("Cancel a transaction, rollback its session. transactionId={}", transactionId); try { t.getSession().rollback(); @@ -158,10 +158,9 @@ public class HttpRemoteSiteListener implements RemoteSiteListener { } } - @Override public void stop() { - if(transactionMaintenanceTask != null) { + if (transactionMaintenanceTask != null) { logger.debug("Stopping transactionMaintenanceTask..."); transactionMaintenanceTask.cancel(true); } @@ -191,10 +190,9 @@ public class HttpRemoteSiteListener implements RemoteSiteListener { /** * @param transactionId transactionId to check - * @return Returns a HandshakeProperties instance which is created when this transaction is started, - only if the transaction is active, - and it holds a HandshakeProperties, - otherwise return null + * @return Returns a HandshakeProperties instance which is created when this + * transaction is started, only if the transaction is active, and it holds a + * HandshakeProperties, otherwise return null */ public HandshakeProperties getHandshakenProperties(final String transactionId) { TransactionWrapper transaction = transactions.get(transactionId); @@ -205,7 +203,7 @@ public class HttpRemoteSiteListener implements RemoteSiteListener { } public void holdTransaction(final String transactionId, final FlowFileTransaction transaction, - final HandshakeProperties handshakenProperties) throws IllegalStateException { + final HandshakeProperties handshakenProperties) throws IllegalStateException { // We don't check expiration of the transaction here, to support large file transport or slow network. // The availability of current transaction is already checked when the HTTP request was received at SiteToSiteResource. TransactionWrapper currentTransaction = transactions.remove(transactionId); @@ -224,7 +222,7 @@ public class HttpRemoteSiteListener implements RemoteSiteListener { } public FlowFileTransaction finalizeTransaction(final String transactionId) throws IllegalStateException { - if (!isTransactionActive(transactionId)){ + if (!isTransactionActive(transactionId)) { throw new IllegalStateException("Transaction was not found or not active anymore. transactionId=" + transactionId); } TransactionWrapper transaction = transactions.remove(transactionId); @@ -239,7 +237,7 @@ public class HttpRemoteSiteListener implements RemoteSiteListener { } public void extendTransaction(final String transactionId) throws IllegalStateException { - if (!isTransactionActive(transactionId)){ + if (!isTransactionActive(transactionId)) { throw new IllegalStateException("Transaction was not found or not active anymore. transactionId=" + transactionId); } TransactionWrapper transaction = transactions.get(transactionId); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/RemoteResourceFactory.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/RemoteResourceFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/RemoteResourceFactory.java index b0ce357..0e29aef 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/RemoteResourceFactory.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/RemoteResourceFactory.java @@ -122,8 +122,8 @@ public class RemoteResourceFactory extends RemoteResourceInitiator { } public static <T extends VersionedRemoteResource> T - receiveResourceNegotiation(final Class<T> cls, final DataInputStream dis, final DataOutputStream dos, final Class<?>[] constructorArgClasses, final Object[] constructorArgs) - throws IOException, HandshakeException { + receiveResourceNegotiation(final Class<T> cls, final DataInputStream dis, final DataOutputStream dos, final Class<?>[] constructorArgClasses, final Object[] constructorArgs) + throws IOException, HandshakeException { final String resourceClassName = dis.readUTF(); final T resource; try { http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/SocketRemoteSiteListener.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/SocketRemoteSiteListener.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/SocketRemoteSiteListener.java index bd9d204..9f9e3aa 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/SocketRemoteSiteListener.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/SocketRemoteSiteListener.java @@ -46,6 +46,8 @@ import java.util.Arrays; import java.util.Optional; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import org.apache.nifi.remote.cluster.ClusterNodeInformation; +import org.apache.nifi.util.NiFiProperties; public class SocketRemoteSiteListener implements RemoteSiteListener { @@ -55,18 +57,20 @@ public class SocketRemoteSiteListener implements RemoteSiteListener { private final SSLContext sslContext; private final NodeInformant nodeInformant; private final AtomicReference<ProcessGroup> rootGroup = new AtomicReference<>(); + private final NiFiProperties nifiProperties; private final AtomicBoolean stopped = new AtomicBoolean(false); private static final Logger LOG = LoggerFactory.getLogger(SocketRemoteSiteListener.class); - public SocketRemoteSiteListener(final int socketPort, final SSLContext sslContext) { - this(socketPort, sslContext, null); + public SocketRemoteSiteListener(final int socketPort, final SSLContext sslContext, final NiFiProperties nifiProperties) { + this(socketPort, sslContext, nifiProperties, null); } - public SocketRemoteSiteListener(final int socketPort, final SSLContext sslContext, final NodeInformant nodeInformant) { + public SocketRemoteSiteListener(final int socketPort, final SSLContext sslContext, final NiFiProperties nifiProperties, final NodeInformant nodeInformant) { this.socketPort = socketPort; this.sslContext = sslContext; + this.nifiProperties = nifiProperties; this.nodeInformant = nodeInformant; } @@ -267,7 +271,14 @@ public class SocketRemoteSiteListener implements RemoteSiteListener { protocol.getPort().receiveFlowFiles(peer, protocol); break; case REQUEST_PEER_LIST: - protocol.sendPeerList(peer, nodeInformant == null ? Optional.empty() : Optional.of(nodeInformant.getNodeInformation())); + final Optional<ClusterNodeInformation> nodeInfo = (nodeInformant == null) ? Optional.empty() : Optional.of(nodeInformant.getNodeInformation()); + protocol.sendPeerList( + peer, + nodeInfo, + nifiProperties.getRemoteInputHost(), + nifiProperties.getRemoteInputPort(), + nifiProperties.getRemoteInputHttpPort(), + nifiProperties.isSiteToSiteSecure()); break; case SHUTDOWN: protocol.shutdown(peer); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/StandardRemoteGroupPort.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/StandardRemoteGroupPort.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/StandardRemoteGroupPort.java index 8f115f7..1996357 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/StandardRemoteGroupPort.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/StandardRemoteGroupPort.java @@ -74,14 +74,17 @@ public class StandardRemoteGroupPort extends RemoteGroupPort { private final AtomicBoolean targetRunning = new AtomicBoolean(true); private final SSLContext sslContext; private final TransferDirection transferDirection; + private final NiFiProperties nifiProperties; private final AtomicReference<SiteToSiteClient> clientRef = new AtomicReference<>(); + SiteToSiteClient getSiteToSiteClient() { return clientRef.get(); } public StandardRemoteGroupPort(final String id, final String name, final ProcessGroup processGroup, final RemoteProcessGroup remoteGroup, - final TransferDirection direction, final ConnectableType type, final SSLContext sslContext, final ProcessScheduler scheduler) { + final TransferDirection direction, final ConnectableType type, final SSLContext sslContext, final ProcessScheduler scheduler, + final NiFiProperties nifiProperties) { // remote group port id needs to be unique but cannot just be the id of the port // in the remote group instance. this supports referencing the same remote // instance more than once. @@ -90,11 +93,12 @@ public class StandardRemoteGroupPort extends RemoteGroupPort { this.remoteGroup = remoteGroup; this.transferDirection = direction; this.sslContext = sslContext; + this.nifiProperties = nifiProperties; setScheduldingPeriod(MINIMUM_SCHEDULING_NANOS + " nanos"); } - private static File getPeerPersistenceFile(final String portId) { - final File stateDir = NiFiProperties.getInstance().getPersistentStateDirectory(); + private static File getPeerPersistenceFile(final String portId, final NiFiProperties nifiProperties) { + final File stateDir = nifiProperties.getPersistentStateDirectory(); return new File(stateDir, portId + ".peers"); } @@ -138,17 +142,17 @@ public class StandardRemoteGroupPort extends RemoteGroupPort { final long penalizationMillis = FormatUtils.getTimeDuration(remoteGroup.getYieldDuration(), TimeUnit.MILLISECONDS); final SiteToSiteClient client = new SiteToSiteClient.Builder() - .url(remoteGroup.getTargetUri().toString()) - .portIdentifier(getIdentifier()) - .sslContext(sslContext) - .useCompression(isUseCompression()) - .eventReporter(remoteGroup.getEventReporter()) - .peerPersistenceFile(getPeerPersistenceFile(getIdentifier())) - .nodePenalizationPeriod(penalizationMillis, TimeUnit.MILLISECONDS) - .timeout(remoteGroup.getCommunicationsTimeout(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS) - .transportProtocol(remoteGroup.getTransportProtocol()) - .httpProxy(new HttpProxy(remoteGroup.getProxyHost(), remoteGroup.getProxyPort(), remoteGroup.getProxyUser(), remoteGroup.getProxyPassword())) - .build(); + .url(remoteGroup.getTargetUri().toString()) + .portIdentifier(getIdentifier()) + .sslContext(sslContext) + .useCompression(isUseCompression()) + .eventReporter(remoteGroup.getEventReporter()) + .peerPersistenceFile(getPeerPersistenceFile(getIdentifier(), nifiProperties)) + .nodePenalizationPeriod(penalizationMillis, TimeUnit.MILLISECONDS) + .timeout(remoteGroup.getCommunicationsTimeout(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS) + .transportProtocol(remoteGroup.getTransportProtocol()) + .httpProxy(new HttpProxy(remoteGroup.getProxyHost(), remoteGroup.getProxyPort(), remoteGroup.getProxyUser(), remoteGroup.getProxyPassword())) + .build(); clientRef.set(client); } @@ -306,7 +310,7 @@ public class StandardRemoteGroupPort extends RemoteGroupPort { final String flowFileDescription = (flowFilesSent.size() < 20) ? flowFilesSent.toString() : flowFilesSent.size() + " FlowFiles"; logger.info("{} Successfully sent {} ({}) to {} in {} milliseconds at a rate of {}", new Object[]{ - this, flowFileDescription, dataSize, transaction.getCommunicant().getUrl(), uploadMillis, uploadDataRate}); + this, flowFileDescription, dataSize, transaction.getCommunicant().getUrl(), uploadMillis, uploadDataRate}); return flowFilesSent.size(); } catch (final Exception e) { @@ -364,7 +368,7 @@ public class StandardRemoteGroupPort extends RemoteGroupPort { final long uploadMillis = stopWatch.getDuration(TimeUnit.MILLISECONDS); final String dataSize = FormatUtils.formatDataSize(bytesReceived); logger.info("{} Successfully receveied {} ({}) from {} in {} milliseconds at a rate of {}", new Object[]{ - this, flowFileDescription, dataSize, transaction.getCommunicant().getUrl(), uploadMillis, uploadDataRate}); + this, flowFileDescription, dataSize, transaction.getCommunicant().getUrl(), uploadMillis, uploadDataRate}); } return flowFilesReceived.size(); @@ -386,16 +390,16 @@ public class StandardRemoteGroupPort extends RemoteGroupPort { ValidationResult error = null; if (!targetExists.get()) { error = new ValidationResult.Builder() - .explanation(String.format("Remote instance indicates that port '%s' no longer exists.", getName())) - .subject(String.format("Remote port '%s'", getName())) - .valid(false) - .build(); + .explanation(String.format("Remote instance indicates that port '%s' no longer exists.", getName())) + .subject(String.format("Remote port '%s'", getName())) + .valid(false) + .build(); } else if (getConnectableType() == ConnectableType.REMOTE_OUTPUT_PORT && getConnections(Relationship.ANONYMOUS).isEmpty()) { error = new ValidationResult.Builder() - .explanation(String.format("Port '%s' has no outbound connections", getName())) - .subject(String.format("Remote port '%s'", getName())) - .valid(false) - .build(); + .explanation(String.format("Port '%s' has no outbound connections", getName())) + .subject(String.format("Remote port '%s'", getName())) + .valid(false) + .build(); } if (error != null) { http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/AbstractFlowFileServerProtocol.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/AbstractFlowFileServerProtocol.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/AbstractFlowFileServerProtocol.java index 8600368..fe324ad 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/AbstractFlowFileServerProtocol.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/AbstractFlowFileServerProtocol.java @@ -201,7 +201,7 @@ public abstract class AbstractFlowFileServerProtocol implements ServerProtocol { handshakeCompleted = true; } - abstract protected HandshakeProperties doHandshake(final Peer peer) throws IOException, HandshakeException; + abstract protected HandshakeProperties doHandshake(final Peer peer) throws IOException, HandshakeException; @Override public int transferFlowFiles(final Peer peer, final ProcessContext context, final ProcessSession session, final FlowFileCodec codec) throws IOException, ProtocolException { @@ -394,9 +394,10 @@ public abstract class AbstractFlowFileServerProtocol implements ServerProtocol { protected final void writeTransactionResponse(boolean isTransfer, ResponseCode response, CommunicationsSession commsSession) throws IOException { writeTransactionResponse(isTransfer, response, commsSession, null); } + protected void writeTransactionResponse(boolean isTransfer, ResponseCode response, CommunicationsSession commsSession, String explanation) throws IOException { final DataOutputStream dos = new DataOutputStream(commsSession.getOutput().getOutputStream()); - if(explanation == null){ + if (explanation == null) { response.writeResponse(dos); } else { response.writeResponse(dos, explanation); @@ -436,7 +437,7 @@ public abstract class AbstractFlowFileServerProtocol implements ServerProtocol { final CheckedInputStream checkedInputStream = new CheckedInputStream(flowFileInputStream, crc); final DataPacket dataPacket = codec.decode(checkedInputStream); - if(dataPacket == null){ + if (dataPacket == null) { logger.debug("{} Received null dataPacket indicating the end of transaction from {}", this, peer); break; } @@ -528,7 +529,7 @@ public abstract class AbstractFlowFileServerProtocol implements ServerProtocol { final long uploadMillis = stopWatch.getDuration(TimeUnit.MILLISECONDS); final String dataSize = FormatUtils.formatDataSize(bytesReceived); logger.info("{} Successfully received {} ({}) from {} in {} milliseconds at a rate of {}", new Object[]{ - this, flowFileDescription, dataSize, peer, uploadMillis, uploadDataRate}); + this, flowFileDescription, dataSize, peer, uploadMillis, uploadDataRate}); return flowFilesReceived.size(); } http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/FlowFileTransaction.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/FlowFileTransaction.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/FlowFileTransaction.java index 560cbaf..4f14fbb 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/FlowFileTransaction.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/FlowFileTransaction.java @@ -33,7 +33,7 @@ public class FlowFileTransaction { private final String calculatedCRC; public FlowFileTransaction() { - this(null, null, new StopWatch(true), 0, null, null); + this(null, null, new StopWatch(true), 0, null, null); } public FlowFileTransaction(ProcessSession session, ProcessContext context, StopWatch stopWatch, long bytesSent, Set<FlowFile> flowFilesSent, String calculatedCRC) { http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/HandshakeProperties.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/HandshakeProperties.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/HandshakeProperties.java index c4538da..b12a5b5 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/HandshakeProperties.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/HandshakeProperties.java @@ -28,7 +28,6 @@ public class HandshakeProperties { private long batchBytes = 0L; private long batchDurationNanos = 0L; - public String getCommsIdentifier() { return commsIdentifier; } http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/http/StandardHttpFlowFileServerProtocol.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/http/StandardHttpFlowFileServerProtocol.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/http/StandardHttpFlowFileServerProtocol.java index b2171df..e71ac1d 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/http/StandardHttpFlowFileServerProtocol.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/http/StandardHttpFlowFileServerProtocol.java @@ -40,6 +40,7 @@ import java.io.DataInputStream; import java.io.DataOutputStream; import java.io.IOException; import java.util.Optional; +import org.apache.nifi.util.NiFiProperties; public class StandardHttpFlowFileServerProtocol extends AbstractFlowFileServerProtocol implements HttpFlowFileServerProtocol { @@ -47,11 +48,12 @@ public class StandardHttpFlowFileServerProtocol extends AbstractFlowFileServerPr private final FlowFileCodec codec = new StandardFlowFileCodec(); private final VersionNegotiator versionNegotiator; - private final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(); + private final HttpRemoteSiteListener transactionManager; - public StandardHttpFlowFileServerProtocol(final VersionNegotiator versionNegotiator) { + public StandardHttpFlowFileServerProtocol(final VersionNegotiator versionNegotiator, final NiFiProperties nifiProperties) { super(); this.versionNegotiator = versionNegotiator; + this.transactionManager = HttpRemoteSiteListener.getInstance(nifiProperties); } @Override @@ -91,7 +93,7 @@ public class StandardHttpFlowFileServerProtocol extends AbstractFlowFileServerPr HttpServerCommunicationsSession commSession = (HttpServerCommunicationsSession) commsSession; commSession.setResponseCode(response); - if(isTransfer){ + if (isTransfer) { switch (response) { case NO_MORE_DATA: logger.debug("{} There's no data to send.", this); @@ -136,8 +138,8 @@ public class StandardHttpFlowFileServerProtocol extends AbstractFlowFileServerPr ByteArrayOutputStream bos = new ByteArrayOutputStream(); Transaction.TransactionState currentStatus = commSession.getStatus(); - if(isTransfer){ - switch (currentStatus){ + if (isTransfer) { + switch (currentStatus) { case DATA_EXCHANGED: String clientChecksum = commSession.getChecksum(); logger.debug("readTransactionResponse. clientChecksum={}", clientChecksum); @@ -149,7 +151,7 @@ public class StandardHttpFlowFileServerProtocol extends AbstractFlowFileServerPr break; } } else { - switch (currentStatus){ + switch (currentStatus) { case TRANSACTION_STARTED: logger.debug("readTransactionResponse. returning CONTINUE_TRANSACTION."); // We don't know if there's more data to receive, so just continue it. @@ -159,7 +161,7 @@ public class StandardHttpFlowFileServerProtocol extends AbstractFlowFileServerPr // Checksum was successfully validated at client side, or BAD_CHECKSUM is returned. ResponseCode responseCode = commSession.getResponseCode(); logger.debug("readTransactionResponse. responseCode={}", responseCode); - if(responseCode.containsMessage()){ + if (responseCode.containsMessage()) { responseCode.writeResponse(new DataOutputStream(bos), ""); } else { responseCode.writeResponse(new DataOutputStream(bos)); @@ -226,7 +228,8 @@ public class StandardHttpFlowFileServerProtocol extends AbstractFlowFileServerPr } @Override - public void sendPeerList(final Peer peer, final Optional<ClusterNodeInformation> clusterNodeInformation) throws IOException { + public void sendPeerList(Peer peer, Optional<ClusterNodeInformation> clusterNodeInfo, String remoteInputHost, + int remoteInputPort, int remoteInputHttpPort, boolean isSiteToSiteSecure) throws IOException { } @Override http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/socket/SocketFlowFileServerProtocol.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/socket/SocketFlowFileServerProtocol.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/socket/SocketFlowFileServerProtocol.java index 574c726..e965cf4 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/socket/SocketFlowFileServerProtocol.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/socket/SocketFlowFileServerProtocol.java @@ -30,7 +30,6 @@ import org.apache.nifi.remote.protocol.CommunicationsSession; import org.apache.nifi.remote.protocol.HandshakeProperties; import org.apache.nifi.remote.protocol.RequestType; import org.apache.nifi.remote.protocol.ResponseCode; -import org.apache.nifi.util.NiFiProperties; import java.io.DataInputStream; import java.io.DataOutputStream; @@ -84,7 +83,7 @@ public class SocketFlowFileServerProtocol extends AbstractFlowFileServerProtocol validateHandshakeRequest(confirmed, peer, properties); } catch (HandshakeException e) { ResponseCode handshakeResult = e.getResponseCode(); - if(handshakeResult.containsMessage()){ + if (handshakeResult.containsMessage()) { handshakeResult.writeResponse(dos, e.getMessage()); } else { handshakeResult.writeResponse(dos); @@ -136,7 +135,6 @@ public class SocketFlowFileServerProtocol extends AbstractFlowFileServerProtocol } } - @Override public RequestType getRequestType(final Peer peer) throws IOException { if (!handshakeCompleted) { @@ -154,7 +152,13 @@ public class SocketFlowFileServerProtocol extends AbstractFlowFileServerProtocol } @Override - public void sendPeerList(final Peer peer, final Optional<ClusterNodeInformation> clusterNodeInfo) throws IOException { + public void sendPeerList( + final Peer peer, + final Optional<ClusterNodeInformation> clusterNodeInfo, + final String remoteInputHost, + final int remoteInputPort, + final int remoteInputHttpPort, + final boolean isSiteToSiteSecure) throws IOException { if (!handshakeCompleted) { throw new IllegalStateException("Handshake has not been completed"); } @@ -166,11 +170,9 @@ public class SocketFlowFileServerProtocol extends AbstractFlowFileServerProtocol final CommunicationsSession commsSession = peer.getCommunicationsSession(); final DataOutputStream dos = new DataOutputStream(commsSession.getOutput().getOutputStream()); - final NiFiProperties properties = NiFiProperties.getInstance(); - - String remoteInputHost = properties.getRemoteInputHost(); - if (remoteInputHost == null) { - remoteInputHost = InetAddress.getLocalHost().getHostName(); + String remoteInputHostVal = remoteInputHost; + if (remoteInputHostVal == null) { + remoteInputHostVal = InetAddress.getLocalHost().getHostName(); } logger.debug("{} Advertising Remote Input host name {}", this, peer); @@ -178,8 +180,8 @@ public class SocketFlowFileServerProtocol extends AbstractFlowFileServerProtocol if (clusterNodeInfo.isPresent()) { nodeInfos = new ArrayList<>(clusterNodeInfo.get().getNodeInformation()); } else { - final NodeInformation self = new NodeInformation(remoteInputHost, properties.getRemoteInputPort(), properties.getRemoteInputHttpPort(), properties.getRemoteInputHttpPort(), - properties.isSiteToSiteSecure(), 0); + final NodeInformation self = new NodeInformation(remoteInputHostVal, remoteInputPort, remoteInputHttpPort, remoteInputHttpPort, + isSiteToSiteSecure, 0); nodeInfos = Collections.singletonList(self); } @@ -212,7 +214,6 @@ public class SocketFlowFileServerProtocol extends AbstractFlowFileServerProtocol return RESOURCE_NAME; } - @Override public VersionNegotiator getVersionNegotiator() { return versionNegotiator; http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestHttpRemoteSiteListener.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestHttpRemoteSiteListener.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestHttpRemoteSiteListener.java index e485095..58d4d26 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestHttpRemoteSiteListener.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestHttpRemoteSiteListener.java @@ -39,7 +39,7 @@ public class TestHttpRemoteSiteListener { @Test public void testNormalTransactionProgress() { - HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(); + HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null)); String transactionId = transactionManager.createTransaction(); assertTrue("Transaction should be active.", transactionManager.isTransactionActive(transactionId)); @@ -59,7 +59,7 @@ public class TestHttpRemoteSiteListener { @Test public void testDuplicatedTransactionId() { - HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(); + HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null)); String transactionId = transactionManager.createTransaction(); assertTrue("Transaction should be active.", transactionManager.isTransactionActive(transactionId)); @@ -78,7 +78,7 @@ public class TestHttpRemoteSiteListener { @Test public void testNoneExistingTransaction() { - HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(); + HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null)); String transactionId = "does-not-exist-1"; assertFalse("Transaction should not be active.", transactionManager.isTransactionActive(transactionId)); @@ -88,8 +88,8 @@ public class TestHttpRemoteSiteListener { try { transactionManager.holdTransaction(transactionId, transaction, null); } catch (IllegalStateException e) { - fail("Transaction can be held even if the transaction id is not valid anymore," + - " in order to support large file or slow network."); + fail("Transaction can be held even if the transaction id is not valid anymore," + + " in order to support large file or slow network."); } transactionId = "does-not-exist-2"; @@ -100,5 +100,4 @@ public class TestHttpRemoteSiteListener { } } - } http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestStandardRemoteGroupPort.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestStandardRemoteGroupPort.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestStandardRemoteGroupPort.java index 4209c93..b44f118 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestStandardRemoteGroupPort.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestStandardRemoteGroupPort.java @@ -43,6 +43,7 @@ import java.net.URI; import java.nio.channels.SocketChannel; import java.util.HashMap; import java.util.Map; +import org.apache.nifi.util.NiFiProperties; import static org.mockito.Matchers.any; import static org.mockito.Matchers.eq; @@ -70,17 +71,18 @@ public class TestStandardRemoteGroupPort { @BeforeClass public static void setup() throws Exception { + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties"); System.setProperty("org.slf4j.simpleLogger.log.org.apache.nifi.remote", "DEBUG"); } private void setupMock(final SiteToSiteTransportProtocol protocol, - final TransferDirection direction) throws Exception { + final TransferDirection direction) throws Exception { setupMock(protocol, direction, mock(Transaction.class)); } private void setupMock(final SiteToSiteTransportProtocol protocol, - final TransferDirection direction, - final Transaction transaction) throws Exception { + final TransferDirection direction, + final Transaction transaction) throws Exception { processGroup = null; remoteGroup = mock(RemoteProcessGroup.class); scheduler = null; @@ -102,7 +104,7 @@ public class TestStandardRemoteGroupPort { break; } port = spy(new StandardRemoteGroupPort(ID, NAME, - processGroup, remoteGroup, direction, connectableType, null, scheduler)); + processGroup, remoteGroup, direction, connectableType, null, scheduler, NiFiProperties.createBasicNiFiProperties(null, null))); doReturn(true).when(remoteGroup).isTransmitting(); doReturn(protocol).when(remoteGroup).getTransportProtocol(); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/TestSocketChannelStreams.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/TestSocketChannelStreams.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/TestSocketChannelStreams.java index 03f8190..27a8807 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/TestSocketChannelStreams.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/TestSocketChannelStreams.java @@ -76,7 +76,7 @@ package org.apache.nifi.remote.io.socket; // dos.flush(); // // final EventReporter eventReporter = Mockito.mock(EventReporter.class); -// final StandardSiteToSiteProtocol proposedProtocol = new StandardSiteToSiteProtocol(commsSession, eventReporter, NiFiProperties.getInstance()); +// final StandardSiteToSiteProtocol proposedProtocol = new StandardSiteToSiteProtocol(commsSession, eventReporter, nifiProperties); // // final StandardSiteToSiteProtocol negotiatedProtocol = (StandardSiteToSiteProtocol) RemoteResourceFactory.initiateResourceNegotiation(proposedProtocol, dis, dos); // System.out.println(negotiatedProtocol); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/ssl/TestSSLSocketChannel.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/ssl/TestSSLSocketChannel.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/ssl/TestSSLSocketChannel.java index 8fe7149..2ead857 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/ssl/TestSSLSocketChannel.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/ssl/TestSSLSocketChannel.java @@ -67,7 +67,7 @@ package org.apache.nifi.remote.io.socket.ssl; // public void testSendingToLocalInstance() throws IOException, InterruptedException, HandshakeException, UnknownPortException, PortNotRunningException, URISyntaxException { // System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties"); // -// final NiFiProperties properties = NiFiProperties.getInstance(); +// final NiFiProperties properties; // final SSLContext sslContext = SslContextFactory.createSslContext(properties); // // final SSLSocketChannel channel = new SSLSocketChannel(sslContext, "localhost", 5000, true); @@ -84,7 +84,7 @@ package org.apache.nifi.remote.io.socket.ssl; // dos.flush(); // // final EventReporter eventReporter = Mockito.mock(EventReporter.class); -// final StandardSiteToSiteProtocol proposedProtocol = new StandardSiteToSiteProtocol(commsSession, eventReporter, NiFiProperties.getInstance()); +// final StandardSiteToSiteProtocol proposedProtocol = new StandardSiteToSiteProtocol(commsSession, eventReporter, nifiProperties); // final StandardSiteToSiteProtocol negotiatedProtocol = (StandardSiteToSiteProtocol) RemoteResourceFactory.initiateResourceNegotiation(proposedProtocol, dis, dos); // System.out.println(negotiatedProtocol); // @@ -104,7 +104,7 @@ package org.apache.nifi.remote.io.socket.ssl; // public void testWithSimpleSSLSocket() throws IOException, InterruptedException { // System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties"); // -// final NiFiProperties properties = NiFiProperties.getInstance(); +// final NiFiProperties properties; // final SSLContext sslContext = SslContextFactory.createSslContext(properties); // // final ServerThread server = new ServerThread(sslContext); @@ -138,7 +138,7 @@ package org.apache.nifi.remote.io.socket.ssl; // public void testDirectChannelComms() throws IOException, InterruptedException { // System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties"); // -// final NiFiProperties properties = NiFiProperties.getInstance(); +// final NiFiProperties properties; // final SSLContext sslContext = SslContextFactory.createSslContext(properties); // // final ServerThread server = new ServerThread(sslContext); @@ -193,7 +193,7 @@ package org.apache.nifi.remote.io.socket.ssl; // public void testWriteTimesOut() throws IOException, InterruptedException { // System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties"); // -// final NiFiProperties properties = NiFiProperties.getInstance(); +// final NiFiProperties properties; // final SSLContext sslContext = SslContextFactory.createSslContext(properties); // // final ServerThread server = new ServerThread(sslContext); @@ -238,7 +238,7 @@ package org.apache.nifi.remote.io.socket.ssl; // public void testInputOutputStreams() throws IOException, InterruptedException { // System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties"); // -// final NiFiProperties properties = NiFiProperties.getInstance(); +// final NiFiProperties properties; // final SSLContext sslContext = SslContextFactory.createSslContext(properties); // // final ServerThread server = new ServerThread(sslContext); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/protocol/http/TestHttpFlowFileServerProtocol.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/protocol/http/TestHttpFlowFileServerProtocol.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/protocol/http/TestHttpFlowFileServerProtocol.java index 7c9d30b..f5e803d 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/protocol/http/TestHttpFlowFileServerProtocol.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/protocol/http/TestHttpFlowFileServerProtocol.java @@ -90,7 +90,7 @@ public class TestHttpFlowFileServerProtocol { private HttpFlowFileServerProtocol getDefaultHttpFlowFileServerProtocol() { final StandardVersionNegotiator versionNegotiator = new StandardVersionNegotiator(5, 4, 3, 2, 1); - return new StandardHttpFlowFileServerProtocol(versionNegotiator); + return new StandardHttpFlowFileServerProtocol(versionNegotiator, NiFiProperties.createBasicNiFiProperties(null, null)); } @Test @@ -294,7 +294,7 @@ public class TestHttpFlowFileServerProtocol { } private Peer transferOneFile(final HttpFlowFileServerProtocol serverProtocol, final String transactionId) throws IOException { - final HttpRemoteSiteListener remoteSiteListener = HttpRemoteSiteListener.getInstance(); + final HttpRemoteSiteListener remoteSiteListener = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null)); final Peer peer = getDefaultPeer(transactionId); final HttpServerCommunicationsSession commsSession = (HttpServerCommunicationsSession) peer.getCommunicationsSession(); final String endpointUri = "https://peer-host:8443/nifi-api/output-ports/port-id/transactions/"; @@ -338,7 +338,7 @@ public class TestHttpFlowFileServerProtocol { @Test public void testTransferTwoFiles() throws Exception { - final HttpRemoteSiteListener remoteSiteListener = HttpRemoteSiteListener.getInstance(); + final HttpRemoteSiteListener remoteSiteListener = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null)); final String transactionId = "testTransferTwoFiles"; final Peer peer = getDefaultPeer(transactionId); @@ -470,7 +470,7 @@ public class TestHttpFlowFileServerProtocol { } private void receiveOneFile(final HttpFlowFileServerProtocol serverProtocol, final String transactionId, final Peer peer) throws IOException { - final HttpRemoteSiteListener remoteSiteListener = HttpRemoteSiteListener.getInstance(); + final HttpRemoteSiteListener remoteSiteListener = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null)); final String endpointUri = "https://peer-host:8443/nifi-api/input-ports/port-id/transactions/"; + transactionId + "/flow-files"; final HttpServerCommunicationsSession commsSession = (HttpServerCommunicationsSession) peer.getCommunicationsSession(); @@ -530,7 +530,7 @@ public class TestHttpFlowFileServerProtocol { @Test public void testReceiveTwoFiles() throws Exception { - final HttpRemoteSiteListener remoteSiteListener = HttpRemoteSiteListener.getInstance(); + final HttpRemoteSiteListener remoteSiteListener = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null)); final String transactionId = "testReceiveTwoFile"; final String endpointUri = "https://peer-host:8443/nifi-api/input-ports/port-id/transactions/"; http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/test/java/org/apache/nifi/web/server/JettyServerTest.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/test/java/org/apache/nifi/web/server/JettyServerTest.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/test/java/org/apache/nifi/web/server/JettyServerTest.java index 314e331..f4f0bf8 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/test/java/org/apache/nifi/web/server/JettyServerTest.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/test/java/org/apache/nifi/web/server/JettyServerTest.java @@ -17,13 +17,13 @@ package org.apache.nifi.web.server; +import java.lang.reflect.InvocationTargetException; +import java.util.HashMap; +import java.util.Map; import org.eclipse.jetty.util.ssl.SslContextFactory; import org.apache.nifi.util.NiFiProperties; import org.junit.Test; -import java.lang.reflect.Constructor; -import java.lang.reflect.InvocationTargetException; - import static org.mockito.Matchers.anyString; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -31,17 +31,17 @@ import static org.mockito.Mockito.verify; public class JettyServerTest { @Test - public void testConfigureSslContextFactoryWithKeystorePasswordAndKeyPassword() throws InvocationTargetException, NoSuchMethodException, InstantiationException, IllegalAccessException { + public void testConfigureSslContextFactoryWithKeystorePasswordAndKeyPassword() { // Expect that if we set both passwords, KeyStore password is used for KeyStore, Key password is used for Key Manager String testKeystorePassword = "testKeystorePassword"; String testKeyPassword = "testKeyPassword"; - NiFiProperties nifiProperties = createNifiProperties(); + final Map<String, String> addProps = new HashMap<>(); + addProps.put(NiFiProperties.SECURITY_KEYSTORE_PASSWD, testKeystorePassword); + addProps.put(NiFiProperties.SECURITY_KEY_PASSWD, testKeyPassword); + NiFiProperties nifiProperties = NiFiProperties.createBasicNiFiProperties(null, addProps); SslContextFactory contextFactory = mock(SslContextFactory.class); - nifiProperties.setProperty(NiFiProperties.SECURITY_KEYSTORE_PASSWD, testKeystorePassword); - nifiProperties.setProperty(NiFiProperties.SECURITY_KEY_PASSWD, testKeyPassword); - JettyServer.configureSslContextFactory(contextFactory, nifiProperties); verify(contextFactory).setKeyStorePassword(testKeystorePassword); @@ -53,11 +53,11 @@ public class JettyServerTest { // Expect that with no KeyStore password, we will only need to set Key Manager Password String testKeyPassword = "testKeyPassword"; - NiFiProperties nifiProperties = createNifiProperties(); + final Map<String, String> addProps = new HashMap<>(); + addProps.put(NiFiProperties.SECURITY_KEY_PASSWD, testKeyPassword); + NiFiProperties nifiProperties = NiFiProperties.createBasicNiFiProperties(null, addProps); SslContextFactory contextFactory = mock(SslContextFactory.class); - nifiProperties.setProperty(NiFiProperties.SECURITY_KEY_PASSWD, testKeyPassword); - JettyServer.configureSslContextFactory(contextFactory, nifiProperties); verify(contextFactory).setKeyManagerPassword(testKeyPassword); @@ -69,20 +69,15 @@ public class JettyServerTest { // Expect that with no KeyPassword, we use the same one from the KeyStore String testKeystorePassword = "testKeystorePassword"; - NiFiProperties nifiProperties = createNifiProperties(); + final Map<String, String> addProps = new HashMap<>(); + addProps.put(NiFiProperties.SECURITY_KEYSTORE_PASSWD, testKeystorePassword); + NiFiProperties nifiProperties = NiFiProperties.createBasicNiFiProperties(null, addProps); SslContextFactory contextFactory = mock(SslContextFactory.class); - nifiProperties.setProperty(NiFiProperties.SECURITY_KEYSTORE_PASSWD, testKeystorePassword); - JettyServer.configureSslContextFactory(contextFactory, nifiProperties); verify(contextFactory).setKeyStorePassword(testKeystorePassword); verify(contextFactory).setKeyManagerPassword(testKeystorePassword); } - private NiFiProperties createNifiProperties() throws NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException { - Constructor<NiFiProperties> constructor = NiFiProperties.class.getDeclaredConstructor(); - constructor.setAccessible(true); - return constructor.newInstance(); - } }
