From e11b9867260fbff3e7eb93ffed571636f1c58af7 Mon Sep 17 00:00:00 2001 From: ggivo Date: Fri, 1 Aug 2025 11:49:44 +0300 Subject: [PATCH 01/16] Support for Client-side opt-in A client can tell the server if it wants to receive maintenance push notifications via the following command: CLIENT MAINT_NOTIFICATIONS [parameter value parameter value ...] --- .../io/lettuce/core/AbstractRedisClient.java | 10 +- .../java/io/lettuce/core/ClientOptions.java | 32 +-- .../io/lettuce/core/ConnectionBuilder.java | 2 +- .../io/lettuce/core/ConnectionMetadata.java | 12 ++ .../core/MaintenanceEventsOptions.java | 191 ++++++++++++++++++ .../java/io/lettuce/core/RedisHandshake.java | 47 ++++- .../java/io/lettuce/core/TimeoutOptions.java | 4 +- .../core/protocol/CommandExpiryWriter.java | 2 +- .../lettuce/core/protocol/CommandKeyword.java | 2 +- .../protocol/MaintenanceAwareComponent.java | 2 +- .../MaintenanceAwareConnectionWatchdog.java | 94 +++++---- .../MaintenanceAwareExpiryWriter.java | 4 +- .../io/lettuce/core/protocol/RebindState.java | 2 +- .../LettuceMaintenanceEventsDemo.java | 3 +- .../lettuce/core/RedisHandshakeUnitTests.java | 12 +- ...nanceAwareConnectionWatchdogUnitTests.java | 118 ++++++++--- ...MaintenanceAwareExpiryWriterUnitTests.java | 14 +- 17 files changed, 444 insertions(+), 107 deletions(-) create mode 100644 src/main/java/io/lettuce/core/MaintenanceEventsOptions.java diff --git a/src/main/java/io/lettuce/core/AbstractRedisClient.java b/src/main/java/io/lettuce/core/AbstractRedisClient.java index fcb65f2914..b2a01c632c 100644 --- a/src/main/java/io/lettuce/core/AbstractRedisClient.java +++ b/src/main/java/io/lettuce/core/AbstractRedisClient.java @@ -20,6 +20,8 @@ package io.lettuce.core; import java.io.Closeable; +import java.net.InetAddress; +import java.net.InetSocketAddress; import java.net.SocketAddress; import java.time.Duration; import java.util.ArrayList; @@ -34,6 +36,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import io.lettuce.core.MaintenanceEventsOptions.AddressTypeSource; import reactor.core.publisher.Mono; import io.lettuce.core.event.command.CommandListener; import io.lettuce.core.event.connection.ConnectEvent; @@ -629,8 +632,13 @@ private CompletableFuture closeClientResources(long quietPeriod, long time } protected RedisHandshake createHandshake(ConnectionState state) { + AddressTypeSource source = null; + if (clientOptions.getMaintenanceEventsOptions().supportsMaintenanceEvents()) { + source = clientOptions.getMaintenanceEventsOptions().getAddressTypeSource(); + } + return new RedisHandshake(clientOptions.getConfiguredProtocolVersion(), clientOptions.isPingBeforeActivateConnection(), - state); + state, source); } } diff --git a/src/main/java/io/lettuce/core/ClientOptions.java b/src/main/java/io/lettuce/core/ClientOptions.java index 4fb91c9506..31ef2fa059 100644 --- a/src/main/java/io/lettuce/core/ClientOptions.java +++ b/src/main/java/io/lettuce/core/ClientOptions.java @@ -52,7 +52,7 @@ public class ClientOptions implements Serializable { public static final boolean DEFAULT_AUTO_RECONNECT = true; - public static final boolean DEFAULT_SUPPORT_MAINTENANCE_EVENTS = false; + public static final MaintenanceEventsOptions DEFAULT_MAINTENANCE_EVENTS_OPTIONS = MaintenanceEventsOptions.disabled(); public static final Predicate> DEFAULT_REPLAY_FILTER = (cmd) -> false; @@ -98,7 +98,7 @@ public class ClientOptions implements Serializable { private final boolean autoReconnect; - private final boolean supportMaintenanceEvents; + private final MaintenanceEventsOptions maintenanceEventsOptions; private final Predicate> replayFilter; @@ -136,7 +136,7 @@ public class ClientOptions implements Serializable { protected ClientOptions(Builder builder) { this.autoReconnect = builder.autoReconnect; - this.supportMaintenanceEvents = builder.supportMaintenanceEvents; + this.maintenanceEventsOptions = builder.maintenanceEventsOptions; this.replayFilter = builder.replayFilter; this.cancelCommandsOnReconnectFailure = builder.cancelCommandsOnReconnectFailure; this.decodeBufferPolicy = builder.decodeBufferPolicy; @@ -158,7 +158,7 @@ protected ClientOptions(Builder builder) { protected ClientOptions(ClientOptions original) { this.autoReconnect = original.isAutoReconnect(); - this.supportMaintenanceEvents = original.supportsMaintenanceEvents(); + this.maintenanceEventsOptions = original.getMaintenanceEventsOptions(); this.replayFilter = original.getReplayFilter(); this.cancelCommandsOnReconnectFailure = original.isCancelCommandsOnReconnectFailure(); this.decodeBufferPolicy = original.getDecodeBufferPolicy(); @@ -213,7 +213,7 @@ public static class Builder { private boolean autoReconnect = DEFAULT_AUTO_RECONNECT; - private boolean supportMaintenanceEvents = DEFAULT_SUPPORT_MAINTENANCE_EVENTS; + private MaintenanceEventsOptions maintenanceEventsOptions = DEFAULT_MAINTENANCE_EVENTS_OPTIONS; private Predicate> replayFilter = DEFAULT_REPLAY_FILTER; @@ -268,14 +268,14 @@ public Builder autoReconnect(boolean autoReconnect) { * Configure whether the driver should listen for server events that notify on current maintenance activities. When * enabled, this option will help with the connection handover and reduce the number of failed commands. This feature * requires the server to support maintenance events. Defaults to {@code false}. See - * {@link #DEFAULT_SUPPORT_MAINTENANCE_EVENTS}. + * {@link #DEFAULT_MAINTENANCE_EVENTS_OPTIONS}. * - * @param supportEvents true/false + * @param maintenanceEventsOptions true/false * @return {@code this} * @since 7.0 */ - public Builder supportMaintenanceEvents(boolean supportEvents) { - this.supportMaintenanceEvents = supportEvents; + public Builder supportMaintenanceEvents(MaintenanceEventsOptions maintenanceEventsOptions) { + this.maintenanceEventsOptions = maintenanceEventsOptions; return this; } @@ -574,7 +574,7 @@ public ClientOptions build() { public ClientOptions.Builder mutate() { Builder builder = new Builder(); - builder.autoReconnect(isAutoReconnect()).supportMaintenanceEvents(supportsMaintenanceEvents()) + builder.autoReconnect(isAutoReconnect()).supportMaintenanceEvents(getMaintenanceEventsOptions()) .cancelCommandsOnReconnectFailure(isCancelCommandsOnReconnectFailure()).replayFilter(getReplayFilter()) .decodeBufferPolicy(getDecodeBufferPolicy()).disconnectedBehavior(getDisconnectedBehavior()) .reauthenticateBehavior(getReauthenticateBehaviour()).readOnlyCommands(getReadOnlyCommands()) @@ -601,15 +601,15 @@ public boolean isAutoReconnect() { } /** - * Returns whether the client supports maintenance events. + * Returns the {@link MaintenanceEventsOptions} to listen for server events that notify on current maintenance activities. * - * @return {@code true} if maintenance events are supported. + * @return {@link MaintenanceEventsOptions} * @since 7.0 - * @see #DEFAULT_SUPPORT_MAINTENANCE_EVENTS - * @see #supportsMaintenanceEvents() + * @see #DEFAULT_MAINTENANCE_EVENTS_OPTIONS + * @see #getMaintenanceEventsOptions() */ - public boolean supportsMaintenanceEvents() { - return supportMaintenanceEvents; + public MaintenanceEventsOptions getMaintenanceEventsOptions() { + return maintenanceEventsOptions; } /** diff --git a/src/main/java/io/lettuce/core/ConnectionBuilder.java b/src/main/java/io/lettuce/core/ConnectionBuilder.java index b86b276ea8..bb5732fa7e 100644 --- a/src/main/java/io/lettuce/core/ConnectionBuilder.java +++ b/src/main/java/io/lettuce/core/ConnectionBuilder.java @@ -155,7 +155,7 @@ protected ConnectionWatchdog createConnectionWatchdog() { LettuceAssert.assertState(socketAddressSupplier != null, "SocketAddressSupplier must be set for autoReconnect=true"); ConnectionWatchdog watchdog; - if (clientOptions.supportsMaintenanceEvents()) { + if (clientOptions.getMaintenanceEventsOptions().supportsMaintenanceEvents()) { watchdog = new MaintenanceAwareConnectionWatchdog(clientResources.reconnectDelay(), clientOptions, bootstrap, clientResources.timer(), clientResources.eventExecutorGroup(), socketAddressSupplier, reconnectionListener, connection, clientResources.eventBus(), endpoint); diff --git a/src/main/java/io/lettuce/core/ConnectionMetadata.java b/src/main/java/io/lettuce/core/ConnectionMetadata.java index d74c839b25..db9b87b721 100644 --- a/src/main/java/io/lettuce/core/ConnectionMetadata.java +++ b/src/main/java/io/lettuce/core/ConnectionMetadata.java @@ -11,6 +11,8 @@ class ConnectionMetadata { private volatile String libraryVersion; + private volatile boolean sslEnabled; + public ConnectionMetadata() { } @@ -23,6 +25,7 @@ public void apply(RedisURI redisURI) { setClientName(redisURI.getClientName()); setLibraryName(redisURI.getLibraryName()); setLibraryVersion(redisURI.getLibraryVersion()); + setSslEnabled(redisURI.isSsl()); } public void apply(ConnectionMetadata metadata) { @@ -30,6 +33,7 @@ public void apply(ConnectionMetadata metadata) { setClientName(metadata.getClientName()); setLibraryName(metadata.getLibraryName()); setLibraryVersion(metadata.getLibraryVersion()); + setSslEnabled(metadata.isSslEnabled()); } protected void setClientName(String clientName) { @@ -56,4 +60,12 @@ String getLibraryVersion() { return libraryVersion; } + boolean isSslEnabled() { + return sslEnabled; + } + + void setSslEnabled(boolean sslEnabled) { + this.sslEnabled = sslEnabled; + } + } diff --git a/src/main/java/io/lettuce/core/MaintenanceEventsOptions.java b/src/main/java/io/lettuce/core/MaintenanceEventsOptions.java new file mode 100644 index 0000000000..4e68e790d2 --- /dev/null +++ b/src/main/java/io/lettuce/core/MaintenanceEventsOptions.java @@ -0,0 +1,191 @@ +/* + * Copyright 2011-Present, Redis Ltd. and Contributors + * All rights reserved. + * + * Licensed under the MIT License. + * + * This file contains contributions from third-party contributors + * licensed 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 + * + * https://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 io.lettuce.core; + +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.SocketAddress; + +public class MaintenanceEventsOptions { + + public static final boolean DEFAULT_SUPPORT_MAINTENANCE_EVENTS = false; + + private final boolean supportMaintenanceEvents; + + private final AddressTypeSource addressTypeSource; + + protected MaintenanceEventsOptions(MaintenanceEventsOptions.Builder builder) { + this.addressTypeSource = builder.addressTypeSource; + this.supportMaintenanceEvents = builder.supportMaintenanceEvents; + } + + public static MaintenanceEventsOptions.Builder builder() { + return new MaintenanceEventsOptions.Builder(); + } + + public static MaintenanceEventsOptions create() { + return builder().build(); + } + + public static MaintenanceEventsOptions disabled() { + return builder().supportMaintenanceEvents(false).build(); + } + + public static MaintenanceEventsOptions enabled() { + return builder().supportMaintenanceEvents().autoResolveAddressType().build(); + } + + public static MaintenanceEventsOptions enabled(AddressType addressType) { + return builder().supportMaintenanceEvents().fixedAddressType(addressType).build(); + } + + public boolean supportsMaintenanceEvents() { + return supportMaintenanceEvents; + } + + /** + * @return the address type source to determine the requested address type when maintenance events are enabled . Can be + * {@code null} if {@link #supportsMaintenanceEvents()} is {@code false}. + */ + public AddressTypeSource getAddressTypeSource() { + return addressTypeSource; + } + + public static class Builder { + + private boolean supportMaintenanceEvents = DEFAULT_SUPPORT_MAINTENANCE_EVENTS; + + private AddressTypeSource addressTypeSource; + + public MaintenanceEventsOptions.Builder supportMaintenanceEvents() { + return supportMaintenanceEvents(true); + } + + public MaintenanceEventsOptions.Builder supportMaintenanceEvents(boolean supportMaintenanceEvents) { + this.supportMaintenanceEvents = supportMaintenanceEvents; + return this; + } + + public Builder fixedAddressType(AddressType addressType) { + this.addressTypeSource = new FixedAddressTypeSource(addressType); + return this; + } + + public Builder autoResolveAddressType() { + this.addressTypeSource = new AutoresolveAddressTypeSource(); + return this; + } + + public MaintenanceEventsOptions build() { + return new MaintenanceEventsOptions(this); + } + + } + + public enum AddressType { + INTERNAL_IP, INTERNAL_FQDN, PUBLIC_IP, PUBLIC_FQDN + } + + private static class FixedAddressTypeSource extends MaintenanceEventsOptions.AddressTypeSource { + + private final AddressType addressType; + + FixedAddressTypeSource(AddressType addressType) { + + this.addressType = addressType; + } + + @Override + public AddressType getAddressType(SocketAddress socketAddress, boolean sslEnabled) { + return addressType; + } + + } + + private static class AutoresolveAddressTypeSource extends MaintenanceEventsOptions.AddressTypeSource { + + AutoresolveAddressTypeSource() { + } + + @Override + public MaintenanceEventsOptions.AddressType getAddressType(SocketAddress socketAddress, boolean sslEnabled) { + if (isReservedIp(socketAddress)) { + // use private + if (sslEnabled) { + return MaintenanceEventsOptions.AddressType.INTERNAL_FQDN; + } else { + return MaintenanceEventsOptions.AddressType.INTERNAL_IP; + } + } else { + // use public + if (sslEnabled) { + return MaintenanceEventsOptions.AddressType.PUBLIC_FQDN; + } else { + return MaintenanceEventsOptions.AddressType.PUBLIC_IP; + } + } + } + + public static boolean isReservedIp(SocketAddress socketAddress) { + if (!(socketAddress instanceof InetSocketAddress)) { + return false; + } + + InetSocketAddress inetSocketAddress = (InetSocketAddress) socketAddress; + InetAddress address = inetSocketAddress.getAddress(); + + if (address == null || address.isAnyLocalAddress() || address.isLoopbackAddress()) { + return false; + } + + byte[] bytes = address.getAddress(); + + // IPv4 only + if (bytes.length != 4) { + return false; + } + + int firstByte = bytes[0] & 0xFF; + int secondByte = bytes[1] & 0xFF; + + // 10.0.0.0/8 + if (firstByte == 10) + return true; + + // 172.16.0.0/12 + if (firstByte == 172 && (secondByte >= 16 && secondByte <= 31)) + return true; + + // 192.168.0.0/16 + if (firstByte == 192 && secondByte == 168) + return true; + + return false; + } + + } + + public static abstract class AddressTypeSource { + + public abstract AddressType getAddressType(SocketAddress socketAddress, boolean sslEnabled); + + } + +} diff --git a/src/main/java/io/lettuce/core/RedisHandshake.java b/src/main/java/io/lettuce/core/RedisHandshake.java index bf5ad7e4b5..e07d69d2b5 100644 --- a/src/main/java/io/lettuce/core/RedisHandshake.java +++ b/src/main/java/io/lettuce/core/RedisHandshake.java @@ -29,18 +29,24 @@ import java.util.regex.Matcher; import java.util.regex.Pattern; +import io.lettuce.core.MaintenanceEventsOptions.AddressTypeSource; import io.lettuce.core.codec.StringCodec; import io.lettuce.core.internal.Futures; import io.lettuce.core.internal.LettuceAssert; import io.lettuce.core.internal.LettuceStrings; +import io.lettuce.core.output.StatusOutput; import io.lettuce.core.protocol.AsyncCommand; import io.lettuce.core.protocol.Command; +import io.lettuce.core.protocol.CommandArgs; import io.lettuce.core.protocol.ConnectionInitializer; import io.lettuce.core.protocol.ProtocolVersion; import io.netty.channel.Channel; import io.netty.util.internal.logging.InternalLogger; import io.netty.util.internal.logging.InternalLoggerFactory; +import static io.lettuce.core.protocol.CommandKeyword.MAINT_NOTIFICATIONS; +import static io.lettuce.core.protocol.CommandType.CLIENT; + /** * Redis RESP2/RESP3 handshake using the configured {@link ProtocolVersion} and other options for connection initialization and * connection state restoration. This class is part of the internal API. @@ -63,8 +69,12 @@ class RedisHandshake implements ConnectionInitializer { private volatile ProtocolVersion negotiatedProtocolVersion; - RedisHandshake(ProtocolVersion requestedProtocolVersion, boolean pingOnConnect, ConnectionState connectionState) { + private final MaintenanceEventsOptions.AddressTypeSource addressTypeSource; + + RedisHandshake(ProtocolVersion requestedProtocolVersion, boolean pingOnConnect, ConnectionState connectionState, + MaintenanceEventsOptions.AddressTypeSource addressTypeSource) { + this.addressTypeSource = addressTypeSource; this.requestedProtocolVersion = requestedProtocolVersion; this.pingOnConnect = pingOnConnect; this.connectionState = connectionState; @@ -261,6 +271,19 @@ private CompletableFuture applyPostHandshake(Channel channel) { postHandshake.add(new AsyncCommand<>(this.commandBuilder.readOnly())); } + if (addressTypeSource != null) { + CommandArgs args = new CommandArgs<>(StringCodec.UTF8).add(MAINT_NOTIFICATIONS).add("on"); + String addressType = addressType(channel, connectionState, addressTypeSource); + + if (addressType != null) { + args.add("moving-endpoint-type").add(addressType); + } + + Command maintNotificationsOn = new Command<>(CLIENT, new StatusOutput<>(StringCodec.UTF8), + args); + postHandshake.add(new AsyncCommand<>(maintNotificationsOn)); + } + if (postHandshake.isEmpty()) { return CompletableFuture.completedFuture(null); } @@ -268,6 +291,28 @@ private CompletableFuture applyPostHandshake(Channel channel) { return dispatch(channel, postHandshake); } + private String addressType(Channel channel, ConnectionState state, AddressTypeSource addressTypeSource) { + MaintenanceEventsOptions.AddressType addressType = addressTypeSource.getAddressType(channel.remoteAddress(), + state.getConnectionMetadata().isSslEnabled()); + + if (addressType == null) { + return null; + } + + switch (addressType) { + case INTERNAL_IP: + return "internal-ip"; + case INTERNAL_FQDN: + return "internal-fqdn"; + case PUBLIC_IP: + return "external-ip"; + case PUBLIC_FQDN: + return "external-fqdn"; + default: + throw new IllegalArgumentException("Unknown moving endpoint address type:" + addressType); + } + } + private CompletionStage applyConnectionMetadataSafely(Channel channel) { return applyConnectionMetadata(channel).handle((result, error) -> { if (error != null) { diff --git a/src/main/java/io/lettuce/core/TimeoutOptions.java b/src/main/java/io/lettuce/core/TimeoutOptions.java index 434db7cc9e..ac96cbace3 100644 --- a/src/main/java/io/lettuce/core/TimeoutOptions.java +++ b/src/main/java/io/lettuce/core/TimeoutOptions.java @@ -121,7 +121,7 @@ public Builder timeoutCommands(boolean enabled) { * Enable timeout relaxing during maintenance events. Disabled by default, see {@link #DEFAULT_RELAXED_TIMEOUT}. *

* If the Redis server supports sending maintenance events, and the client is set up to use that by the - * {@link ClientOptions#supportsMaintenanceEvents()} option, the client would listen to notifications that the current + * {@link ClientOptions#getMaintenanceEventsOptions()} option, the client would listen to notifications that the current * endpoint is about to go down (as part of some maintenance activity, for example). In such cases, the driver could * extend the existing timeout settings for newly issued commands, or such that are in flight, to make sure they do not * time out during this process. These commands could be either a part of the offline buffer or waiting for a reply. @@ -129,7 +129,7 @@ public Builder timeoutCommands(boolean enabled) { * @param duration {@link Duration} to relax timeouts proactively, must not be {@code null}. * @return {@code this} * @since 7.0 - * @see ClientOptions#supportsMaintenanceEvents() + * @see ClientOptions#getMaintenanceEventsOptions() */ public Builder timeoutsRelaxingDuringMaintenance(Duration duration) { LettuceAssert.notNull(duration, "Duration must not be null"); diff --git a/src/main/java/io/lettuce/core/protocol/CommandExpiryWriter.java b/src/main/java/io/lettuce/core/protocol/CommandExpiryWriter.java index b637ada08c..c57ef515cb 100644 --- a/src/main/java/io/lettuce/core/protocol/CommandExpiryWriter.java +++ b/src/main/java/io/lettuce/core/protocol/CommandExpiryWriter.java @@ -95,7 +95,7 @@ public CommandExpiryWriter(RedisChannelWriter delegate, ClientOptions clientOpti */ public static RedisChannelWriter buildCommandExpiryWriter(RedisChannelWriter delegate, ClientOptions clientOptions, ClientResources clientResources) { - if (clientOptions.supportsMaintenanceEvents()) { + if (clientOptions.getMaintenanceEventsOptions().supportsMaintenanceEvents()) { return new MaintenanceAwareExpiryWriter(delegate, clientOptions, clientResources); } else { return new CommandExpiryWriter(delegate, clientOptions, clientResources); diff --git a/src/main/java/io/lettuce/core/protocol/CommandKeyword.java b/src/main/java/io/lettuce/core/protocol/CommandKeyword.java index 628365ca4e..fb1cde32c6 100644 --- a/src/main/java/io/lettuce/core/protocol/CommandKeyword.java +++ b/src/main/java/io/lettuce/core/protocol/CommandKeyword.java @@ -47,7 +47,7 @@ public enum CommandKeyword implements ProtocolKeyword { RESETSTAT, RESTART, RETRYCOUNT, REWRITE, RIGHT, SAVECONFIG, SDSLEN, SETINFO, SETNAME, SETSLOT, SHARDS, SLOTS, STABLE, - MIGRATING, IMPORTING, SAVE, SKIPME, SLAVES, STREAM, STORE, SUM, SEGFAULT, SETUSER, TAKEOVER, TRACKING, TRACKINGINFO, TYPE, UNBLOCK, USERS, USAGE, WEIGHTS, WHOAMI, + MIGRATING, IMPORTING, SAVE, SKIPME, SLAVES, STREAM, STORE, SUM, SEGFAULT, SETUSER, TAKEOVER, TRACKING, MAINT_NOTIFICATIONS, TRACKINGINFO, TYPE, UNBLOCK, USERS, USAGE, WEIGHTS, WHOAMI, WITHMATCHLEN, WITHSCORE, WITHSCORES, WITHVALUES, XOR, XX, FXX, YES, INDENT, NEWLINE, SPACE, GT, LT, diff --git a/src/main/java/io/lettuce/core/protocol/MaintenanceAwareComponent.java b/src/main/java/io/lettuce/core/protocol/MaintenanceAwareComponent.java index b405d23e2d..2a24ab12dd 100644 --- a/src/main/java/io/lettuce/core/protocol/MaintenanceAwareComponent.java +++ b/src/main/java/io/lettuce/core/protocol/MaintenanceAwareComponent.java @@ -13,7 +13,7 @@ * * @author Tihomir Mateev * @since 7.0 - * @see ClientOptions#supportsMaintenanceEvents() + * @see ClientOptions#getMaintenanceEventsOptions() */ public interface MaintenanceAwareComponent { diff --git a/src/main/java/io/lettuce/core/protocol/MaintenanceAwareConnectionWatchdog.java b/src/main/java/io/lettuce/core/protocol/MaintenanceAwareConnectionWatchdog.java index b81cff6338..8da1458d6e 100644 --- a/src/main/java/io/lettuce/core/protocol/MaintenanceAwareConnectionWatchdog.java +++ b/src/main/java/io/lettuce/core/protocol/MaintenanceAwareConnectionWatchdog.java @@ -37,7 +37,7 @@ * * @author Tihomir Mateev * @since 7.0 - * @see ClientOptions#supportsMaintenanceEvents() + * @see ClientOptions#getMaintenanceEventsOptions() */ @ChannelHandler.Sharable public class MaintenanceAwareConnectionWatchdog extends ConnectionWatchdog implements PushListener { @@ -54,13 +54,17 @@ public class MaintenanceAwareConnectionWatchdog extends ConnectionWatchdog imple private static final String FAILED_OVER_MESSAGE_TYPE = "FAILED_OVER"; - private static final int REBIND_ADDRESS_INDEX = 2; + private static final int REBIND_ADDRESS_INDEX = 3; public static final AttributeKey REBIND_ATTRIBUTE = AttributeKey.newInstance("rebindAddress"); - private static final int FAILING_OVER_SHARDS_INDEX = 2; + private static final int MIGRATING_SHARDS_INDEX = 3; - private static final int FAILED_OVER_SHARDS_INDEX = 1; + private static final int MIGRATED_SHARDS_INDEX = 2; + + private static final int FAILING_OVER_SHARDS_INDEX = 3; + + private static final int FAILED_OVER_SHARDS_INDEX = 2; private Channel channel; @@ -124,10 +128,10 @@ public void onPushMessage(PushMessage message) { } } else if (MIGRATING_MESSAGE_TYPE.equals(mType)) { logger.debug("Shard migration started"); - notifyMigrateStarted(); + notifyMigrateStarted(getMigratingShards(message)); } else if (MIGRATED_MESSAGE_TYPE.equals(mType)) { logger.debug("Shard migration completed"); - notifyMigrateCompleted(); + notifyMigrateCompleted(getMigratedShards(message)); } else if (FAILING_OVER_MESSAGE_TYPE.equals(mType)) { logger.debug("Failover started"); notifyFailoverStarted(getFailingOverShards(message)); @@ -137,55 +141,71 @@ public void onPushMessage(PushMessage message) { } } - private String getFailingOverShards(PushMessage message) { - List content = message.getContent(StringCodec.UTF8::decodeValue); + private String getMigratingShards(PushMessage message) { + List content = message.getContent(); - if (content.size() < 3) { - logger.warn("Invalid failing over message format, expected at least 3 elements, got {}", content.size()); + if (isInvalidMaintenanceEvent(content, 4)) return null; - } - Object shardsObject = content.get(FAILING_OVER_SHARDS_INDEX); + return getShards(content, MIGRATING_SHARDS_INDEX, MIGRATING_MESSAGE_TYPE); + } + + private String getMigratedShards(PushMessage message) { + List content = message.getContent(); - if (!(shardsObject instanceof String)) { - logger.warn("Invalid failing over message format, expected 3rd element to be a List, got {}", - shardsObject != null ? shardsObject.getClass() : "null"); + if (isInvalidMaintenanceEvent(content, 3)) return null; + + return getShards(content, MIGRATED_SHARDS_INDEX, MIGRATED_MESSAGE_TYPE); + } + + private static boolean isInvalidMaintenanceEvent(List content, int expectedSize) { + if (content.size() < expectedSize) { + logger.warn("Invalid maintenance message format, expected at least {} elements, got {}", expectedSize, + content.size()); + return true; } - @SuppressWarnings("unchecked") - String shards = (String) shardsObject; - return shards; + return false; } - private String getFailedOverShards(PushMessage message) { - List content = message.getContent(StringCodec.UTF8::decodeValue); + private String getFailingOverShards(PushMessage message) { + List content = message.getContent(); - if (content.size() < 2) { - logger.warn("Invalid failed over message format, expected at least 2 elements, got {}", content.size()); + if (isInvalidMaintenanceEvent(content, 3)) return null; - } - Object shardsObject = content.get(FAILED_OVER_SHARDS_INDEX); + return getShards(content, FAILING_OVER_SHARDS_INDEX, FAILING_OVER_MESSAGE_TYPE); + } - if (!(shardsObject instanceof String)) { - logger.warn("Invalid failed over message format, expected 2rd element to be a String, got {}", - shardsObject != null ? shardsObject.getClass() : "null"); + private static String getShards(List content, int shardsIndex, String maintenanceEvent) { + Object shardsObject = content.get(shardsIndex); + + if (!(shardsObject instanceof ByteBuffer)) { + logger.warn("Invalid shards format, expected ByteBuffer, got {} for {} maintenance event", + shardsObject != null ? shardsObject.getClass() : "null", maintenanceEvent); return null; } - // expected to be a list of strings ["1","2"] + return StringCodec.UTF8.decodeKey((ByteBuffer) shardsObject); + } + + private String getFailedOverShards(PushMessage message) { + List content = message.getContent(); + + if (content.size() < 3) { + logger.warn("Invalid failed over message format, expected at least 2 elements, got {}", content.size()); + return null; + } - @SuppressWarnings("unchecked") - String shards = (String) shardsObject; - return shards; + return getShards(content, FAILED_OVER_SHARDS_INDEX, FAILED_OVER_MESSAGE_TYPE); } private SocketAddress getRemoteAddress(PushMessage message) { List content = message.getContent(); - if (content.size() != 3) { - logger.warn("Invalid re-bind message format, expected 3 elements, got {}", content.size()); + if (content.size() != 4) { + logger.warn("Invalid re-bind message format, expected 4 elements, got {}", content.size()); return null; } @@ -226,12 +246,12 @@ private void notifyRebindStarted() { this.componentListeners.forEach(MaintenanceAwareComponent::onRebindStarted); } - private void notifyMigrateStarted() { - this.componentListeners.forEach(MaintenanceAwareComponent::onMigrateStarted); + private void notifyMigrateStarted(String shards) { + this.componentListeners.forEach(component -> component.onMigrateStarted(shards)); } - private void notifyMigrateCompleted() { - this.componentListeners.forEach(MaintenanceAwareComponent::onMigrateCompleted); + private void notifyMigrateCompleted(String shards) { + this.componentListeners.forEach(component -> component.onMigrateCompleted(shards)); } private void notifyFailoverStarted(String shards) { diff --git a/src/main/java/io/lettuce/core/protocol/MaintenanceAwareExpiryWriter.java b/src/main/java/io/lettuce/core/protocol/MaintenanceAwareExpiryWriter.java index 911bfaeeef..16f2085371 100644 --- a/src/main/java/io/lettuce/core/protocol/MaintenanceAwareExpiryWriter.java +++ b/src/main/java/io/lettuce/core/protocol/MaintenanceAwareExpiryWriter.java @@ -32,14 +32,14 @@ * progress. The relaxation is done by starting a new timer with the relaxed timeout value. The relaxed timeout is configured * via {@link TimeoutOptions#getRelaxedTimeout()}. *

- * The logic is only applied when the {@link ClientOptions#supportsMaintenanceEvents()} is enabled. + * The logic is only applied when the {@link ClientOptions#getMaintenanceEventsOptions()} is enabled. * * @author Tihomir Mateev * @since 7.0 * @see TimeoutOptions * @see MaintenanceAwareComponent * @see MaintenanceAwareConnectionWatchdog - * @see ClientOptions#supportsMaintenanceEvents() + * @see ClientOptions#getMaintenanceEventsOptions() */ public class MaintenanceAwareExpiryWriter extends CommandExpiryWriter implements MaintenanceAwareComponent { diff --git a/src/main/java/io/lettuce/core/protocol/RebindState.java b/src/main/java/io/lettuce/core/protocol/RebindState.java index 367b5b207b..67d59b7d1b 100644 --- a/src/main/java/io/lettuce/core/protocol/RebindState.java +++ b/src/main/java/io/lettuce/core/protocol/RebindState.java @@ -13,7 +13,7 @@ * * @author Tihomir Mateev * @since 7.0 - * @see ClientOptions#supportsMaintenanceEvents() + * @see ClientOptions#getMaintenanceEventsOptions() */ public enum RebindState { /** diff --git a/src/test/java/biz/paluch/redis/extensibility/LettuceMaintenanceEventsDemo.java b/src/test/java/biz/paluch/redis/extensibility/LettuceMaintenanceEventsDemo.java index 83c4db10a1..16615fdc50 100644 --- a/src/test/java/biz/paluch/redis/extensibility/LettuceMaintenanceEventsDemo.java +++ b/src/test/java/biz/paluch/redis/extensibility/LettuceMaintenanceEventsDemo.java @@ -1,6 +1,7 @@ package biz.paluch.redis.extensibility; import io.lettuce.core.ClientOptions; +import io.lettuce.core.MaintenanceEventsOptions; import io.lettuce.core.RedisClient; import io.lettuce.core.RedisURI; import io.lettuce.core.TimeoutOptions; @@ -32,7 +33,7 @@ public static void main(String[] args) throws ExecutionException, InterruptedExc // (optional) relax timeouts during re-bind to decrease risk of timeouts .timeoutsRelaxingDuringMaintenance(Duration.ofMillis(750)).build(); // (required) enable proactive re-bind by enabling it in the ClientOptions - ClientOptions options = ClientOptions.builder().timeoutOptions(timeoutOpts).supportMaintenanceEvents(true).build(); + ClientOptions options = ClientOptions.builder().timeoutOptions(timeoutOpts).supportMaintenanceEvents(null).build(); RedisClient redisClient = RedisClient.create(RedisURI.create(ADDRESS == null ? "redis://localhost:6379" : ADDRESS)); redisClient.setOptions(options); diff --git a/src/test/java/io/lettuce/core/RedisHandshakeUnitTests.java b/src/test/java/io/lettuce/core/RedisHandshakeUnitTests.java index 7bfd187601..8e56a71f22 100644 --- a/src/test/java/io/lettuce/core/RedisHandshakeUnitTests.java +++ b/src/test/java/io/lettuce/core/RedisHandshakeUnitTests.java @@ -37,7 +37,7 @@ void handshakeWithResp3ShouldPass() { ConnectionState state = new ConnectionState(); state.setCredentialsProvider(new StaticCredentialsProvider("foo", "bar".toCharArray())); - RedisHandshake handshake = new RedisHandshake(ProtocolVersion.RESP3, false, state); + RedisHandshake handshake = new RedisHandshake(ProtocolVersion.RESP3, false, state, null); handshake.initialize(channel); AsyncCommand> hello = channel.readOutbound(); @@ -54,7 +54,7 @@ void handshakeWithDiscoveryShouldPass() { ConnectionState state = new ConnectionState(); state.setCredentialsProvider(new StaticCredentialsProvider("foo", "bar".toCharArray())); - RedisHandshake handshake = new RedisHandshake(null, false, state); + RedisHandshake handshake = new RedisHandshake(null, false, state, null); handshake.initialize(channel); AsyncCommand> hello = channel.readOutbound(); @@ -71,7 +71,7 @@ void handshakeWithDiscoveryShouldDowngrade() { ConnectionState state = new ConnectionState(); state.setCredentialsProvider(new StaticCredentialsProvider(null, null)); - RedisHandshake handshake = new RedisHandshake(null, false, state); + RedisHandshake handshake = new RedisHandshake(null, false, state, null); handshake.initialize(channel); AsyncCommand> hello = channel.readOutbound(); @@ -94,7 +94,7 @@ void handshakeFireAndForgetPostHandshake() { ConnectionState state = new ConnectionState(); state.setCredentialsProvider(new StaticCredentialsProvider(null, null)); state.apply(connectionMetdata); - RedisHandshake handshake = new RedisHandshake(null, false, state); + RedisHandshake handshake = new RedisHandshake(null, false, state, null); CompletionStage handshakeInit = handshake.initialize(channel); AsyncCommand> hello = channel.readOutbound(); @@ -117,7 +117,7 @@ void handshakeWithInvalidResponseShouldPropagateException() { ConnectionState state = new ConnectionState(); state.setCredentialsProvider(new StaticCredentialsProvider(null, null)); - RedisHandshake handshake = new RedisHandshake(null, false, state); + RedisHandshake handshake = new RedisHandshake(null, false, state, null); CompletionStage handshakeInit = handshake.initialize(channel); AsyncCommand> hello = channel.readOutbound(); @@ -142,7 +142,7 @@ void handshakeDelayedCredentialProvider() { ConnectionState state = new ConnectionState(); state.setCredentialsProvider(cp); state.apply(connectionMetdata); - RedisHandshake handshake = new RedisHandshake(null, false, state); + RedisHandshake handshake = new RedisHandshake(null, false, state, null); CompletionStage handshakeInit = handshake.initialize(channel); cp.completeCredentials(RedisCredentials.just("foo", "bar")); diff --git a/src/test/java/io/lettuce/core/protocol/MaintenanceAwareConnectionWatchdogUnitTests.java b/src/test/java/io/lettuce/core/protocol/MaintenanceAwareConnectionWatchdogUnitTests.java index 7753755c0e..a2d6855cae 100644 --- a/src/test/java/io/lettuce/core/protocol/MaintenanceAwareConnectionWatchdogUnitTests.java +++ b/src/test/java/io/lettuce/core/protocol/MaintenanceAwareConnectionWatchdogUnitTests.java @@ -204,9 +204,7 @@ void testChannelReadCompleteWithoutRebindCompleted() throws Exception { @Test void testOnPushMessageMovingWithEmptyStack() { // Given - String addressAndPort = "127.0.0.1:6380"; - ByteBuffer addressBuffer = StringCodec.UTF8.encodeKey(addressAndPort); - List content = Arrays.asList("MOVING", "slot", addressBuffer); + List content = movingPushContent(1, 15, "127.0.0.1:6380"); when(pushMessage.getType()).thenReturn("MOVING"); when(pushMessage.getContent()).thenReturn(content); @@ -233,12 +231,75 @@ void testOnPushMessageMovingWithEmptyStack() { verify(component1, never()).onRebindStarted(); // Not called when stack is empty } + /** + * MOVING