[maintenance events] Push notification listener #4194
Conversation
- 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
Register PushInvalidateConsumer after cache is initialised
ConenctionFactory should be rebound before triggering the disposal of Idle connection, so that any newly creaetd are using the proper hostname
There was a problem hiding this comment.
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_HANDLERPUBSUB_ONLY_HANDLERMaintenanceEventConsumerListenerNotificationConsumer- helps to plug customer app push listenersPushInvalidateConsumer(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 aCONSUME_ALL_HANDLER. Then current naming of flag "isProcessed" might be suboptimal in this case. Alternatives would beforwardToClientorisRequesterBound
| maintenanceEventHandler.addListener(new ConnectionRebindHandler()); | ||
| } | ||
|
|
||
| if (TimeoutOptions.isRelaxedTimeoutEnabled(config.getTimeoutOptions().getRelaxedTimeout())) { |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
I look at them as a functionality not bound to maintenance events (hitless upgrade. Maintenance events are just a way to trigger a relaxed timeout for a period of time, but this can also be done from different external triggers.
Moreover, similar to Lettuce, at some point, we can add a configurable command timeout per command type.
src/main/java/redis/clients/jedis/util/ReflectionTestUtils.java
Outdated
Show resolved
Hide resolved
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
changed the title
- Mark all pushes by default as processed - Remove CONSUME_ALL_HANDLER
We can simplify the logic by switching the default and removing the need for CONSUME_ALL_HANDLER. However, all applicable consumers must still be invoked if present. For example, if a user registers a listener for Until we refactor the Pub/Sub system and verify that no external code relies on propagated push events, we still need a mechanism to determine whether a given Push event should be propagated. Any consumer may update this decision, so the order of consumers remains important. Implementing an option for skipping a consumer is possible, but if propagation is disabled too early, again we end up with the same uncertainty. So I think the final resolution is to make sure no push messages are propagated at all, but don't think this can be handled easily and need to be tackled separately. My idea is that PushConsumers should be kept internal and should not be exposed to users. Instead, users should interact with the system via PushListener. |
|
Discussed with @atakavci and we came up with the idea that it will be more optimal if instead of invoking the consumers, for each push message and each consumer decides if it is interested to process it, it will be better for consumers to register themself for a specific message type e.g. ("INVALIDATE") and get invoked only for it. Considering this an optimisation, we will try to improve it with a separate PR. |
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:
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:
CLIENT TRACKING ONis enabled in a context other than CSC, or if other RESP3 push events are received.Changes Introduced