Push notification listener #4194

ggivo · 2025-07-01T12:17:17Z

Description

This PR enhances Jedis and UnifiedJedis with improved support for RESP3 push messages and introduces a configurable option for relaxed timeout behavior.

These changes are part of a broader effort to support more robust and controlled upgrade and maintenance scenarios. Specifically, they lay the groundwork for:

Temporarily relaxing timeouts during planned maintenance windows.
Gracefully re-binding pooled connections to updated endpoints without disrupting active operations.

The goal is to increase resilience and flexibility in production environments that require connection stability during topology changes or Redis upgrades.

Backround

Starting with RESP3, Redis servers can send push messages to clients out-of-band (i.e., outside the standard request/response flow). See the RESP3 Push Specification for more details.

In the current state of the client:

Jedis supports only known Push notifications like invalidate. Push notifications are supported using dedicated client-side caching (CSC) connection (CacheConnection)
This leads to runtime errors if CLIENT TRACKING ON is enabled in a context other than CSC, or if other RESP3 push events are received.

Changes Introduced

Added support for processing and discarding all RESP3 push messages, preventing errors in non-CSC scenarios.
Introduced support for registering a custom PushListener, enabling applications to handle push notifications as needed.
Pub/Sub push messages are still handled as before and propagated to the client for backward compatibility.
Added a configurable option to enable relaxed timeout behavior, intended for usage during maintenance events or when expecting slow responses from the server.

- Preparation step for processing custom push notifications - Push notification can appear out-of band in-between executed commands - Current Connection implementation does not support out of band Push notifications - Meaning it will crash if "CLIENT TRACKING ON is enabled" on regular Jedis Connection and "invalidation" push event is triggered This commit provides a way to register push handler for the connection which process incoming push messages, before actual command is executed. To preserve backward compatibility unprocessed push messages are forward to application logic as before. - By default Connection will start with NOOP push handler which marks any incoming push event as processed and skips it - On subcsribe/psubscribe a dedicated push handler is registered which propagates to the app only supported push vents such as (message, subscribe, unsubscribe ...) - CacheConection is refactored to use a push handler handling "invalidate" push events only, and skipping any other

This commit adds a new PushHandlerChain class that implements the Chain of Responsibility pattern for Redis RESP3 push message handling. Key features: - Allows composing multiple PushHandlers in a processing chain - Push events propagate through the complete chain in sequence - Events marked as not processed are propagated to the client application - Provides both constructor-based and fluent builder API for chain creation - Includes predefined handlers for common use cases (CONSUME_ALL, PROPAGATE_ALL) - Supports immutable chain transformations via methods like then(), The chain approach provides a flexible way to handle different types of push messages (invalidations, pub/sub, etc.) with specialized handlers while maintaining a clean separation of concerns. Example usage: PushHandlerChain chain = PushHandlerChain.of(loggingHandler) .then(invalidationHandler) .then(PushHandlerChain.PROPAGATE_PUB_SUB_PUSH_HANDLER);

- code clean up - added relaxed timeout configuration - fix unit tests

… constructors.

Register PushInvalidateConsumer after cache is initialised

…y leak

src/main/java/redis/clients/jedis/csc/CacheConnection.java

src/main/java/redis/clients/jedis/Jedis.java

src/main/java/redis/clients/jedis/JedisShardedPubSubBase.java

src/main/java/redis/clients/jedis/JedisClientConfig.java

src/main/java/redis/clients/jedis/Protocol.java

src/main/java/redis/clients/jedis/util/JedisAsserts.java

ConenctionFactory should be rebound before triggering the disposal of Idle connection, so that any newly creaetd are using the proper hostname

atakavci

looks good!! there are couple of things not entirely clear to me. Just to clear them on my mind i will summarize as i understand;
all push messages goes through all the consumers in the chain,
in current impl, they go through;

CONSUME_ALL_HANDLER
PUBSUB_ONLY_HANDLER
MaintenanceEventConsumer
ListenerNotificationConsumer - helps to plug customer app push listeners
PushInvalidateConsumer (in case of CSC -CacheConnection-)

what i don't like is each push messages visit all consumers no matter what. And the last one in the list, it kind of decides to set if data handed back to command execution or not.
so we expect each consumer to do whatever it wants to do with message and set the context appropriately.
this setup brings both flexibility and chaotic behaviour as well, because if someone shows up and marks it as they wish, they challenge to other consumer implementations plugged previously.

i think no push messages, other than pub/sub ones should be handed back to command executer anyway.
(and at some point, even they are not going to following the command response path in the case we introduce a way to customer app handle pub/sub messages via push listeners).
IMHO there should be something separate from letting client received/notified of push messages as well. May be with PUBSUB_ONLY_HANDLER functions the same way with ListenerNotificationConsumer but only for pub/sub messages.

let me try to explain the way i see it;

a consumer chooses if each message propagates further to next consumer or not, and returns it as a value of consume result.
Each consumer along the chain should receive only what is let by its preceding consumers. this secures if messages should be seen or not, by any other consumer. Also this is more efficient in case there is a load of invalidation and/or pubsub messages.

interface PushConsumer {
   bool consume(PushMessageContext);
}

no messages other than pub/sub ones handed back to customer app (or received by command executor). isProcessed flag in context still remains the same to identify if message is chosen to be returned to the app
each push message initializes with default isProcessed=true and there is no need for a CONSUME_ALL_HANDLER. Then current naming of flag "isProcessed" might be suboptimal in this case. Alternatives would be forwardToClient or isRequesterBound

atakavci · 2025-07-21T07:54:59Z

src/main/java/redis/clients/jedis/Connection.java

+          maintenanceEventHandler.addListener(new ConnectionRebindHandler());
+        }
+
+        if (TimeoutOptions.isRelaxedTimeoutEnabled(config.getTimeoutOptions().getRelaxedTimeout())) {


are we sure relaxedTimeout s are something that makes sense as standalone item? All this should be considered part of hitless upgrade setup and shouldnt be some standalone item in the configuration.

atakavci · 2025-07-21T08:25:42Z

src/main/java/redis/clients/jedis/Connection.java

+  }
+
+
+  private static class MaintenanceEventConsumer implements PushConsumer {


nit: these types(ListenerNotificationConsumer, MaintenanceEventConsumer, ..) better be located in their own files for the sake of readability, maintainability etc..
May be as package private.

atakavci · 2025-07-21T08:33:49Z

src/main/java/redis/clients/jedis/PushConsumerChain.java

+      }
+    }
+
+    private boolean isPubSubType(String type) {


nit: not sure it adds any value to have this as a method.

atakavci · 2025-07-21T08:53:31Z

src/main/java/redis/clients/jedis/util/ReflectionTestUtils.java

+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+
+public class ReflectionTestUtils {


lets move this to test folder

atakavci · 2025-07-21T09:17:39Z

src/main/java/redis/clients/jedis/TimeoutOptions.java

+  public static boolean isRelaxedTimeoutEnabled(Duration relaxedTimeout) {
+    return  relaxedTimeout != null && !relaxedTimeout.equals(DISABLED_TIMEOUT);
+  }
+
+  public static boolean isRelaxedTimeoutDisabled(Duration relaxedTimeout) {
+    return  relaxedTimeout == null || relaxedTimeout.equals(DISABLED_TIMEOUT);
+  }
+
+  public static boolean isRelaxedTimeoutDisabled(int relaxedTimeout) {
+    return  relaxedTimeout == DISABLED_TIMEOUT_MS;
+  }


do we really need all of this? first one should be enough, likely to be the optimal one.

personal POV; if there is nothing planned to built/add more on TimeoutOptions at the moment, i'd say this options class itself is an overkill.

atakavci · 2025-07-21T12:29:25Z

src/main/java/redis/clients/jedis/Protocol.java

@@ -211,22 +214,52 @@ private static List<KeyValue> processMapKeyValueReply(final RedisInputStream is)
      default:
        final List<KeyValue> ret = new ArrayList<>(num);
        for (int i = 0; i < num; i++) {
-          ret.add(new KeyValue(process(is), process(is)));
+          ret.add(new KeyValue(process(is, null), process(is,null)));
        }
        return ret;
    }
  }

  public static Object read(final RedisInputStream is) {


is this method remains just for test purposes?

atakavci · 2025-07-22T07:30:11Z

src/main/java/redis/clients/jedis/Protocol.java

+    // or push-event is not handled and need to be propagated to application
+    Object reply;
+    do {
+      reply = process(is, pushConsumer);


its only here that we expect a push message (instance of PushConsumerContext) can be returned. so lets return a wrapper type that contains if its a push message or command response. Then, we wont need to use class type checking like within isPush.

atakavci · 2025-07-22T07:42:48Z

src/main/java/redis/clients/jedis/PushHandler.java

+ * @author Ivo Gaydajiev
+ * @since 6.1
+ */
+public interface PushHandler {


same for MaintenanceEventHandler

atakavci · 2025-07-22T07:57:20Z

src/main/java/redis/clients/jedis/ConnectionPool.java

+    @Override
+    public void onRebind(HostAndPort target) {
+      HostAndPort previous = rebindTarget.getAndSet(target);
+      if (previous != target) {


is it possible to trigger same handler with multiple rebind on different targets?
not sure how it should be but if we rely on host+port equality, lets use equals.

atakavci · 2025-07-22T08:13:28Z

src/main/java/redis/clients/jedis/Connection.java

+    }
+  }
+
+  private static int safeToInt(long millis) {


nit: move it to some util class.. may be a new one like NumberUtils and which also exposes DoublePrecision methods and retire DoublePrecision as well.

Issue : If Maintenace notifications are received during blocking command, relaxTimeout is enforced instead of infinit timeout. Fix: Introduce dedicated relax timeout setting for blocking commands. It will fall back to infinit timeout if not set

ggivo added 3 commits June 23, 2025 13:23

Handle relax timeout for maintenance events

4aec187

- code clean up - added relaxed timeout configuration - fix unit tests

ggivo marked this pull request as draft July 1, 2025 12:17

ggivo added 6 commits July 2, 2025 13:22

Merge branch 'master' into feature/hu-notifications

67d4afa

Support custom Push listeners for Jedis client

ab62e9d

Add proactiveRebindEnabled configuration option

b36d7f8

PushHandler is now provided through JedisClientConfig instead through…

0452a93

… constructors.

Fix NPE in CacheConnection

4bd38da

Register PushInvalidateConsumer after cache is initialised

[cleanup] Use weak reference in AdaptiveTimeoutHandler to avoid memor…

4ad3fce

…y leak

ggivo self-assigned this Jul 7, 2025

ggivo added 2 commits July 7, 2025 13:33

[cleanup] Fix javadoc errors

f475177

[cleanup] Fix TransactionCommandsTest mocked test

56cd409

a-TODO-rov reviewed Jul 8, 2025

View reviewed changes

ggivo added 8 commits July 9, 2025 09:23

Moving/Rebind initial support

175b773

Mocked relaxed timeout test

c19d8a5

Mocked rebind test

4ee525d

Fix : wrong order connection.rebind pool.clear

b940ad7

ConenctionFactory should be rebound before triggering the disposal of Idle connection, so that any newly creaetd are using the proper hostname

[clean up] Address review comments from a-TODO-rov

7efbe72

add more rebind tests

b0987e0

clean up

7e11737

clean up remove unused test method

b769492

atakavci reviewed Jul 22, 2025

View reviewed changes

ggivo added 2 commits July 24, 2025 09:00

fix relaxed timeout on blocking command

e7ffd1d

Issue : If Maintenace notifications are received during blocking command, relaxTimeout is enforced instead of infinit timeout. Fix: Introduce dedicated relax timeout setting for blocking commands. It will fall back to infinit timeout if not set

format

a67b5cb

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Push notification listener #4194

Push notification listener #4194

Uh oh!

ggivo commented Jul 1, 2025 •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

atakavci left a comment

Uh oh!

atakavci Jul 21, 2025

Uh oh!

atakavci Jul 21, 2025

Uh oh!

atakavci Jul 21, 2025

Uh oh!

atakavci Jul 21, 2025

Uh oh!

atakavci Jul 21, 2025

Uh oh!

atakavci Jul 21, 2025

Uh oh!

atakavci Jul 22, 2025

Uh oh!

atakavci Jul 22, 2025

Uh oh!

atakavci Jul 22, 2025

Uh oh!

atakavci Jul 22, 2025

Uh oh!

Uh oh!

		}


		private static class MaintenanceEventConsumer implements PushConsumer {

Push notification listener #4194

Are you sure you want to change the base?

Push notification listener #4194

Uh oh!

Conversation

ggivo commented Jul 1, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Backround

Changes Introduced

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

atakavci left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

ggivo commented Jul 1, 2025 •

edited

Loading