Adds AddressResolver, and implements it for Java, fixing #4396

Author: tavomaciel

CHANGELOG.md

@@ -20,6 +20,7 @@
 * Python: Move OpenTelemetry config classes to glide_shared for code reuse between async and sync clients
 * JAVA: Add dynamic PubSub methods (subscribe, psubscribe, unsubscribe, punsubscribe, ssubscribe, sunsubscribe), getSubscriptions() for subscription state tracking, pubsubReconciliationIntervalMs configuration option, and subscription_out_of_sync_count and subscription_last_sync_timestamp metrics ([#5267](https://github.com/valkey-io/valkey-glide/issues/5267))
 * Go: Add ALLOW_NON_COVERED_SLOTS flag support for cluster scan ([#4895](https://github.com/valkey-io/valkey-glide/issues/4895))
+* CORE: Support custom socket address resolution when connecting to valkey ([#4396](https://github.com/valkey-io/valkey-glide/issues/4396))
 
 #### Fixes
 * Node: Fix to handle non-string types in toBuffersArray ([#4842](https://github.com/valkey-io/valkey-glide/issues/4842))

glide-core/redis-rs/redis/src/cluster.rs

@@ -48,7 +48,9 @@ use crate::connection::{
     connect, Connection, ConnectionAddr, ConnectionInfo, ConnectionLike, RedisConnectionInfo,
 };
 use crate::parser::parse_redis_value;
-use crate::types::{ErrorKind, HashMap, RedisError, RedisResult, RetryMethod, Value};
+use crate::types::{
+    AddressResolver, ErrorKind, HashMap, RedisError, RedisResult, RetryMethod, Value,
+};
 pub use crate::TlsMode; // Pub for backwards compatibility
 use crate::{
     cluster_client::ClusterParams,
@@ -383,7 +385,16 @@ where
                 ErrorKind::ClientError,
                 "can't parse node address",
             )))?;
-            match parse_and_count_slots(&value, self.cluster_params.tls, addr).map(
+            match parse_and_count_slots(
+                &value,
+                self.cluster_params.tls,
+                addr,
+                self.cluster_params
+                    .address_resolver
+                    .as_ref()
+                    .map(Arc::as_ref),
+            )
+            .map(
                 |ParsedSlotsResult {
                      slots,
                      address_to_ip_map,
@@ -1006,7 +1017,8 @@ pub(crate) fn get_connection_info(
             host.to_string(),
             port,
             cluster_params.tls,
-            cluster_params.tls_params,
+            cluster_params.tls_params.clone(),
+            cluster_params.address_resolver.as_ref().map(Arc::as_ref),
         ),
         redis: RedisConnectionInfo {
             password: cluster_params.password,
@@ -1024,21 +1036,29 @@ pub(crate) fn get_connection_addr(
     port: u16,
     tls: Option<TlsMode>,
     tls_params: Option<TlsConnParams>,
+    address_resolver: Option<&dyn AddressResolver>,
 ) -> ConnectionAddr {
+    // Resolve the address if a resolver is provided
+    let (resolved_host, resolved_port) = if let Some(resolver) = address_resolver {
+        resolver.resolve(&host, port)
+    } else {
+        (host, port)
+    };
+
     match tls {
         Some(TlsMode::Secure) => ConnectionAddr::TcpTls {
-            host,
-            port,
+            host: resolved_host,
+            port: resolved_port,
             insecure: false,
             tls_params,
         },
         Some(TlsMode::Insecure) => ConnectionAddr::TcpTls {
-            host,
-            port,
+            host: resolved_host,
+            port: resolved_port,
             insecure: true,
             tls_params,
         },
-        _ => ConnectionAddr::Tcp(host, port),
+        _ => ConnectionAddr::Tcp(resolved_host, resolved_port),
     }
 }
 

glide-core/redis-rs/redis/src/cluster_async/mod.rs

@@ -289,9 +289,9 @@ where
     }
 
     /// Send commands in `pipeline` to the given `route`. If `route` is [None], it will be computed from `pipeline`.
-    /// - `pipeline_retry_strategy`: Configures retry behavior for pipeline commands.  
-    ///   - `retry_server_error`: If `true`, retries commands on server errors (may cause reordering).  
-    ///   - `retry_connection_error`: If `true`, retries on connection errors (may lead to duplicate executions).  
+    /// - `pipeline_retry_strategy`: Configures retry behavior for pipeline commands.
+    ///   - `retry_server_error`: If `true`, retries commands on server errors (may cause reordering).
+    ///   - `retry_connection_error`: If `true`, retries on connection errors (may lead to duplicate executions).
     ///     TODO: add wiki link.
     pub async fn route_pipeline<'a>(
         &'a mut self,
@@ -664,9 +664,9 @@ enum CmdArg<C> {
         count: usize,
         route: Option<InternalSingleNodeRouting<C>>,
         sub_pipeline: bool,
-        /// Configures retry behavior for pipeline commands.  
-        ///   - `retry_server_error`: If `true`, retries commands on server errors (may cause reordering).  
-        ///   - `retry_connection_error`: If `true`, retries on connection errors (may lead to duplicate executions).  
+        /// Configures retry behavior for pipeline commands.
+        ///   - `retry_server_error`: If `true`, retries commands on server errors (may cause reordering).
+        ///   - `retry_connection_error`: If `true`, retries on connection errors (may lead to duplicate executions).
         pipeline_retry_strategy: PipelineRetryStrategy,
     },
     ClusterScan {
@@ -3698,13 +3698,17 @@ where
     let read_from_replicas = inner
         .get_cluster_param(|params| params.read_from_replicas.clone())
         .expect(MUTEX_READ_ERR);
+    let address_resolver = inner
+        .get_cluster_param(|params| params.address_resolver.clone())
+        .expect(MUTEX_READ_ERR);
     TopologyQueryResult {
         topology_result: calculate_topology(
             topology_values,
             curr_retry,
             tls_mode,
             num_of_nodes_to_query,
             read_from_replicas,
+            address_resolver.as_ref().map(Arc::as_ref),
         ),
         failed_connections: Some(failed_addresses),
     }

glide-core/redis-rs/redis/src/cluster_client.rs

@@ -4,7 +4,7 @@ use crate::cluster_topology::{
     DEFAULT_SLOTS_REFRESH_MAX_JITTER_MILLI, DEFAULT_SLOTS_REFRESH_WAIT_DURATION,
 };
 use crate::connection::{ConnectionAddr, ConnectionInfo, IntoConnectionInfo};
-use crate::types::{ErrorKind, ProtocolVersion, RedisError, RedisResult};
+use crate::types::{AddressResolver, ErrorKind, ProtocolVersion, RedisError, RedisResult};
 use crate::{cluster, cluster::TlsMode};
 use crate::{PushInfo, RetryStrategy};
 use rand::Rng;
@@ -48,6 +48,7 @@ struct BuilderParams {
     refresh_topology_from_initial_nodes: bool,
     database_id: i64,
     tcp_nodelay: bool,
+    address_resolver: Option<Arc<dyn AddressResolver>>,
 }
 
 #[derive(Clone)]
@@ -151,6 +152,8 @@ pub struct ClusterParams {
     pub(crate) refresh_topology_from_initial_nodes: bool,
     pub(crate) database_id: i64,
     pub(crate) tcp_nodelay: bool,
+    /// Optional callback for resolving addresses before connection.
+    pub(crate) address_resolver: Option<Arc<dyn AddressResolver>>,
 }
 
 impl ClusterParams {
@@ -183,6 +186,7 @@ impl ClusterParams {
             refresh_topology_from_initial_nodes: value.refresh_topology_from_initial_nodes,
             database_id: value.database_id,
             tcp_nodelay: value.tcp_nodelay,
+            address_resolver: value.address_resolver,
         })
     }
 }
@@ -506,6 +510,16 @@ impl ClusterClientBuilder {
         self
     }
 
+    /// Sets an address resolver callback for resolving node addresses.
+    ///
+    /// When set, the resolver will be called to resolve host:port pairs
+    /// before establishing connections to cluster nodes. This allows custom
+    /// DNS resolution or address translation logic.
+    pub fn address_resolver(mut self, resolver: Arc<dyn AddressResolver>) -> ClusterClientBuilder {
+        self.builder_params.address_resolver = Some(resolver);
+        self
+    }
+
     /// Enables timing out on slow connection time.
     ///
     /// If enabled, the cluster will only wait the given time on each connection attempt to each node.