This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 34a6d8a0a1 NIFI-8128 Added Sentinel Password to Redis Service
34a6d8a0a1 is described below
commit 34a6d8a0a14cdc0cb4e0e779ee63dd11942386e2
Author: EndzeitBegins <[email protected]>
AuthorDate: Fri Jul 14 18:07:43 2023 +0200
NIFI-8128 Added Sentinel Password to Redis Service
- Replaced embedded-redis and with Testcontainers for Redis integration
tests
This closes #7481
Signed-off-by: David Handermann <[email protected]>
---
.../nifi-redis-extensions/pom.xml | 17 +-
.../org/apache/nifi/redis/util/RedisUtils.java | 26 ++-
.../ITRedisDistributedMapCacheClientService.java | 228 +++++++++++++++++----
.../nifi/redis/state/ITRedisStateProvider.java | 42 ++--
.../nifi/redis/testcontainers/RedisContainer.java | 113 ++++++++++
.../testcontainers/RedisReplicaContainer.java | 60 ++++++
.../testcontainers/RedisSentinelContainer.java | 113 ++++++++++
7 files changed, 516 insertions(+), 83 deletions(-)
diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/pom.xml
b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/pom.xml
index 80baf8d1d2..82a3b96607 100644
--- a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/pom.xml
+++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/pom.xml
@@ -82,6 +82,11 @@
<artifactId>nifi-record-path</artifactId>
<version>2.0.0-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-ssl-context-service-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
<!-- Test dependencies -->
<dependency>
<groupId>org.apache.nifi</groupId>
@@ -96,15 +101,15 @@
<scope>test</scope>
</dependency>
<dependency>
- <groupId>com.github.kstyrc</groupId>
- <artifactId>embedded-redis</artifactId>
- <version>0.6</version>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers</artifactId>
+ <version>${testcontainers.version}</version>
<scope>test</scope>
</dependency>
<dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-ssl-context-service-api</artifactId>
- <scope>compile</scope>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <scope>test</scope>
</dependency>
</dependencies>
</project>
diff --git
a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java
b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java
index af9b61e7a0..30e6f7b009 100644
---
a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java
+++
b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java
@@ -35,6 +35,7 @@ import
org.springframework.data.redis.connection.RedisSentinelConfiguration;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import
org.springframework.data.redis.connection.jedis.JedisClientConfiguration;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.lang.Nullable;
import redis.clients.jedis.JedisPoolConfig;
import javax.net.ssl.SSLContext;
@@ -134,7 +135,16 @@ public class RedisUtils {
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.")
+ .description("The password used to authenticate to the Redis
server. See the 'requirepass' property in redis.conf.")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+ .sensitive(true)
+ .build();
+
+ public static final PropertyDescriptor SENTINEL_PASSWORD = new
PropertyDescriptor.Builder()
+ .name("Sentinel Password")
+ .displayName("Sentinel Password")
+ .description("The password used to authenticate to the Redis
Sentinel server. See the 'requirepass' and 'sentinel sentinel-pass' properties
in sentinel.conf.")
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
.sensitive(true)
@@ -275,6 +285,7 @@ public class RedisUtils {
props.add(RedisUtils.CLUSTER_MAX_REDIRECTS);
props.add(RedisUtils.SENTINEL_MASTER);
props.add(RedisUtils.PASSWORD);
+ props.add(RedisUtils.SENTINEL_PASSWORD);
props.add(RedisUtils.SSL_CONTEXT_SERVICE);
props.add(RedisUtils.POOL_MAX_TOTAL);
props.add(RedisUtils.POOL_MAX_IDLE);
@@ -297,6 +308,7 @@ public class RedisUtils {
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 String sentinelPassword =
context.getProperty(RedisUtils.SENTINEL_PASSWORD).evaluateAttributeExpressions().getValue();
final Integer timeout =
context.getProperty(RedisUtils.COMMUNICATION_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue();
final JedisPoolConfig poolConfig = createJedisPoolConfig(context);
@@ -323,7 +335,7 @@ public class RedisUtils {
final String host = hostAndPortSplit[0].trim();
final Integer port = Integer.parseInt(hostAndPortSplit[1].trim());
final RedisStandaloneConfiguration redisStandaloneConfiguration =
new RedisStandaloneConfiguration(host, port);
- enrichRedisConfiguration(redisStandaloneConfiguration, dbIndex,
password);
+ enrichRedisConfiguration(redisStandaloneConfiguration, dbIndex,
password, sentinelPassword);
connectionFactory = new
JedisConnectionFactory(redisStandaloneConfiguration, jedisClientConfiguration);
@@ -331,7 +343,7 @@ public class RedisUtils {
final String[] sentinels = connectionString.split("[,]");
final String sentinelMaster =
context.getProperty(RedisUtils.SENTINEL_MASTER).evaluateAttributeExpressions().getValue();
final RedisSentinelConfiguration sentinelConfiguration = new
RedisSentinelConfiguration(sentinelMaster, new
HashSet<>(getTrimmedValues(sentinels)));
- enrichRedisConfiguration(sentinelConfiguration, dbIndex, password);
+ enrichRedisConfiguration(sentinelConfiguration, dbIndex, password,
sentinelPassword);
logger.info("Connecting to Redis in sentinel mode...");
logger.info("Redis master = " + sentinelMaster);
@@ -347,7 +359,7 @@ public class RedisUtils {
final Integer maxRedirects =
context.getProperty(RedisUtils.CLUSTER_MAX_REDIRECTS).asInteger();
final RedisClusterConfiguration clusterConfiguration = new
RedisClusterConfiguration(getTrimmedValues(clusterNodes));
- enrichRedisConfiguration(clusterConfiguration, dbIndex, password);
+ enrichRedisConfiguration(clusterConfiguration, dbIndex, password,
sentinelPassword);
clusterConfiguration.setMaxRedirects(maxRedirects);
logger.info("Connecting to Redis in clustered mode...");
@@ -373,13 +385,17 @@ public class RedisUtils {
private static void enrichRedisConfiguration(final RedisConfiguration
redisConfiguration,
final Integer dbIndex,
- final String password) {
+ @Nullable final String
password,
+ @Nullable final String
sentinelPassword) {
if (redisConfiguration instanceof
RedisConfiguration.WithDatabaseIndex) {
((RedisConfiguration.WithDatabaseIndex)
redisConfiguration).setDatabase(dbIndex);
}
if (redisConfiguration instanceof RedisConfiguration.WithPassword) {
((RedisConfiguration.WithPassword)
redisConfiguration).setPassword(RedisPassword.of(password));
}
+ if (redisConfiguration instanceof
RedisConfiguration.SentinelConfiguration) {
+ ((RedisConfiguration.SentinelConfiguration)
redisConfiguration).setSentinelPassword(RedisPassword.of(sentinelPassword));
+ }
}
private static JedisPoolConfig createJedisPoolConfig(final PropertyContext
context) {
diff --git
a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/ITRedisDistributedMapCacheClientService.java
b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/ITRedisDistributedMapCacheClientService.java
index aec018954e..f14187f6ed 100644
---
a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/ITRedisDistributedMapCacheClientService.java
+++
b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/ITRedisDistributedMapCacheClientService.java
@@ -30,14 +30,19 @@ import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.redis.RedisConnectionPool;
+import org.apache.nifi.redis.testcontainers.RedisContainer;
+import org.apache.nifi.redis.testcontainers.RedisReplicaContainer;
+import org.apache.nifi.redis.testcontainers.RedisSentinelContainer;
import org.apache.nifi.redis.util.RedisUtils;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import redis.embedded.RedisServer;
+import org.junit.jupiter.api.io.TempDir;
+import org.springframework.lang.NonNull;
+import org.springframework.lang.Nullable;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
@@ -46,6 +51,8 @@ import java.net.InetSocketAddress;
import java.net.StandardSocketOptions;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -53,8 +60,10 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.function.Consumer;
import static org.apache.nifi.redis.util.RedisUtils.REDIS_CONNECTION_POOL;
+import static org.apache.nifi.redis.util.RedisUtils.REDIS_MODE_SENTINEL;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -64,22 +73,156 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
*/
public class ITRedisDistributedMapCacheClientService {
- private TestRedisProcessor proc;
- private TestRunner testRunner;
- private RedisServer redisServer;
+ public static final String CONTAINER_IMAGE_TAG = "redis:7.0.12-alpine";
+
+ private static final String masterName = "redisLeader";
+ @TempDir
+
+ private Path testDirectory;
+
+ private final TestRedisProcessor proc = new TestRedisProcessor();
+ private final TestRunner testRunner = TestRunners.newTestRunner(proc);
+
+ private final List<RedisContainer> redisContainers = new ArrayList<>();
private RedisConnectionPoolService redisConnectionPool;
- private RedisDistributedMapCacheClientService redisMapCacheClientService;
- private int redisPort;
- @BeforeEach
- public void setup() throws IOException {
- this.redisPort = getAvailablePort();
+ @Test
+ public void testStandaloneRedis() throws InitializationException,
IOException {
+ int redisPort = setupStandaloneRedis(null).port;
+ setUpRedisConnectionPool(portsToConnectionString(redisPort), pool -> {
+ // uncomment this to test using a different database index than
the default 0
+ // testRunner.setProperty(pool, RedisUtils.DATABASE, "1");
+ });
+ setupRedisMapCacheClientService();
+
+ executeProcessor();
+ }
+
+ @Test
+ public void testStandaloneRedisWithAuthentication() throws
InitializationException, IOException {
+ final String redisPassword = "foobared";
+ final int redisPort = setupStandaloneRedis(redisPassword).port;
+ setUpRedisConnectionPool(portsToConnectionString(redisPort), pool -> {
+ testRunner.setProperty(redisConnectionPool, RedisUtils.PASSWORD,
redisPassword);
+ });
+ setupRedisMapCacheClientService();
+
+ executeProcessor();
+ }
+
+ @Test
+ public void testSentinelRedis() throws InitializationException,
IOException {
+ RedisContainer redisMasterContainer = setupStandaloneRedis(null);
+ String masterHost = "127.0.0.1";
+ int masterPort = redisMasterContainer.port;
+ setUpRedisReplica(masterHost, masterPort, null);
+ setUpRedisReplica(masterHost, masterPort, null);
+
+ int sentinelAPort = setUpSentinel(masterHost, masterPort, null, 2,
null).port;
+ int sentinelBPort = setUpSentinel(masterHost, masterPort, null, 2,
null).port;
+ int sentinelCPort = setUpSentinel(masterHost, masterPort, null, 2,
null).port;
+
+ setUpRedisConnectionPool(portsToConnectionString(sentinelAPort,
sentinelBPort, sentinelCPort), pool -> {
+ testRunner.setProperty(redisConnectionPool, RedisUtils.REDIS_MODE,
REDIS_MODE_SENTINEL);
+ testRunner.setProperty(redisConnectionPool,
RedisUtils.SENTINEL_MASTER, masterName);
+ });
+
+ setupRedisMapCacheClientService();
+
+ executeProcessor();
+ }
- this.redisServer = new RedisServer(redisPort);
- redisServer.start();
+ @Test
+ public void testSentinelRedisWithAuthentication() throws
InitializationException, IOException {
+ String redisPassword = "t0p_53cr35";
+ String sentinelPassword = "otherPassword";
+
+ RedisContainer redisMasterContainer =
setupStandaloneRedis(redisPassword);
+ String masterHost = "127.0.0.1";
+ int masterPort = redisMasterContainer.port;
+ setUpRedisReplica(masterHost, masterPort, redisPassword);
+ setUpRedisReplica(masterHost, masterPort, redisPassword);
+
+ int sentinelAPort = setUpSentinel(masterHost, masterPort,
redisPassword, 2, sentinelPassword).port;
+ int sentinelBPort = setUpSentinel(masterHost, masterPort,
redisPassword, 2, sentinelPassword).port;
+ int sentinelCPort = setUpSentinel(masterHost, masterPort,
redisPassword, 2, sentinelPassword).port;
+
+ setUpRedisConnectionPool(portsToConnectionString(sentinelAPort,
sentinelBPort, sentinelCPort), pool -> {
+ testRunner.setProperty(redisConnectionPool, RedisUtils.REDIS_MODE,
REDIS_MODE_SENTINEL);
+ testRunner.setProperty(redisConnectionPool,
RedisUtils.SENTINEL_MASTER, masterName);
+
+ testRunner.setProperty(redisConnectionPool, RedisUtils.PASSWORD,
redisPassword);
+ testRunner.setProperty(redisConnectionPool,
RedisUtils.SENTINEL_PASSWORD, sentinelPassword);
+ });
+ setupRedisMapCacheClientService();
+
+ executeProcessor();
+ }
+
+ @AfterEach
+ public void teardown() {
+ if (redisConnectionPool != null) {
+ redisConnectionPool.onDisabled();
+ }
- proc = new TestRedisProcessor();
- testRunner = TestRunners.newTestRunner(proc);
+ redisContainers.forEach(RedisContainer::stop);
+ }
+
+ private RedisContainer setupStandaloneRedis(final @Nullable String
redisPassword) throws IOException {
+ int redisPort = getAvailablePort();
+
+ RedisContainer redisContainer = new
RedisContainer(CONTAINER_IMAGE_TAG);
+ redisContainer.mountConfigurationFrom(testDirectory);
+ redisContainer.setPort(redisPort);
+ redisContainer.addPortBinding(redisPort, redisPort);
+ redisContainer.setPassword(redisPassword);
+
+ redisContainers.add(redisContainer);
+ redisContainer.start();
+
+ return redisContainer;
+ }
+
+ private RedisReplicaContainer setUpRedisReplica(final @NonNull String
masterHost,
+ final int masterPort,
+ final @Nullable String
redisPassword) throws IOException {
+ int replicaPort = getAvailablePort();
+
+ RedisReplicaContainer redisReplicaContainer = new
RedisReplicaContainer(CONTAINER_IMAGE_TAG);
+ redisReplicaContainer.mountConfigurationFrom(testDirectory);
+ redisReplicaContainer.setPort(replicaPort);
+ redisReplicaContainer.addPortBinding(replicaPort, replicaPort);
+ redisReplicaContainer.setReplicaOf(masterHost, masterPort);
+ redisReplicaContainer.setPassword(redisPassword);
+
+ redisContainers.add(redisReplicaContainer);
+ redisReplicaContainer.start();
+
+ return redisReplicaContainer;
+ }
+
+ private RedisSentinelContainer setUpSentinel(final @NonNull String
masterHost,
+ final int masterPort,
+ final @Nullable String
redisPassword,
+ final int quorumSize,
+ final @Nullable String
sentinelPassword) throws IOException {
+ int sentinelPort = getAvailablePort();
+
+ RedisSentinelContainer redisSentinelContainer = new
RedisSentinelContainer(CONTAINER_IMAGE_TAG);
+ redisSentinelContainer.mountConfigurationFrom(testDirectory);
+ redisSentinelContainer.setPort(sentinelPort);
+ redisSentinelContainer.addPortBinding(sentinelPort, sentinelPort);
+ redisSentinelContainer.setMasterHost(masterHost);
+ redisSentinelContainer.setMasterPort(masterPort);
+ redisSentinelContainer.setMasterName(masterName);
+ redisSentinelContainer.setQuorumSize(quorumSize);
+ redisSentinelContainer.setPassword(redisPassword);
+ redisSentinelContainer.setSentinelPassword(sentinelPassword);
+
+ redisContainers.add(redisSentinelContainer);
+ redisSentinelContainer.start();
+
+ return redisSentinelContainer;
}
private int getAvailablePort() throws IOException {
@@ -90,41 +233,37 @@ public class ITRedisDistributedMapCacheClientService {
}
}
- @AfterEach
- public void teardown() throws IOException {
- if (redisServer != null) {
- redisServer.stop();
- }
- }
-
- @Test
- public void testStandaloneRedis() throws InitializationException {
- try {
- // create, configure, and enable the RedisConnectionPool service
- redisConnectionPool = new RedisConnectionPoolService();
- testRunner.addControllerService("redis-connection-pool",
redisConnectionPool);
- testRunner.setProperty(redisConnectionPool,
RedisUtils.CONNECTION_STRING, "localhost:" + redisPort);
+ private String portsToConnectionString(int... ports) {
+ StringBuilder stringBuilder = new StringBuilder();
- // uncomment this to test using a different database index than
the default 0
- //testRunner.setProperty(redisConnectionPool, RedisUtils.DATABASE,
"1");
+ for (int i = 0; i < ports.length; i++) {
+ if (i > 0) {
+ stringBuilder.append(',');
+ }
+ stringBuilder.append("localhost");
+ stringBuilder.append(':');
+ stringBuilder.append(ports[i]);
+ }
- // uncomment this to test using a password to authenticate to redis
- //testRunner.setProperty(redisConnectionPool, RedisUtils.PASSWORD,
"foobared");
+ return stringBuilder.toString();
+ }
- testRunner.enableControllerService(redisConnectionPool);
+ public void setUpRedisConnectionPool(String connectionString,
Consumer<RedisConnectionPool> initializer) throws InitializationException {
+ // create, configure, and enable the RedisConnectionPool service
+ redisConnectionPool = new RedisConnectionPoolService();
+ testRunner.addControllerService("redis-connection-pool",
redisConnectionPool);
+ testRunner.setProperty(redisConnectionPool,
RedisUtils.CONNECTION_STRING, connectionString);
- setupRedisMapCacheClientService();
- executeProcessor();
- } finally {
- if (redisConnectionPool != null) {
- redisConnectionPool.onDisabled();
- }
+ if (initializer != null) {
+ initializer.accept(redisConnectionPool);
}
+
+ testRunner.enableControllerService(redisConnectionPool);
}
private void setupRedisMapCacheClientService() throws
InitializationException {
// create, configure, and enable the RedisDistributedMapCacheClient
service
- redisMapCacheClientService = new
RedisDistributedMapCacheClientService();
+ RedisDistributedMapCacheClientService redisMapCacheClientService = new
RedisDistributedMapCacheClientService();
testRunner.addControllerService("redis-map-cache-client",
redisMapCacheClientService);
testRunner.setProperty(redisMapCacheClientService,
REDIS_CONNECTION_POOL, "redis-connection-pool");
testRunner.enableControllerService(redisMapCacheClientService);
@@ -218,12 +357,12 @@ public class ITRedisDistributedMapCacheClientService {
assertEquals(value, cacheClient.get(keyThatDoesntExist,
stringSerializer, stringDeserializer));
// verify atomic fetch returns the correct entry
- final AtomicCacheEntry<String,String,byte[]> entry =
cacheClient.fetch(key, stringSerializer, stringDeserializer);
+ final AtomicCacheEntry<String, String, byte[]> entry =
cacheClient.fetch(key, stringSerializer, stringDeserializer);
assertEquals(key, entry.getKey());
assertEquals(value, entry.getValue());
assertTrue(Arrays.equals(value.getBytes(StandardCharsets.UTF_8),
entry.getRevision().orElse(null)));
- final AtomicCacheEntry<String,String,byte[]> notLatestEntry =
new AtomicCacheEntry<>(entry.getKey(), entry.getValue(), "not
previous".getBytes(StandardCharsets.UTF_8));
+ final AtomicCacheEntry<String, String, byte[]> notLatestEntry
= new AtomicCacheEntry<>(entry.getKey(), entry.getValue(), "not
previous".getBytes(StandardCharsets.UTF_8));
// verify atomic replace does not replace when previous value
is not equal
assertFalse(cacheClient.replace(notLatestEntry,
stringSerializer, stringSerializer));
@@ -237,12 +376,12 @@ public class ITRedisDistributedMapCacheClientService {
// verify atomic replace does replace no value previous existed
final String replaceKeyDoesntExist = key +
"_REPLACE_DOES_NOT_EXIST";
- final AtomicCacheEntry<String,String,byte[]> entryDoesNotExist
= new AtomicCacheEntry<>(replaceKeyDoesntExist, replacementValue, null);
+ final AtomicCacheEntry<String, String, byte[]>
entryDoesNotExist = new AtomicCacheEntry<>(replaceKeyDoesntExist,
replacementValue, null);
assertTrue(cacheClient.replace(entryDoesNotExist,
stringSerializer, stringSerializer));
assertEquals(replacementValue,
cacheClient.get(replaceKeyDoesntExist, stringSerializer, stringDeserializer));
final int numToDelete = 2000;
- for (int i=0; i < numToDelete; i++) {
+ for (int i = 0; i < numToDelete; i++) {
cacheClient.put(key + "-" + i, value, stringSerializer,
stringSerializer);
}
@@ -294,3 +433,4 @@ public class ITRedisDistributedMapCacheClientService {
}
}
}
+
diff --git
a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/ITRedisStateProvider.java
b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/ITRedisStateProvider.java
index 599e5ec6ab..5315439f1f 100644
---
a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/ITRedisStateProvider.java
+++
b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/ITRedisStateProvider.java
@@ -23,18 +23,18 @@ 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.logging.ComponentLog;
+import org.apache.nifi.redis.service.ITRedisDistributedMapCacheClientService;
+import org.apache.nifi.redis.testcontainers.RedisContainer;
import org.apache.nifi.redis.util.RedisUtils;
import org.apache.nifi.util.MockComponentLog;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import redis.embedded.RedisServer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
import javax.net.ssl.SSLContext;
import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.net.StandardSocketOptions;
-import java.nio.channels.SocketChannel;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -51,22 +51,21 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
* NOTE: These test cases should be kept in-sync with
AbstractTestStateProvider which is in the framework
* and couldn't be extended here.
*/
+@Testcontainers
public class ITRedisStateProvider {
protected final String componentId =
"111111111-1111-1111-1111-111111111111";
- private RedisServer redisServer;
+ @Container
+ public RedisContainer redisContainer = new
RedisContainer(ITRedisDistributedMapCacheClientService.CONTAINER_IMAGE_TAG)
+ .withExposedPorts(6379);
+
private RedisStateProvider provider;
@BeforeEach
- public void setup() throws Exception {
- final int redisPort = getAvailablePort();
-
- this.redisServer = new RedisServer(redisPort);
- redisServer.start();
-
+ public void setup() {
final Map<PropertyDescriptor, String> properties = new HashMap<>();
- properties.put(RedisUtils.CONNECTION_STRING, "localhost:" + redisPort);
+ properties.put(RedisUtils.CONNECTION_STRING, redisContainer.getHost()
+ ":" + redisContainer.getFirstMappedPort());
this.provider = createProvider(properties);
}
@@ -80,10 +79,6 @@ public class ITRedisStateProvider {
provider.disable();
provider.shutdown();
}
-
- if (redisServer != null) {
- redisServer.stop();
- }
}
public StateProvider getProvider() {
@@ -173,7 +168,7 @@ public class ITRedisStateProvider {
assertEquals(1, map.size());
assertEquals("value", map.get(key));
- provider.setState(Collections.<String, String> emptyMap(),
componentId);
+ provider.setState(Collections.<String, String>emptyMap(), componentId);
final StateMap stateMap = provider.getState(componentId);
map = stateMap.toMap();
@@ -284,8 +279,8 @@ public class ITRedisStateProvider {
}
@Override
- public Map<String,String> getAllProperties() {
- final Map<String,String> propValueMap = new LinkedHashMap<>();
+ public Map<String, String> getAllProperties() {
+ final Map<String, String> propValueMap = new LinkedHashMap<>();
for (final Map.Entry<PropertyDescriptor, String> entry :
properties.entrySet()) {
propValueMap.put(entry.getKey().getName(),
entry.getValue());
}
@@ -321,13 +316,4 @@ public class ITRedisStateProvider {
provider.enable();
return provider;
}
-
- private int getAvailablePort() throws IOException {
- try (SocketChannel socket = SocketChannel.open()) {
- socket.setOption(StandardSocketOptions.SO_REUSEADDR, true);
- socket.bind(new InetSocketAddress("localhost", 0));
- return socket.socket().getLocalPort();
- }
- }
-
}
diff --git
a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisContainer.java
b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisContainer.java
new file mode 100644
index 0000000000..24f5b68a98
--- /dev/null
+++
b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisContainer.java
@@ -0,0 +1,113 @@
+/*
+ * 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.testcontainers;
+
+import org.springframework.lang.NonNull;
+import org.springframework.lang.Nullable;
+import org.testcontainers.containers.BindMode;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.utility.DockerImageName;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+public class RedisContainer extends GenericContainer<RedisContainer> {
+
+ public static final int REDIS_PORT = 6379;
+
+ public RedisContainer(@NonNull DockerImageName dockerImageName) {
+ super(dockerImageName);
+ }
+
+ public RedisContainer(@NonNull String fullImageName) {
+ this(DockerImageName.parse(fullImageName));
+ }
+
+ public int port = REDIS_PORT;
+
+ @Nullable
+ protected String password = null;
+ @Nullable
+ protected Path configurationMountDirectory = null;
+
+ protected final List<String> configurationOptions = new ArrayList<>();
+
+ public void setPassword(final @Nullable String password) {
+ this.password = password;
+ }
+
+ public void setPort(final int port) {
+ this.port = port;
+ }
+
+ public void mountConfigurationFrom(final Path mountDirectory) {
+ this.configurationMountDirectory = mountDirectory;
+ }
+
+ public void addConfigurationOption(final String configurationOption) {
+ this.configurationOptions.add(configurationOption);
+ }
+
+ protected void adjustConfiguration() {
+ addConfigurationOption("port " + port);
+
+ if (password != null) {
+ addConfigurationOption("requirepass " + password);
+ }
+ }
+
+ /**
+ * Sets up a static binding between a port on the host and one in the
container.
+ * In order for auto-discovery mechanisms of Redis to work, 1-to-1 mapped
ports are useful.
+ */
+ public void addPortBinding(int hostPort, int containerPort) {
+ addFixedExposedPort(hostPort, containerPort);
+ }
+
+ @Override
+ protected void configure() {
+ adjustConfiguration();
+
+ Path configurationFilePath = writeConfigurationFile().toAbsolutePath();
+ String hostPath = configurationFilePath.toString();
+ String containerPath = "/usr/local/etc/redis/redis.conf";
+ addFileSystemBind(hostPath, containerPath, BindMode.READ_WRITE);
+
+ setCommand(containerPath);
+ }
+
+ protected Path writeConfigurationFile() {
+ try {
+ Path mountDirectory = this.configurationMountDirectory;
+ if (mountDirectory == null) {
+ mountDirectory =
Files.createTempDirectory("redis-container-configuration");
+ }
+
+ Path configFile = mountDirectory.resolve("redis-" +
UUID.randomUUID() + ".conf");
+ Files.write(configFile, configurationOptions,
StandardCharsets.UTF_8);
+
+ return configFile;
+ } catch (IOException ioException) {
+ throw new IllegalStateException("Cannot start container because
configuration could not be written", ioException);
+ }
+ }
+}
diff --git
a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisReplicaContainer.java
b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisReplicaContainer.java
new file mode 100644
index 0000000000..815fe33f85
--- /dev/null
+++
b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisReplicaContainer.java
@@ -0,0 +1,60 @@
+/*
+ * 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.testcontainers;
+
+import org.springframework.lang.NonNull;
+import org.testcontainers.utility.DockerImageName;
+
+public class RedisReplicaContainer extends RedisContainer {
+
+ public RedisReplicaContainer(@NonNull DockerImageName dockerImageName) {
+ super(dockerImageName);
+ }
+
+ public RedisReplicaContainer(@NonNull String fullImageName) {
+ this(DockerImageName.parse(fullImageName));
+ }
+
+ @NonNull
+ protected String masterHost = "localhost";
+ protected int masterPort = REDIS_PORT;
+
+ public void setMasterHost(@NonNull String masterHost) {
+ this.masterHost = masterHost;
+ }
+
+ public void setMasterPort(int masterPort) {
+ this.masterPort = masterPort;
+ }
+
+ public void setReplicaOf(@NonNull String masterHost, int masterPort) {
+ setMasterHost(masterHost);
+ setMasterPort(masterPort);
+ }
+
+ @Override
+ protected void adjustConfiguration() {
+ addConfigurationOption("port " + port);
+
+ if (password != null) {
+ addConfigurationOption("requirepass " + password);
+ addConfigurationOption("masterauth " + password);
+ }
+
+ addConfigurationOption("replicaof " + masterHost + " " + masterPort);
+ }
+}
diff --git
a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisSentinelContainer.java
b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisSentinelContainer.java
new file mode 100644
index 0000000000..06b802dc3b
--- /dev/null
+++
b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisSentinelContainer.java
@@ -0,0 +1,113 @@
+/*
+ * 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.testcontainers;
+
+import org.springframework.lang.NonNull;
+import org.springframework.lang.Nullable;
+import org.testcontainers.utility.DockerImageName;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+public class RedisSentinelContainer extends RedisContainer {
+
+ public static final int REDIS_SENTINEL_PORT = 26379;
+
+ public RedisSentinelContainer(final @NonNull DockerImageName
dockerImageName) {
+ super(dockerImageName);
+
+ setPort(REDIS_SENTINEL_PORT);
+ }
+
+ public RedisSentinelContainer(final @NonNull String fullImageName) {
+ this(DockerImageName.parse(fullImageName));
+ }
+
+ @NonNull
+ protected String masterHost = "localhost";
+ protected int masterPort = REDIS_PORT;
+ @NonNull
+ protected String masterName = "mymaster";
+ @Nullable
+ protected String sentinelPassword = null;
+ private long downAfterMilliseconds = 60000L;
+ private long failoverTimeout = 180000L;
+ private int parallelSyncs = 1;
+ private int quorumSize = 1;
+
+ public void setMasterHost(final @NonNull String masterHost) {
+ this.masterHost = masterHost;
+ }
+
+ public void setMasterPort(final int masterPort) {
+ this.masterPort = masterPort;
+ }
+
+ public void setMasterName(final @NonNull String masterName) {
+ this.masterName = masterName;
+ }
+
+ public void setSentinelPassword(final @Nullable String sentinelPassword) {
+ this.sentinelPassword = sentinelPassword;
+ }
+
+ public void setQuorumSize(final int quorumSize) {
+ this.quorumSize = quorumSize;
+ }
+
+ public void setDownAfterMilliseconds(final long downAfterMilliseconds) {
+ this.downAfterMilliseconds = downAfterMilliseconds;
+ }
+
+ public void setFailoverTimeout(final long failoverTimeout) {
+ this.failoverTimeout = failoverTimeout;
+ }
+
+ public void setParallelSyncs(final int parallelSyncs) {
+ this.parallelSyncs = parallelSyncs;
+ }
+
+
+ @Override
+ protected void adjustConfiguration() {
+ addConfigurationOption("port " + port);
+
+ addConfigurationOption(String.format("sentinel monitor %s %s %d %d",
masterName, masterHost, masterPort, quorumSize));
+ addConfigurationOption(String.format("sentinel down-after-milliseconds
%s %d", masterName, downAfterMilliseconds));
+ addConfigurationOption(String.format("sentinel failover-timeout %s
%d", masterName, failoverTimeout));
+ addConfigurationOption(String.format("sentinel parallel-syncs %s %d",
masterName, parallelSyncs));
+
+ if (password != null) {
+ addConfigurationOption("sentinel auth-pass " + masterName + " " +
password);
+ }
+ if (sentinelPassword != null) {
+ addConfigurationOption("requirepass " + sentinelPassword);
+ addConfigurationOption("sentinel sentinel-pass " +
sentinelPassword);
+ }
+ }
+
+ @Override
+ protected void configure() {
+ super.configure();
+
+
+ List<String> commandParts = new
ArrayList<>(Arrays.asList(getCommandParts()));
+ commandParts.add("--sentinel");
+ setCommandParts(commandParts.toArray(new String[0]));
+ }
+}
