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 ca29d160c44 NIFI-16043 Downgrade spring-data-redis from 4.1.0 to 4.0.6
(#11395)
ca29d160c44 is described below
commit ca29d160c440fb6f233dc52769540a24bab5a383
Author: Pierre Villard <[email protected]>
AuthorDate: Mon Jul 6 15:16:45 2026 +0200
NIFI-16043 Downgrade spring-data-redis from 4.1.0 to 4.0.6 (#11395)
Signed-off-by: David Handermann <[email protected]>
---
...impleRedisDistributedMapCacheClientService.java | 4 +-
.../nifi/redis/state/RedisStateProvider.java | 13 +-
.../ITRedisDistributedMapCacheClientService.java | 204 +++++++++++++++++++++
.../nifi/redis/state/ITRedisStateProvider.java | 2 +-
.../nifi/redis/testcontainers/RedisContainer.java | 3 +
nifi-extension-bundles/nifi-redis-bundle/pom.xml | 2 +-
6 files changed, 214 insertions(+), 14 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/service/SimpleRedisDistributedMapCacheClientService.java
b/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/service/SimpleRedisDistributedMapCacheClientService.java
index 064a3cf35de..a3b144f8d09 100644
---
a/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/service/SimpleRedisDistributedMapCacheClientService.java
+++
b/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/service/SimpleRedisDistributedMapCacheClientService.java
@@ -31,7 +31,7 @@ import org.apache.nifi.redis.RedisConnectionPool;
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.connection.SetCondition;
+import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.core.types.Expiration;
import java.io.ByteArrayOutputStream;
@@ -149,7 +149,7 @@ public class SimpleRedisDistributedMapCacheClientService
extends AbstractControl
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.stringCommands().set(kv.getKey(), kv.getValue(),
SetCondition.upsert(), Expiration.seconds(ttl));
+ redisConnection.stringCommands().set(kv.getKey(), kv.getValue(),
Expiration.seconds(ttl), RedisStringCommands.SetOption.upsert());
return null;
});
}
diff --git
a/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java
b/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java
index c13ff694d66..0b78b3d989c 100644
---
a/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java
+++
b/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java
@@ -225,21 +225,14 @@ public class RedisStateProvider extends
AbstractConfigurableComponent implements
boolean replaced = false;
- // start a watch on the key so the transaction will abort if the
value is modified concurrently
+ // start a watch on the key and retrieve the current value
final byte[] key =
getComponentKey(componentId).getBytes(StandardCharsets.UTF_8);
redisConnection.watch(key);
final Optional<String> previousVersion = oldValue == null ?
Optional.empty() : oldValue.getStateVersion();
- // Once a key is being watched the connection is in queueing mode,
so a read issued on the same connection
- // is enqueued into the pending transaction and returns no value.
The current value must therefore be read
- // on a separate connection in order to execute immediately. The
watch placed above still guarantees that
- // the transaction below aborts if the value is modified after
this read.
- final RedisStateMap currStateMap;
- try (final RedisConnection readConnection = getRedis()) {
- final byte[] currValue =
readConnection.stringCommands().get(key);
- currStateMap = serDe.deserialize(currValue);
- }
+ final byte[] currValue = redisConnection.stringCommands().get(key);
+ final RedisStateMap currStateMap = serDe.deserialize(currValue);
final Optional<String> currentVersion = currStateMap == null ?
Optional.empty() : currStateMap.getStateVersion();
// start a transaction
diff --git
a/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/ITRedisDistributedMapCacheClientService.java
b/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/ITRedisDistributedMapCacheClientService.java
new file mode 100644
index 00000000000..5a0be03f166
--- /dev/null
+++
b/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/ITRedisDistributedMapCacheClientService.java
@@ -0,0 +1,204 @@
+/*
+ * 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.commons.lang3.SerializationException;
+import org.apache.nifi.distributed.cache.client.AtomicCacheEntry;
+import org.apache.nifi.distributed.cache.client.Deserializer;
+import org.apache.nifi.distributed.cache.client.Serializer;
+import
org.apache.nifi.distributed.cache.client.exception.DeserializationException;
+import org.apache.nifi.redis.testcontainers.RedisContainer;
+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.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.Container;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Integration test for {@link RedisDistributedMapCacheClientService} against
a real Redis, covering the atomic
+ * operations {@code getAndPutIfAbsent} (used by DetectDuplicate) and {@code
replace} (used by Wait/Notify).
+ * A non-zero TTL is configured so the tests exercise the
value-plus-expiration paths.
+ */
+public class ITRedisDistributedMapCacheClientService {
+
+ private static final Serializer<String> STRING_SERIALIZER = new
StringSerializer();
+ private static final Deserializer<String> STRING_DESERIALIZER = new
StringDeserializer();
+
+ public static RedisContainer redisContainer = new
RedisContainer(RedisContainer.DEFAULT_IMAGE_NAME).withExposedPorts(6379);
+
+ private TestRunner testRunner;
+ private RedisConnectionPoolService redisConnectionPool;
+ private RedisDistributedMapCacheClientService cacheClient;
+
+ @BeforeAll
+ public static void startContainer() {
+ redisContainer.start();
+ }
+
+ @AfterAll
+ public static void stopContainer() {
+ redisContainer.stop();
+ }
+
+ @BeforeEach
+ public void setup() throws InitializationException {
+ flushDatabase();
+
+ testRunner = TestRunners.newTestRunner(new FakeRedisProcessor());
+
+ redisConnectionPool = new RedisConnectionPoolService();
+ testRunner.addControllerService("redis-pool", redisConnectionPool);
+ testRunner.setProperty(redisConnectionPool, RedisUtils.REDIS_MODE,
"Standalone");
+ testRunner.setProperty(redisConnectionPool,
RedisUtils.CONNECTION_STRING,
+ redisContainer.getHost() + ":" +
redisContainer.getFirstMappedPort());
+ testRunner.enableControllerService(redisConnectionPool);
+
+ cacheClient = new RedisDistributedMapCacheClientService();
+ testRunner.addControllerService("redis-cache", cacheClient);
+ testRunner.setProperty(cacheClient, RedisUtils.REDIS_CONNECTION_POOL,
"redis-pool");
+ // A non-zero TTL exercises the value-plus-expiration path of the
atomic operations.
+ testRunner.setProperty(cacheClient, RedisUtils.TTL, "60 secs");
+ testRunner.enableControllerService(cacheClient);
+ }
+
+ @AfterEach
+ public void teardown() {
+ if (cacheClient != null) {
+ testRunner.disableControllerService(cacheClient);
+ }
+ if (redisConnectionPool != null) {
+ testRunner.disableControllerService(redisConnectionPool);
+ }
+ flushDatabase();
+ }
+
+ @Test
+ public void testGetAndPutIfAbsentReturnsNullWhenKeyAbsent() throws
IOException {
+ final String key = "detect-duplicate-absent";
+
+ final String previous = cacheClient.getAndPutIfAbsent(key, "v1",
+ STRING_SERIALIZER, STRING_SERIALIZER, STRING_DESERIALIZER);
+
+ assertNull(previous);
+ assertEquals("v1", cacheClient.get(key, STRING_SERIALIZER,
STRING_DESERIALIZER));
+ }
+
+ @Test
+ public void testGetAndPutIfAbsentReturnsExistingWhenKeyPresent() throws
IOException {
+ final String key = "detect-duplicate-present";
+
+ // first occurrence: absent, stored, returns null
+ assertNull(cacheClient.getAndPutIfAbsent(key, "v1",
+ STRING_SERIALIZER, STRING_SERIALIZER, STRING_DESERIALIZER));
+
+ // second occurrence: present -> must return the existing value and
NOT overwrite it.
+ final String previous = cacheClient.getAndPutIfAbsent(key, "v2",
+ STRING_SERIALIZER, STRING_SERIALIZER, STRING_DESERIALIZER);
+
+ assertEquals("v1", previous);
+ assertEquals("v1", cacheClient.get(key, STRING_SERIALIZER,
STRING_DESERIALIZER));
+ }
+
+ @Test
+ public void testReplaceSucceedsWithMatchingRevision() throws IOException {
+ final String key = "replace-match";
+ cacheClient.put(key, "v1", STRING_SERIALIZER, STRING_SERIALIZER);
+
+ final AtomicCacheEntry<String, String, byte[]> entry =
+ cacheClient.fetch(key, STRING_SERIALIZER, STRING_DESERIALIZER);
+ assertEquals("v1", entry.getValue());
+ assertTrue(entry.getRevision().isPresent());
+
+ entry.setValue("v2");
+ assertTrue(cacheClient.replace(entry, STRING_SERIALIZER,
STRING_SERIALIZER));
+ assertEquals("v2", cacheClient.get(key, STRING_SERIALIZER,
STRING_DESERIALIZER));
+ }
+
+ @Test
+ public void testReplaceFailsWithStaleRevision() throws IOException {
+ final String key = "replace-stale";
+ cacheClient.put(key, "v1", STRING_SERIALIZER, STRING_SERIALIZER);
+
+ final AtomicCacheEntry<String, String, byte[]> entry =
+ cacheClient.fetch(key, STRING_SERIALIZER, STRING_DESERIALIZER);
+
+ // another writer changes the key after the fetch -> the entry's
revision is now stale
+ cacheClient.put(key, "changed", STRING_SERIALIZER, STRING_SERIALIZER);
+
+ entry.setValue("v2");
+ assertFalse(cacheClient.replace(entry, STRING_SERIALIZER,
STRING_SERIALIZER));
+ assertEquals("changed", cacheClient.get(key, STRING_SERIALIZER,
STRING_DESERIALIZER));
+ }
+
+ @Test
+ public void testReplaceCreatesWhenKeyAbsent() throws IOException {
+ final String key = "replace-create";
+
+ // no revision -> create-if-absent semantics
+ final AtomicCacheEntry<String, String, byte[]> createEntry = new
AtomicCacheEntry<>(key, "v1", null);
+ assertTrue(cacheClient.replace(createEntry, STRING_SERIALIZER,
STRING_SERIALIZER));
+ assertEquals("v1", cacheClient.get(key, STRING_SERIALIZER,
STRING_DESERIALIZER));
+
+ // a second create against a now-present key must fail
+ final AtomicCacheEntry<String, String, byte[]> secondCreate = new
AtomicCacheEntry<>(key, "v2", null);
+ assertFalse(cacheClient.replace(secondCreate, STRING_SERIALIZER,
STRING_SERIALIZER));
+ assertEquals("v1", cacheClient.get(key, STRING_SERIALIZER,
STRING_DESERIALIZER));
+ }
+
+ private static void flushDatabase() {
+ try {
+ final Container.ExecResult execResult =
redisContainer.execInContainer("redis-cli", "flushall");
+ if (execResult.getExitCode() != 0) {
+ throw new IllegalStateException(String.format("Failed to flush
Redis container: %s%s",
+ execResult.getStdout(), execResult.getStderr()));
+ }
+ } catch (InterruptedException interruptedException) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("Interrupted while flushing Redis
container", interruptedException);
+ } catch (IOException ioException) {
+ throw new IllegalStateException("Failed to flush Redis container",
ioException);
+ }
+ }
+
+ private static class StringSerializer implements Serializer<String> {
+ @Override
+ public void serialize(final String value, final OutputStream output)
throws SerializationException, IOException {
+ output.write(value.getBytes(StandardCharsets.UTF_8));
+ }
+ }
+
+ private static class StringDeserializer implements Deserializer<String> {
+ @Override
+ public String deserialize(final byte[] input) throws
DeserializationException {
+ return input == null || input.length == 0 ? null : new
String(input, StandardCharsets.UTF_8);
+ }
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/ITRedisStateProvider.java
b/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/ITRedisStateProvider.java
index 3eb7bfcc22f..7968515f416 100644
---
a/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/ITRedisStateProvider.java
+++
b/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/ITRedisStateProvider.java
@@ -55,7 +55,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class ITRedisStateProvider {
protected final String componentId =
"111111111-1111-1111-1111-111111111111";
- public static RedisContainer redisContainer = new
RedisContainer("redis:8.8-alpine").withExposedPorts(6379);
+ public static RedisContainer redisContainer = new
RedisContainer(RedisContainer.DEFAULT_IMAGE_NAME).withExposedPorts(6379);
private RedisStateProvider provider;
diff --git
a/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisContainer.java
b/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisContainer.java
index 3fde902ca23..305db2c87d1 100644
---
a/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisContainer.java
+++
b/nifi-extension-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisContainer.java
@@ -34,6 +34,9 @@ public class RedisContainer extends
GenericContainer<RedisContainer> {
public static final int REDIS_PORT = 6379;
+ // Single source of the Redis image used across the bundle's integration
tests.
+ public static final String DEFAULT_IMAGE_NAME = "redis:8.8-alpine";
+
public RedisContainer(@NonNull DockerImageName dockerImageName) {
super(dockerImageName);
}
diff --git a/nifi-extension-bundles/nifi-redis-bundle/pom.xml
b/nifi-extension-bundles/nifi-redis-bundle/pom.xml
index ab0f9049810..6025f40c6ba 100644
--- a/nifi-extension-bundles/nifi-redis-bundle/pom.xml
+++ b/nifi-extension-bundles/nifi-redis-bundle/pom.xml
@@ -25,7 +25,7 @@
<packaging>pom</packaging>
<properties>
- <spring.data.redis.version>4.1.0</spring.data.redis.version>
+ <spring.data.redis.version>4.0.6</spring.data.redis.version>
<jedis.version>7.5.3</jedis.version>
</properties>
