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 2020/09/22 20:53:26 UTC
[nifi] branch support/nifi-1.12.x updated (b1b5ccd -> 67e2b90)
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a change to branch support/nifi-1.12.x
in repository https://gitbox.apache.org/repos/asf/nifi.git.
from b1b5ccd NIFI-7787 updating to latest nifi nar maven plugin
new 683f3c4 NIFI-7811: Prevent NPE and bad behavior when PrometheusReportingTask can't start
new d4b0c1b NIFI-5061-NiFi documentation incomplete/wrong for EL hierarchy.
new 52f219b NIFI-7816: Correct documentation example for urlEncode function in Expression Language Guide (#4536)
new 1fb8bc3 NIFI-7794 Adding a property to PutAzureStorageBlob to determine whether to check the existence of a container and create it if necessary.
new be3e918 NIFI-7832 Resetting boolean that indicates password is being used when service is disabled (#4543)
new 67e2b90 NIFI-7834 - InvokeAWSGatewayAPI should not be annotated with @TriggerWhenEmpty
The 6 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:
.../main/asciidoc/expression-language-guide.adoc | 24 ++++++++---
.../processors/aws/wag/InvokeAWSGatewayApi.java | 2 -
.../azure/storage/PutAzureBlobStorage.java | 20 ++++++++-
.../azure/storage/utils/AzureStorageUtils.java | 2 +-
.../reporting/prometheus/PrometheusRecordSink.java | 20 ++++++---
.../prometheus/PrometheusReportingTask.java | 44 ++++++++++++-------
.../reporting/prometheus/PrometheusServer.java | 50 +++++++++++++---------
.../prometheus/PrometheusReportingTaskIT.java | 33 ++++++++++----
.../prometheus/TestPrometheusRecordSink.java | 18 ++++++++
.../hortonworks/HortonworksSchemaRegistry.java | 2 +
10 files changed, 157 insertions(+), 58 deletions(-)
[nifi] 04/06: NIFI-7794 Adding a property to PutAzureStorageBlob to
determine whether to check the existence of a container and create it if
necessary.
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.12.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 1fb8bc3acc2e422d8e335124864326dc1a976926
Author: Paul Kelly <pk...@gmail.com>
AuthorDate: Thu Sep 17 18:52:38 2020 +0000
NIFI-7794 Adding a property to PutAzureStorageBlob to determine whether to check the existence of a container and create it if necessary.
This closes #4535.
Signed-off-by: Joey Frazee <jf...@apache.org>
---
.../azure/storage/PutAzureBlobStorage.java | 20 +++++++++++++++++++-
.../azure/storage/utils/AzureStorageUtils.java | 2 +-
2 files changed, 20 insertions(+), 2 deletions(-)
diff --git a/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage.java b/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage.java
index 7441f9c..5301700 100644
--- a/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage.java
+++ b/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage.java
@@ -73,11 +73,25 @@ public class PutAzureBlobStorage extends AbstractAzureBlobProcessor {
.required(true)
.build();
+ public static final PropertyDescriptor CREATE_CONTAINER = new PropertyDescriptor.Builder()
+ .name("azure-create-container")
+ .displayName("Create Container")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .description("Specifies whether to check if the container exists and to automatically create it if it does not. " +
+ "Permission to list containers is required. If false, this check is not made, but the Put operation " +
+ "will fail if the container does not exist.")
+ .build();
+
@Override
public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
List<PropertyDescriptor> properties = new ArrayList<>(super.getSupportedPropertyDescriptors());
properties.remove(BLOB);
properties.add(BLOB_NAME);
+ properties.add(CREATE_CONTAINER);
return properties;
}
@@ -93,11 +107,15 @@ public class PutAzureBlobStorage extends AbstractAzureBlobProcessor {
String blobPath = context.getProperty(BLOB_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean createContainer = context.getProperty(CREATE_CONTAINER).asBoolean();
+
AtomicReference<Exception> storedException = new AtomicReference<>();
try {
CloudBlobClient blobClient = AzureStorageUtils.createCloudBlobClient(context, getLogger(), flowFile);
CloudBlobContainer container = blobClient.getContainerReference(containerName);
- container.createIfNotExists();
+
+ if (createContainer)
+ container.createIfNotExists();
CloudBlob blob = container.getBlockBlobReference(blobPath);
diff --git a/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java b/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
index 34be7a3..8eebb2c 100644
--- a/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
+++ b/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java
@@ -109,7 +109,7 @@ public final class AzureStorageUtils {
public static final PropertyDescriptor CONTAINER = new PropertyDescriptor.Builder()
.name("container-name")
.displayName("Container Name")
- .description("Name of the Azure storage container. In case of PutAzureBlobStorage processor, container will be created if it does not exist.")
+ .description("Name of the Azure storage container. In case of PutAzureBlobStorage processor, container can be created if it does not exist.")
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
.required(true)
[nifi] 06/06: NIFI-7834 - InvokeAWSGatewayAPI should not be
annotated with @TriggerWhenEmpty
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.12.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 67e2b9025afaca468df795f4d5c7faff8b05a81c
Author: Pierre Villard <pi...@gmail.com>
AuthorDate: Tue Sep 22 18:37:28 2020 +0200
NIFI-7834 - InvokeAWSGatewayAPI should not be annotated with @TriggerWhenEmpty
---
.../java/org/apache/nifi/processors/aws/wag/InvokeAWSGatewayApi.java | 2 --
1 file changed, 2 deletions(-)
diff --git a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/wag/InvokeAWSGatewayApi.java b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/wag/InvokeAWSGatewayApi.java
index de990e4..d138dcc 100644
--- a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/wag/InvokeAWSGatewayApi.java
+++ b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/wag/InvokeAWSGatewayApi.java
@@ -22,7 +22,6 @@ import org.apache.nifi.annotation.behavior.DynamicProperty;
import org.apache.nifi.annotation.behavior.InputRequirement;
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
import org.apache.nifi.annotation.behavior.SupportsBatching;
-import org.apache.nifi.annotation.behavior.TriggerWhenEmpty;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
@@ -52,7 +51,6 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
-@TriggerWhenEmpty
@SupportsBatching
@InputRequirement(Requirement.INPUT_ALLOWED)
@Tags({"Amazon", "AWS", "Client", "Gateway-API", "Rest", "http", "https"})
[nifi] 03/06: NIFI-7816: Correct documentation example for
urlEncode function in Expression Language Guide (#4536)
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.12.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 52f219bf0838f667a07e7109dae486a1439c719c
Author: Mohammed Nadeem <na...@gmail.com>
AuthorDate: Tue Sep 22 00:42:23 2020 +0530
NIFI-7816: Correct documentation example for urlEncode function in Expression Language Guide (#4536)
Signed-off-by: Andy LoPresto <al...@apache.org>
---
nifi-docs/src/main/asciidoc/expression-language-guide.adoc | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/nifi-docs/src/main/asciidoc/expression-language-guide.adoc b/nifi-docs/src/main/asciidoc/expression-language-guide.adoc
index 7217f2a..b79629e 100644
--- a/nifi-docs/src/main/asciidoc/expression-language-guide.adoc
+++ b/nifi-docs/src/main/asciidoc/expression-language-guide.adoc
@@ -1380,7 +1380,7 @@ Each of the following functions will encode a string according the rules of the
*Examples*: We can URL-Encode an attribute named "url" by using the Expression `${url:urlEncode()}`. If
the value of the "url" attribute is "https://nifi.apache.org/some value with spaces", this
- Expression will then return "https://nifi.apache.org/some%20value%20with%20spaces".
+ Expression will then return "https%3A%2F%2Fnifi.apache.org%2Fsome+value+with+spaces".
@@ -1397,7 +1397,7 @@ Each of the following functions will encode a string according the rules of the
*Return Type*: [.returnType]#String#
*Examples*: If we have a URL-Encoded attribute named "url" with the value
- "https://nifi.apache.org/some%20value%20with%20spaces", then the Expression
+ "https://nifi.apache.org/some%20value%20with%20spaces" or "https%3A%2F%2Fnifi.apache.org%2Fsome+value+with+spaces", then the Expression
`${url:urlDecode()}` will return "https://nifi.apache.org/some value with spaces".
[nifi] 02/06: NIFI-5061-NiFi documentation incomplete/wrong for EL
hierarchy.
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.12.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit d4b0c1bada78ae7184b8cc174f06544c4d63cf00
Author: abrown <ab...@cloudera.com>
AuthorDate: Tue Sep 15 17:55:13 2020 +0100
NIFI-5061-NiFi documentation incomplete/wrong for EL hierarchy.
Co-authored-by: Pierre Villard <pi...@gmail.com>
Signed-off-by: Pierre Villard <pi...@gmail.com>
This closes #4530.
---
.../src/main/asciidoc/expression-language-guide.adoc | 20 ++++++++++++++++----
1 file changed, 16 insertions(+), 4 deletions(-)
diff --git a/nifi-docs/src/main/asciidoc/expression-language-guide.adoc b/nifi-docs/src/main/asciidoc/expression-language-guide.adoc
index aad0db2..7217f2a 100644
--- a/nifi-docs/src/main/asciidoc/expression-language-guide.adoc
+++ b/nifi-docs/src/main/asciidoc/expression-language-guide.adoc
@@ -108,10 +108,9 @@ If any of these special characters is present in an attribute is quoted by using
The Expression Language allows single quotes and double quotes to be used interchangeably. For example, the following
can be used to escape an attribute named `my attribute`: `${"my attribute"}` or `${'my attribute'}`.
-In this example, the value to be returned is the value of the "my attribute" value, if it exists. If that attribute
-does not exist, the Expression Language will then look for a System Environment Variable named "my attribute." If
-unable to find this, it will look for a JVM System Property named "my attribute." Finally, if none of these exists,
-the Expression Language will return a `null` value.
+In this example, the value to be returned is the value of the "my attribute" attribute, if it exists. The Expression
+Language will search through a hierarchy for a matching property. See <<expression-language-hierarchy>>
+for a description of the hierarchy.
There also exist some functions that expect to have no subject. These functions are invoked simply
by calling the function at the beginning of the Expression, such as `${hostname()}`. These functions
@@ -131,7 +130,20 @@ the Expression Language between delimiters. Therefore, we can use the Expression
mean the same thing. We cannot, however, use `${file name:equals(${uuid})}`, because this results
in `file` and `name` being interpreted as different tokens, rather than a single token, `filename`.
+[[expression-language-hierarchy]]
+=== Expression Language Hierarchy
+When using Expression Language to reference a property by name there is a defined hierarchy within which NiFi
+will search for the value.
+The current hierarchy in NiFi is as follows:
+
+1. Search FlowFile for attribute/key
+2. Search Process Group Variables for attribute/key
+3. Search File Registry file for attribute/key
+4. Search NiFi JVM Properties for attribute/key
+5. Search System Environment Variables for attribute/key
+
+NiFi will search for, and return, the first occurrence of a matching property. If no matching property is found, `null` is returned.
[[usage]]
== Expression Language in the Application
[nifi] 05/06: NIFI-7832 Resetting boolean that indicates password
is being used when service is disabled (#4543)
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.12.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit be3e918ae370b6496711df3697eddd9ce058b12a
Author: Bryan Bende <bb...@apache.org>
AuthorDate: Tue Sep 22 12:11:46 2020 -0400
NIFI-7832 Resetting boolean that indicates password is being used when service is disabled (#4543)
Signed-off-by: Andy LoPresto <al...@apache.org>
---
.../nifi/schemaregistry/hortonworks/HortonworksSchemaRegistry.java | 2 ++
1 file changed, 2 insertions(+)
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-hwx-schema-registry-bundle/nifi-hwx-schema-registry-service/src/main/java/org/apache/nifi/schemaregistry/hortonworks/HortonworksSchemaRegistry.java b/nifi-nar-bundles/nifi-standard-services/nifi-hwx-schema-registry-bundle/nifi-hwx-schema-registry-service/src/main/java/org/apache/nifi/schemaregistry/hortonworks/HortonworksSchemaRegistry.java
index 317c5e6..f48f250 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-hwx-schema-registry-bundle/nifi-hwx-schema-registry-service/src/main/java/org/apache/nifi/schemaregistry/hortonworks/HortonworksSchemaRegistry.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-hwx-schema-registry-bundle/nifi-hwx-schema-registry-service/src/main/java/org/apache/nifi/schemaregistry/hortonworks/HortonworksSchemaRegistry.java
@@ -222,6 +222,7 @@ public class HortonworksSchemaRegistry extends AbstractControllerService impleme
final String keytab = kerberosCredentialsService.getKeytab();
final String jaasConfigString = getKeytabJaasConfig(principal, keytab);
schemaRegistryConfig.put(SchemaRegistryClient.Configuration.SASL_JAAS_CONFIG.name(), jaasConfigString);
+ usingKerberosWithPassword = false;
} else if (!StringUtils.isBlank(kerberosPrincipal) && !StringUtils.isBlank(kerberosPassword)) {
schemaRegistryConfig.put(SchemaRegistryClientWithKerberosPassword.SCHEMA_REGISTRY_CLIENT_KERBEROS_PRINCIPAL, kerberosPrincipal);
schemaRegistryConfig.put(SchemaRegistryClientWithKerberosPassword.SCHEMA_REGISTRY_CLIENT_KERBEROS_PASSWORD, kerberosPassword);
@@ -268,6 +269,7 @@ public class HortonworksSchemaRegistry extends AbstractControllerService impleme
}
initialized = false;
+ usingKerberosWithPassword = false;
}
[nifi] 01/06: NIFI-7811: Prevent NPE and bad behavior when
PrometheusReportingTask can't start
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.12.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 683f3c49c902769653ce6b9eb31ff029a194d393
Author: Matthew Burgess <ma...@apache.org>
AuthorDate: Tue Sep 15 13:31:32 2020 -0400
NIFI-7811: Prevent NPE and bad behavior when PrometheusReportingTask can't start
Signed-off-by: Pierre Villard <pi...@gmail.com>
This closes #4531.
---
.../reporting/prometheus/PrometheusRecordSink.java | 20 ++++++---
.../prometheus/PrometheusReportingTask.java | 44 ++++++++++++-------
.../reporting/prometheus/PrometheusServer.java | 50 +++++++++++++---------
.../prometheus/PrometheusReportingTaskIT.java | 33 ++++++++++----
.../prometheus/TestPrometheusRecordSink.java | 18 ++++++++
5 files changed, 117 insertions(+), 48 deletions(-)
diff --git a/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/main/java/org/apache/nifi/reporting/prometheus/PrometheusRecordSink.java b/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/main/java/org/apache/nifi/reporting/prometheus/PrometheusRecordSink.java
index 91fd7c2..bc4009d 100644
--- a/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/main/java/org/apache/nifi/reporting/prometheus/PrometheusRecordSink.java
+++ b/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/main/java/org/apache/nifi/reporting/prometheus/PrometheusRecordSink.java
@@ -26,6 +26,7 @@ import org.apache.nifi.annotation.lifecycle.OnShutdown;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.AbstractControllerService;
import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.record.sink.RecordSinkService;
import org.apache.nifi.reporting.ReportingContext;
import org.apache.nifi.prometheus.util.PrometheusMetricsUtil;
@@ -120,7 +121,8 @@ public class PrometheusRecordSink extends AbstractControllerService implements R
prometheusServer.setMetricsCollectors(metricsCollectors);
getLogger().info("Started JETTY server");
} catch (Exception e) {
- getLogger().error("Failed to start Jetty server", e);
+ // Don't allow this to finish successfully, onTrigger should not be called if the Jetty server wasn't started
+ throw new ProcessException("Failed to start Jetty server", e);
}
}
@@ -181,15 +183,23 @@ public class PrometheusRecordSink extends AbstractControllerService implements R
@OnDisabled
public void onStopped() throws Exception {
- Server server = prometheusServer.getServer();
- server.stop();
+ if (prometheusServer != null) {
+ Server server = prometheusServer.getServer();
+ if (server != null) {
+ server.stop();
+ }
+ }
recordSchema = null;
}
@OnShutdown
public void onShutDown() throws Exception {
- Server server = prometheusServer.getServer();
- server.stop();
+ if (prometheusServer != null) {
+ Server server = prometheusServer.getServer();
+ if (server != null) {
+ server.stop();
+ }
+ }
recordSchema = null;
}
diff --git a/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/main/java/org/apache/nifi/reporting/prometheus/PrometheusReportingTask.java b/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/main/java/org/apache/nifi/reporting/prometheus/PrometheusReportingTask.java
index 60c64f7..774d036 100644
--- a/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/main/java/org/apache/nifi/reporting/prometheus/PrometheusReportingTask.java
+++ b/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/main/java/org/apache/nifi/reporting/prometheus/PrometheusReportingTask.java
@@ -17,12 +17,6 @@
package org.apache.nifi.reporting.prometheus;
-import java.net.InetSocketAddress;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.function.Function;
-
import io.prometheus.client.CollectorRegistry;
import org.apache.nifi.annotation.configuration.DefaultSchedule;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
@@ -34,23 +28,32 @@ import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.controller.status.ProcessGroupStatus;
import org.apache.nifi.metrics.jvm.JmxJvmMetrics;
+import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.prometheus.util.JvmMetricsRegistry;
import org.apache.nifi.prometheus.util.NiFiMetricsRegistry;
+import org.apache.nifi.prometheus.util.PrometheusMetricsUtil;
import org.apache.nifi.reporting.AbstractReportingTask;
import org.apache.nifi.reporting.ReportingContext;
-import org.apache.nifi.prometheus.util.PrometheusMetricsUtil;
import org.apache.nifi.scheduling.SchedulingStrategy;
import org.apache.nifi.ssl.RestrictedSSLContextService;
import org.apache.nifi.ssl.SSLContextService;
import org.eclipse.jetty.server.Server;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
+
import static org.apache.nifi.prometheus.util.PrometheusMetricsUtil.METRICS_STRATEGY_COMPONENTS;
import static org.apache.nifi.prometheus.util.PrometheusMetricsUtil.METRICS_STRATEGY_PG;
import static org.apache.nifi.prometheus.util.PrometheusMetricsUtil.METRICS_STRATEGY_ROOT;
@Tags({ "reporting", "prometheus", "metrics", "time series data" })
-@CapabilityDescription("Reports metrics in Prometheus format by creating /metrics http endpoint which can be used for external monitoring of the application."
- + " The reporting task reports a set of metrics regarding the JVM (optional) and the NiFi instance")
+@CapabilityDescription("Reports metrics in Prometheus format by creating a /metrics HTTP(S) endpoint which can be used for external monitoring of the application."
+ + " The reporting task reports a set of metrics regarding the JVM (optional) and the NiFi instance. Note that if the underlying Jetty server (i.e. the "
+ + "Prometheus endpoint) cannot be started (for example if two PrometheusReportingTask instances are started on the same port), this may cause a delay in "
+ + "shutting down NiFi while it waits for the server resources to be cleaned up.")
@DefaultSchedule(strategy = SchedulingStrategy.TIMER_DRIVEN, period = "60 sec")
public class PrometheusReportingTask extends AbstractReportingTask {
@@ -145,24 +148,35 @@ public class PrometheusReportingTask extends AbstractReportingTask {
this.prometheusServer.setMetricsCollectors(metricsCollectors);
getLogger().info("Started JETTY server");
} catch (Exception e) {
- getLogger().error("Failed to start Jetty server", e);
+ // Don't allow this to finish successfully, onTrigger should not be called if the Jetty server wasn't started
+ throw new ProcessException("Failed to start Jetty server", e);
}
}
@OnStopped
public void OnStopped() throws Exception {
- Server server = this.prometheusServer.getServer();
- server.stop();
+ if (prometheusServer != null) {
+ Server server = prometheusServer.getServer();
+ if (server != null) {
+ server.stop();
+ }
+ }
}
@OnShutdown
public void onShutDown() throws Exception {
- Server server = prometheusServer.getServer();
- server.stop();
+ if (prometheusServer != null) {
+ Server server = prometheusServer.getServer();
+ if (server != null) {
+ server.stop();
+ }
+ }
}
@Override
public void onTrigger(final ReportingContext context) {
- this.prometheusServer.setReportingContext(context);
+ if (prometheusServer != null) {
+ prometheusServer.setReportingContext(context);
+ }
}
}
diff --git a/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/main/java/org/apache/nifi/reporting/prometheus/PrometheusServer.java b/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/main/java/org/apache/nifi/reporting/prometheus/PrometheusServer.java
index d57f1c1..cd815f5 100644
--- a/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/main/java/org/apache/nifi/reporting/prometheus/PrometheusServer.java
+++ b/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/main/java/org/apache/nifi/reporting/prometheus/PrometheusServer.java
@@ -17,20 +17,8 @@
package org.apache.nifi.reporting.prometheus;
-import java.io.IOException;
-import java.io.OutputStreamWriter;
-import java.net.HttpURLConnection;
-import java.net.InetSocketAddress;
-import java.util.Collections;
-import java.util.List;
-import java.util.function.Function;
-
-import javax.servlet.ServletException;
-import javax.servlet.ServletOutputStream;
-import javax.servlet.http.HttpServlet;
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
-
+import io.prometheus.client.CollectorRegistry;
+import io.prometheus.client.exporter.common.TextFormat;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.reporting.ReportingContext;
import org.apache.nifi.ssl.SSLContextService;
@@ -45,8 +33,18 @@ import org.eclipse.jetty.servlet.ServletContextHandler;
import org.eclipse.jetty.servlet.ServletHolder;
import org.eclipse.jetty.util.ssl.SslContextFactory;
-import io.prometheus.client.CollectorRegistry;
-import io.prometheus.client.exporter.common.TextFormat;
+import javax.servlet.ServletException;
+import javax.servlet.ServletOutputStream;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
public class PrometheusServer {
private static ComponentLog logger;
@@ -86,10 +84,16 @@ public class PrometheusServer {
PrometheusServer.logger = logger;
metricsCollectors = Collections.emptyList();
this.server = new Server(addr);
-
this.handler = new ServletContextHandler(server, "/metrics");
this.handler.addServlet(new ServletHolder(new MetricsServlet()), "/");
- this.server.start();
+ try {
+ this.server.start();
+ } catch (Exception e) {
+ // If Jetty couldn't start, stop it explicitly to avoid dangling threads
+ logger.debug("PrometheusServer: Couldn't start Jetty server, stopping manually");
+ this.server.stop();
+ throw e;
+ }
}
public PrometheusServer(int addr, SSLContextService sslContextService, ComponentLog logger, boolean needClientAuth, boolean wantClientAuth) throws Exception {
@@ -108,8 +112,14 @@ public class PrometheusServer {
new HttpConnectionFactory(httpsConfiguration));
https.setPort(addr);
this.server.setConnectors(new Connector[]{https});
- this.server.start();
-
+ try {
+ this.server.start();
+ } catch (Exception e) {
+ // If Jetty couldn't start, stop it explicitly to avoid dangling threads
+ logger.debug("PrometheusServer: Couldn't start Jetty server, stopping manually");
+ this.server.stop();
+ throw e;
+ }
}
private SslContextFactory createSslFactory(final SSLContextService sslService, boolean needClientAuth, boolean wantClientAuth) {
diff --git a/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/test/java/org/apache/nifi/reporting/prometheus/PrometheusReportingTaskIT.java b/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/test/java/org/apache/nifi/reporting/prometheus/PrometheusReportingTaskIT.java
index 52a7444..a23bd30c 100644
--- a/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/test/java/org/apache/nifi/reporting/prometheus/PrometheusReportingTaskIT.java
+++ b/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/test/java/org/apache/nifi/reporting/prometheus/PrometheusReportingTaskIT.java
@@ -16,13 +16,6 @@
*/
package org.apache.nifi.reporting.prometheus;
-import java.io.IOException;
-import java.net.HttpURLConnection;
-import java.net.URL;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
@@ -32,8 +25,9 @@ import org.apache.http.util.EntityUtils;
import org.apache.nifi.controller.status.PortStatus;
import org.apache.nifi.controller.status.ProcessGroupStatus;
import org.apache.nifi.controller.status.RunStatus;
-import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.prometheus.util.PrometheusMetricsUtil;
+import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.state.MockStateManager;
import org.apache.nifi.util.MockComponentLog;
import org.apache.nifi.util.MockConfigurationContext;
@@ -44,6 +38,15 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import java.io.IOException;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+
+import static org.junit.Assert.fail;
+
public class PrometheusReportingTaskIT {
private static final String TEST_INIT_CONTEXT_ID = "test-init-context-id";
private static final String TEST_INIT_CONTEXT_NAME = "test-init-context-name";
@@ -183,4 +186,18 @@ public class PrometheusReportingTaskIT {
// Ignore
}
}
+
+ @Test
+ public void testTwoInstances() throws IOException, InterruptedException, InitializationException {
+ testedReportingTask.initialize(reportingInitContextStub);
+ testedReportingTask.onScheduled(configurationContextStub);
+ PrometheusReportingTask testedReportingTask2 = new PrometheusReportingTask();
+ testedReportingTask2.initialize(reportingInitContextStub);
+ try {
+ testedReportingTask2.onScheduled(configurationContextStub);
+ fail("Should have reported Address In Use");
+ } catch (ProcessException pe) {
+ // Do nothing, this is the expected behavior
+ }
+ }
}
diff --git a/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/test/java/org/apache/nifi/reporting/prometheus/TestPrometheusRecordSink.java b/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/test/java/org/apache/nifi/reporting/prometheus/TestPrometheusRecordSink.java
index 34bb657..7dd3e4a 100644
--- a/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/test/java/org/apache/nifi/reporting/prometheus/TestPrometheusRecordSink.java
+++ b/nifi-nar-bundles/nifi-prometheus-bundle/nifi-prometheus-reporting-task/src/test/java/org/apache/nifi/reporting/prometheus/TestPrometheusRecordSink.java
@@ -29,6 +29,7 @@ import org.apache.nifi.components.state.StateManager;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.controller.ControllerServiceInitializationContext;
import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.prometheus.util.PrometheusMetricsUtil;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.serialization.SimpleRecordSchema;
@@ -59,6 +60,7 @@ import java.util.UUID;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -111,6 +113,22 @@ public class TestPrometheusRecordSink {
}
}
+ @Test
+ public void testTwoInstances() throws InitializationException {
+ PrometheusRecordSink sink1 = initTask();
+ try {
+ PrometheusRecordSink sink2 = initTask();
+ fail("Should have reported Address In Use");
+ } catch (ProcessException pe) {
+ // Do nothing, this is the expected behavior
+ }
+ try {
+ sink1.onStopped();
+ } catch (Exception e) {
+ // Do nothing, just need to shut down the server before the next run
+ }
+ }
+
private String getMetrics() throws IOException {
URL url = new URL("http://localhost:" + portString + "/metrics");
HttpURLConnection con = (HttpURLConnection) url.openConnection();