You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by pv...@apache.org on 2023/07/14 16:46:32 UTC
[nifi] branch support/nifi-1.x updated: NIFI-11794 - Fix NPE + configure max attempts for Redis State Provider (#7473)
This is an automated email from the ASF dual-hosted git repository.
pvillard pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/support/nifi-1.x by this push:
new 00a6478c06 NIFI-11794 - Fix NPE + configure max attempts for Redis State Provider (#7473)
00a6478c06 is described below
commit 00a6478c06bc167d1c6fddd6f6ca3c25f192f49c
Author: Pierre Villard <pi...@gmail.com>
AuthorDate: Fri Jul 14 17:51:18 2023 +0200
NIFI-11794 - Fix NPE + configure max attempts for Redis State Provider (#7473)
Signed-off-by: Otto Fowler<ot...@gmail.com>
This closes #7473.
---
.../src/main/resources/conf/state-management.xml | 3 +++
.../apache/nifi/redis/state/RedisStateProvider.java | 20 +++++++++++++++++---
.../nifi/redis/state/TestRedisStateProvider.java | 1 +
3 files changed, 21 insertions(+), 3 deletions(-)
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/state-management.xml b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/state-management.xml
index 85fb6c6dcf..c16d4fbb18 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/state-management.xml
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/state-management.xml
@@ -99,6 +99,9 @@
nifi.properties. This means that a TLS-enabled Redis connection is only possible if the Apache NiFi instance is running in secure mode.
If this property is false, an insecure Redis connection will be used even if the Apache NiFi instance is secure (default false).
+ Max Attempts - Maximum number of attempts when setting/clearing the state for a component. This number should be higher than the number of nodes
+ in the NiFi cluster to account for the case where each node may concurrently try to clear a state with a local scope (default is 20).
+
Pool - Max Total - The maximum number of connections that can be allocated by the pool (checked out to clients, or idle awaiting checkout).
A negative value indicates that there is no limit.
diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java
index 6d552dc771..07a36a8557 100644
--- a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java
+++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java
@@ -68,17 +68,28 @@ public class RedisStateProvider extends AbstractConfigurableComponent implements
.defaultValue("false")
.addValidator(StandardValidators.BOOLEAN_VALIDATOR)
.build();
+ public static final PropertyDescriptor MAX_ATTEMPTS = new PropertyDescriptor.Builder()
+ .name("Max Attempts")
+ .displayName("Max Attempts")
+ .description("Maximum number of attempts when setting/clearing the state for a component. This number should be higher than the number of nodes "
+ + "in the NiFi cluster to account for the case where each node may concurrently try to clear a state with a local scope.")
+ .required(true)
+ .defaultValue("20")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .build();
static final List<PropertyDescriptor> STATE_PROVIDER_PROPERTIES;
static {
final List<PropertyDescriptor> props = new ArrayList<>(RedisUtils.REDIS_CONNECTION_PROPERTY_DESCRIPTORS);
props.add(KEY_PREFIX);
props.add(ENABLE_TLS);
+ props.add(MAX_ATTEMPTS);
STATE_PROVIDER_PROPERTIES = Collections.unmodifiableList(props);
}
private String identifier;
private String keyPrefix;
+ private int maxAttempts;
private ComponentLog logger;
private PropertyContext context;
private SSLContext sslContext;
@@ -102,6 +113,8 @@ public class RedisStateProvider extends AbstractConfigurableComponent implements
keyPrefix = keyPrefix + "/";
}
this.keyPrefix = keyPrefix;
+
+ this.maxAttempts = context.getProperty(MAX_ATTEMPTS).asInteger();
}
@Override
@@ -173,7 +186,7 @@ public class RedisStateProvider extends AbstractConfigurableComponent implements
int attempted = 0;
boolean updated = false;
- while (!updated && attempted < 20) {
+ while (!updated && attempted < this.maxAttempts) {
updated = replace(currStateMap, state, componentId, true);
attempted++;
}
@@ -245,7 +258,8 @@ public class RedisStateProvider extends AbstractConfigurableComponent implements
final List<Object> results = redisConnection.exec();
// if we have a result then the replace succeeded
- if (results.size() > 0) {
+ // results can be null if the transaction has been aborted
+ if (results != null && results.size() > 0) {
replaced = true;
}
@@ -258,7 +272,7 @@ public class RedisStateProvider extends AbstractConfigurableComponent implements
int attempted = 0;
boolean updated = false;
- while (!updated && attempted < 20) {
+ while (!updated && attempted < this.maxAttempts) {
final StateMap currStateMap = getState(componentId);
updated = replace(currStateMap, Collections.emptyMap(), componentId, true);
diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/TestRedisStateProvider.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/TestRedisStateProvider.java
index b0c5a16ef0..cac9c0acd5 100644
--- a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/TestRedisStateProvider.java
+++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/TestRedisStateProvider.java
@@ -45,6 +45,7 @@ public class TestRedisStateProvider {
// Set up mock state provider init context
Mockito.when(context.getProperty(RedisStateProvider.KEY_PREFIX)).thenReturn(new MockPropertyValue("/nifi/components/"));
+ Mockito.when(context.getProperty(RedisStateProvider.MAX_ATTEMPTS)).thenReturn(new MockPropertyValue("20"));
// Set up mock validation context
Mockito.when(validationContext.getProperty(RedisUtils.CONNECTION_STRING)).thenReturn(new MockPropertyValue("localhost:6379"));