You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by bbende <gi...@git.apache.org> on 2017/06/15 17:39:03 UTC
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
GitHub user bbende opened a pull request:
https://github.com/apache/nifi/pull/1918
NIFI-4061 Add a RedisStateProvider
This PR adds a RedisStateProvider that can be used as an alternative to the ZooKeeperStateProvider for clustered state.
This PR includes the Redis work from NIFI-4043, so this could be merged to include both NIFI-4043 and NIFI-4061, or the other one could be merged first and then this branch could be rebased.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/bbende/nifi NIFI-4061
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/nifi/pull/1918.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1918
----
commit 3cfe414eabc82192797fbd5d39c6c27860dc3eaa
Author: Bryan Bende <bb...@apache.org>
Date: 2017-06-12T19:53:20Z
NIFI-4043 Initial commit of nifi-redis-bundle
commit 1581cd1662bf6239465fe6fefa472903788a255e
Author: Bryan Bende <bb...@apache.org>
Date: 2017-06-13T19:57:22Z
NIFI-4061 Initial version of RedisStateProvider
- Adding PropertyContext and updating existing contexts to extend it
- Added embedded Redis for unit testing
- Added wrapped StateProvider with NAR ClassLoader in StandardStateManagerProvider
- Updating state-management.xml with config for RedisStateProvider
- Renaming tests that use RedisServer to be IT tests so they don't run all the time
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122836427
--- Diff: nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/service/RedisDistributedMapCacheClientService.java ---
@@ -0,0 +1,313 @@
+/*
+ * 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.redis.service;
+
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnDisabled;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.distributed.cache.client.AtomicCacheEntry;
+import org.apache.nifi.distributed.cache.client.AtomicDistributedMapCacheClient;
+import org.apache.nifi.distributed.cache.client.Deserializer;
+import org.apache.nifi.distributed.cache.client.Serializer;
+import org.apache.nifi.redis.RedisConnectionPool;
+import org.apache.nifi.redis.RedisType;
+import org.apache.nifi.redis.util.RedisAction;
+import org.apache.nifi.util.Tuple;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.ScanOptions;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+@Tags({ "redis", "distributed", "cache", "map" })
+@CapabilityDescription("An implementation of DistributedMapCacheClient that uses Redis as the backing cache. This service relies on " +
+ "the WATCH, MULTI, and EXEC commands in Redis, which are not fully supported when Redis is clustered. As a result, this service " +
+ "can only be used with a Redis Connection Pool that is configured for standalone or sentinel mode. Sentinel mode can be used to " +
+ "provide high-availability configurations.")
+public class RedisDistributedMapCacheClientService extends AbstractControllerService implements AtomicDistributedMapCacheClient<byte[]> {
+
+ public static final PropertyDescriptor REDIS_CONNECTION_POOL = new PropertyDescriptor.Builder()
+ .name("redis-connection-pool")
+ .displayName("Redis Connection Pool")
+ .identifiesControllerService(RedisConnectionPool.class)
+ .required(true)
+ .build();
+
+ static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS;
+ static {
+ final List<PropertyDescriptor> props = new ArrayList<>();
+ props.add(REDIS_CONNECTION_POOL);
+ PROPERTY_DESCRIPTORS = Collections.unmodifiableList(props);
+ }
+
+ private volatile RedisConnectionPool redisConnectionPool;
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTY_DESCRIPTORS;
+ }
+
+ @Override
+ protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
+ final List<ValidationResult> results = new ArrayList<>();
+
+ final RedisConnectionPool redisConnectionPool = validationContext.getProperty(REDIS_CONNECTION_POOL).asControllerService(RedisConnectionPool.class);
+ if (redisConnectionPool != null) {
+ final RedisType redisType = redisConnectionPool.getRedisType();
+ if (redisType != null && redisType == RedisType.CLUSTER) {
+ results.add(new ValidationResult.Builder()
+ .subject(REDIS_CONNECTION_POOL.getDisplayName())
+ .valid(false)
+ .explanation(REDIS_CONNECTION_POOL.getDisplayName()
+ + " is configured in clustered mode, and this service requires a non-clustered Redis")
+ .build());
+ }
+ }
+
+ return results;
+ }
+
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) {
+ this.redisConnectionPool = context.getProperty(REDIS_CONNECTION_POOL).asControllerService(RedisConnectionPool.class);
+ }
+
+ @OnDisabled
+ public void onDisabled() {
+ this.redisConnectionPool = null;
+ }
+
+ @Override
+ public <K, V> boolean putIfAbsent(final K key, final V value, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) throws IOException {
+ return withConnection(redisConnection -> {
+ final Tuple<byte[],byte[]> kv = serialize(key, value, keySerializer, valueSerializer);
+ return redisConnection.setNX(kv.getKey(), kv.getValue());
+ });
+ }
+
+ @Override
+ public <K, V> V getAndPutIfAbsent(final K key, final V value, final Serializer<K> keySerializer, final Serializer<V> valueSerializer, final Deserializer<V> valueDeserializer) throws IOException {
+ return withConnection(redisConnection -> {
+ final Tuple<byte[],byte[]> kv = serialize(key, value, keySerializer, valueSerializer);
+ do {
+ // start a watch on the key and retrieve the current value
+ redisConnection.watch(kv.getKey());
+ final byte[] existingValue = redisConnection.get(kv.getKey());
+
+ // start a transaction and perform the put-if-absent
+ redisConnection.multi();
+ redisConnection.setNX(kv.getKey(), kv.getValue());
+
+ // execute the transaction
+ final List<Object> results = redisConnection.exec();
+
+ // if the results list was empty, then the transaction failed (i.e. key was modified after we started watching), so keep looping to retry
+ // if the results list has results, then the transaction succeeded and it should have the result of the setNX operation
+ if (results.size() > 0) {
+ final Object firstResult = results.get(0);
+ if (firstResult instanceof Boolean) {
+ final Boolean absent = (Boolean) firstResult;
+ return absent ? null : valueDeserializer.deserialize(existingValue);
+ } else {
+ // this shouldn't really happen, but just in case there is a non-boolean result then bounce out of the loop
+ throw new IOException("Unexpected result from Redis transaction: Expected Boolean result, but got "
+ + firstResult.getClass().getName() + " with value " + firstResult.toString());
+ }
+ }
+ } while (true);
--- End diff --
Good idea, made the suggested changed in AbstractControllerService and changed this loop to use the new isEnabled() method.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122761028
--- Diff: nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java ---
@@ -0,0 +1,422 @@
+/*
+ * 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.redis.util;
+
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.redis.RedisType;
+import org.apache.nifi.util.StringUtils;
+import org.springframework.data.redis.connection.RedisClusterConfiguration;
+import org.springframework.data.redis.connection.RedisSentinelConfiguration;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import redis.clients.jedis.JedisPoolConfig;
+import redis.clients.jedis.JedisShardInfo;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+public class RedisUtils {
+
+ // These properties are shared between the connection pool controller service and the state provider, the name
+ // is purposely set to be more human-readable since that will be referenced in state-management.xml
+
+ public static final AllowableValue REDIS_MODE_STANDALONE = new AllowableValue(RedisType.STANDALONE.getDisplayName(), RedisType.STANDALONE.getDisplayName(), RedisType.STANDALONE.getDescription());
+ public static final AllowableValue REDIS_MODE_SENTINEL = new AllowableValue(RedisType.SENTINEL.getDisplayName(), RedisType.SENTINEL.getDisplayName(), RedisType.SENTINEL.getDescription());
+ public static final AllowableValue REDIS_MODE_CLUSTER = new AllowableValue(RedisType.CLUSTER.getDisplayName(), RedisType.CLUSTER.getDisplayName(), RedisType.CLUSTER.getDescription());
+
+ public static final PropertyDescriptor REDIS_MODE = new PropertyDescriptor.Builder()
+ .name("Redis Mode")
+ .displayName("Redis Mode")
+ .description("The type of Redis being communicated with - standalone, sentinel, or clustered.")
+ .allowableValues(REDIS_MODE_STANDALONE, REDIS_MODE_SENTINEL, REDIS_MODE_CLUSTER)
+ .defaultValue(REDIS_MODE_STANDALONE.getValue())
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor CONNECTION_STRING = new PropertyDescriptor.Builder()
+ .name("Connection String")
+ .displayName("Connection String")
+ .description("The connection string for Redis. In a standalone instance this value will be of the form hostname:port. " +
+ "In a sentinel instance this value will be the comma-separated list of sentinels, such as host1:port1,host2:port2,host3:port3. " +
+ "In a clustered instance this value will be the comma-separated list of cluster masters, such as host1:port,host2:port,host3:port.")
+ .required(true)
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ public static final PropertyDescriptor DATABASE = new PropertyDescriptor.Builder()
+ .name("Database Index")
+ .displayName("Database Index")
+ .description("The database index to be used by connections created from this connection pool. " +
+ "See the databases property in redis.conf, by default databases 0-15 will be available.")
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .defaultValue("0")
+ .expressionLanguageSupported(true)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor COMMUNICATION_TIMEOUT = new PropertyDescriptor.Builder()
+ .name("Communication Timeout")
+ .displayName("Communication Timeout")
+ .description("The timeout to use when attempting to communicate with Redis.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("10 seconds")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor CLUSTER_MAX_REDIRECTS = new PropertyDescriptor.Builder()
+ .name("Cluster Max Redirects")
+ .displayName("Cluster Max Redirects")
+ .description("The maximum number of redirects that can be performed when clustered.")
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor SENTINEL_MASTER = new PropertyDescriptor.Builder()
+ .name("Sentinel Master")
+ .displayName("Sentinel Master")
+ .description("The name of the sentinel master, require when Mode is set to Sentinel")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder()
+ .name("Password")
+ .displayName("Password")
+ .description("The password used to authenticate to the Redis server. See the requirepass property in redis.conf.")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .sensitive(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MAX_TOTAL = new PropertyDescriptor.Builder()
+ .name("Pool - Max Total")
+ .displayName("Pool - Max Total")
+ .description("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.")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .defaultValue("8")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MAX_IDLE = new PropertyDescriptor.Builder()
+ .name("Pool - Max Idle")
+ .displayName("Pool - Max Idle")
+ .description("The maximum number of idle connections that can be held in the pool, or a negative value if there is no limit.")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .defaultValue("8")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MIN_IDLE = new PropertyDescriptor.Builder()
+ .name("Pool - Min Idle")
+ .displayName("Pool - Min Idle")
+ .description("The target for the minimum number of idle connections to maintain in the pool. If the configured value of Min Idle is " +
+ "greater than the configured value for Max Idle, then the value of Max Idle will be used instead.")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .defaultValue("0")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_BLOCK_WHEN_EXHAUSTED = new PropertyDescriptor.Builder()
+ .name("Pool - Block When Exhausted")
+ .displayName("Pool - Block When Exhausted")
+ .description("Whether or not clients should block and wait when trying to obtain a connection from the pool when the pool has no available connections. " +
+ "Setting this to false means an error will occur immediately when a client requests a connection and none are available.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MAX_WAIT_TIME = new PropertyDescriptor.Builder()
+ .name("Pool - Max Wait Time")
+ .displayName("Pool - Max Wait Time")
+ .description("The amount of time to wait for an available connection when Block When Exhausted is set to true.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("10 seconds")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MIN_EVICTABLE_IDLE_TIME = new PropertyDescriptor.Builder()
+ .name("Pool - Min Evictable Idle Time")
+ .displayName("Pool - Min Evictable Idle Time")
+ .description("The minimum amount of time an object may sit idle in the pool before it is eligible for eviction.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("60 seconds")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TIME_BETWEEN_EVICTION_RUNS = new PropertyDescriptor.Builder()
+ .name("Pool - Time Between Eviction Runs")
+ .displayName("Pool - Time Between Eviction Runs")
+ .description("The amount of time between attempting to evict idle connections from the pool.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("30 seconds")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_NUM_TESTS_PER_EVICTION_RUN = new PropertyDescriptor.Builder()
+ .name("Pool - Num Tests Per Eviction Run")
+ .displayName("Pool - Num Tests Per Eviction Run")
+ .description("The number of connections to tests per eviction attempt. A negative value indicates to test all connections.")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .defaultValue("-1")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TEST_ON_CREATE = new PropertyDescriptor.Builder()
+ .name("Pool - Test On Create")
+ .displayName("Pool - Test On Create")
+ .description("Whether or not connections should be tested upon creation.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TEST_ON_BORROW = new PropertyDescriptor.Builder()
+ .name("Pool - Test On Borrow")
+ .displayName("Pool - Test On Borrow")
+ .description("Whether or not connections should be tested upon borrowing from the pool.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TEST_ON_RETURN = new PropertyDescriptor.Builder()
+ .name("Pool - Test On Return")
+ .displayName("Pool - Test On Return")
+ .description("Whether or not connections should be tested upon returning to the pool.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TEST_WHILE_IDLE = new PropertyDescriptor.Builder()
+ .name("Pool - Test While Idle")
+ .displayName("Pool - Test While Idle")
+ .description("Whether or not connections should be tested while idle.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .required(true)
+ .build();
+
+ public static final List<PropertyDescriptor> REDIS_CONNECTION_PROPERTY_DESCRIPTORS;
+ static {
+ final List<PropertyDescriptor> props = new ArrayList<>();
+ props.add(RedisUtils.REDIS_MODE);
+ props.add(RedisUtils.CONNECTION_STRING);
+ props.add(RedisUtils.DATABASE);
+ props.add(RedisUtils.COMMUNICATION_TIMEOUT);
+ props.add(RedisUtils.CLUSTER_MAX_REDIRECTS);
+ props.add(RedisUtils.SENTINEL_MASTER);
+ props.add(RedisUtils.PASSWORD);
+ props.add(RedisUtils.POOL_MAX_TOTAL);
+ props.add(RedisUtils.POOL_MAX_IDLE);
+ props.add(RedisUtils.POOL_MIN_IDLE);
+ props.add(RedisUtils.POOL_BLOCK_WHEN_EXHAUSTED);
+ props.add(RedisUtils.POOL_MAX_WAIT_TIME);
+ props.add(RedisUtils.POOL_MIN_EVICTABLE_IDLE_TIME);
+ props.add(RedisUtils.POOL_TIME_BETWEEN_EVICTION_RUNS);
+ props.add(RedisUtils.POOL_NUM_TESTS_PER_EVICTION_RUN);
+ props.add(RedisUtils.POOL_TEST_ON_CREATE);
+ props.add(RedisUtils.POOL_TEST_ON_BORROW);
+ props.add(RedisUtils.POOL_TEST_ON_RETURN);
+ props.add(RedisUtils.POOL_TEST_WHILE_IDLE);
+ REDIS_CONNECTION_PROPERTY_DESCRIPTORS = Collections.unmodifiableList(props);
+ }
+
+
+ public static JedisConnectionFactory createConnectionFactory(final PropertyContext context, final ComponentLog logger) {
+ final String redisMode = context.getProperty(RedisUtils.REDIS_MODE).getValue();
+ final String connectionString = context.getProperty(RedisUtils.CONNECTION_STRING).evaluateAttributeExpressions().getValue();
+ final Integer dbIndex = context.getProperty(RedisUtils.DATABASE).evaluateAttributeExpressions().asInteger();
+ final String password = context.getProperty(RedisUtils.PASSWORD).evaluateAttributeExpressions().getValue();
+ final Integer timeout = context.getProperty(RedisUtils.COMMUNICATION_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue();
+ final JedisPoolConfig poolConfig = createJedisPoolConfig(context);
+
+ JedisConnectionFactory connectionFactory;
+
+ if (RedisUtils.REDIS_MODE_STANDALONE.getValue().equals(redisMode)) {
+ final JedisShardInfo jedisShardInfo = createJedisShardInfo(connectionString, timeout, password);
+
+ logger.info("Connecting to Redis in standalone mode at " + connectionString);
+ connectionFactory = new JedisConnectionFactory(jedisShardInfo);
+
+ } else if (RedisUtils.REDIS_MODE_SENTINEL.getValue().equals(redisMode)) {
+ final String[] sentinels = connectionString.split("[,]");
+ final String sentinelMaster = context.getProperty(RedisUtils.SENTINEL_MASTER).evaluateAttributeExpressions().getValue();
+ final RedisSentinelConfiguration sentinelConfiguration = new RedisSentinelConfiguration(sentinelMaster, new HashSet<>(Arrays.asList(sentinels)));
+ final JedisShardInfo jedisShardInfo = createJedisShardInfo(sentinels[0], timeout, password);
+
+ logger.info("Connecting to Redis in sentinel mode...");
+ logger.info("Redis master = " + sentinelMaster);
+
+ for (final String sentinel : sentinels) {
+ logger.info("Redis sentinel at " + sentinel);
+ }
+
+ connectionFactory = new JedisConnectionFactory(sentinelConfiguration, poolConfig);
+ connectionFactory.setShardInfo(jedisShardInfo);
+
+ } else {
+ final String[] clusterNodes = connectionString.split("[,]");
+ final Integer maxRedirects = context.getProperty(RedisUtils.CLUSTER_MAX_REDIRECTS).asInteger();
+
+ final RedisClusterConfiguration clusterConfiguration = new RedisClusterConfiguration(Arrays.asList(clusterNodes));
+ clusterConfiguration.setMaxRedirects(maxRedirects);
+
+ logger.info("Connecting to Redis in clustered mode...");
+ for (final String clusterNode : clusterNodes) {
+ logger.info("Redis cluster node at " + clusterNode);
+ }
+
+ connectionFactory = new JedisConnectionFactory(clusterConfiguration, poolConfig);
+ }
+
+ connectionFactory.setUsePool(true);
+ connectionFactory.setPoolConfig(poolConfig);
+ connectionFactory.setDatabase(dbIndex);
+ connectionFactory.setTimeout(timeout);
+
+ if (!StringUtils.isBlank(password)) {
+ connectionFactory.setPassword(password);
+ }
+
+ // need to call this to initialize the pool/connections
+ connectionFactory.afterPropertiesSet();
+ return connectionFactory;
+ }
+
+ private static JedisShardInfo createJedisShardInfo(final String hostAndPort, final Integer timeout, final String password) {
+ final String[] hostAndPortSplit = hostAndPort.split("[:]");
+ final String host = hostAndPortSplit[0].trim();
+ final Integer port = Integer.parseInt(hostAndPortSplit[1].trim());
+
+ final JedisShardInfo jedisShardInfo = new JedisShardInfo(host, port);
+ jedisShardInfo.setConnectionTimeout(timeout);
+ jedisShardInfo.setSoTimeout(timeout);
+
+ if (!StringUtils.isEmpty(password)) {
+ jedisShardInfo.setPassword(password);
+ }
+
+ return jedisShardInfo;
+ }
+
+ private static JedisPoolConfig createJedisPoolConfig(final PropertyContext context) {
+ final JedisPoolConfig poolConfig = new JedisPoolConfig();
+ poolConfig.setMaxTotal(context.getProperty(RedisUtils.POOL_MAX_TOTAL).asInteger());
+ poolConfig.setMaxIdle(context.getProperty(RedisUtils.POOL_MAX_IDLE).asInteger());
+ poolConfig.setMinIdle(context.getProperty(RedisUtils.POOL_MIN_IDLE).asInteger());
+ poolConfig.setBlockWhenExhausted(context.getProperty(RedisUtils.POOL_BLOCK_WHEN_EXHAUSTED).asBoolean());
+ poolConfig.setMaxWaitMillis(context.getProperty(RedisUtils.POOL_MAX_WAIT_TIME).asTimePeriod(TimeUnit.MILLISECONDS));
+ poolConfig.setMinEvictableIdleTimeMillis(context.getProperty(RedisUtils.POOL_MIN_EVICTABLE_IDLE_TIME).asTimePeriod(TimeUnit.MILLISECONDS));
+ poolConfig.setTimeBetweenEvictionRunsMillis(context.getProperty(RedisUtils.POOL_TIME_BETWEEN_EVICTION_RUNS).asTimePeriod(TimeUnit.MILLISECONDS));
+ poolConfig.setNumTestsPerEvictionRun(context.getProperty(RedisUtils.POOL_NUM_TESTS_PER_EVICTION_RUN).asInteger());
+ poolConfig.setTestOnCreate(context.getProperty(RedisUtils.POOL_TEST_ON_CREATE).asBoolean());
+ poolConfig.setTestOnBorrow(context.getProperty(RedisUtils.POOL_TEST_ON_BORROW).asBoolean());
+ poolConfig.setTestOnReturn(context.getProperty(RedisUtils.POOL_TEST_ON_RETURN).asBoolean());
+ poolConfig.setTestWhileIdle(context.getProperty(RedisUtils.POOL_TEST_WHILE_IDLE).asBoolean());
+ return poolConfig;
+ }
+
+ public static Collection<ValidationResult> validate(ValidationContext validationContext) {
+ final List<ValidationResult> results = new ArrayList<>();
+
+ final String redisMode = validationContext.getProperty(RedisUtils.REDIS_MODE).getValue();
+ final String connectionString = validationContext.getProperty(RedisUtils.CONNECTION_STRING).evaluateAttributeExpressions().getValue();
+ final Integer dbIndex = validationContext.getProperty(RedisUtils.DATABASE).evaluateAttributeExpressions().asInteger();
+
+ if (StringUtils.isBlank(connectionString)) {
+ results.add(new ValidationResult.Builder()
+ .subject(RedisUtils.CONNECTION_STRING.getDisplayName())
+ .valid(false)
+ .explanation("Connection String cannot be blank")
+ .build());
+ } else if (RedisUtils.REDIS_MODE_STANDALONE.getValue().equals(redisMode)) {
+ final String[] hostAndPort = connectionString.split("[:]");
+ if (hostAndPort == null || hostAndPort.length != 2 || !isInteger(hostAndPort[1])) {
+ results.add(new ValidationResult.Builder()
+ .subject(RedisUtils.CONNECTION_STRING.getDisplayName())
+ .input(connectionString)
+ .valid(false)
+ .explanation("Standalone Connection String must be in the form host:port")
+ .build());
+ }
+ } else {
+ for (final String connection : connectionString.split("[,]")) {
+ final String[] hostAndPort = connection.split("[:]");
+ if (hostAndPort == null || hostAndPort.length != 2 || !isInteger(hostAndPort[1])) {
+ results.add(new ValidationResult.Builder()
+ .subject(RedisUtils.CONNECTION_STRING.getDisplayName())
+ .input(connection)
+ .valid(false)
+ .explanation("Connection String must be in the form host:port")
--- End diff --
Should probably indicate that connection string must be in form of host:port,host:port,host:port,... instead of just "host:port" because it may imply to users that only a single host & port can be provided.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/nifi/pull/1918
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122836730
--- Diff: nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/service/RedisDistributedMapCacheClientService.java ---
@@ -0,0 +1,313 @@
+/*
+ * 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.redis.service;
+
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnDisabled;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.distributed.cache.client.AtomicCacheEntry;
+import org.apache.nifi.distributed.cache.client.AtomicDistributedMapCacheClient;
+import org.apache.nifi.distributed.cache.client.Deserializer;
+import org.apache.nifi.distributed.cache.client.Serializer;
+import org.apache.nifi.redis.RedisConnectionPool;
+import org.apache.nifi.redis.RedisType;
+import org.apache.nifi.redis.util.RedisAction;
+import org.apache.nifi.util.Tuple;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.ScanOptions;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+@Tags({ "redis", "distributed", "cache", "map" })
+@CapabilityDescription("An implementation of DistributedMapCacheClient that uses Redis as the backing cache. This service relies on " +
+ "the WATCH, MULTI, and EXEC commands in Redis, which are not fully supported when Redis is clustered. As a result, this service " +
+ "can only be used with a Redis Connection Pool that is configured for standalone or sentinel mode. Sentinel mode can be used to " +
+ "provide high-availability configurations.")
+public class RedisDistributedMapCacheClientService extends AbstractControllerService implements AtomicDistributedMapCacheClient<byte[]> {
+
+ public static final PropertyDescriptor REDIS_CONNECTION_POOL = new PropertyDescriptor.Builder()
+ .name("redis-connection-pool")
+ .displayName("Redis Connection Pool")
+ .identifiesControllerService(RedisConnectionPool.class)
+ .required(true)
+ .build();
+
+ static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS;
+ static {
+ final List<PropertyDescriptor> props = new ArrayList<>();
+ props.add(REDIS_CONNECTION_POOL);
+ PROPERTY_DESCRIPTORS = Collections.unmodifiableList(props);
+ }
+
+ private volatile RedisConnectionPool redisConnectionPool;
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTY_DESCRIPTORS;
+ }
+
+ @Override
+ protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
+ final List<ValidationResult> results = new ArrayList<>();
+
+ final RedisConnectionPool redisConnectionPool = validationContext.getProperty(REDIS_CONNECTION_POOL).asControllerService(RedisConnectionPool.class);
+ if (redisConnectionPool != null) {
+ final RedisType redisType = redisConnectionPool.getRedisType();
+ if (redisType != null && redisType == RedisType.CLUSTER) {
+ results.add(new ValidationResult.Builder()
+ .subject(REDIS_CONNECTION_POOL.getDisplayName())
+ .valid(false)
+ .explanation(REDIS_CONNECTION_POOL.getDisplayName()
+ + " is configured in clustered mode, and this service requires a non-clustered Redis")
+ .build());
+ }
+ }
+
+ return results;
+ }
+
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) {
+ this.redisConnectionPool = context.getProperty(REDIS_CONNECTION_POOL).asControllerService(RedisConnectionPool.class);
+ }
+
+ @OnDisabled
+ public void onDisabled() {
+ this.redisConnectionPool = null;
+ }
+
+ @Override
+ public <K, V> boolean putIfAbsent(final K key, final V value, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) throws IOException {
+ return withConnection(redisConnection -> {
+ final Tuple<byte[],byte[]> kv = serialize(key, value, keySerializer, valueSerializer);
+ return redisConnection.setNX(kv.getKey(), kv.getValue());
+ });
+ }
+
+ @Override
+ public <K, V> V getAndPutIfAbsent(final K key, final V value, final Serializer<K> keySerializer, final Serializer<V> valueSerializer, final Deserializer<V> valueDeserializer) throws IOException {
+ return withConnection(redisConnection -> {
+ final Tuple<byte[],byte[]> kv = serialize(key, value, keySerializer, valueSerializer);
+ do {
+ // start a watch on the key and retrieve the current value
+ redisConnection.watch(kv.getKey());
+ final byte[] existingValue = redisConnection.get(kv.getKey());
+
+ // start a transaction and perform the put-if-absent
+ redisConnection.multi();
+ redisConnection.setNX(kv.getKey(), kv.getValue());
+
+ // execute the transaction
+ final List<Object> results = redisConnection.exec();
+
+ // if the results list was empty, then the transaction failed (i.e. key was modified after we started watching), so keep looping to retry
+ // if the results list has results, then the transaction succeeded and it should have the result of the setNX operation
+ if (results.size() > 0) {
+ final Object firstResult = results.get(0);
+ if (firstResult instanceof Boolean) {
+ final Boolean absent = (Boolean) firstResult;
+ return absent ? null : valueDeserializer.deserialize(existingValue);
+ } else {
+ // this shouldn't really happen, but just in case there is a non-boolean result then bounce out of the loop
+ throw new IOException("Unexpected result from Redis transaction: Expected Boolean result, but got "
+ + firstResult.getClass().getName() + " with value " + firstResult.toString());
+ }
+ }
+ } while (true);
+ });
+ }
+
+ @Override
+ public <K> boolean containsKey(final K key, final Serializer<K> keySerializer) throws IOException {
+ return withConnection(redisConnection -> {
+ final byte[] k = serialize(key, keySerializer);
+ return redisConnection.exists(k);
+ });
+ }
+
+ @Override
+ public <K, V> void put(final K key, final V value, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) throws IOException {
+ withConnection(redisConnection -> {
+ final Tuple<byte[],byte[]> kv = serialize(key, value, keySerializer, valueSerializer);
+ redisConnection.set(kv.getKey(), kv.getValue());
+ return null;
+ });
+ }
+
+ @Override
+ public <K, V> V get(final K key, final Serializer<K> keySerializer, final Deserializer<V> valueDeserializer) throws IOException {
+ return withConnection(redisConnection -> {
+ final byte[] k = serialize(key, keySerializer);
+ final byte[] v = redisConnection.get(k);
+ return valueDeserializer.deserialize(v);
+ });
+ }
+
+ @Override
+ public void close() throws IOException {
+ // nothing to do
+ }
+
+ @Override
+ public <K> boolean remove(final K key, final Serializer<K> keySerializer) throws IOException {
+ return withConnection(redisConnection -> {
+ final byte[] k = serialize(key, keySerializer);
+ final long numRemoved = redisConnection.del(k);
+ return numRemoved > 0;
+ });
+ }
+
+ @Override
+ public long removeByPattern(final String regex) throws IOException {
+ return withConnection(redisConnection -> {
+ final List<byte[]> allKeys = new ArrayList<>();
--- End diff --
That makes sense, I hadn't thought about the heap usage... Since the cursor returned from scan is already paging through smaller key ranges at a time, I changed the logic to perform a delete for batches of 1000 and set the hint for the scan to return 1000 keys at a time.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122837003
--- Diff: nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java ---
@@ -0,0 +1,271 @@
+/*
+ * 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.redis.state;
+
+import org.apache.nifi.components.AbstractConfigurableComponent;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.components.state.StateProvider;
+import org.apache.nifi.components.state.StateProviderInitializationContext;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.redis.RedisType;
+import org.apache.nifi.redis.util.RedisAction;
+import org.apache.nifi.redis.util.RedisUtils;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A StateProvider backed by Redis.
+ */
+public class RedisStateProvider extends AbstractConfigurableComponent implements StateProvider {
+
+ static final int ENCODING_VERSION = 1;
+
+ private String identifier;
+ private PropertyContext context;
+ private ComponentLog logger;
+
+ private volatile boolean enabled;
+ private volatile JedisConnectionFactory connectionFactory;
+
+ private final RedisStateMapSerDe serDe = new RedisStateMapJsonSerDe();
+
+ @Override
+ public final void initialize(final StateProviderInitializationContext context) throws IOException {
+ this.context = context;
+ this.identifier = context.getIdentifier();
+ this.logger = context.getLogger();
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return RedisUtils.REDIS_CONNECTION_PROPERTY_DESCRIPTORS;
+ }
+
+ @Override
+ protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
+ final List<ValidationResult> results = new ArrayList<>(RedisUtils.validate(validationContext));
+
+ final RedisType redisType = RedisType.fromDisplayName(validationContext.getProperty(RedisUtils.REDIS_MODE).getValue());
+ if (redisType != null && redisType == RedisType.CLUSTER) {
+ results.add(new ValidationResult.Builder()
+ .subject(RedisUtils.REDIS_MODE.getDisplayName())
+ .valid(false)
+ .explanation(RedisUtils.REDIS_MODE.getDisplayName()
+ + " is configured in clustered mode, and this service requires a non-clustered Redis")
+ .build());
+ }
+
+ return results;
+ }
+
+ @Override
+ public String getIdentifier() {
+ return identifier;
+ }
+
+ @Override
+ public void enable() {
+ enabled = true;
+ }
+
+ @Override
+ public void disable() {
+ enabled = false;
+ }
+
+ @Override
+ public boolean isEnabled() {
+ return enabled;
+ }
+
+ @Override
+ public void shutdown() {
+ if (connectionFactory != null) {
+ connectionFactory.destroy();
+ connectionFactory = null;
+ }
+ }
+
+ @Override
+ public void setState(final Map<String, String> state, final String componentId) throws IOException {
+ verifyEnabled();
+
+ final StateMap currStateMap = getState(componentId);
+
+ int attempted = 0;
+ boolean updated = false;
+
+ while (!updated && attempted < 20) {
+ updated = replace(currStateMap, state, componentId, true);
+ attempted++;
+ }
+
+ if (!updated) {
+ throw new IOException("Unable to update state due to concurrent modifications");
+ }
+ }
+
+ @Override
+ public StateMap getState(final String componentId) throws IOException {
+ return withConnection(redisConnection -> {
+ final byte[] key = getComponentPath(componentId).getBytes(StandardCharsets.UTF_8);
+ final byte[] value = redisConnection.get(key);
+
+ final RedisStateMap stateMap = serDe.deserialize(value);
+ if (stateMap == null) {
+ return new RedisStateMap.Builder().encodingVersion(ENCODING_VERSION).build();
+ } else {
+ return stateMap;
+ }
+ });
+ }
+
+ @Override
+ public boolean replace(final StateMap oldValue, final Map<String, String> newValue, final String componentId) throws IOException {
+ return replace(oldValue, newValue, componentId, false);
+ }
+
+ private boolean replace(final StateMap oldValue, final Map<String, String> newValue, final String componentId, final boolean allowReplaceMissing) throws IOException {
+ return withConnection(redisConnection -> {
+
+ boolean replaced = false;
+
+ // start a watch on the key and retrieve the current value
+ final byte[] key = getComponentPath(componentId).getBytes(StandardCharsets.UTF_8);
+ redisConnection.watch(key);
+
+ final long prevVersion = oldValue == null ? -1L : oldValue.getVersion();
+
+ final byte[] currValue = redisConnection.get(key);
+ final RedisStateMap currStateMap = serDe.deserialize(currValue);
+ final long currVersion = currStateMap == null ? -1L : currStateMap.getVersion();
+
+ // the replace API expects that you can't call replace on a non-existing value, so unwatch and return
+ if (!allowReplaceMissing && currVersion == -1) {
+ redisConnection.unwatch();
+ return false;
+ }
+
+ // start a transaction
+ redisConnection.multi();
+
+ // compare-and-set
+ if (prevVersion == currVersion) {
+ // build the new RedisStateMap incrementing the version, using latest encoding, and using the passed in values
+ final RedisStateMap newStateMap = new RedisStateMap.Builder()
+ .version(currVersion + 1)
+ .encodingVersion(ENCODING_VERSION)
+ .stateValues(newValue)
+ .build();
+
+ // if we use set(k, newVal) then the results list will always have size == 0 b/c when convertPipelineAndTxResults is set to true,
+ // status responses like "OK" are skipped over, so by using getSet we can rely on the results list to know if the transaction succeeded
+ redisConnection.getSet(key, serDe.serialize(newStateMap));
+ }
+
+ // execute the transaction
+ final List<Object> results = redisConnection.exec();
+
+ // if we have a result then the replace succeeded
+ if (results.size() > 0) {
+ replaced = true;
+ }
+
+ return replaced;
+ });
+ }
+
+ @Override
+ public void clear(final String componentId) throws IOException {
+ final StateMap currStateMap = getState(componentId);
+
+ int attempted = 0;
+ boolean updated = false;
+
+ while (!updated && attempted < 20) {
+ updated = replace(currStateMap, Collections.emptyMap(), componentId, true);
+ attempted++;
+ }
+
+ if (!updated) {
+ throw new IOException("Unable to update state due to concurrent modifications");
+ }
+ }
+
+ @Override
+ public void onComponentRemoved(final String componentId) throws IOException {
+ withConnection(redisConnection -> {
+ final byte[] key = getComponentPath(componentId).getBytes(StandardCharsets.UTF_8);
+ redisConnection.del(key);
+ return true;
+ });
+ }
+
+ @Override
+ public Scope[] getSupportedScopes() {
+ return new Scope[] {Scope.CLUSTER};
+ }
+
+ private String getComponentPath(final String componentId) {
--- End diff --
I like that idea, I added a property called "Key Prefix" since with Redis it isn't as much of a path as it is with ZooKeeper. Even without the Key Prefix they could still use the Database Index to segregate the keys, but I still like making the prefix configurable.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #1918: NIFI-4061 Add a RedisStateProvider
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on the issue:
https://github.com/apache/nifi/pull/1918
@markap14 thanks for reviewing! I pushed up a commit that should address your initial comments and I tried to write a reply to each one so you have an idea of what I did to address it. Let me know if there is anything else.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122836798
--- Diff: nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateMapJsonSerDe.java ---
@@ -0,0 +1,103 @@
+/*
+ * 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.redis.state;
+
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonToken;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * A RedisStateMapSerDe that uses JSON as the underlying representation.
+ */
+public class RedisStateMapJsonSerDe implements RedisStateMapSerDe {
+
+ public static final String FIELD_VERSION = "version";
+ public static final String FIELD_ENCODING = "encodingVersion";
+ public static final String FIELD_STATE_VALUES = "stateValues";
+
+ private final JsonFactory jsonFactory = new JsonFactory();
+
+ @Override
+ public byte[] serialize(final RedisStateMap stateMap) throws IOException {
+ if (stateMap == null) {
+ return null;
+ }
+
+ try (final ByteArrayOutputStream out = new ByteArrayOutputStream()) {
+ final JsonGenerator jsonGenerator = jsonFactory.createGenerator(out);
+ jsonGenerator.writeStartObject();
+ jsonGenerator.writeNumberField(FIELD_VERSION, stateMap.getVersion());
+ jsonGenerator.writeNumberField(FIELD_ENCODING, stateMap.getEncodingVersion());
+
+ jsonGenerator.writeObjectFieldStart(FIELD_STATE_VALUES);
+ for (Map.Entry<String,String> entry : stateMap.toMap().entrySet()) {
+ jsonGenerator.writeStringField(entry.getKey(), entry.getValue());
+ }
+ jsonGenerator.writeEndObject();
+
+ jsonGenerator.writeEndObject();
+ jsonGenerator.flush();
+
+ return out.toByteArray();
+ }
+ }
+
+ @Override
+ public RedisStateMap deserialize(final byte[] data) throws IOException {
+ if (data == null || data.length == 0) {
+ return null;
+ }
+
+ final RedisStateMap.Builder builder = new RedisStateMap.Builder();
+
+ try (final JsonParser jsonParser = jsonFactory.createParser(data)) {
+ while (jsonParser.nextToken() != JsonToken.END_OBJECT) {
--- End diff --
I knew that code should have been simpler :) updated it to use the readValueAsTree
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122728877
--- Diff: nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/service/RedisConnectionPoolService.java ---
@@ -0,0 +1,95 @@
+/*
+ * 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.redis.service;
+
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnDisabled;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.redis.RedisConnectionPool;
+import org.apache.nifi.redis.RedisType;
+import org.apache.nifi.redis.util.RedisUtils;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+@Tags({"redis", "cache"})
+@CapabilityDescription("A service that provides connections to Redis.")
+public class RedisConnectionPoolService extends AbstractControllerService implements RedisConnectionPool {
+
+ private volatile PropertyContext context;
+ private volatile RedisType redisType;
+ private volatile JedisConnectionFactory connectionFactory;
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return RedisUtils.REDIS_CONNECTION_PROPERTY_DESCRIPTORS;
+ }
+
+ @Override
+ protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
+ final List<ValidationResult> results = new ArrayList<>(RedisUtils.validate(validationContext));
--- End diff --
I wonder if it makes sense to update RedisUtils to return a List instead of a Collection and just return that here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122836075
--- Diff: nifi-api/src/main/java/org/apache/nifi/context/PropertyContext.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.context;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.PropertyValue;
+
+/**
+ * A context for retrieving a PropertyValue from a PropertyDescriptor.
+ */
+public interface PropertyContext {
--- End diff --
Good call. I added getAllProperties since there a lot of the contexts already had a getProperties that returned Map<PropertyDescriptor,String> and would conflict.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122731744
--- Diff: nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/service/RedisDistributedMapCacheClientService.java ---
@@ -0,0 +1,313 @@
+/*
+ * 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.redis.service;
+
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnDisabled;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.distributed.cache.client.AtomicCacheEntry;
+import org.apache.nifi.distributed.cache.client.AtomicDistributedMapCacheClient;
+import org.apache.nifi.distributed.cache.client.Deserializer;
+import org.apache.nifi.distributed.cache.client.Serializer;
+import org.apache.nifi.redis.RedisConnectionPool;
+import org.apache.nifi.redis.RedisType;
+import org.apache.nifi.redis.util.RedisAction;
+import org.apache.nifi.util.Tuple;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.ScanOptions;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+@Tags({ "redis", "distributed", "cache", "map" })
+@CapabilityDescription("An implementation of DistributedMapCacheClient that uses Redis as the backing cache. This service relies on " +
+ "the WATCH, MULTI, and EXEC commands in Redis, which are not fully supported when Redis is clustered. As a result, this service " +
+ "can only be used with a Redis Connection Pool that is configured for standalone or sentinel mode. Sentinel mode can be used to " +
+ "provide high-availability configurations.")
+public class RedisDistributedMapCacheClientService extends AbstractControllerService implements AtomicDistributedMapCacheClient<byte[]> {
+
+ public static final PropertyDescriptor REDIS_CONNECTION_POOL = new PropertyDescriptor.Builder()
+ .name("redis-connection-pool")
+ .displayName("Redis Connection Pool")
+ .identifiesControllerService(RedisConnectionPool.class)
+ .required(true)
+ .build();
+
+ static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS;
+ static {
+ final List<PropertyDescriptor> props = new ArrayList<>();
+ props.add(REDIS_CONNECTION_POOL);
+ PROPERTY_DESCRIPTORS = Collections.unmodifiableList(props);
+ }
+
+ private volatile RedisConnectionPool redisConnectionPool;
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTY_DESCRIPTORS;
+ }
+
+ @Override
+ protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
+ final List<ValidationResult> results = new ArrayList<>();
+
+ final RedisConnectionPool redisConnectionPool = validationContext.getProperty(REDIS_CONNECTION_POOL).asControllerService(RedisConnectionPool.class);
+ if (redisConnectionPool != null) {
+ final RedisType redisType = redisConnectionPool.getRedisType();
+ if (redisType != null && redisType == RedisType.CLUSTER) {
+ results.add(new ValidationResult.Builder()
+ .subject(REDIS_CONNECTION_POOL.getDisplayName())
+ .valid(false)
+ .explanation(REDIS_CONNECTION_POOL.getDisplayName()
+ + " is configured in clustered mode, and this service requires a non-clustered Redis")
+ .build());
+ }
+ }
+
+ return results;
+ }
+
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) {
+ this.redisConnectionPool = context.getProperty(REDIS_CONNECTION_POOL).asControllerService(RedisConnectionPool.class);
+ }
+
+ @OnDisabled
+ public void onDisabled() {
+ this.redisConnectionPool = null;
+ }
+
+ @Override
+ public <K, V> boolean putIfAbsent(final K key, final V value, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) throws IOException {
+ return withConnection(redisConnection -> {
+ final Tuple<byte[],byte[]> kv = serialize(key, value, keySerializer, valueSerializer);
+ return redisConnection.setNX(kv.getKey(), kv.getValue());
+ });
+ }
+
+ @Override
+ public <K, V> V getAndPutIfAbsent(final K key, final V value, final Serializer<K> keySerializer, final Serializer<V> valueSerializer, final Deserializer<V> valueDeserializer) throws IOException {
+ return withConnection(redisConnection -> {
+ final Tuple<byte[],byte[]> kv = serialize(key, value, keySerializer, valueSerializer);
+ do {
+ // start a watch on the key and retrieve the current value
+ redisConnection.watch(kv.getKey());
+ final byte[] existingValue = redisConnection.get(kv.getKey());
+
+ // start a transaction and perform the put-if-absent
+ redisConnection.multi();
+ redisConnection.setNX(kv.getKey(), kv.getValue());
+
+ // execute the transaction
+ final List<Object> results = redisConnection.exec();
+
+ // if the results list was empty, then the transaction failed (i.e. key was modified after we started watching), so keep looping to retry
+ // if the results list has results, then the transaction succeeded and it should have the result of the setNX operation
+ if (results.size() > 0) {
+ final Object firstResult = results.get(0);
+ if (firstResult instanceof Boolean) {
+ final Boolean absent = (Boolean) firstResult;
+ return absent ? null : valueDeserializer.deserialize(existingValue);
+ } else {
+ // this shouldn't really happen, but just in case there is a non-boolean result then bounce out of the loop
+ throw new IOException("Unexpected result from Redis transaction: Expected Boolean result, but got "
+ + firstResult.getClass().getName() + " with value " + firstResult.toString());
+ }
+ }
+ } while (true);
--- End diff --
Rather than doing this `while (true)` we should probably do this while the Controller Service is still enabled. Would involve having an @OnEnabled method that sets a volatile boolean to enabled = true and an @OnDisabled method that sets the value to false... I thought we had this in AbstractControllerService already but it looks like there is no `boolean isEnabled()` method. Probably worth just adding that to AbstractControllerService and call it here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122836349
--- Diff: nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/service/RedisConnectionPoolService.java ---
@@ -0,0 +1,95 @@
+/*
+ * 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.redis.service;
+
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnDisabled;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.redis.RedisConnectionPool;
+import org.apache.nifi.redis.RedisType;
+import org.apache.nifi.redis.util.RedisUtils;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+@Tags({"redis", "cache"})
+@CapabilityDescription("A service that provides connections to Redis.")
+public class RedisConnectionPoolService extends AbstractControllerService implements RedisConnectionPool {
+
+ private volatile PropertyContext context;
+ private volatile RedisType redisType;
+ private volatile JedisConnectionFactory connectionFactory;
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return RedisUtils.REDIS_CONNECTION_PROPERTY_DESCRIPTORS;
+ }
+
+ @Override
+ protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
+ final List<ValidationResult> results = new ArrayList<>(RedisUtils.validate(validationContext));
--- End diff --
Good call, made that change.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122837047
--- Diff: nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java ---
@@ -0,0 +1,422 @@
+/*
+ * 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.redis.util;
+
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.redis.RedisType;
+import org.apache.nifi.util.StringUtils;
+import org.springframework.data.redis.connection.RedisClusterConfiguration;
+import org.springframework.data.redis.connection.RedisSentinelConfiguration;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import redis.clients.jedis.JedisPoolConfig;
+import redis.clients.jedis.JedisShardInfo;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+public class RedisUtils {
+
+ // These properties are shared between the connection pool controller service and the state provider, the name
+ // is purposely set to be more human-readable since that will be referenced in state-management.xml
+
+ public static final AllowableValue REDIS_MODE_STANDALONE = new AllowableValue(RedisType.STANDALONE.getDisplayName(), RedisType.STANDALONE.getDisplayName(), RedisType.STANDALONE.getDescription());
+ public static final AllowableValue REDIS_MODE_SENTINEL = new AllowableValue(RedisType.SENTINEL.getDisplayName(), RedisType.SENTINEL.getDisplayName(), RedisType.SENTINEL.getDescription());
+ public static final AllowableValue REDIS_MODE_CLUSTER = new AllowableValue(RedisType.CLUSTER.getDisplayName(), RedisType.CLUSTER.getDisplayName(), RedisType.CLUSTER.getDescription());
+
+ public static final PropertyDescriptor REDIS_MODE = new PropertyDescriptor.Builder()
+ .name("Redis Mode")
+ .displayName("Redis Mode")
+ .description("The type of Redis being communicated with - standalone, sentinel, or clustered.")
+ .allowableValues(REDIS_MODE_STANDALONE, REDIS_MODE_SENTINEL, REDIS_MODE_CLUSTER)
+ .defaultValue(REDIS_MODE_STANDALONE.getValue())
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor CONNECTION_STRING = new PropertyDescriptor.Builder()
+ .name("Connection String")
+ .displayName("Connection String")
+ .description("The connection string for Redis. In a standalone instance this value will be of the form hostname:port. " +
+ "In a sentinel instance this value will be the comma-separated list of sentinels, such as host1:port1,host2:port2,host3:port3. " +
+ "In a clustered instance this value will be the comma-separated list of cluster masters, such as host1:port,host2:port,host3:port.")
+ .required(true)
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ public static final PropertyDescriptor DATABASE = new PropertyDescriptor.Builder()
+ .name("Database Index")
+ .displayName("Database Index")
+ .description("The database index to be used by connections created from this connection pool. " +
+ "See the databases property in redis.conf, by default databases 0-15 will be available.")
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .defaultValue("0")
+ .expressionLanguageSupported(true)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor COMMUNICATION_TIMEOUT = new PropertyDescriptor.Builder()
+ .name("Communication Timeout")
+ .displayName("Communication Timeout")
+ .description("The timeout to use when attempting to communicate with Redis.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("10 seconds")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor CLUSTER_MAX_REDIRECTS = new PropertyDescriptor.Builder()
+ .name("Cluster Max Redirects")
+ .displayName("Cluster Max Redirects")
+ .description("The maximum number of redirects that can be performed when clustered.")
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor SENTINEL_MASTER = new PropertyDescriptor.Builder()
+ .name("Sentinel Master")
+ .displayName("Sentinel Master")
+ .description("The name of the sentinel master, require when Mode is set to Sentinel")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder()
+ .name("Password")
+ .displayName("Password")
+ .description("The password used to authenticate to the Redis server. See the requirepass property in redis.conf.")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .sensitive(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MAX_TOTAL = new PropertyDescriptor.Builder()
+ .name("Pool - Max Total")
+ .displayName("Pool - Max Total")
+ .description("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.")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .defaultValue("8")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MAX_IDLE = new PropertyDescriptor.Builder()
+ .name("Pool - Max Idle")
+ .displayName("Pool - Max Idle")
+ .description("The maximum number of idle connections that can be held in the pool, or a negative value if there is no limit.")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .defaultValue("8")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MIN_IDLE = new PropertyDescriptor.Builder()
+ .name("Pool - Min Idle")
+ .displayName("Pool - Min Idle")
+ .description("The target for the minimum number of idle connections to maintain in the pool. If the configured value of Min Idle is " +
+ "greater than the configured value for Max Idle, then the value of Max Idle will be used instead.")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .defaultValue("0")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_BLOCK_WHEN_EXHAUSTED = new PropertyDescriptor.Builder()
+ .name("Pool - Block When Exhausted")
+ .displayName("Pool - Block When Exhausted")
+ .description("Whether or not clients should block and wait when trying to obtain a connection from the pool when the pool has no available connections. " +
+ "Setting this to false means an error will occur immediately when a client requests a connection and none are available.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MAX_WAIT_TIME = new PropertyDescriptor.Builder()
+ .name("Pool - Max Wait Time")
+ .displayName("Pool - Max Wait Time")
+ .description("The amount of time to wait for an available connection when Block When Exhausted is set to true.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("10 seconds")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MIN_EVICTABLE_IDLE_TIME = new PropertyDescriptor.Builder()
+ .name("Pool - Min Evictable Idle Time")
+ .displayName("Pool - Min Evictable Idle Time")
+ .description("The minimum amount of time an object may sit idle in the pool before it is eligible for eviction.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("60 seconds")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TIME_BETWEEN_EVICTION_RUNS = new PropertyDescriptor.Builder()
+ .name("Pool - Time Between Eviction Runs")
+ .displayName("Pool - Time Between Eviction Runs")
+ .description("The amount of time between attempting to evict idle connections from the pool.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("30 seconds")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_NUM_TESTS_PER_EVICTION_RUN = new PropertyDescriptor.Builder()
+ .name("Pool - Num Tests Per Eviction Run")
+ .displayName("Pool - Num Tests Per Eviction Run")
+ .description("The number of connections to tests per eviction attempt. A negative value indicates to test all connections.")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .defaultValue("-1")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TEST_ON_CREATE = new PropertyDescriptor.Builder()
+ .name("Pool - Test On Create")
+ .displayName("Pool - Test On Create")
+ .description("Whether or not connections should be tested upon creation.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TEST_ON_BORROW = new PropertyDescriptor.Builder()
+ .name("Pool - Test On Borrow")
+ .displayName("Pool - Test On Borrow")
+ .description("Whether or not connections should be tested upon borrowing from the pool.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TEST_ON_RETURN = new PropertyDescriptor.Builder()
+ .name("Pool - Test On Return")
+ .displayName("Pool - Test On Return")
+ .description("Whether or not connections should be tested upon returning to the pool.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TEST_WHILE_IDLE = new PropertyDescriptor.Builder()
+ .name("Pool - Test While Idle")
+ .displayName("Pool - Test While Idle")
+ .description("Whether or not connections should be tested while idle.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .required(true)
+ .build();
+
+ public static final List<PropertyDescriptor> REDIS_CONNECTION_PROPERTY_DESCRIPTORS;
+ static {
+ final List<PropertyDescriptor> props = new ArrayList<>();
+ props.add(RedisUtils.REDIS_MODE);
+ props.add(RedisUtils.CONNECTION_STRING);
+ props.add(RedisUtils.DATABASE);
+ props.add(RedisUtils.COMMUNICATION_TIMEOUT);
+ props.add(RedisUtils.CLUSTER_MAX_REDIRECTS);
+ props.add(RedisUtils.SENTINEL_MASTER);
+ props.add(RedisUtils.PASSWORD);
+ props.add(RedisUtils.POOL_MAX_TOTAL);
+ props.add(RedisUtils.POOL_MAX_IDLE);
+ props.add(RedisUtils.POOL_MIN_IDLE);
+ props.add(RedisUtils.POOL_BLOCK_WHEN_EXHAUSTED);
+ props.add(RedisUtils.POOL_MAX_WAIT_TIME);
+ props.add(RedisUtils.POOL_MIN_EVICTABLE_IDLE_TIME);
+ props.add(RedisUtils.POOL_TIME_BETWEEN_EVICTION_RUNS);
+ props.add(RedisUtils.POOL_NUM_TESTS_PER_EVICTION_RUN);
+ props.add(RedisUtils.POOL_TEST_ON_CREATE);
+ props.add(RedisUtils.POOL_TEST_ON_BORROW);
+ props.add(RedisUtils.POOL_TEST_ON_RETURN);
+ props.add(RedisUtils.POOL_TEST_WHILE_IDLE);
+ REDIS_CONNECTION_PROPERTY_DESCRIPTORS = Collections.unmodifiableList(props);
+ }
+
+
+ public static JedisConnectionFactory createConnectionFactory(final PropertyContext context, final ComponentLog logger) {
+ final String redisMode = context.getProperty(RedisUtils.REDIS_MODE).getValue();
+ final String connectionString = context.getProperty(RedisUtils.CONNECTION_STRING).evaluateAttributeExpressions().getValue();
+ final Integer dbIndex = context.getProperty(RedisUtils.DATABASE).evaluateAttributeExpressions().asInteger();
+ final String password = context.getProperty(RedisUtils.PASSWORD).evaluateAttributeExpressions().getValue();
+ final Integer timeout = context.getProperty(RedisUtils.COMMUNICATION_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue();
+ final JedisPoolConfig poolConfig = createJedisPoolConfig(context);
+
+ JedisConnectionFactory connectionFactory;
+
+ if (RedisUtils.REDIS_MODE_STANDALONE.getValue().equals(redisMode)) {
+ final JedisShardInfo jedisShardInfo = createJedisShardInfo(connectionString, timeout, password);
+
+ logger.info("Connecting to Redis in standalone mode at " + connectionString);
+ connectionFactory = new JedisConnectionFactory(jedisShardInfo);
+
+ } else if (RedisUtils.REDIS_MODE_SENTINEL.getValue().equals(redisMode)) {
+ final String[] sentinels = connectionString.split("[,]");
+ final String sentinelMaster = context.getProperty(RedisUtils.SENTINEL_MASTER).evaluateAttributeExpressions().getValue();
+ final RedisSentinelConfiguration sentinelConfiguration = new RedisSentinelConfiguration(sentinelMaster, new HashSet<>(Arrays.asList(sentinels)));
+ final JedisShardInfo jedisShardInfo = createJedisShardInfo(sentinels[0], timeout, password);
+
+ logger.info("Connecting to Redis in sentinel mode...");
+ logger.info("Redis master = " + sentinelMaster);
+
+ for (final String sentinel : sentinels) {
+ logger.info("Redis sentinel at " + sentinel);
+ }
+
+ connectionFactory = new JedisConnectionFactory(sentinelConfiguration, poolConfig);
+ connectionFactory.setShardInfo(jedisShardInfo);
+
+ } else {
+ final String[] clusterNodes = connectionString.split("[,]");
+ final Integer maxRedirects = context.getProperty(RedisUtils.CLUSTER_MAX_REDIRECTS).asInteger();
+
+ final RedisClusterConfiguration clusterConfiguration = new RedisClusterConfiguration(Arrays.asList(clusterNodes));
+ clusterConfiguration.setMaxRedirects(maxRedirects);
+
+ logger.info("Connecting to Redis in clustered mode...");
+ for (final String clusterNode : clusterNodes) {
+ logger.info("Redis cluster node at " + clusterNode);
+ }
+
+ connectionFactory = new JedisConnectionFactory(clusterConfiguration, poolConfig);
+ }
+
+ connectionFactory.setUsePool(true);
+ connectionFactory.setPoolConfig(poolConfig);
+ connectionFactory.setDatabase(dbIndex);
+ connectionFactory.setTimeout(timeout);
+
+ if (!StringUtils.isBlank(password)) {
+ connectionFactory.setPassword(password);
+ }
+
+ // need to call this to initialize the pool/connections
+ connectionFactory.afterPropertiesSet();
+ return connectionFactory;
+ }
+
+ private static JedisShardInfo createJedisShardInfo(final String hostAndPort, final Integer timeout, final String password) {
+ final String[] hostAndPortSplit = hostAndPort.split("[:]");
+ final String host = hostAndPortSplit[0].trim();
+ final Integer port = Integer.parseInt(hostAndPortSplit[1].trim());
+
+ final JedisShardInfo jedisShardInfo = new JedisShardInfo(host, port);
+ jedisShardInfo.setConnectionTimeout(timeout);
+ jedisShardInfo.setSoTimeout(timeout);
+
+ if (!StringUtils.isEmpty(password)) {
+ jedisShardInfo.setPassword(password);
+ }
+
+ return jedisShardInfo;
+ }
+
+ private static JedisPoolConfig createJedisPoolConfig(final PropertyContext context) {
+ final JedisPoolConfig poolConfig = new JedisPoolConfig();
+ poolConfig.setMaxTotal(context.getProperty(RedisUtils.POOL_MAX_TOTAL).asInteger());
+ poolConfig.setMaxIdle(context.getProperty(RedisUtils.POOL_MAX_IDLE).asInteger());
+ poolConfig.setMinIdle(context.getProperty(RedisUtils.POOL_MIN_IDLE).asInteger());
+ poolConfig.setBlockWhenExhausted(context.getProperty(RedisUtils.POOL_BLOCK_WHEN_EXHAUSTED).asBoolean());
+ poolConfig.setMaxWaitMillis(context.getProperty(RedisUtils.POOL_MAX_WAIT_TIME).asTimePeriod(TimeUnit.MILLISECONDS));
+ poolConfig.setMinEvictableIdleTimeMillis(context.getProperty(RedisUtils.POOL_MIN_EVICTABLE_IDLE_TIME).asTimePeriod(TimeUnit.MILLISECONDS));
+ poolConfig.setTimeBetweenEvictionRunsMillis(context.getProperty(RedisUtils.POOL_TIME_BETWEEN_EVICTION_RUNS).asTimePeriod(TimeUnit.MILLISECONDS));
+ poolConfig.setNumTestsPerEvictionRun(context.getProperty(RedisUtils.POOL_NUM_TESTS_PER_EVICTION_RUN).asInteger());
+ poolConfig.setTestOnCreate(context.getProperty(RedisUtils.POOL_TEST_ON_CREATE).asBoolean());
+ poolConfig.setTestOnBorrow(context.getProperty(RedisUtils.POOL_TEST_ON_BORROW).asBoolean());
+ poolConfig.setTestOnReturn(context.getProperty(RedisUtils.POOL_TEST_ON_RETURN).asBoolean());
+ poolConfig.setTestWhileIdle(context.getProperty(RedisUtils.POOL_TEST_WHILE_IDLE).asBoolean());
+ return poolConfig;
+ }
+
+ public static Collection<ValidationResult> validate(ValidationContext validationContext) {
+ final List<ValidationResult> results = new ArrayList<>();
+
+ final String redisMode = validationContext.getProperty(RedisUtils.REDIS_MODE).getValue();
+ final String connectionString = validationContext.getProperty(RedisUtils.CONNECTION_STRING).evaluateAttributeExpressions().getValue();
+ final Integer dbIndex = validationContext.getProperty(RedisUtils.DATABASE).evaluateAttributeExpressions().asInteger();
+
+ if (StringUtils.isBlank(connectionString)) {
+ results.add(new ValidationResult.Builder()
+ .subject(RedisUtils.CONNECTION_STRING.getDisplayName())
+ .valid(false)
+ .explanation("Connection String cannot be blank")
+ .build());
+ } else if (RedisUtils.REDIS_MODE_STANDALONE.getValue().equals(redisMode)) {
+ final String[] hostAndPort = connectionString.split("[:]");
+ if (hostAndPort == null || hostAndPort.length != 2 || !isInteger(hostAndPort[1])) {
+ results.add(new ValidationResult.Builder()
+ .subject(RedisUtils.CONNECTION_STRING.getDisplayName())
+ .input(connectionString)
+ .valid(false)
+ .explanation("Standalone Connection String must be in the form host:port")
+ .build());
+ }
+ } else {
+ for (final String connection : connectionString.split("[,]")) {
+ final String[] hostAndPort = connection.split("[:]");
+ if (hostAndPort == null || hostAndPort.length != 2 || !isInteger(hostAndPort[1])) {
+ results.add(new ValidationResult.Builder()
+ .subject(RedisUtils.CONNECTION_STRING.getDisplayName())
+ .input(connection)
+ .valid(false)
+ .explanation("Connection String must be in the form host:port")
--- End diff --
Good call, updated
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #1918: NIFI-4061 Add a RedisStateProvider
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on the issue:
https://github.com/apache/nifi/pull/1918
@markap14 good catch! The issue appears to be with the logic in the clear method of the RedisStateProvider, I pushed up a commit that should address this problem.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122732793
--- Diff: nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/service/RedisDistributedMapCacheClientService.java ---
@@ -0,0 +1,313 @@
+/*
+ * 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.redis.service;
+
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnDisabled;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.distributed.cache.client.AtomicCacheEntry;
+import org.apache.nifi.distributed.cache.client.AtomicDistributedMapCacheClient;
+import org.apache.nifi.distributed.cache.client.Deserializer;
+import org.apache.nifi.distributed.cache.client.Serializer;
+import org.apache.nifi.redis.RedisConnectionPool;
+import org.apache.nifi.redis.RedisType;
+import org.apache.nifi.redis.util.RedisAction;
+import org.apache.nifi.util.Tuple;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.ScanOptions;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+@Tags({ "redis", "distributed", "cache", "map" })
+@CapabilityDescription("An implementation of DistributedMapCacheClient that uses Redis as the backing cache. This service relies on " +
+ "the WATCH, MULTI, and EXEC commands in Redis, which are not fully supported when Redis is clustered. As a result, this service " +
+ "can only be used with a Redis Connection Pool that is configured for standalone or sentinel mode. Sentinel mode can be used to " +
+ "provide high-availability configurations.")
+public class RedisDistributedMapCacheClientService extends AbstractControllerService implements AtomicDistributedMapCacheClient<byte[]> {
+
+ public static final PropertyDescriptor REDIS_CONNECTION_POOL = new PropertyDescriptor.Builder()
+ .name("redis-connection-pool")
+ .displayName("Redis Connection Pool")
+ .identifiesControllerService(RedisConnectionPool.class)
+ .required(true)
+ .build();
+
+ static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS;
+ static {
+ final List<PropertyDescriptor> props = new ArrayList<>();
+ props.add(REDIS_CONNECTION_POOL);
+ PROPERTY_DESCRIPTORS = Collections.unmodifiableList(props);
+ }
+
+ private volatile RedisConnectionPool redisConnectionPool;
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTY_DESCRIPTORS;
+ }
+
+ @Override
+ protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
+ final List<ValidationResult> results = new ArrayList<>();
+
+ final RedisConnectionPool redisConnectionPool = validationContext.getProperty(REDIS_CONNECTION_POOL).asControllerService(RedisConnectionPool.class);
+ if (redisConnectionPool != null) {
+ final RedisType redisType = redisConnectionPool.getRedisType();
+ if (redisType != null && redisType == RedisType.CLUSTER) {
+ results.add(new ValidationResult.Builder()
+ .subject(REDIS_CONNECTION_POOL.getDisplayName())
+ .valid(false)
+ .explanation(REDIS_CONNECTION_POOL.getDisplayName()
+ + " is configured in clustered mode, and this service requires a non-clustered Redis")
+ .build());
+ }
+ }
+
+ return results;
+ }
+
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) {
+ this.redisConnectionPool = context.getProperty(REDIS_CONNECTION_POOL).asControllerService(RedisConnectionPool.class);
+ }
+
+ @OnDisabled
+ public void onDisabled() {
+ this.redisConnectionPool = null;
+ }
+
+ @Override
+ public <K, V> boolean putIfAbsent(final K key, final V value, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) throws IOException {
+ return withConnection(redisConnection -> {
+ final Tuple<byte[],byte[]> kv = serialize(key, value, keySerializer, valueSerializer);
+ return redisConnection.setNX(kv.getKey(), kv.getValue());
+ });
+ }
+
+ @Override
+ public <K, V> V getAndPutIfAbsent(final K key, final V value, final Serializer<K> keySerializer, final Serializer<V> valueSerializer, final Deserializer<V> valueDeserializer) throws IOException {
+ return withConnection(redisConnection -> {
+ final Tuple<byte[],byte[]> kv = serialize(key, value, keySerializer, valueSerializer);
+ do {
+ // start a watch on the key and retrieve the current value
+ redisConnection.watch(kv.getKey());
+ final byte[] existingValue = redisConnection.get(kv.getKey());
+
+ // start a transaction and perform the put-if-absent
+ redisConnection.multi();
+ redisConnection.setNX(kv.getKey(), kv.getValue());
+
+ // execute the transaction
+ final List<Object> results = redisConnection.exec();
+
+ // if the results list was empty, then the transaction failed (i.e. key was modified after we started watching), so keep looping to retry
+ // if the results list has results, then the transaction succeeded and it should have the result of the setNX operation
+ if (results.size() > 0) {
+ final Object firstResult = results.get(0);
+ if (firstResult instanceof Boolean) {
+ final Boolean absent = (Boolean) firstResult;
+ return absent ? null : valueDeserializer.deserialize(existingValue);
+ } else {
+ // this shouldn't really happen, but just in case there is a non-boolean result then bounce out of the loop
+ throw new IOException("Unexpected result from Redis transaction: Expected Boolean result, but got "
+ + firstResult.getClass().getName() + " with value " + firstResult.toString());
+ }
+ }
+ } while (true);
+ });
+ }
+
+ @Override
+ public <K> boolean containsKey(final K key, final Serializer<K> keySerializer) throws IOException {
+ return withConnection(redisConnection -> {
+ final byte[] k = serialize(key, keySerializer);
+ return redisConnection.exists(k);
+ });
+ }
+
+ @Override
+ public <K, V> void put(final K key, final V value, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) throws IOException {
+ withConnection(redisConnection -> {
+ final Tuple<byte[],byte[]> kv = serialize(key, value, keySerializer, valueSerializer);
+ redisConnection.set(kv.getKey(), kv.getValue());
+ return null;
+ });
+ }
+
+ @Override
+ public <K, V> V get(final K key, final Serializer<K> keySerializer, final Deserializer<V> valueDeserializer) throws IOException {
+ return withConnection(redisConnection -> {
+ final byte[] k = serialize(key, keySerializer);
+ final byte[] v = redisConnection.get(k);
+ return valueDeserializer.deserialize(v);
+ });
+ }
+
+ @Override
+ public void close() throws IOException {
+ // nothing to do
+ }
+
+ @Override
+ public <K> boolean remove(final K key, final Serializer<K> keySerializer) throws IOException {
+ return withConnection(redisConnection -> {
+ final byte[] k = serialize(key, keySerializer);
+ final long numRemoved = redisConnection.del(k);
+ return numRemoved > 0;
+ });
+ }
+
+ @Override
+ public long removeByPattern(final String regex) throws IOException {
+ return withConnection(redisConnection -> {
+ final List<byte[]> allKeys = new ArrayList<>();
--- End diff --
I'm a bit concerned about the potential heap utilization that this could involve. If the regex is .*, for example, this would return millions of objects. Perhaps we should break out of the do/while if we hit some threshold (say 1,000 or 10,000 or something) elements, and then repeat the process until we have nothing left to remove.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122723598
--- Diff: nifi-api/src/main/java/org/apache/nifi/context/PropertyContext.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.context;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.PropertyValue;
+
+/**
+ * A context for retrieving a PropertyValue from a PropertyDescriptor.
+ */
+public interface PropertyContext {
--- End diff --
I like this. We have several different 'context' objects that all have this same type of thing. I think it probably also makes sense to define a method in this interface:
`Map<String, String> getProperties()`
Thoughts?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122837202
--- Diff: nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java ---
@@ -0,0 +1,422 @@
+/*
+ * 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.redis.util;
+
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.redis.RedisType;
+import org.apache.nifi.util.StringUtils;
+import org.springframework.data.redis.connection.RedisClusterConfiguration;
+import org.springframework.data.redis.connection.RedisSentinelConfiguration;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import redis.clients.jedis.JedisPoolConfig;
+import redis.clients.jedis.JedisShardInfo;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+public class RedisUtils {
+
+ // These properties are shared between the connection pool controller service and the state provider, the name
+ // is purposely set to be more human-readable since that will be referenced in state-management.xml
+
+ public static final AllowableValue REDIS_MODE_STANDALONE = new AllowableValue(RedisType.STANDALONE.getDisplayName(), RedisType.STANDALONE.getDisplayName(), RedisType.STANDALONE.getDescription());
+ public static final AllowableValue REDIS_MODE_SENTINEL = new AllowableValue(RedisType.SENTINEL.getDisplayName(), RedisType.SENTINEL.getDisplayName(), RedisType.SENTINEL.getDescription());
+ public static final AllowableValue REDIS_MODE_CLUSTER = new AllowableValue(RedisType.CLUSTER.getDisplayName(), RedisType.CLUSTER.getDisplayName(), RedisType.CLUSTER.getDescription());
+
+ public static final PropertyDescriptor REDIS_MODE = new PropertyDescriptor.Builder()
+ .name("Redis Mode")
+ .displayName("Redis Mode")
+ .description("The type of Redis being communicated with - standalone, sentinel, or clustered.")
+ .allowableValues(REDIS_MODE_STANDALONE, REDIS_MODE_SENTINEL, REDIS_MODE_CLUSTER)
+ .defaultValue(REDIS_MODE_STANDALONE.getValue())
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor CONNECTION_STRING = new PropertyDescriptor.Builder()
+ .name("Connection String")
+ .displayName("Connection String")
+ .description("The connection string for Redis. In a standalone instance this value will be of the form hostname:port. " +
+ "In a sentinel instance this value will be the comma-separated list of sentinels, such as host1:port1,host2:port2,host3:port3. " +
+ "In a clustered instance this value will be the comma-separated list of cluster masters, such as host1:port,host2:port,host3:port.")
+ .required(true)
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ public static final PropertyDescriptor DATABASE = new PropertyDescriptor.Builder()
+ .name("Database Index")
+ .displayName("Database Index")
+ .description("The database index to be used by connections created from this connection pool. " +
+ "See the databases property in redis.conf, by default databases 0-15 will be available.")
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .defaultValue("0")
+ .expressionLanguageSupported(true)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor COMMUNICATION_TIMEOUT = new PropertyDescriptor.Builder()
+ .name("Communication Timeout")
+ .displayName("Communication Timeout")
+ .description("The timeout to use when attempting to communicate with Redis.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("10 seconds")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor CLUSTER_MAX_REDIRECTS = new PropertyDescriptor.Builder()
+ .name("Cluster Max Redirects")
+ .displayName("Cluster Max Redirects")
+ .description("The maximum number of redirects that can be performed when clustered.")
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor SENTINEL_MASTER = new PropertyDescriptor.Builder()
+ .name("Sentinel Master")
+ .displayName("Sentinel Master")
+ .description("The name of the sentinel master, require when Mode is set to Sentinel")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder()
+ .name("Password")
+ .displayName("Password")
+ .description("The password used to authenticate to the Redis server. See the requirepass property in redis.conf.")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .sensitive(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MAX_TOTAL = new PropertyDescriptor.Builder()
+ .name("Pool - Max Total")
+ .displayName("Pool - Max Total")
+ .description("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.")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .defaultValue("8")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MAX_IDLE = new PropertyDescriptor.Builder()
+ .name("Pool - Max Idle")
+ .displayName("Pool - Max Idle")
+ .description("The maximum number of idle connections that can be held in the pool, or a negative value if there is no limit.")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .defaultValue("8")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MIN_IDLE = new PropertyDescriptor.Builder()
+ .name("Pool - Min Idle")
+ .displayName("Pool - Min Idle")
+ .description("The target for the minimum number of idle connections to maintain in the pool. If the configured value of Min Idle is " +
+ "greater than the configured value for Max Idle, then the value of Max Idle will be used instead.")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .defaultValue("0")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_BLOCK_WHEN_EXHAUSTED = new PropertyDescriptor.Builder()
+ .name("Pool - Block When Exhausted")
+ .displayName("Pool - Block When Exhausted")
+ .description("Whether or not clients should block and wait when trying to obtain a connection from the pool when the pool has no available connections. " +
+ "Setting this to false means an error will occur immediately when a client requests a connection and none are available.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MAX_WAIT_TIME = new PropertyDescriptor.Builder()
+ .name("Pool - Max Wait Time")
+ .displayName("Pool - Max Wait Time")
+ .description("The amount of time to wait for an available connection when Block When Exhausted is set to true.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("10 seconds")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MIN_EVICTABLE_IDLE_TIME = new PropertyDescriptor.Builder()
+ .name("Pool - Min Evictable Idle Time")
+ .displayName("Pool - Min Evictable Idle Time")
+ .description("The minimum amount of time an object may sit idle in the pool before it is eligible for eviction.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("60 seconds")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TIME_BETWEEN_EVICTION_RUNS = new PropertyDescriptor.Builder()
+ .name("Pool - Time Between Eviction Runs")
+ .displayName("Pool - Time Between Eviction Runs")
+ .description("The amount of time between attempting to evict idle connections from the pool.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("30 seconds")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_NUM_TESTS_PER_EVICTION_RUN = new PropertyDescriptor.Builder()
+ .name("Pool - Num Tests Per Eviction Run")
+ .displayName("Pool - Num Tests Per Eviction Run")
+ .description("The number of connections to tests per eviction attempt. A negative value indicates to test all connections.")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .defaultValue("-1")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TEST_ON_CREATE = new PropertyDescriptor.Builder()
+ .name("Pool - Test On Create")
+ .displayName("Pool - Test On Create")
+ .description("Whether or not connections should be tested upon creation.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TEST_ON_BORROW = new PropertyDescriptor.Builder()
+ .name("Pool - Test On Borrow")
+ .displayName("Pool - Test On Borrow")
+ .description("Whether or not connections should be tested upon borrowing from the pool.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TEST_ON_RETURN = new PropertyDescriptor.Builder()
+ .name("Pool - Test On Return")
+ .displayName("Pool - Test On Return")
+ .description("Whether or not connections should be tested upon returning to the pool.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TEST_WHILE_IDLE = new PropertyDescriptor.Builder()
+ .name("Pool - Test While Idle")
+ .displayName("Pool - Test While Idle")
+ .description("Whether or not connections should be tested while idle.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .required(true)
+ .build();
+
+ public static final List<PropertyDescriptor> REDIS_CONNECTION_PROPERTY_DESCRIPTORS;
+ static {
+ final List<PropertyDescriptor> props = new ArrayList<>();
+ props.add(RedisUtils.REDIS_MODE);
+ props.add(RedisUtils.CONNECTION_STRING);
+ props.add(RedisUtils.DATABASE);
+ props.add(RedisUtils.COMMUNICATION_TIMEOUT);
+ props.add(RedisUtils.CLUSTER_MAX_REDIRECTS);
+ props.add(RedisUtils.SENTINEL_MASTER);
+ props.add(RedisUtils.PASSWORD);
+ props.add(RedisUtils.POOL_MAX_TOTAL);
+ props.add(RedisUtils.POOL_MAX_IDLE);
+ props.add(RedisUtils.POOL_MIN_IDLE);
+ props.add(RedisUtils.POOL_BLOCK_WHEN_EXHAUSTED);
+ props.add(RedisUtils.POOL_MAX_WAIT_TIME);
+ props.add(RedisUtils.POOL_MIN_EVICTABLE_IDLE_TIME);
+ props.add(RedisUtils.POOL_TIME_BETWEEN_EVICTION_RUNS);
+ props.add(RedisUtils.POOL_NUM_TESTS_PER_EVICTION_RUN);
+ props.add(RedisUtils.POOL_TEST_ON_CREATE);
+ props.add(RedisUtils.POOL_TEST_ON_BORROW);
+ props.add(RedisUtils.POOL_TEST_ON_RETURN);
+ props.add(RedisUtils.POOL_TEST_WHILE_IDLE);
+ REDIS_CONNECTION_PROPERTY_DESCRIPTORS = Collections.unmodifiableList(props);
+ }
+
+
+ public static JedisConnectionFactory createConnectionFactory(final PropertyContext context, final ComponentLog logger) {
+ final String redisMode = context.getProperty(RedisUtils.REDIS_MODE).getValue();
+ final String connectionString = context.getProperty(RedisUtils.CONNECTION_STRING).evaluateAttributeExpressions().getValue();
+ final Integer dbIndex = context.getProperty(RedisUtils.DATABASE).evaluateAttributeExpressions().asInteger();
+ final String password = context.getProperty(RedisUtils.PASSWORD).evaluateAttributeExpressions().getValue();
+ final Integer timeout = context.getProperty(RedisUtils.COMMUNICATION_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue();
+ final JedisPoolConfig poolConfig = createJedisPoolConfig(context);
+
+ JedisConnectionFactory connectionFactory;
+
+ if (RedisUtils.REDIS_MODE_STANDALONE.getValue().equals(redisMode)) {
+ final JedisShardInfo jedisShardInfo = createJedisShardInfo(connectionString, timeout, password);
+
+ logger.info("Connecting to Redis in standalone mode at " + connectionString);
+ connectionFactory = new JedisConnectionFactory(jedisShardInfo);
+
+ } else if (RedisUtils.REDIS_MODE_SENTINEL.getValue().equals(redisMode)) {
+ final String[] sentinels = connectionString.split("[,]");
+ final String sentinelMaster = context.getProperty(RedisUtils.SENTINEL_MASTER).evaluateAttributeExpressions().getValue();
+ final RedisSentinelConfiguration sentinelConfiguration = new RedisSentinelConfiguration(sentinelMaster, new HashSet<>(Arrays.asList(sentinels)));
+ final JedisShardInfo jedisShardInfo = createJedisShardInfo(sentinels[0], timeout, password);
+
+ logger.info("Connecting to Redis in sentinel mode...");
+ logger.info("Redis master = " + sentinelMaster);
+
+ for (final String sentinel : sentinels) {
+ logger.info("Redis sentinel at " + sentinel);
+ }
+
+ connectionFactory = new JedisConnectionFactory(sentinelConfiguration, poolConfig);
+ connectionFactory.setShardInfo(jedisShardInfo);
+
+ } else {
+ final String[] clusterNodes = connectionString.split("[,]");
+ final Integer maxRedirects = context.getProperty(RedisUtils.CLUSTER_MAX_REDIRECTS).asInteger();
+
+ final RedisClusterConfiguration clusterConfiguration = new RedisClusterConfiguration(Arrays.asList(clusterNodes));
+ clusterConfiguration.setMaxRedirects(maxRedirects);
+
+ logger.info("Connecting to Redis in clustered mode...");
+ for (final String clusterNode : clusterNodes) {
+ logger.info("Redis cluster node at " + clusterNode);
+ }
+
+ connectionFactory = new JedisConnectionFactory(clusterConfiguration, poolConfig);
+ }
+
+ connectionFactory.setUsePool(true);
+ connectionFactory.setPoolConfig(poolConfig);
+ connectionFactory.setDatabase(dbIndex);
+ connectionFactory.setTimeout(timeout);
+
+ if (!StringUtils.isBlank(password)) {
+ connectionFactory.setPassword(password);
+ }
+
+ // need to call this to initialize the pool/connections
+ connectionFactory.afterPropertiesSet();
+ return connectionFactory;
+ }
+
+ private static JedisShardInfo createJedisShardInfo(final String hostAndPort, final Integer timeout, final String password) {
+ final String[] hostAndPortSplit = hostAndPort.split("[:]");
+ final String host = hostAndPortSplit[0].trim();
+ final Integer port = Integer.parseInt(hostAndPortSplit[1].trim());
+
+ final JedisShardInfo jedisShardInfo = new JedisShardInfo(host, port);
+ jedisShardInfo.setConnectionTimeout(timeout);
+ jedisShardInfo.setSoTimeout(timeout);
+
+ if (!StringUtils.isEmpty(password)) {
+ jedisShardInfo.setPassword(password);
+ }
+
+ return jedisShardInfo;
+ }
+
+ private static JedisPoolConfig createJedisPoolConfig(final PropertyContext context) {
+ final JedisPoolConfig poolConfig = new JedisPoolConfig();
+ poolConfig.setMaxTotal(context.getProperty(RedisUtils.POOL_MAX_TOTAL).asInteger());
+ poolConfig.setMaxIdle(context.getProperty(RedisUtils.POOL_MAX_IDLE).asInteger());
+ poolConfig.setMinIdle(context.getProperty(RedisUtils.POOL_MIN_IDLE).asInteger());
+ poolConfig.setBlockWhenExhausted(context.getProperty(RedisUtils.POOL_BLOCK_WHEN_EXHAUSTED).asBoolean());
+ poolConfig.setMaxWaitMillis(context.getProperty(RedisUtils.POOL_MAX_WAIT_TIME).asTimePeriod(TimeUnit.MILLISECONDS));
+ poolConfig.setMinEvictableIdleTimeMillis(context.getProperty(RedisUtils.POOL_MIN_EVICTABLE_IDLE_TIME).asTimePeriod(TimeUnit.MILLISECONDS));
+ poolConfig.setTimeBetweenEvictionRunsMillis(context.getProperty(RedisUtils.POOL_TIME_BETWEEN_EVICTION_RUNS).asTimePeriod(TimeUnit.MILLISECONDS));
+ poolConfig.setNumTestsPerEvictionRun(context.getProperty(RedisUtils.POOL_NUM_TESTS_PER_EVICTION_RUN).asInteger());
+ poolConfig.setTestOnCreate(context.getProperty(RedisUtils.POOL_TEST_ON_CREATE).asBoolean());
+ poolConfig.setTestOnBorrow(context.getProperty(RedisUtils.POOL_TEST_ON_BORROW).asBoolean());
+ poolConfig.setTestOnReturn(context.getProperty(RedisUtils.POOL_TEST_ON_RETURN).asBoolean());
+ poolConfig.setTestWhileIdle(context.getProperty(RedisUtils.POOL_TEST_WHILE_IDLE).asBoolean());
+ return poolConfig;
+ }
+
+ public static Collection<ValidationResult> validate(ValidationContext validationContext) {
+ final List<ValidationResult> results = new ArrayList<>();
+
+ final String redisMode = validationContext.getProperty(RedisUtils.REDIS_MODE).getValue();
+ final String connectionString = validationContext.getProperty(RedisUtils.CONNECTION_STRING).evaluateAttributeExpressions().getValue();
+ final Integer dbIndex = validationContext.getProperty(RedisUtils.DATABASE).evaluateAttributeExpressions().asInteger();
+
+ if (StringUtils.isBlank(connectionString)) {
+ results.add(new ValidationResult.Builder()
+ .subject(RedisUtils.CONNECTION_STRING.getDisplayName())
+ .valid(false)
+ .explanation("Connection String cannot be blank")
+ .build());
+ } else if (RedisUtils.REDIS_MODE_STANDALONE.getValue().equals(redisMode)) {
+ final String[] hostAndPort = connectionString.split("[:]");
+ if (hostAndPort == null || hostAndPort.length != 2 || !isInteger(hostAndPort[1])) {
+ results.add(new ValidationResult.Builder()
+ .subject(RedisUtils.CONNECTION_STRING.getDisplayName())
+ .input(connectionString)
+ .valid(false)
+ .explanation("Standalone Connection String must be in the form host:port")
+ .build());
+ }
+ } else {
+ for (final String connection : connectionString.split("[,]")) {
+ final String[] hostAndPort = connection.split("[:]");
--- End diff --
I tried to make the validation and trimming consistent since I was already trimming the values on our end.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122736750
--- Diff: nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateMapJsonSerDe.java ---
@@ -0,0 +1,103 @@
+/*
+ * 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.redis.state;
+
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonToken;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * A RedisStateMapSerDe that uses JSON as the underlying representation.
+ */
+public class RedisStateMapJsonSerDe implements RedisStateMapSerDe {
+
+ public static final String FIELD_VERSION = "version";
+ public static final String FIELD_ENCODING = "encodingVersion";
+ public static final String FIELD_STATE_VALUES = "stateValues";
+
+ private final JsonFactory jsonFactory = new JsonFactory();
+
+ @Override
+ public byte[] serialize(final RedisStateMap stateMap) throws IOException {
+ if (stateMap == null) {
+ return null;
+ }
+
+ try (final ByteArrayOutputStream out = new ByteArrayOutputStream()) {
+ final JsonGenerator jsonGenerator = jsonFactory.createGenerator(out);
+ jsonGenerator.writeStartObject();
+ jsonGenerator.writeNumberField(FIELD_VERSION, stateMap.getVersion());
+ jsonGenerator.writeNumberField(FIELD_ENCODING, stateMap.getEncodingVersion());
+
+ jsonGenerator.writeObjectFieldStart(FIELD_STATE_VALUES);
+ for (Map.Entry<String,String> entry : stateMap.toMap().entrySet()) {
+ jsonGenerator.writeStringField(entry.getKey(), entry.getValue());
+ }
+ jsonGenerator.writeEndObject();
+
+ jsonGenerator.writeEndObject();
+ jsonGenerator.flush();
+
+ return out.toByteArray();
+ }
+ }
+
+ @Override
+ public RedisStateMap deserialize(final byte[] data) throws IOException {
+ if (data == null || data.length == 0) {
+ return null;
+ }
+
+ final RedisStateMap.Builder builder = new RedisStateMap.Builder();
+
+ try (final JsonParser jsonParser = jsonFactory.createParser(data)) {
+ while (jsonParser.nextToken() != JsonToken.END_OBJECT) {
--- End diff --
This code can probably be a lot simpler if we just use jsonParser.readValueAsTree() here. That returns a JsonNode object. We could then just call jsonNode.get(FIELD_VERSION), jsonNode.get(FIELD_ENCODING), jsonNode.get(FIELD_STATE_VALUES)... but will leave it up to you if you think the current approach is better. Just wanted to ensure that you know that this exists :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #1918: NIFI-4061 Add a RedisStateProvider
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on the issue:
https://github.com/apache/nifi/pull/1918
@bbende great thanks! All seems to work well now. +1 merged to master!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122761385
--- Diff: nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java ---
@@ -0,0 +1,422 @@
+/*
+ * 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.redis.util;
+
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.redis.RedisType;
+import org.apache.nifi.util.StringUtils;
+import org.springframework.data.redis.connection.RedisClusterConfiguration;
+import org.springframework.data.redis.connection.RedisSentinelConfiguration;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import redis.clients.jedis.JedisPoolConfig;
+import redis.clients.jedis.JedisShardInfo;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+public class RedisUtils {
+
+ // These properties are shared between the connection pool controller service and the state provider, the name
+ // is purposely set to be more human-readable since that will be referenced in state-management.xml
+
+ public static final AllowableValue REDIS_MODE_STANDALONE = new AllowableValue(RedisType.STANDALONE.getDisplayName(), RedisType.STANDALONE.getDisplayName(), RedisType.STANDALONE.getDescription());
+ public static final AllowableValue REDIS_MODE_SENTINEL = new AllowableValue(RedisType.SENTINEL.getDisplayName(), RedisType.SENTINEL.getDisplayName(), RedisType.SENTINEL.getDescription());
+ public static final AllowableValue REDIS_MODE_CLUSTER = new AllowableValue(RedisType.CLUSTER.getDisplayName(), RedisType.CLUSTER.getDisplayName(), RedisType.CLUSTER.getDescription());
+
+ public static final PropertyDescriptor REDIS_MODE = new PropertyDescriptor.Builder()
+ .name("Redis Mode")
+ .displayName("Redis Mode")
+ .description("The type of Redis being communicated with - standalone, sentinel, or clustered.")
+ .allowableValues(REDIS_MODE_STANDALONE, REDIS_MODE_SENTINEL, REDIS_MODE_CLUSTER)
+ .defaultValue(REDIS_MODE_STANDALONE.getValue())
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor CONNECTION_STRING = new PropertyDescriptor.Builder()
+ .name("Connection String")
+ .displayName("Connection String")
+ .description("The connection string for Redis. In a standalone instance this value will be of the form hostname:port. " +
+ "In a sentinel instance this value will be the comma-separated list of sentinels, such as host1:port1,host2:port2,host3:port3. " +
+ "In a clustered instance this value will be the comma-separated list of cluster masters, such as host1:port,host2:port,host3:port.")
+ .required(true)
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ public static final PropertyDescriptor DATABASE = new PropertyDescriptor.Builder()
+ .name("Database Index")
+ .displayName("Database Index")
+ .description("The database index to be used by connections created from this connection pool. " +
+ "See the databases property in redis.conf, by default databases 0-15 will be available.")
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .defaultValue("0")
+ .expressionLanguageSupported(true)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor COMMUNICATION_TIMEOUT = new PropertyDescriptor.Builder()
+ .name("Communication Timeout")
+ .displayName("Communication Timeout")
+ .description("The timeout to use when attempting to communicate with Redis.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("10 seconds")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor CLUSTER_MAX_REDIRECTS = new PropertyDescriptor.Builder()
+ .name("Cluster Max Redirects")
+ .displayName("Cluster Max Redirects")
+ .description("The maximum number of redirects that can be performed when clustered.")
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor SENTINEL_MASTER = new PropertyDescriptor.Builder()
+ .name("Sentinel Master")
+ .displayName("Sentinel Master")
+ .description("The name of the sentinel master, require when Mode is set to Sentinel")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder()
+ .name("Password")
+ .displayName("Password")
+ .description("The password used to authenticate to the Redis server. See the requirepass property in redis.conf.")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .sensitive(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MAX_TOTAL = new PropertyDescriptor.Builder()
+ .name("Pool - Max Total")
+ .displayName("Pool - Max Total")
+ .description("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.")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .defaultValue("8")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MAX_IDLE = new PropertyDescriptor.Builder()
+ .name("Pool - Max Idle")
+ .displayName("Pool - Max Idle")
+ .description("The maximum number of idle connections that can be held in the pool, or a negative value if there is no limit.")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .defaultValue("8")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MIN_IDLE = new PropertyDescriptor.Builder()
+ .name("Pool - Min Idle")
+ .displayName("Pool - Min Idle")
+ .description("The target for the minimum number of idle connections to maintain in the pool. If the configured value of Min Idle is " +
+ "greater than the configured value for Max Idle, then the value of Max Idle will be used instead.")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .defaultValue("0")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_BLOCK_WHEN_EXHAUSTED = new PropertyDescriptor.Builder()
+ .name("Pool - Block When Exhausted")
+ .displayName("Pool - Block When Exhausted")
+ .description("Whether or not clients should block and wait when trying to obtain a connection from the pool when the pool has no available connections. " +
+ "Setting this to false means an error will occur immediately when a client requests a connection and none are available.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MAX_WAIT_TIME = new PropertyDescriptor.Builder()
+ .name("Pool - Max Wait Time")
+ .displayName("Pool - Max Wait Time")
+ .description("The amount of time to wait for an available connection when Block When Exhausted is set to true.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("10 seconds")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_MIN_EVICTABLE_IDLE_TIME = new PropertyDescriptor.Builder()
+ .name("Pool - Min Evictable Idle Time")
+ .displayName("Pool - Min Evictable Idle Time")
+ .description("The minimum amount of time an object may sit idle in the pool before it is eligible for eviction.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("60 seconds")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TIME_BETWEEN_EVICTION_RUNS = new PropertyDescriptor.Builder()
+ .name("Pool - Time Between Eviction Runs")
+ .displayName("Pool - Time Between Eviction Runs")
+ .description("The amount of time between attempting to evict idle connections from the pool.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("30 seconds")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_NUM_TESTS_PER_EVICTION_RUN = new PropertyDescriptor.Builder()
+ .name("Pool - Num Tests Per Eviction Run")
+ .displayName("Pool - Num Tests Per Eviction Run")
+ .description("The number of connections to tests per eviction attempt. A negative value indicates to test all connections.")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .defaultValue("-1")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TEST_ON_CREATE = new PropertyDescriptor.Builder()
+ .name("Pool - Test On Create")
+ .displayName("Pool - Test On Create")
+ .description("Whether or not connections should be tested upon creation.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TEST_ON_BORROW = new PropertyDescriptor.Builder()
+ .name("Pool - Test On Borrow")
+ .displayName("Pool - Test On Borrow")
+ .description("Whether or not connections should be tested upon borrowing from the pool.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TEST_ON_RETURN = new PropertyDescriptor.Builder()
+ .name("Pool - Test On Return")
+ .displayName("Pool - Test On Return")
+ .description("Whether or not connections should be tested upon returning to the pool.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor POOL_TEST_WHILE_IDLE = new PropertyDescriptor.Builder()
+ .name("Pool - Test While Idle")
+ .displayName("Pool - Test While Idle")
+ .description("Whether or not connections should be tested while idle.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .required(true)
+ .build();
+
+ public static final List<PropertyDescriptor> REDIS_CONNECTION_PROPERTY_DESCRIPTORS;
+ static {
+ final List<PropertyDescriptor> props = new ArrayList<>();
+ props.add(RedisUtils.REDIS_MODE);
+ props.add(RedisUtils.CONNECTION_STRING);
+ props.add(RedisUtils.DATABASE);
+ props.add(RedisUtils.COMMUNICATION_TIMEOUT);
+ props.add(RedisUtils.CLUSTER_MAX_REDIRECTS);
+ props.add(RedisUtils.SENTINEL_MASTER);
+ props.add(RedisUtils.PASSWORD);
+ props.add(RedisUtils.POOL_MAX_TOTAL);
+ props.add(RedisUtils.POOL_MAX_IDLE);
+ props.add(RedisUtils.POOL_MIN_IDLE);
+ props.add(RedisUtils.POOL_BLOCK_WHEN_EXHAUSTED);
+ props.add(RedisUtils.POOL_MAX_WAIT_TIME);
+ props.add(RedisUtils.POOL_MIN_EVICTABLE_IDLE_TIME);
+ props.add(RedisUtils.POOL_TIME_BETWEEN_EVICTION_RUNS);
+ props.add(RedisUtils.POOL_NUM_TESTS_PER_EVICTION_RUN);
+ props.add(RedisUtils.POOL_TEST_ON_CREATE);
+ props.add(RedisUtils.POOL_TEST_ON_BORROW);
+ props.add(RedisUtils.POOL_TEST_ON_RETURN);
+ props.add(RedisUtils.POOL_TEST_WHILE_IDLE);
+ REDIS_CONNECTION_PROPERTY_DESCRIPTORS = Collections.unmodifiableList(props);
+ }
+
+
+ public static JedisConnectionFactory createConnectionFactory(final PropertyContext context, final ComponentLog logger) {
+ final String redisMode = context.getProperty(RedisUtils.REDIS_MODE).getValue();
+ final String connectionString = context.getProperty(RedisUtils.CONNECTION_STRING).evaluateAttributeExpressions().getValue();
+ final Integer dbIndex = context.getProperty(RedisUtils.DATABASE).evaluateAttributeExpressions().asInteger();
+ final String password = context.getProperty(RedisUtils.PASSWORD).evaluateAttributeExpressions().getValue();
+ final Integer timeout = context.getProperty(RedisUtils.COMMUNICATION_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue();
+ final JedisPoolConfig poolConfig = createJedisPoolConfig(context);
+
+ JedisConnectionFactory connectionFactory;
+
+ if (RedisUtils.REDIS_MODE_STANDALONE.getValue().equals(redisMode)) {
+ final JedisShardInfo jedisShardInfo = createJedisShardInfo(connectionString, timeout, password);
+
+ logger.info("Connecting to Redis in standalone mode at " + connectionString);
+ connectionFactory = new JedisConnectionFactory(jedisShardInfo);
+
+ } else if (RedisUtils.REDIS_MODE_SENTINEL.getValue().equals(redisMode)) {
+ final String[] sentinels = connectionString.split("[,]");
+ final String sentinelMaster = context.getProperty(RedisUtils.SENTINEL_MASTER).evaluateAttributeExpressions().getValue();
+ final RedisSentinelConfiguration sentinelConfiguration = new RedisSentinelConfiguration(sentinelMaster, new HashSet<>(Arrays.asList(sentinels)));
+ final JedisShardInfo jedisShardInfo = createJedisShardInfo(sentinels[0], timeout, password);
+
+ logger.info("Connecting to Redis in sentinel mode...");
+ logger.info("Redis master = " + sentinelMaster);
+
+ for (final String sentinel : sentinels) {
+ logger.info("Redis sentinel at " + sentinel);
+ }
+
+ connectionFactory = new JedisConnectionFactory(sentinelConfiguration, poolConfig);
+ connectionFactory.setShardInfo(jedisShardInfo);
+
+ } else {
+ final String[] clusterNodes = connectionString.split("[,]");
+ final Integer maxRedirects = context.getProperty(RedisUtils.CLUSTER_MAX_REDIRECTS).asInteger();
+
+ final RedisClusterConfiguration clusterConfiguration = new RedisClusterConfiguration(Arrays.asList(clusterNodes));
+ clusterConfiguration.setMaxRedirects(maxRedirects);
+
+ logger.info("Connecting to Redis in clustered mode...");
+ for (final String clusterNode : clusterNodes) {
+ logger.info("Redis cluster node at " + clusterNode);
+ }
+
+ connectionFactory = new JedisConnectionFactory(clusterConfiguration, poolConfig);
+ }
+
+ connectionFactory.setUsePool(true);
+ connectionFactory.setPoolConfig(poolConfig);
+ connectionFactory.setDatabase(dbIndex);
+ connectionFactory.setTimeout(timeout);
+
+ if (!StringUtils.isBlank(password)) {
+ connectionFactory.setPassword(password);
+ }
+
+ // need to call this to initialize the pool/connections
+ connectionFactory.afterPropertiesSet();
+ return connectionFactory;
+ }
+
+ private static JedisShardInfo createJedisShardInfo(final String hostAndPort, final Integer timeout, final String password) {
+ final String[] hostAndPortSplit = hostAndPort.split("[:]");
+ final String host = hostAndPortSplit[0].trim();
+ final Integer port = Integer.parseInt(hostAndPortSplit[1].trim());
+
+ final JedisShardInfo jedisShardInfo = new JedisShardInfo(host, port);
+ jedisShardInfo.setConnectionTimeout(timeout);
+ jedisShardInfo.setSoTimeout(timeout);
+
+ if (!StringUtils.isEmpty(password)) {
+ jedisShardInfo.setPassword(password);
+ }
+
+ return jedisShardInfo;
+ }
+
+ private static JedisPoolConfig createJedisPoolConfig(final PropertyContext context) {
+ final JedisPoolConfig poolConfig = new JedisPoolConfig();
+ poolConfig.setMaxTotal(context.getProperty(RedisUtils.POOL_MAX_TOTAL).asInteger());
+ poolConfig.setMaxIdle(context.getProperty(RedisUtils.POOL_MAX_IDLE).asInteger());
+ poolConfig.setMinIdle(context.getProperty(RedisUtils.POOL_MIN_IDLE).asInteger());
+ poolConfig.setBlockWhenExhausted(context.getProperty(RedisUtils.POOL_BLOCK_WHEN_EXHAUSTED).asBoolean());
+ poolConfig.setMaxWaitMillis(context.getProperty(RedisUtils.POOL_MAX_WAIT_TIME).asTimePeriod(TimeUnit.MILLISECONDS));
+ poolConfig.setMinEvictableIdleTimeMillis(context.getProperty(RedisUtils.POOL_MIN_EVICTABLE_IDLE_TIME).asTimePeriod(TimeUnit.MILLISECONDS));
+ poolConfig.setTimeBetweenEvictionRunsMillis(context.getProperty(RedisUtils.POOL_TIME_BETWEEN_EVICTION_RUNS).asTimePeriod(TimeUnit.MILLISECONDS));
+ poolConfig.setNumTestsPerEvictionRun(context.getProperty(RedisUtils.POOL_NUM_TESTS_PER_EVICTION_RUN).asInteger());
+ poolConfig.setTestOnCreate(context.getProperty(RedisUtils.POOL_TEST_ON_CREATE).asBoolean());
+ poolConfig.setTestOnBorrow(context.getProperty(RedisUtils.POOL_TEST_ON_BORROW).asBoolean());
+ poolConfig.setTestOnReturn(context.getProperty(RedisUtils.POOL_TEST_ON_RETURN).asBoolean());
+ poolConfig.setTestWhileIdle(context.getProperty(RedisUtils.POOL_TEST_WHILE_IDLE).asBoolean());
+ return poolConfig;
+ }
+
+ public static Collection<ValidationResult> validate(ValidationContext validationContext) {
+ final List<ValidationResult> results = new ArrayList<>();
+
+ final String redisMode = validationContext.getProperty(RedisUtils.REDIS_MODE).getValue();
+ final String connectionString = validationContext.getProperty(RedisUtils.CONNECTION_STRING).evaluateAttributeExpressions().getValue();
+ final Integer dbIndex = validationContext.getProperty(RedisUtils.DATABASE).evaluateAttributeExpressions().asInteger();
+
+ if (StringUtils.isBlank(connectionString)) {
+ results.add(new ValidationResult.Builder()
+ .subject(RedisUtils.CONNECTION_STRING.getDisplayName())
+ .valid(false)
+ .explanation("Connection String cannot be blank")
+ .build());
+ } else if (RedisUtils.REDIS_MODE_STANDALONE.getValue().equals(redisMode)) {
+ final String[] hostAndPort = connectionString.split("[:]");
+ if (hostAndPort == null || hostAndPort.length != 2 || !isInteger(hostAndPort[1])) {
+ results.add(new ValidationResult.Builder()
+ .subject(RedisUtils.CONNECTION_STRING.getDisplayName())
+ .input(connectionString)
+ .valid(false)
+ .explanation("Standalone Connection String must be in the form host:port")
+ .build());
+ }
+ } else {
+ for (final String connection : connectionString.split("[,]")) {
+ final String[] hostAndPort = connection.split("[:]");
--- End diff --
Does the Redis connection support trimming values, such as host1:port1, host2:port2, ... instead of host1:port1,host2:port2 ?? If so, we should make sure that we allow that here. If not, then that's ok
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122742378
--- Diff: nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java ---
@@ -0,0 +1,271 @@
+/*
+ * 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.redis.state;
+
+import org.apache.nifi.components.AbstractConfigurableComponent;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.components.state.StateProvider;
+import org.apache.nifi.components.state.StateProviderInitializationContext;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.redis.RedisType;
+import org.apache.nifi.redis.util.RedisAction;
+import org.apache.nifi.redis.util.RedisUtils;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A StateProvider backed by Redis.
+ */
+public class RedisStateProvider extends AbstractConfigurableComponent implements StateProvider {
+
+ static final int ENCODING_VERSION = 1;
+
+ private String identifier;
+ private PropertyContext context;
+ private ComponentLog logger;
+
+ private volatile boolean enabled;
+ private volatile JedisConnectionFactory connectionFactory;
+
+ private final RedisStateMapSerDe serDe = new RedisStateMapJsonSerDe();
+
+ @Override
+ public final void initialize(final StateProviderInitializationContext context) throws IOException {
+ this.context = context;
+ this.identifier = context.getIdentifier();
+ this.logger = context.getLogger();
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return RedisUtils.REDIS_CONNECTION_PROPERTY_DESCRIPTORS;
+ }
+
+ @Override
+ protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
+ final List<ValidationResult> results = new ArrayList<>(RedisUtils.validate(validationContext));
+
+ final RedisType redisType = RedisType.fromDisplayName(validationContext.getProperty(RedisUtils.REDIS_MODE).getValue());
+ if (redisType != null && redisType == RedisType.CLUSTER) {
+ results.add(new ValidationResult.Builder()
+ .subject(RedisUtils.REDIS_MODE.getDisplayName())
+ .valid(false)
+ .explanation(RedisUtils.REDIS_MODE.getDisplayName()
+ + " is configured in clustered mode, and this service requires a non-clustered Redis")
+ .build());
+ }
+
+ return results;
+ }
+
+ @Override
+ public String getIdentifier() {
+ return identifier;
+ }
+
+ @Override
+ public void enable() {
+ enabled = true;
+ }
+
+ @Override
+ public void disable() {
+ enabled = false;
+ }
+
+ @Override
+ public boolean isEnabled() {
+ return enabled;
+ }
+
+ @Override
+ public void shutdown() {
+ if (connectionFactory != null) {
+ connectionFactory.destroy();
+ connectionFactory = null;
+ }
+ }
+
+ @Override
+ public void setState(final Map<String, String> state, final String componentId) throws IOException {
+ verifyEnabled();
+
+ final StateMap currStateMap = getState(componentId);
+
+ int attempted = 0;
+ boolean updated = false;
+
+ while (!updated && attempted < 20) {
+ updated = replace(currStateMap, state, componentId, true);
+ attempted++;
+ }
+
+ if (!updated) {
+ throw new IOException("Unable to update state due to concurrent modifications");
+ }
+ }
+
+ @Override
+ public StateMap getState(final String componentId) throws IOException {
+ return withConnection(redisConnection -> {
+ final byte[] key = getComponentPath(componentId).getBytes(StandardCharsets.UTF_8);
+ final byte[] value = redisConnection.get(key);
+
+ final RedisStateMap stateMap = serDe.deserialize(value);
+ if (stateMap == null) {
+ return new RedisStateMap.Builder().encodingVersion(ENCODING_VERSION).build();
+ } else {
+ return stateMap;
+ }
+ });
+ }
+
+ @Override
+ public boolean replace(final StateMap oldValue, final Map<String, String> newValue, final String componentId) throws IOException {
+ return replace(oldValue, newValue, componentId, false);
+ }
+
+ private boolean replace(final StateMap oldValue, final Map<String, String> newValue, final String componentId, final boolean allowReplaceMissing) throws IOException {
+ return withConnection(redisConnection -> {
+
+ boolean replaced = false;
+
+ // start a watch on the key and retrieve the current value
+ final byte[] key = getComponentPath(componentId).getBytes(StandardCharsets.UTF_8);
+ redisConnection.watch(key);
+
+ final long prevVersion = oldValue == null ? -1L : oldValue.getVersion();
+
+ final byte[] currValue = redisConnection.get(key);
+ final RedisStateMap currStateMap = serDe.deserialize(currValue);
+ final long currVersion = currStateMap == null ? -1L : currStateMap.getVersion();
+
+ // the replace API expects that you can't call replace on a non-existing value, so unwatch and return
+ if (!allowReplaceMissing && currVersion == -1) {
+ redisConnection.unwatch();
+ return false;
+ }
+
+ // start a transaction
+ redisConnection.multi();
+
+ // compare-and-set
+ if (prevVersion == currVersion) {
+ // build the new RedisStateMap incrementing the version, using latest encoding, and using the passed in values
+ final RedisStateMap newStateMap = new RedisStateMap.Builder()
+ .version(currVersion + 1)
+ .encodingVersion(ENCODING_VERSION)
+ .stateValues(newValue)
+ .build();
+
+ // if we use set(k, newVal) then the results list will always have size == 0 b/c when convertPipelineAndTxResults is set to true,
+ // status responses like "OK" are skipped over, so by using getSet we can rely on the results list to know if the transaction succeeded
+ redisConnection.getSet(key, serDe.serialize(newStateMap));
+ }
+
+ // execute the transaction
+ final List<Object> results = redisConnection.exec();
+
+ // if we have a result then the replace succeeded
+ if (results.size() > 0) {
+ replaced = true;
+ }
+
+ return replaced;
+ });
+ }
+
+ @Override
+ public void clear(final String componentId) throws IOException {
+ final StateMap currStateMap = getState(componentId);
+
+ int attempted = 0;
+ boolean updated = false;
+
+ while (!updated && attempted < 20) {
+ updated = replace(currStateMap, Collections.emptyMap(), componentId, true);
+ attempted++;
+ }
+
+ if (!updated) {
+ throw new IOException("Unable to update state due to concurrent modifications");
+ }
+ }
+
+ @Override
+ public void onComponentRemoved(final String componentId) throws IOException {
+ withConnection(redisConnection -> {
+ final byte[] key = getComponentPath(componentId).getBytes(StandardCharsets.UTF_8);
+ redisConnection.del(key);
+ return true;
+ });
+ }
+
+ @Override
+ public Scope[] getSupportedScopes() {
+ return new Scope[] {Scope.CLUSTER};
+ }
+
+ private String getComponentPath(final String componentId) {
--- End diff --
Would it make sense to have a configurable property for a "root path"? So instead of "nifi/components/123" for instance we could use a root path of "production-nifi-5" and get a path of "production-nifi-5/components/123". As is, since components have unique ID's, I think you'll be okay. But this provides the benefit of being able to see which nifi instance a key belongs to when looking at redis and allows you to more easily blow away all state for a specific nifi instead of you need to (for instance, you decommission the instance). Would also be safer if a static value like "framework" were ever used for the component ID.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122836288
--- Diff: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/state-management.xml ---
@@ -63,4 +63,68 @@
<property name="Session Timeout">10 seconds</property>
<property name="Access Control">Open</property>
</cluster-provider>
+
+ <!--
+ Cluster State Provider that stores state in Redis. This can be used as an alternative to the ZooKeeper State Provider.
+
+ This provider requires the following properties:
+
+ Redis Mode - The type of Redis instance:
+ - Standalone
+ - Sentinel
+
+ Connection String - The connection string for Redis.
+ - In a standalone instance this value will be of the form hostname:port.
+ - In a sentinel instance this value will be the comma-separated list of sentinels, such as host1:port1,host2:port2,host3:port3.
+ - In a clustered instance this value will be the comma-separated list of cluster masters, such as host1:port,host2:port,host3:port.
--- End diff --
I added the "Cluster" value to the Redis Mode property and stated how it is not currently supported for the state provider due limitations of Redis transactions in clustered mode.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #1918: NIFI-4061 Add a RedisStateProvider
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1918#discussion_r122725606
--- Diff: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/state-management.xml ---
@@ -63,4 +63,68 @@
<property name="Session Timeout">10 seconds</property>
<property name="Access Control">Open</property>
</cluster-provider>
+
+ <!--
+ Cluster State Provider that stores state in Redis. This can be used as an alternative to the ZooKeeper State Provider.
+
+ This provider requires the following properties:
+
+ Redis Mode - The type of Redis instance:
+ - Standalone
+ - Sentinel
+
+ Connection String - The connection string for Redis.
+ - In a standalone instance this value will be of the form hostname:port.
+ - In a sentinel instance this value will be the comma-separated list of sentinels, such as host1:port1,host2:port2,host3:port3.
+ - In a clustered instance this value will be the comma-separated list of cluster masters, such as host1:port,host2:port,host3:port.
--- End diff --
Should we have a 'Cluster' value for the Redis Mode? Or should this be removed from the 'Connection String' documentation? It seems like there's a mismatch here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #1918: NIFI-4061 Add a RedisStateProvider
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on the issue:
https://github.com/apache/nifi/pull/1918
@bbende thanks! All looks good from a code review perspective. I did some testing by standing up a docker image of redis. Then I created a 2 node cluster and added 2 processors: ListFile (File Location = Remote, Primary Node Only) -> UpdateAttribute. Ran ListFile and saw that state was updated appropriately. Clicked Clear state. Now, Node 1 was kicked out of my cluster because it failed to clear state. No indication in the logs, though, as to why it failed. Node 2 had the following stack trace (which is interesting, given that Node 1 was the one kicked out of the cluster):
{code}
2017-06-21 10:55:33,161 ERROR [StandardProcessScheduler Thread-2] o.a.n.controller.StandardProcessorNode Failed to invoke @OnScheduled method due to java.lang.RuntimeException: Failed while executing one of processor's OnScheduled task.
java.lang.RuntimeException: Failed while executing one of processor's OnScheduled task.
at org.apache.nifi.controller.StandardProcessorNode.invokeTaskAsCancelableFuture(StandardProcessorNode.java:1482)
at org.apache.nifi.controller.StandardProcessorNode.access$000(StandardProcessorNode.java:102)
at org.apache.nifi.controller.StandardProcessorNode$1.run(StandardProcessorNode.java:1303)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.util.concurrent.ExecutionException: java.lang.reflect.InvocationTargetException
at java.util.concurrent.FutureTask.report(FutureTask.java:122)
at java.util.concurrent.FutureTask.get(FutureTask.java:206)
at org.apache.nifi.controller.StandardProcessorNode.invokeTaskAsCancelableFuture(StandardProcessorNode.java:1465)
... 9 common frames omitted
Caused by: java.lang.reflect.InvocationTargetException: null
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at org.apache.nifi.util.ReflectionUtils.invokeMethodsWithAnnotations(ReflectionUtils.java:137)
at org.apache.nifi.util.ReflectionUtils.invokeMethodsWithAnnotations(ReflectionUtils.java:125)
at org.apache.nifi.util.ReflectionUtils.invokeMethodsWithAnnotations(ReflectionUtils.java:70)
at org.apache.nifi.util.ReflectionUtils.invokeMethodsWithAnnotation(ReflectionUtils.java:47)
at org.apache.nifi.controller.StandardProcessorNode$1$1.call(StandardProcessorNode.java:1307)
at org.apache.nifi.controller.StandardProcessorNode$1$1.call(StandardProcessorNode.java:1303)
... 6 common frames omitted
Caused by: java.io.IOException: Unable to update state due to concurrent modifications
at org.apache.nifi.redis.state.RedisStateProvider.clear(RedisStateProvider.java:243)
at org.apache.nifi.controller.state.manager.StandardStateManagerProvider$1.clear(StandardStateManagerProvider.java:292)
at org.apache.nifi.controller.state.StandardStateManager.clear(StandardStateManager.java:85)
at org.apache.nifi.processor.util.list.AbstractListProcessor.updateState(AbstractListProcessor.java:217)
... 16 common frames omitted
{code}
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---