You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jo...@apache.org on 2022/05/20 16:32:10 UTC
[nifi] branch support/nifi-1.16 updated (c6e3cdd555 -> 3440921bb6)
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a change to branch support/nifi-1.16
in repository https://gitbox.apache.org/repos/asf/nifi.git
from c6e3cdd555 NIFI-10001: When enabling a collection of Controller Services, change… (#6042)
new 9314799935 NIFI-10015 Upgraded Hadoop from 3.3.0 to 3.3.2
new 7ed0637bb6 NIFI-10036: Corrected Elasticsearch Client Service to prefix paths for all requests
new 7268e638dd NIFI-10037: When system test fails to clean up flow, destroy the entire environment so that the next test starts in a healthy state. Name troubleshooting directories with the name of the test class to avoid ambiguity. Also added a log statement so that we know which test is running when looking at the log output from the tests themselves. Finally, found an issue in AbstractComponentNode in which we iterate over the elements in a Map and call setProperty, which can upda [...]
new 0394270007 NIFI-3869 Added HTTP/2 support to ListenHTTP and HandleHttpRequest
new 3440921bb6 NIFI-10005 fixing version references
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../pom.xml | 19 +-
.../connector/ApplicationLayerProtocol.java} | 22 +--
.../connector/ServerConnectorFactory.java | 14 +-
.../connector/StandardServerConnectorFactory.java | 193 +++++++++++++++++++++
.../alpn/ALPNServerConnectionFactory.java | 64 +++++++
.../connector/alpn/StandardALPNProcessor.java | 123 +++++++++++++
.../StandardServerConnectorFactoryTest.java | 172 ++++++++++++++++++
nifi-commons/pom.xml | 1 +
.../ElasticSearchClientServiceImpl.java | 10 +-
.../nifi/controller/AbstractComponentNode.java | 6 +-
nifi-nar-bundles/nifi-jetty-bundle/pom.xml | 10 ++
.../nifi-standard-processors/pom.xml | 13 ++
.../processors/standard/HandleHttpRequest.java | 99 +++--------
.../processors/standard/HandleHttpResponse.java | 1 -
.../nifi/processors/standard/ListenHTTP.java | 86 ++++-----
.../standard/http/HttpProtocolStrategy.java | 68 ++++++++
.../nifi/processors/standard/TestListenHTTP.java | 10 +-
.../org/apache/nifi/tests/system/NiFiSystemIT.java | 8 +
.../tests/system/TroubleshootingTestWatcher.java | 11 +-
.../tests/system/loadbalance/LoadBalanceIT.java | 54 ++++--
.../resources/conf/clustered/node1/nifi.properties | 2 +-
.../resources/conf/clustered/node2/nifi.properties | 2 +-
.../test/resources/conf/default/nifi.properties | 2 +-
pom.xml | 15 +-
24 files changed, 826 insertions(+), 179 deletions(-)
copy nifi-commons/{nifi-write-ahead-log => nifi-jetty-configuration}/pom.xml (77%)
copy nifi-commons/{nifi-utils/src/main/java/org/apache/nifi/stream/io/BufferedInputStream.java => nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/ApplicationLayerProtocol.java} (68%)
copy nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/main/java/org/apache/nifi/web/server/log/RequestLogProvider.java => nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/ServerConnectorFactory.java (74%)
create mode 100644 nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/StandardServerConnectorFactory.java
create mode 100644 nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/alpn/ALPNServerConnectionFactory.java
create mode 100644 nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/alpn/StandardALPNProcessor.java
create mode 100644 nifi-commons/nifi-jetty-configuration/src/test/java/org/apache/nifi/jetty/configuration/connector/StandardServerConnectorFactoryTest.java
create mode 100644 nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/http/HttpProtocolStrategy.java
[nifi] 04/05: NIFI-3869 Added HTTP/2 support to ListenHTTP and HandleHttpRequest
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a commit to branch support/nifi-1.16
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 0394270007fd8e0e66c36dd3ffb3cd31ad712670
Author: exceptionfactory <ex...@apache.org>
AuthorDate: Fri May 13 13:27:05 2022 -0500
NIFI-3869 Added HTTP/2 support to ListenHTTP and HandleHttpRequest
Signed-off-by: Nathan Gough <th...@gmail.com>
This closes #6048.
---
nifi-commons/nifi-jetty-configuration/pom.xml | 39 +++++
.../connector/ApplicationLayerProtocol.java | 36 ++++
.../connector/ServerConnectorFactory.java | 31 ++++
.../connector/StandardServerConnectorFactory.java | 193 +++++++++++++++++++++
.../alpn/ALPNServerConnectionFactory.java | 64 +++++++
.../connector/alpn/StandardALPNProcessor.java | 123 +++++++++++++
.../StandardServerConnectorFactoryTest.java | 172 ++++++++++++++++++
nifi-commons/pom.xml | 1 +
nifi-nar-bundles/nifi-jetty-bundle/pom.xml | 10 ++
.../nifi-standard-processors/pom.xml | 13 ++
.../processors/standard/HandleHttpRequest.java | 99 +++--------
.../processors/standard/HandleHttpResponse.java | 1 -
.../nifi/processors/standard/ListenHTTP.java | 86 ++++-----
.../standard/http/HttpProtocolStrategy.java | 68 ++++++++
.../nifi/processors/standard/TestListenHTTP.java | 10 +-
pom.xml | 13 ++
16 files changed, 835 insertions(+), 124 deletions(-)
diff --git a/nifi-commons/nifi-jetty-configuration/pom.xml b/nifi-commons/nifi-jetty-configuration/pom.xml
new file mode 100644
index 0000000000..8d8e2e7390
--- /dev/null
+++ b/nifi-commons/nifi-jetty-configuration/pom.xml
@@ -0,0 +1,39 @@
+<?xml version="1.0"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-commons</artifactId>
+ <version>1.17.0-SNAPSHOT</version>
+ </parent>
+ <artifactId>nifi-jetty-configuration</artifactId>
+ <packaging>jar</packaging>
+ <dependencies>
+ <dependency>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-server</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty.http2</groupId>
+ <artifactId>http2-server</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-alpn-server</artifactId>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/ApplicationLayerProtocol.java b/nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/ApplicationLayerProtocol.java
new file mode 100644
index 0000000000..9561d75471
--- /dev/null
+++ b/nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/ApplicationLayerProtocol.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.jetty.configuration.connector;
+
+/**
+ * Application Layer Protocols supported for Server Connectors
+ */
+public enum ApplicationLayerProtocol {
+ HTTP_1_1("http/1.1"),
+
+ H2("h2");
+
+ private String protocol;
+
+ ApplicationLayerProtocol(final String protocol) {
+ this.protocol = protocol;
+ }
+
+ public String getProtocol() {
+ return protocol;
+ }
+}
diff --git a/nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/ServerConnectorFactory.java b/nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/ServerConnectorFactory.java
new file mode 100644
index 0000000000..20440d2055
--- /dev/null
+++ b/nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/ServerConnectorFactory.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.jetty.configuration.connector;
+
+import org.eclipse.jetty.server.ServerConnector;
+
+/**
+ * Jetty Server Connector Factory
+ */
+public interface ServerConnectorFactory {
+ /**
+ * Get Server Connector
+ *
+ * @return Configured Server Connector
+ */
+ ServerConnector getServerConnector();
+}
diff --git a/nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/StandardServerConnectorFactory.java b/nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/StandardServerConnectorFactory.java
new file mode 100644
index 0000000000..e670ac71ea
--- /dev/null
+++ b/nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/StandardServerConnectorFactory.java
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.jetty.configuration.connector;
+
+import org.apache.nifi.jetty.configuration.connector.alpn.ALPNServerConnectionFactory;
+import org.eclipse.jetty.http.HttpScheme;
+import org.eclipse.jetty.http2.HTTP2Cipher;
+import org.eclipse.jetty.http2.server.HTTP2ServerConnectionFactory;
+import org.eclipse.jetty.server.ConnectionFactory;
+import org.eclipse.jetty.server.HttpConfiguration;
+import org.eclipse.jetty.server.HttpConnectionFactory;
+import org.eclipse.jetty.server.SecureRequestCustomizer;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.ServerConnector;
+import org.eclipse.jetty.server.SslConnectionFactory;
+import org.eclipse.jetty.util.ssl.SslContextFactory;
+
+import javax.net.ssl.SSLContext;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Standard implementation of Server Connector Factory supporting HTTP/2 and HTTP/1.1 with TLS or simple HTTP/1.1
+ */
+public class StandardServerConnectorFactory implements ServerConnectorFactory {
+ private static final boolean SEND_SERVER_VERSION = false;
+
+ private static final String[] INCLUDE_ALL_SECURITY_PROTOCOLS = new String[0];
+
+ private static final Set<ApplicationLayerProtocol> DEFAULT_APPLICATION_LAYER_PROTOCOLS = Collections.singleton(ApplicationLayerProtocol.HTTP_1_1);
+
+ private final Server server;
+
+ private final int port;
+
+ private Set<ApplicationLayerProtocol> applicationLayerProtocols = DEFAULT_APPLICATION_LAYER_PROTOCOLS;
+
+ private SSLContext sslContext;
+
+ private boolean needClientAuth;
+
+ private boolean wantClientAuth;
+
+ private String[] includeSecurityProtocols = INCLUDE_ALL_SECURITY_PROTOCOLS;
+
+ /**
+ * Standard Server Connector Factory Constructor with required properties
+ *
+ * @param server Jetty Server
+ * @param port Secure Port Number
+ */
+ public StandardServerConnectorFactory(
+ final Server server,
+ final int port
+ ) {
+ this.server = Objects.requireNonNull(server, "Server required");
+ this.port = port;
+ }
+
+ /**
+ * Get Server Connector configured with HTTP/2 and ALPN as well as fallback to HTTP/1.1 with TLS
+ *
+ * @return Secure Server Connector
+ */
+ @Override
+ public ServerConnector getServerConnector() {
+ final HttpConfiguration httpConfiguration = getHttpConfiguration();
+ final HttpConnectionFactory httpConnectionFactory = new HttpConnectionFactory(httpConfiguration);
+
+ final ServerConnector serverConnector;
+ if (sslContext == null) {
+ serverConnector = new ServerConnector(server, httpConnectionFactory);
+ } else {
+ final List<ConnectionFactory> connectionFactories = new ArrayList<>();
+ if (applicationLayerProtocols.contains(ApplicationLayerProtocol.H2)) {
+ final ALPNServerConnectionFactory alpnServerConnectionFactory = new ALPNServerConnectionFactory();
+ final HTTP2ServerConnectionFactory http2ServerConnectionFactory = new HTTP2ServerConnectionFactory(httpConfiguration);
+
+ connectionFactories.add(alpnServerConnectionFactory);
+ connectionFactories.add(http2ServerConnectionFactory);
+ }
+ // Add HTTP/1.1 Connection Factory after HTTP/2
+ if (applicationLayerProtocols.contains(ApplicationLayerProtocol.HTTP_1_1)) {
+ connectionFactories.add(httpConnectionFactory);
+ }
+
+ // SslConnectionFactory must be first and must indicate the next protocol
+ final String nextProtocol = connectionFactories.get(0).getProtocol();
+ final SslConnectionFactory sslConnectionFactory = new SslConnectionFactory(getSslContextFactory(), nextProtocol);
+ connectionFactories.add(0, sslConnectionFactory);
+
+ final ConnectionFactory[] factories = connectionFactories.toArray(new ConnectionFactory[0]);
+ serverConnector = new ServerConnector(server, factories);
+ }
+
+ serverConnector.setPort(port);
+ return serverConnector;
+ }
+
+ /**
+ * Set SSL Context enables TLS communication
+ *
+ * @param sslContext SSL Context
+ */
+ public void setSslContext(final SSLContext sslContext) {
+ this.sslContext = sslContext;
+ }
+
+ /**
+ * Set Need Client Authentication requires clients to provide certificates for mutual TLS
+ *
+ * @param needClientAuth Need Client Authentication status
+ */
+ public void setNeedClientAuth(final boolean needClientAuth) {
+ this.needClientAuth = needClientAuth;
+ }
+
+ /**
+ * Set Want Client Authentication requests clients to provide certificates for mutual TLS but does not require certificates
+ *
+ * @param wantClientAuth Want Client Authentication status
+ */
+ public void setWantClientAuth(final boolean wantClientAuth) {
+ this.wantClientAuth = wantClientAuth;
+ }
+
+ /**
+ * Set Include Security Protocols limits enabled TLS Protocols to the values provided
+ *
+ * @param includeSecurityProtocols Security Protocols with null or empty enabling all standard TLS protocol versions
+ */
+ public void setIncludeSecurityProtocols(final String[] includeSecurityProtocols) {
+ this.includeSecurityProtocols = includeSecurityProtocols == null ? INCLUDE_ALL_SECURITY_PROTOCOLS : includeSecurityProtocols;
+ }
+
+ /**
+ * Set Application Layer Protocols applicable when TLS is enabled
+ *
+ * @param applicationLayerProtocols Protocols requires at one Application Layer Protocol
+ */
+ public void setApplicationLayerProtocols(final Set<ApplicationLayerProtocol> applicationLayerProtocols) {
+ if (Objects.requireNonNull(applicationLayerProtocols, "Application Layer Protocols required").isEmpty()) {
+ throw new IllegalArgumentException("Application Layer Protocols not specified");
+ }
+ this.applicationLayerProtocols = applicationLayerProtocols;
+ }
+
+ private HttpConfiguration getHttpConfiguration() {
+ final HttpConfiguration httpConfiguration = new HttpConfiguration();
+
+ if (sslContext != null) {
+ httpConfiguration.setSecurePort(port);
+ httpConfiguration.setSecureScheme(HttpScheme.HTTPS.asString());
+ httpConfiguration.setSendServerVersion(SEND_SERVER_VERSION);
+
+ final SecureRequestCustomizer secureRequestCustomizer = new SecureRequestCustomizer();
+ httpConfiguration.addCustomizer(secureRequestCustomizer);
+ }
+
+ return httpConfiguration;
+ }
+
+ private SslContextFactory.Server getSslContextFactory() {
+ final SslContextFactory.Server sslContextFactory = new SslContextFactory.Server();
+ sslContextFactory.setSslContext(sslContext);
+ sslContextFactory.setNeedClientAuth(needClientAuth);
+ sslContextFactory.setWantClientAuth(wantClientAuth);
+ sslContextFactory.setIncludeProtocols(includeSecurityProtocols);
+
+ if (applicationLayerProtocols.contains(ApplicationLayerProtocol.H2)) {
+ sslContextFactory.setCipherComparator(HTTP2Cipher.COMPARATOR);
+ }
+
+ return sslContextFactory;
+ }
+}
diff --git a/nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/alpn/ALPNServerConnectionFactory.java b/nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/alpn/ALPNServerConnectionFactory.java
new file mode 100644
index 0000000000..c5da7a5091
--- /dev/null
+++ b/nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/alpn/ALPNServerConnectionFactory.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.jetty.configuration.connector.alpn;
+
+import org.eclipse.jetty.alpn.server.ALPNServerConnection;
+import org.eclipse.jetty.io.AbstractConnection;
+import org.eclipse.jetty.io.EndPoint;
+import org.eclipse.jetty.io.ssl.ALPNProcessor;
+import org.eclipse.jetty.server.Connector;
+import org.eclipse.jetty.server.NegotiatingServerConnectionFactory;
+
+import javax.net.ssl.SSLEngine;
+import java.util.List;
+
+/**
+ * ALPN Server Connection Factory with standard ALPN Processor implementation
+ */
+public class ALPNServerConnectionFactory extends NegotiatingServerConnectionFactory {
+ private static final String ALPN_PROTOCOL = "alpn";
+
+ private final ALPNProcessor.Server processor;
+
+ public ALPNServerConnectionFactory() {
+ super(ALPN_PROTOCOL);
+ processor = new StandardALPNProcessor();
+ }
+
+ /**
+ * Create new Server Connection and configure the connection using ALPN Processor
+ *
+ * @param connector Connector for the Connection
+ * @param endPoint End Point for the Connection
+ * @param sslEngine SSL Engine for the Connection
+ * @param protocols Application Protocols
+ * @param defaultProtocol Default Application Protocol
+ * @return ALPN Server Connection
+ */
+ @Override
+ protected AbstractConnection newServerConnection(
+ final Connector connector,
+ final EndPoint endPoint,
+ final SSLEngine sslEngine,
+ final List<String> protocols,
+ final String defaultProtocol
+ ) {
+ final ALPNServerConnection connection = new ALPNServerConnection(connector, endPoint, sslEngine, protocols, defaultProtocol);
+ processor.configure(sslEngine, connection);
+ return connection;
+ }
+}
diff --git a/nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/alpn/StandardALPNProcessor.java b/nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/alpn/StandardALPNProcessor.java
new file mode 100644
index 0000000000..0c8825226d
--- /dev/null
+++ b/nifi-commons/nifi-jetty-configuration/src/main/java/org/apache/nifi/jetty/configuration/connector/alpn/StandardALPNProcessor.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.jetty.configuration.connector.alpn;
+
+import org.eclipse.jetty.alpn.server.ALPNServerConnection;
+import org.eclipse.jetty.io.Connection;
+import org.eclipse.jetty.io.ssl.ALPNProcessor;
+import org.eclipse.jetty.io.ssl.SslConnection;
+import org.eclipse.jetty.io.ssl.SslHandshakeListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLSession;
+import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.function.BiFunction;
+
+/**
+ * Standard ALPN Processor supporting JDK 1.8.0-251 and higher based on Jetty JDK9ServerALPNProcessor
+ */
+public class StandardALPNProcessor implements ALPNProcessor.Server, SslHandshakeListener {
+ private static final Logger logger = LoggerFactory.getLogger(StandardALPNProcessor.class);
+
+ /**
+ * Applies to SSL Engine instances regardless of implementation
+ *
+ * @param sslEngine SSL Engine to be evaluated
+ * @return Applicable Status
+ */
+ @Override
+ public boolean appliesTo(final SSLEngine sslEngine) {
+ return true;
+ }
+
+ /**
+ * Configure ALPN negotiation for Connection
+ *
+ * @param sslEngine SSL Engine to be configured
+ * @param connection Connection to be configured
+ */
+ @Override
+ public void configure(final SSLEngine sslEngine, final Connection connection) {
+ logger.debug("Configuring Connection Remote Address [{}]", connection.getEndPoint().getRemoteAddress());
+ final ALPNServerConnection serverConnection = (ALPNServerConnection) connection;
+ final ProtocolSelector protocolSelector = new ProtocolSelector(serverConnection);
+ sslEngine.setHandshakeApplicationProtocolSelector(protocolSelector);
+
+ final SslConnection.DecryptedEndPoint endPoint = (SslConnection.DecryptedEndPoint) serverConnection.getEndPoint();
+ endPoint.getSslConnection().addHandshakeListener(protocolSelector);
+ }
+
+ private static final class ProtocolSelector implements BiFunction<SSLEngine, List<String>, String>, SslHandshakeListener {
+ private final ALPNServerConnection serverConnection;
+
+ private ProtocolSelector(final ALPNServerConnection connection) {
+ serverConnection = connection;
+ }
+
+ /**
+ * Select supported Application Layer Protocol based on requested protocols
+ *
+ * @param sslEngine SSL Engine
+ * @param protocols Protocols requested
+ * @return Protocol selected or null when no supported protocol found
+ */
+ @Override
+ public String apply(final SSLEngine sslEngine, final List<String> protocols) {
+ String protocol = null;
+ try {
+ serverConnection.select(protocols);
+ protocol = serverConnection.getProtocol();
+ logger.debug("Connection Remote Address [{}] Application Layer Protocol [{}] selected", serverConnection.getEndPoint().getRemoteAddress(), protocol);
+ } catch (final Throwable e) {
+ logger.debug("Connection Remote Address [{}] Application Layer Protocols {} not supported", serverConnection.getEndPoint().getRemoteAddress(), protocols);
+ }
+ return protocol;
+ }
+
+ /**
+ * Handler for successful handshake checks for selected Application Layer Protocol
+ *
+ * @param event Event is not used
+ */
+ @Override
+ public void handshakeSucceeded(final Event event) {
+ final InetSocketAddress remoteAddress = serverConnection.getEndPoint().getRemoteAddress();
+ final SSLSession session = event.getSSLEngine().getSession();
+ logger.debug("Connection Remote Address [{}] Handshake Succeeded [{}] Cipher Suite [{}]", remoteAddress, session.getProtocol(), session.getCipherSuite());
+
+ final String protocol = serverConnection.getProtocol();
+ if (protocol == null) {
+ logger.debug("Connection Remote Address [{}] Application Layer Protocol not supported", remoteAddress);
+ serverConnection.unsupported();
+ }
+ }
+
+ /**
+ * Handle for failed handshake logs status
+ *
+ * @param event Event is not used
+ * @param failure Failure cause to be logged
+ */
+ @Override
+ public void handshakeFailed(final Event event, final Throwable failure) {
+ logger.warn("Connection Remote Address [{}] Handshake Failed", serverConnection.getEndPoint().getRemoteAddress(), failure);
+ }
+ }
+}
diff --git a/nifi-commons/nifi-jetty-configuration/src/test/java/org/apache/nifi/jetty/configuration/connector/StandardServerConnectorFactoryTest.java b/nifi-commons/nifi-jetty-configuration/src/test/java/org/apache/nifi/jetty/configuration/connector/StandardServerConnectorFactoryTest.java
new file mode 100644
index 0000000000..d465504976
--- /dev/null
+++ b/nifi-commons/nifi-jetty-configuration/src/test/java/org/apache/nifi/jetty/configuration/connector/StandardServerConnectorFactoryTest.java
@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.jetty.configuration.connector;
+
+import org.apache.nifi.jetty.configuration.connector.alpn.ALPNServerConnectionFactory;
+import org.eclipse.jetty.http.HttpScheme;
+import org.eclipse.jetty.http2.server.HTTP2ServerConnectionFactory;
+import org.eclipse.jetty.server.HttpConfiguration;
+import org.eclipse.jetty.server.HttpConnectionFactory;
+import org.eclipse.jetty.server.SecureRequestCustomizer;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.ServerConnector;
+import org.eclipse.jetty.server.SslConnectionFactory;
+import org.eclipse.jetty.util.ssl.SslContextFactory;
+import org.junit.jupiter.api.Test;
+
+import javax.net.ssl.SSLContext;
+
+import java.security.NoSuchAlgorithmException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class StandardServerConnectorFactoryTest {
+ private static final int HTTP_PORT = 8080;
+
+ private static final int HTTPS_PORT = 8443;
+
+ private static final String[] INCLUDE_PROTOCOLS = new String[]{ "TLSv1.2" };
+
+ @Test
+ void testGetServerConnector() {
+ final Server server = new Server();
+ final StandardServerConnectorFactory factory = new StandardServerConnectorFactory(server, HTTP_PORT);
+
+ final ServerConnector serverConnector = factory.getServerConnector();
+
+ assertHttpConnectionFactoryFound(serverConnector);
+ }
+
+ @Test
+ void testGetServerConnectorSecured() throws NoSuchAlgorithmException {
+ final StandardServerConnectorFactory factory = createSecuredStandardServerConnectorFactory();
+
+ final ServerConnector serverConnector = factory.getServerConnector();
+
+ assertHttpConnectionFactoryFound(serverConnector);
+ final SslConnectionFactory sslConnectionFactory = assertSslConnectionFactoryFound(serverConnector);
+
+ final HttpConnectionFactory httpConnectionFactory = assertHttpConnectionFactoryFound(serverConnector);
+ assertHttpConnectionFactorySecured(httpConnectionFactory);
+
+ final SslContextFactory.Server sslContextFactory = (SslContextFactory.Server) sslConnectionFactory.getSslContextFactory();
+ assertFalse(sslContextFactory.getNeedClientAuth());
+ assertFalse(sslContextFactory.getWantClientAuth());
+ assertNotNull(sslContextFactory.getIncludeProtocols());
+
+ final HTTP2ServerConnectionFactory http2ConnectionFactory = serverConnector.getConnectionFactory(HTTP2ServerConnectionFactory.class);
+ assertNull(http2ConnectionFactory);
+ }
+
+ @Test
+ void testGetServerConnectorSecuredNeedClientAuthentication() throws NoSuchAlgorithmException {
+ final StandardServerConnectorFactory factory = createSecuredStandardServerConnectorFactory();
+ factory.setNeedClientAuth(true);
+ factory.setIncludeSecurityProtocols(INCLUDE_PROTOCOLS);
+
+ final ServerConnector serverConnector = factory.getServerConnector();
+
+ assertHttpConnectionFactoryFound(serverConnector);
+ final SslConnectionFactory sslConnectionFactory = assertSslConnectionFactoryFound(serverConnector);
+
+ final HttpConnectionFactory httpConnectionFactory = assertHttpConnectionFactoryFound(serverConnector);
+ assertHttpConnectionFactorySecured(httpConnectionFactory);
+
+ final SslContextFactory.Server sslContextFactory = (SslContextFactory.Server) sslConnectionFactory.getSslContextFactory();
+ assertTrue(sslContextFactory.getNeedClientAuth());
+ assertArrayEquals(INCLUDE_PROTOCOLS, sslContextFactory.getIncludeProtocols());
+ }
+
+ @Test
+ void testGetServerConnectorSecuredHttp2AndHttp1() throws NoSuchAlgorithmException {
+ final StandardServerConnectorFactory factory = createSecuredStandardServerConnectorFactory();
+ factory.setApplicationLayerProtocols(new LinkedHashSet<>(Arrays.asList(ApplicationLayerProtocol.H2, ApplicationLayerProtocol.HTTP_1_1)));
+
+ final ServerConnector serverConnector = factory.getServerConnector();
+
+ final HttpConnectionFactory httpConnectionFactory = assertHttpConnectionFactoryFound(serverConnector);
+ assertHttpConnectionFactorySecured(httpConnectionFactory);
+
+ final SslConnectionFactory sslConnectionFactory = assertSslConnectionFactoryFound(serverConnector);
+ final SslContextFactory.Server sslContextFactory = (SslContextFactory.Server) sslConnectionFactory.getSslContextFactory();
+ assertFalse(sslContextFactory.getNeedClientAuth());
+
+ assertHttp2ConnectionFactoriesFound(serverConnector);
+ }
+
+ @Test
+ void testGetServerConnectorSecuredHttp2() throws NoSuchAlgorithmException {
+ final StandardServerConnectorFactory factory = createSecuredStandardServerConnectorFactory();
+ factory.setApplicationLayerProtocols(Collections.singleton(ApplicationLayerProtocol.H2));
+
+ final ServerConnector serverConnector = factory.getServerConnector();
+
+ final HttpConnectionFactory connectionFactory = serverConnector.getConnectionFactory(HttpConnectionFactory.class);
+ assertNull(connectionFactory);
+
+ final SslConnectionFactory sslConnectionFactory = assertSslConnectionFactoryFound(serverConnector);
+ final SslContextFactory.Server sslContextFactory = (SslContextFactory.Server) sslConnectionFactory.getSslContextFactory();
+ assertFalse(sslContextFactory.getNeedClientAuth());
+
+ assertHttp2ConnectionFactoriesFound(serverConnector);
+ }
+
+ private StandardServerConnectorFactory createSecuredStandardServerConnectorFactory() throws NoSuchAlgorithmException {
+ final Server server = new Server();
+ final StandardServerConnectorFactory factory = new StandardServerConnectorFactory(server, HTTPS_PORT);
+ final SSLContext sslContext = SSLContext.getDefault();
+ factory.setSslContext(sslContext);
+ return factory;
+ }
+
+ private HttpConnectionFactory assertHttpConnectionFactoryFound(final ServerConnector serverConnector) {
+ assertNotNull(serverConnector);
+ final HttpConnectionFactory connectionFactory = serverConnector.getConnectionFactory(HttpConnectionFactory.class);
+ assertNotNull(connectionFactory);
+ return connectionFactory;
+ }
+
+ private void assertHttp2ConnectionFactoriesFound(final ServerConnector serverConnector) {
+ final HTTP2ServerConnectionFactory http2ConnectionFactory = serverConnector.getConnectionFactory(HTTP2ServerConnectionFactory.class);
+ assertNotNull(http2ConnectionFactory);
+
+ final ALPNServerConnectionFactory alpnServerConnectionFactory = serverConnector.getConnectionFactory(ALPNServerConnectionFactory.class);
+ assertNotNull(alpnServerConnectionFactory);
+ }
+
+ private SslConnectionFactory assertSslConnectionFactoryFound(final ServerConnector serverConnector) {
+ final SslConnectionFactory sslConnectionFactory = serverConnector.getConnectionFactory(SslConnectionFactory.class);
+ assertNotNull(sslConnectionFactory);
+ return sslConnectionFactory;
+ }
+
+ private void assertHttpConnectionFactorySecured(final HttpConnectionFactory httpConnectionFactory) {
+ final HttpConfiguration configuration = httpConnectionFactory.getHttpConfiguration();
+ assertEquals(HTTPS_PORT, configuration.getSecurePort());
+ assertEquals(HttpScheme.HTTPS.asString(), configuration.getSecureScheme());
+ final SecureRequestCustomizer secureRequestCustomizer = configuration.getCustomizer(SecureRequestCustomizer.class);
+ assertNotNull(secureRequestCustomizer);
+ }
+}
diff --git a/nifi-commons/pom.xml b/nifi-commons/pom.xml
index cc161ce9a1..02784b7671 100644
--- a/nifi-commons/pom.xml
+++ b/nifi-commons/pom.xml
@@ -32,6 +32,7 @@
<module>nifi-flow-encryptor</module>
<module>nifi-hl7-query-language</module>
<module>nifi-json-utils</module>
+ <module>nifi-jetty-configuration</module>
<module>nifi-logging-utils</module>
<module>nifi-metrics</module>
<module>nifi-parameter</module>
diff --git a/nifi-nar-bundles/nifi-jetty-bundle/pom.xml b/nifi-nar-bundles/nifi-jetty-bundle/pom.xml
index c3a2df3542..76523de05a 100644
--- a/nifi-nar-bundles/nifi-jetty-bundle/pom.xml
+++ b/nifi-nar-bundles/nifi-jetty-bundle/pom.xml
@@ -78,6 +78,16 @@
<artifactId>apache-jstl</artifactId>
<scope>compile</scope>
</dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-alpn-server</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty.http2</groupId>
+ <artifactId>http2-server</artifactId>
+ <scope>compile</scope>
+ </dependency>
</dependencies>
<scm>
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
index 2793581458..b0b840c392 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
@@ -92,6 +92,11 @@
<artifactId>nifi-flowfile-packager</artifactId>
<version>1.16.2-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-jetty-configuration</artifactId>
+ <version>1.17.0-SNAPSHOT</version>
+ </dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-distributed-cache-client-service-api</artifactId>
@@ -180,6 +185,14 @@
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-servlet</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-alpn-server</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty.http2</groupId>
+ <artifactId>http2-server</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
index 96af6c2c7b..7598e39845 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
@@ -33,6 +33,7 @@ import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.http.HttpContextMap;
+import org.apache.nifi.jetty.configuration.connector.StandardServerConnectorFactory;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.processor.ProcessContext;
@@ -40,21 +41,17 @@ import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.http.HttpProtocolStrategy;
import org.apache.nifi.processors.standard.util.HTTPUtils;
import org.apache.nifi.scheduling.ExecutionNode;
import org.apache.nifi.ssl.RestrictedSSLContextService;
import org.apache.nifi.ssl.SSLContextService;
import org.apache.nifi.stream.io.StreamUtils;
import org.eclipse.jetty.server.Connector;
-import org.eclipse.jetty.server.HttpConfiguration;
-import org.eclipse.jetty.server.HttpConnectionFactory;
import org.eclipse.jetty.server.Request;
-import org.eclipse.jetty.server.SecureRequestCustomizer;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
-import org.eclipse.jetty.server.SslConnectionFactory;
import org.eclipse.jetty.server.handler.AbstractHandler;
-import org.eclipse.jetty.util.ssl.SslContextFactory;
import javax.net.ssl.SSLContext;
import javax.servlet.AsyncContext;
@@ -187,6 +184,14 @@ public class HandleHttpRequest extends AbstractProcessor {
.required(false)
.identifiesControllerService(RestrictedSSLContextService.class)
.build();
+ public static final PropertyDescriptor HTTP_PROTOCOL_STRATEGY = new PropertyDescriptor.Builder()
+ .name("HTTP Protocols")
+ .description("HTTP Protocols supported for Application Layer Protocol Negotiation with TLS")
+ .required(true)
+ .allowableValues(HttpProtocolStrategy.class)
+ .defaultValue(HttpProtocolStrategy.HTTP_1_1.getValue())
+ .dependsOn(SSL_CONTEXT)
+ .build();
public static final PropertyDescriptor URL_CHARACTER_SET = new PropertyDescriptor.Builder()
.name("Default URL Character Set")
.description("The character set to use for decoding URL parameters if the HTTP Request does not supply one")
@@ -303,6 +308,7 @@ public class HandleHttpRequest extends AbstractProcessor {
descriptors.add(PORT);
descriptors.add(HOSTNAME);
descriptors.add(SSL_CONTEXT);
+ descriptors.add(HTTP_PROTOCOL_STRATEGY);
descriptors.add(HTTP_CONTEXT_MAP);
descriptors.add(PATH_REGEX);
descriptors.add(URL_CHARACTER_SET);
@@ -356,61 +362,24 @@ public class HandleHttpRequest extends AbstractProcessor {
final long requestTimeout = httpContextMap.getRequestTimeout(TimeUnit.MILLISECONDS);
final String clientAuthValue = context.getProperty(CLIENT_AUTH).getValue();
- final boolean need;
- final boolean want;
- if (CLIENT_NEED.getValue().equals(clientAuthValue)) {
- need = true;
- want = false;
- } else if (CLIENT_WANT.getValue().equals(clientAuthValue)) {
- need = false;
- want = true;
- } else {
- need = false;
- want = false;
- }
-
- final SslContextFactory sslFactory = (sslService == null) ? null : createSslFactory(sslService, need, want);
- final Server server = new Server(port);
-
- // create the http configuration
- final HttpConfiguration httpConfiguration = new HttpConfiguration();
- if (sslFactory == null) {
- // create the connector
- final ServerConnector http = new ServerConnector(server, new HttpConnectionFactory(httpConfiguration));
-
- // set host and port
- if (StringUtils.isNotBlank(host)) {
- http.setHost(host);
- }
- http.setPort(port);
-
- // If request timeout is longer than default Idle Timeout, then increase Idle Timeout as well.
- http.setIdleTimeout(Math.max(http.getIdleTimeout(), requestTimeout));
-
- // add this connector
- server.setConnectors(new Connector[]{http});
- } else {
- // add some secure config
- final HttpConfiguration httpsConfiguration = new HttpConfiguration(httpConfiguration);
- httpsConfiguration.setSecureScheme("https");
- httpsConfiguration.setSecurePort(port);
- httpsConfiguration.addCustomizer(new SecureRequestCustomizer());
-
- // build the connector
- final ServerConnector https = new ServerConnector(server, new SslConnectionFactory(sslFactory, "http/1.1"), new HttpConnectionFactory(httpsConfiguration));
-
- // set host and port
- if (StringUtils.isNotBlank(host)) {
- https.setHost(host);
- }
- https.setPort(port);
-
- // If request timeout is longer than default Idle Timeout, then increase Idle Timeout as well.
- https.setIdleTimeout(Math.max(https.getIdleTimeout(), requestTimeout));
-
- // add this connector
- server.setConnectors(new Connector[]{https});
+ final Server server = new Server();
+
+ final StandardServerConnectorFactory serverConnectorFactory = new StandardServerConnectorFactory(server, port);
+ final boolean needClientAuth = CLIENT_NEED.getValue().equals(clientAuthValue);
+ serverConnectorFactory.setNeedClientAuth(needClientAuth);
+ final boolean wantClientAuth = CLIENT_WANT.getValue().equals(clientAuthValue);
+ serverConnectorFactory.setNeedClientAuth(wantClientAuth);
+ final SSLContext sslContext = sslService == null ? null : sslService.createContext();
+ serverConnectorFactory.setSslContext(sslContext);
+ final HttpProtocolStrategy httpProtocolStrategy = HttpProtocolStrategy.valueOf(context.getProperty(HTTP_PROTOCOL_STRATEGY).getValue());
+ serverConnectorFactory.setApplicationLayerProtocols(httpProtocolStrategy.getApplicationLayerProtocols());
+
+ final ServerConnector serverConnector = serverConnectorFactory.getServerConnector();
+ serverConnector.setIdleTimeout(Math.max(serverConnector.getIdleTimeout(), requestTimeout));
+ if (StringUtils.isNotBlank(host)) {
+ serverConnector.setHost(host);
}
+ server.addConnector(serverConnector);
final Set<String> allowedMethods = new HashSet<>();
if (context.getProperty(ALLOW_GET).asBoolean()) {
@@ -522,18 +491,6 @@ public class HandleHttpRequest extends AbstractProcessor {
return containerQueue.size();
}
- private SslContextFactory createSslFactory(final SSLContextService sslContextService, final boolean needClientAuth, final boolean wantClientAuth) {
- final SslContextFactory.Server sslFactory = new SslContextFactory.Server();
-
- sslFactory.setNeedClientAuth(needClientAuth);
- sslFactory.setWantClientAuth(wantClientAuth);
-
- final SSLContext sslContext = sslContextService.createContext();
- sslFactory.setSslContext(sslContext);
-
- return sslFactory;
- }
-
@OnUnscheduled
public void shutdown() throws Exception {
ready = false;
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpResponse.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpResponse.java
index 96aa93a803..eff8e4fbf0 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpResponse.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpResponse.java
@@ -188,7 +188,6 @@ public class HandleHttpResponse extends AbstractProcessor {
try {
session.exportTo(flowFile, response.getOutputStream());
- response.flushBuffer();
} catch (final ProcessException e) {
getLogger().error("Failed to respond to HTTP request for {} due to {}", new Object[]{flowFile, e});
try {
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenHTTP.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenHTTP.java
index e5f98daeb0..a35b9edc73 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenHTTP.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenHTTP.java
@@ -31,6 +31,7 @@ import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.jetty.configuration.connector.StandardServerConnectorFactory;
import org.apache.nifi.processor.AbstractSessionFactoryProcessor;
import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.processor.ProcessContext;
@@ -39,26 +40,21 @@ import org.apache.nifi.processor.ProcessSessionFactory;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.http.HttpProtocolStrategy;
import org.apache.nifi.processors.standard.servlets.ContentAcknowledgmentServlet;
import org.apache.nifi.processors.standard.servlets.HealthCheckServlet;
import org.apache.nifi.processors.standard.servlets.ListenHTTPServlet;
import org.apache.nifi.scheduling.ExecutionNode;
import org.apache.nifi.security.util.ClientAuth;
-import org.apache.nifi.security.util.TlsConfiguration;
import org.apache.nifi.serialization.RecordReaderFactory;
import org.apache.nifi.serialization.RecordSetWriterFactory;
import org.apache.nifi.ssl.RestrictedSSLContextService;
import org.apache.nifi.ssl.SSLContextService;
import org.apache.nifi.stream.io.LeakyBucketStreamThrottler;
import org.apache.nifi.stream.io.StreamThrottler;
-import org.eclipse.jetty.server.HttpConfiguration;
-import org.eclipse.jetty.server.HttpConnectionFactory;
-import org.eclipse.jetty.server.SecureRequestCustomizer;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
-import org.eclipse.jetty.server.SslConnectionFactory;
import org.eclipse.jetty.servlet.ServletContextHandler;
-import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import javax.net.ssl.SSLContext;
@@ -191,10 +187,18 @@ public class ListenHTTP extends AbstractSessionFactoryProcessor {
.build();
public static final PropertyDescriptor SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder()
.name("SSL Context Service")
- .description("The Controller Service to use in order to obtain an SSL Context")
+ .description("SSL Context Service enables support for HTTPS")
.required(false)
.identifiesControllerService(RestrictedSSLContextService.class)
.build();
+ public static final PropertyDescriptor HTTP_PROTOCOL_STRATEGY = new PropertyDescriptor.Builder()
+ .name("HTTP Protocols")
+ .description("HTTP Protocols supported for Application Layer Protocol Negotiation with TLS")
+ .required(true)
+ .allowableValues(HttpProtocolStrategy.class)
+ .defaultValue(HttpProtocolStrategy.HTTP_1_1.getValue())
+ .dependsOn(SSL_CONTEXT_SERVICE)
+ .build();
public static final PropertyDescriptor HEADERS_AS_ATTRIBUTES_REGEX = new PropertyDescriptor.Builder()
.name("HTTP Headers to receive as Attributes (Regex)")
.description("Specifies the Regular Expression that determines the names of HTTP Headers that should be passed along as FlowFile attributes")
@@ -276,6 +280,7 @@ public class ListenHTTP extends AbstractSessionFactoryProcessor {
HEALTH_CHECK_PORT,
MAX_DATA_RATE,
SSL_CONTEXT_SERVICE,
+ HTTP_PROTOCOL_STRATEGY,
CLIENT_AUTHENTICATION,
AUTHORIZED_DN_PATTERN,
AUTHORIZED_ISSUER_DN_PATTERN,
@@ -396,7 +401,6 @@ public class ListenHTTP extends AbstractSessionFactoryProcessor {
int maxThreadPoolSize = context.getProperty(MAX_THREAD_POOL_SIZE).asInteger();
throttlerRef.set(streamThrottler);
- final boolean sslRequired = sslContextService != null;
final PropertyValue clientAuthenticationProperty = context.getProperty(CLIENT_AUTHENTICATION);
final ClientAuthentication clientAuthentication = getClientAuthentication(sslContextService, clientAuthenticationProperty);
@@ -409,12 +413,13 @@ public class ListenHTTP extends AbstractSessionFactoryProcessor {
// get the configured port
final int port = context.getProperty(PORT).evaluateAttributeExpressions().asInteger();
-
+ final HttpProtocolStrategy httpProtocolStrategy = HttpProtocolStrategy.valueOf(context.getProperty(HTTP_PROTOCOL_STRATEGY).getValue());
final ServerConnector connector = createServerConnector(server,
port,
sslContextService,
- sslRequired,
- clientAuthentication);
+ clientAuthentication,
+ httpProtocolStrategy
+ );
server.addConnector(connector);
// Add a separate connector for the health check port (if specified)
@@ -423,12 +428,14 @@ public class ListenHTTP extends AbstractSessionFactoryProcessor {
final ServerConnector healthCheckConnector = createServerConnector(server,
healthCheckPort,
sslContextService,
- sslRequired,
- ClientAuthentication.NONE);
+ ClientAuthentication.NONE,
+ httpProtocolStrategy
+ );
server.addConnector(healthCheckConnector);
}
- final ServletContextHandler contextHandler = new ServletContextHandler(server, "/", true, sslRequired);
+ final boolean securityEnabled = sslContextService != null;
+ final ServletContextHandler contextHandler = new ServletContextHandler(server, "/", true, securityEnabled);
for (final Class<? extends Servlet> cls : getServerClasses()) {
final Path path = cls.getAnnotation(Path.class);
// Note: servlets must have a path annotation - this will NPE otherwise
@@ -488,41 +495,24 @@ public class ListenHTTP extends AbstractSessionFactoryProcessor {
private ServerConnector createServerConnector(final Server server,
final int port,
final SSLContextService sslContextService,
- final boolean sslRequired,
- final ClientAuthentication clientAuthentication) {
- final ServerConnector connector;
- final HttpConfiguration httpConfiguration = new HttpConfiguration();
- if (sslRequired) {
- httpConfiguration.setSecureScheme("https");
- httpConfiguration.setSecurePort(port);
- httpConfiguration.addCustomizer(new SecureRequestCustomizer());
-
- final SslContextFactory contextFactory = createSslContextFactory(sslContextService, clientAuthentication);
-
- connector = new ServerConnector(server, new SslConnectionFactory(contextFactory, "http/1.1"), new HttpConnectionFactory(httpConfiguration));
- } else {
- connector = new ServerConnector(server, new HttpConnectionFactory(httpConfiguration));
- }
-
- connector.setPort(port);
- return connector;
- }
-
- private SslContextFactory createSslContextFactory(final SSLContextService sslContextService, final ClientAuthentication clientAuthentication) {
- final SslContextFactory.Server contextFactory = new SslContextFactory.Server();
- final SSLContext sslContext = sslContextService.createContext();
- contextFactory.setSslContext(sslContext);
-
- final TlsConfiguration tlsConfiguration = sslContextService.createTlsConfiguration();
- contextFactory.setIncludeProtocols(tlsConfiguration.getEnabledProtocols());
-
- if (ClientAuthentication.REQUIRED.equals(clientAuthentication)) {
- contextFactory.setNeedClientAuth(true);
- } else if (ClientAuthentication.WANT.equals(clientAuthentication)) {
- contextFactory.setWantClientAuth(true);
+ final ClientAuthentication clientAuthentication,
+ final HttpProtocolStrategy httpProtocolStrategy
+ ) {
+ final StandardServerConnectorFactory serverConnectorFactory = new StandardServerConnectorFactory(server, port);
+ final SSLContext sslContext = sslContextService == null ? null : sslContextService.createContext();
+ serverConnectorFactory.setSslContext(sslContext);
+
+ final String[] enabledProtocols = sslContextService == null ? new String[0] : sslContextService.createTlsConfiguration().getEnabledProtocols();
+ serverConnectorFactory.setIncludeSecurityProtocols(enabledProtocols);
+
+ if (ClientAuthentication.REQUIRED == clientAuthentication) {
+ serverConnectorFactory.setNeedClientAuth(true);
+ } else if (ClientAuthentication.WANT == clientAuthentication) {
+ serverConnectorFactory.setWantClientAuth(true);
}
- return contextFactory;
+ serverConnectorFactory.setApplicationLayerProtocols(httpProtocolStrategy.getApplicationLayerProtocols());
+ return serverConnectorFactory.getServerConnector();
}
@OnScheduled
@@ -572,7 +562,7 @@ public class ListenHTTP extends AbstractSessionFactoryProcessor {
for (final String id : findOldFlowFileIds(context)) {
final FlowFileEntryTimeWrapper wrapper = flowFileMap.remove(id);
if (wrapper != null) {
- getLogger().warn("failed to received acknowledgment for HOLD with ID {} sent by {}; rolling back session", new Object[] {id, wrapper.getClientIP()});
+ getLogger().warn("failed to received acknowledgment for HOLD with ID {} sent by {}; rolling back session", id, wrapper.getClientIP());
wrapper.session.rollback();
}
}
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/http/HttpProtocolStrategy.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/http/HttpProtocolStrategy.java
new file mode 100644
index 0000000000..d99d4182ef
--- /dev/null
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/http/HttpProtocolStrategy.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.standard.http;
+
+import org.apache.nifi.components.DescribedValue;
+import org.apache.nifi.jetty.configuration.connector.ApplicationLayerProtocol;
+
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+import static java.util.Arrays.asList;
+import static java.util.Collections.singleton;
+
+/**
+ * HTTP protocol configuration strategy
+ */
+public enum HttpProtocolStrategy implements DescribedValue {
+ HTTP_1_1("http/1.1", "HTTP/1.1", singleton(ApplicationLayerProtocol.HTTP_1_1)),
+
+ H2_HTTP_1_1("h2 http/1.1", "HTTP/2 and HTTP/1.1 negotiated based on requested protocols", new LinkedHashSet<>(asList(ApplicationLayerProtocol.HTTP_1_1, ApplicationLayerProtocol.H2))),
+
+ H2("h2", "HTTP/2", singleton(ApplicationLayerProtocol.H2));
+
+ private final String displayName;
+
+ private final String description;
+
+ private final Set<ApplicationLayerProtocol> applicationLayerProtocols;
+
+ HttpProtocolStrategy(final String displayName, final String description, final Set<ApplicationLayerProtocol> applicationLayerProtocols) {
+ this.displayName = displayName;
+ this.description = description;
+ this.applicationLayerProtocols = applicationLayerProtocols;
+ }
+
+ @Override
+ public String getValue() {
+ return name();
+ }
+
+ @Override
+ public String getDisplayName() {
+ return displayName;
+ }
+
+ @Override
+ public String getDescription() {
+ return description;
+ }
+
+ public Set<ApplicationLayerProtocol> getApplicationLayerProtocols() {
+ return applicationLayerProtocols;
+ }
+}
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenHTTP.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenHTTP.java
index cc006c333d..489a4d3382 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenHTTP.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenHTTP.java
@@ -47,6 +47,7 @@ import okhttp3.RequestBody;
import okhttp3.Response;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processors.standard.http.HttpProtocolStrategy;
import org.apache.nifi.remote.io.socket.NetworkUtils;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.security.util.SslContextFactory;
@@ -99,7 +100,6 @@ public class TestListenHTTP {
private static final Duration CLIENT_CALL_TIMEOUT = Duration.ofSeconds(10);
public static final String LOCALHOST_DN = "CN=localhost";
- private static TlsConfiguration tlsConfiguration;
private static TlsConfiguration serverConfiguration;
private static TlsConfiguration serverTls_1_3_Configuration;
private static TlsConfiguration serverNoTruststoreConfiguration;
@@ -117,7 +117,7 @@ public class TestListenHTTP {
@BeforeClass
public static void setUpSuite() throws GeneralSecurityException {
// generate new keystore and truststore
- tlsConfiguration = new TemporaryKeyStoreBuilder().build();
+ final TlsConfiguration tlsConfiguration = new TemporaryKeyStoreBuilder().build();
serverConfiguration = new StandardTlsConfiguration(
tlsConfiguration.getKeystorePath(),
@@ -223,23 +223,25 @@ public class TestListenHTTP {
}
@Test
- public void testSecurePOSTRequestsReceivedWithoutEL() throws Exception {
+ public void testSecurePOSTRequestsReceivedWithoutELHttp2AndHttp1() throws Exception {
configureProcessorSslContextService(ListenHTTP.ClientAuthentication.AUTO, serverNoTruststoreConfiguration);
runner.setProperty(ListenHTTP.PORT, Integer.toString(availablePort));
runner.setProperty(ListenHTTP.BASE_PATH, HTTP_BASE_PATH);
+ runner.setProperty(ListenHTTP.HTTP_PROTOCOL_STRATEGY, HttpProtocolStrategy.H2_HTTP_1_1.getValue());
runner.assertValid();
testPOSTRequestsReceived(HttpServletResponse.SC_OK, true, false);
}
@Test
- public void testSecurePOSTRequestsReturnCodeReceivedWithoutEL() throws Exception {
+ public void testSecurePOSTRequestsReturnCodeReceivedWithoutELHttp2() throws Exception {
configureProcessorSslContextService(ListenHTTP.ClientAuthentication.AUTO, serverNoTruststoreConfiguration);
runner.setProperty(ListenHTTP.PORT, Integer.toString(availablePort));
runner.setProperty(ListenHTTP.BASE_PATH, HTTP_BASE_PATH);
runner.setProperty(ListenHTTP.RETURN_CODE, Integer.toString(HttpServletResponse.SC_NO_CONTENT));
+ runner.setProperty(ListenHTTP.HTTP_PROTOCOL_STRATEGY, HttpProtocolStrategy.H2.getValue());
runner.assertValid();
testPOSTRequestsReceived(HttpServletResponse.SC_NO_CONTENT, true, false);
diff --git a/pom.xml b/pom.xml
index 26318b1566..682beac129 100644
--- a/pom.xml
+++ b/pom.xml
@@ -403,6 +403,19 @@
<version>${jetty.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-alpn-server</artifactId>
+ <version>${jetty.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty.http2</groupId>
+ <artifactId>http2-server</artifactId>
+ <version>${jetty.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-alpn-client</artifactId>
[nifi] 03/05: NIFI-10037: When system test fails to clean up flow, destroy the entire environment so that the next test starts in a healthy state. Name troubleshooting directories with the name of the test class to avoid ambiguity. Also added a log statement so that we know which test is running when looking at the log output from the tests themselves. Finally, found an issue in AbstractComponentNode in which we iterate over the elements in a Map and call setProperty, which can update the underlying Map [...]
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a commit to branch support/nifi-1.16
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 7268e638dd1f68ccdc21fc2f940991f4e407cd21
Author: Mark Payne <ma...@hotmail.com>
AuthorDate: Thu May 19 11:58:52 2022 -0400
NIFI-10037: When system test fails to clean up flow, destroy the entire environment so that the next test starts in a healthy state. Name troubleshooting directories with the name of the test class to avoid ambiguity. Also added a log statement so that we know which test is running when looking at the log output from the tests themselves. Finally, found an issue in AbstractComponentNode in which we iterate over the elements in a Map and call setProperty, which can update the underlyin [...]
This closes #6059
Signed-off-by: David Handermann <ex...@apache.org>
---
.../nifi/controller/AbstractComponentNode.java | 6 ++-
.../org/apache/nifi/tests/system/NiFiSystemIT.java | 8 ++++
.../tests/system/TroubleshootingTestWatcher.java | 11 +++--
.../tests/system/loadbalance/LoadBalanceIT.java | 54 ++++++++++++++++------
.../resources/conf/clustered/node1/nifi.properties | 2 +-
.../resources/conf/clustered/node2/nifi.properties | 2 +-
.../test/resources/conf/default/nifi.properties | 2 +-
7 files changed, 63 insertions(+), 22 deletions(-)
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractComponentNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractComponentNode.java
index edf26e46da..5f5e78c3cd 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractComponentNode.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractComponentNode.java
@@ -66,6 +66,7 @@ import java.net.URL;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
@@ -600,7 +601,10 @@ public abstract class AbstractComponentNode implements ComponentNode {
// use setProperty instead of setProperties so we can bypass the class loading logic.
// Consider value changed if it is different than the PropertyDescriptor's default value because we need to call the #onPropertiesModified
// method on the component if the current value is not the default value, since the component itself is being reloaded.
- for (final Map.Entry<PropertyDescriptor, PropertyConfiguration> entry : this.properties.entrySet()) {
+ // Also, create a copy of this.properties instead of iterating directly over this.properties since the call to setProperty can change the
+ // underlying map, and the behavior of modifying the map while iterating over its elements is undefined.
+ final Map<PropertyDescriptor, PropertyConfiguration> copyOfPropertiesMap = new HashMap<>(this.properties);
+ for (final Map.Entry<PropertyDescriptor, PropertyConfiguration> entry : copyOfPropertiesMap.entrySet()) {
final PropertyDescriptor propertyDescriptor = entry.getKey();
final PropertyConfiguration configuration = entry.getValue();
diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiSystemIT.java b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiSystemIT.java
index 341e384812..298e3ec473 100644
--- a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiSystemIT.java
+++ b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiSystemIT.java
@@ -73,6 +73,8 @@ public abstract class NiFiSystemIT implements NiFiInstanceProvider {
@BeforeEach
public void setup(final TestInfo testInfo) throws IOException {
this.testInfo = testInfo;
+ final String testClassName = testInfo.getTestClass().map(Class::getSimpleName).orElse("<Unknown Test Class>");
+ logger.info("Beginning Test {}:{}", testClassName, testInfo.getDisplayName());
Thread.currentThread().setContextClassLoader(ClassLoader.getSystemClassLoader());
setupClient();
@@ -116,6 +118,12 @@ public abstract class NiFiSystemIT implements NiFiInstanceProvider {
if (isDestroyEnvironmentAfterEachTest()) {
cleanup();
+ } else if (destroyFlowFailure != null) {
+ // If unable to destroy the flow, we need to shutdown the instance and delete the flow and completely recreate the environment.
+ // Otherwise, we will be left in an unknown state for the next test, and that can cause cascading failures that are very difficult
+ // to understand and troubleshoot.
+ logger.info("Because there was a failure when destroying the flow, will completely tear down the environments and start with a clean environment for the next test.");
+ cleanup();
}
if (destroyFlowFailure != null) {
diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/TroubleshootingTestWatcher.java b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/TroubleshootingTestWatcher.java
index 8d71362359..b2c1ab078f 100644
--- a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/TroubleshootingTestWatcher.java
+++ b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/TroubleshootingTestWatcher.java
@@ -40,7 +40,8 @@ public class TroubleshootingTestWatcher implements TestWatcher {
final NiFiInstanceProvider provider = (NiFiInstanceProvider) testInstance;
final String displayName = context.getDisplayName();
try {
- final File dir = quarantineTroubleshootingInfo(provider, displayName, cause);
+ final String testClassName = context.getTestClass().map(Class::getSimpleName).orElse("TestClassUnknown");
+ final File dir = quarantineTroubleshootingInfo(provider, testClassName, displayName, cause);
logger.info("Test Failed [{}]: Troubleshooting information stored [{}]", displayName, dir.getAbsolutePath());
} catch (final Exception e) {
logger.error("Test Failed [{}]: Troubleshooting information not stored", displayName, e);
@@ -49,17 +50,21 @@ public class TroubleshootingTestWatcher implements TestWatcher {
}
}
- private File quarantineTroubleshootingInfo(final NiFiInstanceProvider provider, final String methodName, final Throwable failureCause) throws IOException {
+ private File quarantineTroubleshootingInfo(final NiFiInstanceProvider provider, final String testClassName, final String methodName, final Throwable failureCause) throws IOException {
NiFiInstance instance = provider.getNiFiInstance();
// The teardown method may or may not have already run at this point. If it has, the instance will be null.
// In that case, just create a new instance and use it - it will map to the same directories.
if (instance == null) {
+ logger.warn("While capturing troubleshooting info for {}, the NiFi Instance is not available. Will create a new one for Diagnostics purposes, but some of the diagnostics may be less " +
+ "accurate, since it's not the same instance that ran the test", methodName);
+
instance = provider.getInstanceFactory().createInstance();
}
final File troubleshooting = new File("target/troubleshooting");
- final File quarantineDir = new File(troubleshooting, methodName);
+ final String quarantineDirName = testClassName + "-" + methodName.replace("()", "");
+ final File quarantineDir = new File(troubleshooting, quarantineDirName);
quarantineDir.mkdirs();
instance.quarantineTroubleshootingInfo(quarantineDir, failureCause);
diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/loadbalance/LoadBalanceIT.java b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/loadbalance/LoadBalanceIT.java
index 2e1b1c0698..5e27f24737 100644
--- a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/loadbalance/LoadBalanceIT.java
+++ b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/loadbalance/LoadBalanceIT.java
@@ -42,12 +42,13 @@ import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.LongSummaryStatistics;
import java.util.Map;
import java.util.Set;
-import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class LoadBalanceIT extends NiFiSystemIT {
private final Logger logger = LoggerFactory.getLogger(getClass());
@@ -277,13 +278,13 @@ public class LoadBalanceIT extends NiFiSystemIT {
private int getQueueSize(final String connectionId) {
final ConnectionStatusEntity statusEntity = getConnectionStatus(connectionId);
final ConnectionStatusDTO connectionStatusDto = statusEntity.getConnectionStatus();
- return connectionStatusDto.getAggregateSnapshot().getFlowFilesQueued().intValue();
+ return connectionStatusDto.getAggregateSnapshot().getFlowFilesQueued();
}
private long getQueueBytes(final String connectionId) {
final ConnectionStatusEntity statusEntity = getConnectionStatus(connectionId);
final ConnectionStatusDTO connectionStatusDto = statusEntity.getConnectionStatus();
- return connectionStatusDto.getAggregateSnapshot().getBytesQueued().longValue();
+ return connectionStatusDto.getAggregateSnapshot().getBytesQueued();
}
private boolean isConnectionDoneLoadBalancing(final String connectionId) {
@@ -372,22 +373,45 @@ public class LoadBalanceIT extends NiFiSystemIT {
instance2.start(true);
waitForAllNodesConnected();
- // Generate the data again
generate = getNifiClient().getProcessorClient().getProcessor(generate.getId());
- getNifiClient().getProcessorClient().startProcessor(generate);
- // Wait until all 20 FlowFiles are queued up
- waitFor(() -> {
- final ConnectionStatusEntity secondRoundStatusEntity = getConnectionStatus(connection.getId());
- return secondRoundStatusEntity.getConnectionStatus().getAggregateSnapshot().getFlowFilesQueued() == 20;
- });
+ // Generate data and wait for it to be spread across the cluster. We do this in an infinite while() loop because
+ // there can be a failure, in which case we'll retry. If that happens, we just want to keep retrying until the test
+ // times out.
+ while (true) {
+ // Generate the data.
+ getNifiClient().getProcessorClient().startProcessor(generate);
- // Wait until load balancing is complete
- waitFor(() -> isConnectionDoneLoadBalancing(connection.getId()));
+ // Wait until all 20 FlowFiles are queued up
+ waitFor(() -> {
+ final ConnectionStatusEntity secondRoundStatusEntity = getConnectionStatus(connection.getId());
+ return secondRoundStatusEntity.getConnectionStatus().getAggregateSnapshot().getFlowFilesQueued() == 20;
+ });
- // Ensure that the FlowFiles are evenly distributed between the nodes.
- final ConnectionStatusEntity afterSecondDataGenerationStatusEntity = getConnectionStatus(connection.getId());
- assertTrue(isEvenlyDistributed(afterSecondDataGenerationStatusEntity));
+ // Wait until load balancing is complete
+ waitFor(() -> isConnectionDoneLoadBalancing(connection.getId()));
+
+ // Log the distribution of data between nodes for easier troubleshooting in case there's a failure.
+ final ConnectionStatusEntity afterSecondDataGenerationStatusEntity = getConnectionStatus(connection.getId());
+ final List<NodeConnectionStatusSnapshotDTO> nodeSnapshots = afterSecondDataGenerationStatusEntity.getConnectionStatus().getNodeSnapshots();
+ logger.info("FlowFiles Queued Per Node:");
+ nodeSnapshots.forEach(snapshot ->
+ logger.info("{}:{} - {}", snapshot.getAddress(), snapshot.getApiPort(), snapshot.getStatusSnapshot().getFlowFilesQueued())
+ );
+
+ // Check if the FlowFiles are evenly distributed between the nodes. If so, we're done.
+ final boolean evenlyDistributed = isEvenlyDistributed(afterSecondDataGenerationStatusEntity);
+ if (evenlyDistributed) {
+ break;
+ }
+
+ // If there's an IOException thrown while communicating between the nodes, the data will be rebalanced and will go to
+ // the local partition. There's nothing we can do about that in this test. However, we can verify that NiFi recovers
+ // from this and continues to distribute data. To do that, we will stop the processor so that it can be started again
+ // (and produce more data) and we can empty the queue so that we know how much data to expect.
+ getNifiClient().getProcessorClient().stopProcessor(generate);
+ getClientUtil().emptyQueue(connection.getId());
+ }
assertEquals(20, getQueueSize(connection.getId()));
assertEquals(20 * 1024 * 1024, getQueueBytes(connection.getId()));
diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/nifi.properties b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/nifi.properties
index 4544f2fc3b..049ac178ef 100644
--- a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/nifi.properties
+++ b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/nifi.properties
@@ -77,7 +77,7 @@ nifi.content.repository.implementation=org.apache.nifi.controller.repository.Fil
nifi.content.claim.max.appendable.size=50 KB
nifi.content.repository.directory.default=./content_repository
nifi.content.repository.archive.max.retention.period=12 hours
-nifi.content.repository.archive.max.usage.percentage=50%
+nifi.content.repository.archive.max.usage.percentage=90%
nifi.content.repository.archive.enabled=true
nifi.content.repository.always.sync=false
nifi.content.viewer.url=../nifi-content-viewer/
diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/nifi.properties b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/nifi.properties
index acd5c6707c..4b7f644058 100644
--- a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/nifi.properties
+++ b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/nifi.properties
@@ -77,7 +77,7 @@ nifi.content.repository.implementation=org.apache.nifi.controller.repository.Fil
nifi.content.claim.max.appendable.size=50 KB
nifi.content.repository.directory.default=./content_repository
nifi.content.repository.archive.max.retention.period=12 hours
-nifi.content.repository.archive.max.usage.percentage=50%
+nifi.content.repository.archive.max.usage.percentage=90%
nifi.content.repository.archive.enabled=true
nifi.content.repository.always.sync=false
nifi.content.viewer.url=../nifi-content-viewer/
diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/nifi.properties b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/nifi.properties
index 7b3de1452f..7c6426c67b 100644
--- a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/nifi.properties
+++ b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/nifi.properties
@@ -77,7 +77,7 @@ nifi.content.repository.implementation=org.apache.nifi.controller.repository.Fil
nifi.content.claim.max.appendable.size=50 KB
nifi.content.repository.directory.default=./content_repository
nifi.content.repository.archive.max.retention.period=12 hours
-nifi.content.repository.archive.max.usage.percentage=50%
+nifi.content.repository.archive.max.usage.percentage=90%
nifi.content.repository.archive.enabled=true
nifi.content.repository.always.sync=false
nifi.content.viewer.url=../nifi-content-viewer/
[nifi] 05/05: NIFI-10005 fixing version references
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a commit to branch support/nifi-1.16
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 3440921bb631650775ebf78e92443fdb26e9b599
Author: Joe Witt <jo...@apache.org>
AuthorDate: Fri May 20 09:31:12 2022 -0700
NIFI-10005 fixing version references
---
nifi-commons/nifi-jetty-configuration/pom.xml | 2 +-
nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/nifi-commons/nifi-jetty-configuration/pom.xml b/nifi-commons/nifi-jetty-configuration/pom.xml
index 8d8e2e7390..f577b79eb2 100644
--- a/nifi-commons/nifi-jetty-configuration/pom.xml
+++ b/nifi-commons/nifi-jetty-configuration/pom.xml
@@ -18,7 +18,7 @@
<parent>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-commons</artifactId>
- <version>1.17.0-SNAPSHOT</version>
+ <version>1.16.2-SNAPSHOT</version>
</parent>
<artifactId>nifi-jetty-configuration</artifactId>
<packaging>jar</packaging>
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
index b0b840c392..46b1435c83 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
@@ -95,7 +95,7 @@
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-jetty-configuration</artifactId>
- <version>1.17.0-SNAPSHOT</version>
+ <version>1.16.2-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
[nifi] 02/05: NIFI-10036: Corrected Elasticsearch Client Service to prefix paths for all requests
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a commit to branch support/nifi-1.16
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 7ed0637bb624b4c2824ef25b121038ae2dc0cd28
Author: Joe Gresock <jg...@gmail.com>
AuthorDate: Thu May 19 10:44:59 2022 -0400
NIFI-10036: Corrected Elasticsearch Client Service to prefix paths for all requests
- Prefixing endpoint paths with a forward slash ensures correct HTTP request formatting required for some deployments with a forwarding proxy
This closes #6058
Signed-off-by: David Handermann <ex...@apache.org>
---
.../nifi/elasticsearch/ElasticSearchClientServiceImpl.java | 10 +++++-----
1 file changed, 5 insertions(+), 5 deletions(-)
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
index 58dc99ea52..2f2ae2a1a2 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
@@ -320,7 +320,7 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
final HttpEntity entity = new NStringEntity(payload.toString(), ContentType.APPLICATION_JSON);
final StopWatch watch = new StopWatch();
watch.start();
- final Response response = performRequest("POST", "_bulk", requestParameters, entity);
+ final Response response = performRequest("POST", "/_bulk", requestParameters, entity);
watch.stop();
final String rawResponse = IOUtils.toString(response.getEntity().getContent(), StandardCharsets.UTF_8);
@@ -360,7 +360,7 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
final HttpEntity entity = new NStringEntity(sb.toString(), ContentType.APPLICATION_JSON);
final StopWatch watch = new StopWatch();
watch.start();
- final Response response = performRequest("POST", "_bulk", requestParameters, entity);
+ final Response response = performRequest("POST", "/_bulk", requestParameters, entity);
watch.stop();
if (getLogger().isDebugEnabled()) {
@@ -406,7 +406,7 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
try {
final StringBuilder endpoint = new StringBuilder();
if (StringUtils.isNotBlank(index) && !"/".equals(index)) {
- endpoint.append(index);
+ endpoint.append("/").append(index);
}
endpoint.append("/_refresh");
final Response response = performRequest("POST", endpoint.toString(), requestParameters, null);
@@ -421,7 +421,7 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
public Map<String, Object> get(final String index, final String type, final String id, final Map<String, String> requestParameters) {
try {
final StringBuilder endpoint = new StringBuilder();
- endpoint.append(index);
+ endpoint.append("/").append(index);
if (StringUtils.isNotBlank(type)) {
endpoint.append("/").append(type);
} else {
@@ -482,7 +482,7 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
put("keep_alive", keepAlive);
}
}};
- final Response response = performRequest("POST", index + "/_pit", params, null);
+ final Response response = performRequest("POST", "/" + index + "/_pit", params, null);
final String body = IOUtils.toString(response.getEntity().getContent(), StandardCharsets.UTF_8);
parseResponseWarningHeaders(response);
[nifi] 01/05: NIFI-10015 Upgraded Hadoop from 3.3.0 to 3.3.2
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a commit to branch support/nifi-1.16
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 931479993503b9d408d3623f27e2f4644236a8a7
Author: exceptionfactory <ex...@apache.org>
AuthorDate: Tue May 17 12:18:53 2022 -0500
NIFI-10015 Upgraded Hadoop from 3.3.0 to 3.3.2
Signed-off-by: Pierre Villard <pi...@gmail.com>
This closes #6051.
---
pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pom.xml b/pom.xml
index 6eaeea037b..26318b1566 100644
--- a/pom.xml
+++ b/pom.xml
@@ -120,7 +120,7 @@
<surefire.version>3.0.0-M5</surefire.version>
<!-- The Hadoop version used by nifi-hadoop-libraries-nar and any NARs that depend on it, other NARs that need
a specific version should override this property, or use a more specific property like abc.hadoop.version -->
- <hadoop.version>3.3.0</hadoop.version>
+ <hadoop.version>3.3.2</hadoop.version>
<ozone.version>1.2.1</ozone.version>
<gcs.version>2.1.5</gcs.version>
<aspectj.version>1.9.6</aspectj.version>