diff --git a/pyproject.toml b/pyproject.toml
index 450df15c6..afdaf3b2c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -169,9 +169,6 @@ exclude = '''
 [tool.pytest.ini_options]
 log_cli = true
 asyncio_mode = "auto"
-filterwarnings = [
-    "ignore:get_async_redis_connection will become async in the next major release:DeprecationWarning",
-]
 
 [tool.mypy]
 warn_unused_configs = true
diff --git a/redisvl/extensions/cache/base.py b/redisvl/extensions/cache/base.py
index 8ff3566db..212faec2b 100644
--- a/redisvl/extensions/cache/base.py
+++ b/redisvl/extensions/cache/base.py
@@ -4,6 +4,7 @@
 specific cache types such as LLM caches and embedding caches.
 """
 
+import asyncio
 from collections.abc import Mapping
 from typing import Any, cast
 
@@ -59,7 +60,13 @@ def __init__(
         # Initialize Redis clients
         self._async_redis_client = async_redis_client
         self._redis_client = redis_client
+        # Guards lazy async client creation, which suspends on an await and so
+        # cannot rely on a bare check-then-set. Mirrors AsyncSearchIndex._lock.
+        self._async_client_lock = asyncio.Lock()
 
+        # Caches never close a caller-supplied client and register no GC
+        # finalizer, so the index's owns_client handover has no cache
+        # equivalent by design.
         if redis_client or async_redis_client:
             self._owns_redis_client = False
         else:
@@ -128,22 +135,26 @@ async def _get_async_redis_client(self) -> AsyncRedisClient:
         Returns:
             AsyncRedisClient: An async Redis client instance.
         """
-        if not hasattr(self, "_async_redis_client") or self._async_redis_client is None:
-            client = self.redis_kwargs.get("redis_client")
-
-            if client and isinstance(client, (Redis, RedisCluster)):
-                self._async_redis_client = RedisConnectionFactory.sync_to_async_redis(
-                    client
-                )
-            else:
-                url = cast(str | None, self.redis_kwargs["redis_url"])
-                kwargs = cast(dict[str, Any], self.redis_kwargs["connection_kwargs"])
-                self._async_redis_client = (
-                    RedisConnectionFactory.get_async_redis_connection(
-                        redis_url=url, **kwargs
-                    )
-                )
-        return self._async_redis_client
+        client = getattr(self, "_async_redis_client", None)
+        if client is None:
+            async with self._async_client_lock:
+                # Double-check: another task may have created the client while
+                # this one waited on the lock or on the factory's round trip.
+                client = getattr(self, "_async_redis_client", None)
+                if client is None:
+                    provided = self.redis_kwargs.get("redis_client")
+                    if provided and isinstance(provided, (Redis, RedisCluster)):
+                        client = RedisConnectionFactory.sync_to_async_redis(provided)
+                    else:
+                        url = cast(str | None, self.redis_kwargs["redis_url"])
+                        kwargs = cast(
+                            dict[str, Any], self.redis_kwargs["connection_kwargs"]
+                        )
+                        client = await RedisConnectionFactory._get_aredis_connection(
+                            redis_url=url, **kwargs
+                        )
+                    self._async_redis_client = client
+        return client
 
     def expire(self, key: str, ttl: int | None = None) -> None:
         """Set or refresh the expiration time for a key in the cache.
diff --git a/redisvl/extensions/router/semantic.py b/redisvl/extensions/router/semantic.py
index 16472615b..f3490d7b4 100644
--- a/redisvl/extensions/router/semantic.py
+++ b/redisvl/extensions/router/semantic.py
@@ -192,7 +192,11 @@ def from_existing(
                 **factory_kwargs,
             )
             index_kwargs["_client_validated"] = True
-            index_kwargs["_owns_redis_client"] = True
+            # index_kwargs wins the merge below, so only claim ownership when
+            # the caller has not already answered. Matches the setdefault in
+            # SearchIndex.from_existing, where an explicit value also wins.
+            if "owns_client" not in init_kwargs:
+                index_kwargs["owns_client"] = True
             if lib_name is not None:
                 index_kwargs["lib_name"] = lib_name
             created_redis_client = True
diff --git a/redisvl/index/index.py b/redisvl/index/index.py
index 5a34c03f0..8eddcfffc 100644
--- a/redisvl/index/index.py
+++ b/redisvl/index/index.py
@@ -748,7 +748,12 @@ def from_dict(cls, schema_dict: dict[str, Any], **kwargs):
         return cls(schema=schema, **kwargs)
 
     def disconnect(self):
-        """Disconnect from the Redis database."""
+        """Close the Redis client if this index owns it.
+
+        Always invalidates the cached SQL schema. When the index does not own
+        the client (see ``owns_client``), the client is left open and the
+        index remains usable.
+        """
         raise NotImplementedError("This method should be implemented by subclasses.")
 
     def key(self, id: str) -> str:
@@ -816,11 +821,11 @@ def __init__(
         redis_url: str | None = None,
         connection_kwargs: dict[str, Any] | None = None,
         validate_on_load: bool = False,
+        owns_client: bool | None = None,
         **kwargs,
     ):
-        """Initialize the RedisVL search index with a schema, Redis client
-        (or URL string with other connection args), connection_args, and other
-        kwargs.
+        """Initialize the RedisVL search index with a schema and either a Redis
+        client or a URL string with other connection kwargs.
 
         Args:
             schema (IndexSchema): Index schema object.
@@ -832,6 +837,12 @@ def __init__(
                 args.
             validate_on_load (bool, optional): Whether to validate data against schema
                 when loading. Defaults to False.
+            owns_client (Optional[bool], optional): Whether the index closes
+                the Redis client when the index is disconnected or garbage
+                collected. Defaults to None, meaning the index owns a client
+                only if it created one itself. Pass True to hand over a client
+                you created, or False to keep one the index would otherwise
+                close, in which case closing it becomes your responsibility.
         """
         if "connection_args" in kwargs:
             connection_kwargs = kwargs.pop("connection_args")
@@ -851,7 +862,18 @@ def __init__(
         self._sql_executors: dict[str, Any] = {}
 
         self._validated_client = kwargs.pop("_client_validated", False)
-        self._owns_redis_client = kwargs.pop("_owns_redis_client", redis_client is None)
+        if "_owns_redis_client" in kwargs:
+            # Underscore-prefixed kwargs are forwarded verbatim by
+            # _split_from_existing_kwargs, so this would otherwise be dropped
+            # in silence and leak the connection it used to control.
+            raise TypeError(
+                "_owns_redis_client is no longer accepted; use owns_client instead"
+            )
+        # Must be assigned before _register_client_finalizer, which gates on
+        # this flag.
+        self._owns_redis_client = (
+            redis_client is None if owns_client is None else bool(owns_client)
+        )
         self._client_finalizer = None
         # Close the owned client when this index is garbage collected. When
         # the client is created lazily, registration happens at creation time
@@ -861,10 +883,15 @@ def __init__(
     _finalizer_close_client = staticmethod(_close_owned_sync_client)
 
     def disconnect(self):
-        """Disconnect from the Redis database."""
+        """Close the Redis client if this index owns it.
+
+        Always invalidates the cached SQL schema. When the index does not own
+        the client (see ``owns_client``), the client is left open and the
+        index remains usable.
+        """
         self.invalidate_sql_schema_cache()
-        if self._owns_redis_client is False:
-            logger.info("Index does not own client, not disconnecting")
+        if not self._owns_redis_client:
+            logger.info("Index does not own its client; leaving it open")
             return
         self._detach_client_finalizer()
         if self.__redis_client:
@@ -888,6 +915,9 @@ def from_existing(
                 instantiated redis client.
             redis_url (Optional[str]): The URL of the Redis server to
                 connect to.
+            owns_client (Optional[bool], optional): Whether the index closes
+                the client. Defaults to True when this method created the
+                client from `redis_url`, and False when you supplied one.
 
         Raises:
             ValueError: If redis_url or redis_client is not provided.
@@ -923,7 +953,7 @@ def from_existing(
         schema_dict = convert_index_info_to_schema(index_info)
         schema = IndexSchema.from_dict(schema_dict)
         if created_redis_client:
-            init_kwargs["_owns_redis_client"] = True
+            init_kwargs.setdefault("owns_client", True)
             return cls(
                 schema,
                 redis_client=redis_client,
@@ -2131,6 +2161,7 @@ def __init__(
         redis_client: AsyncRedisClient | None = None,
         connection_kwargs: dict[str, Any] | None = None,
         validate_on_load: bool = False,
+        owns_client: bool | None = None,
         **kwargs,
     ):
         """Initialize the RedisVL async search index with a schema.
@@ -2145,6 +2176,12 @@ def __init__(
                 args.
             validate_on_load (bool, optional): Whether to validate data against schema
                 when loading. Defaults to False.
+            owns_client (Optional[bool], optional): Whether the index closes
+                the Redis client when the index is disconnected or garbage
+                collected. Defaults to None, meaning the index owns a client
+                only if it created one itself. Pass True to hand over a client
+                you created, or False to keep one the index would otherwise
+                close, in which case closing it becomes your responsibility.
         """
         if "redis_kwargs" in kwargs:
             connection_kwargs = kwargs.pop("redis_kwargs")
@@ -2157,7 +2194,11 @@ def __init__(
         self._validate_on_load = validate_on_load
         self._lib_name: str | None = kwargs.pop("lib_name", None)
 
-        # Store connection parameters
+        # Store connection parameters. Note the asymmetry with SearchIndex:
+        # there, _redis_client is a property that lazily creates the client,
+        # whereas here it is a plain attribute that stays None until
+        # _get_client() creates one. Read it through _get_client(), not
+        # directly.
         self._redis_client = redis_client
         self._redis_url = redis_url
         self._connection_kwargs = connection_kwargs or {}
@@ -2165,7 +2206,18 @@ def __init__(
         self._sql_executors: dict[str, Any] = {}
 
         self._validated_client = kwargs.pop("_client_validated", False)
-        self._owns_redis_client = kwargs.pop("_owns_redis_client", redis_client is None)
+        if "_owns_redis_client" in kwargs:
+            # Underscore-prefixed kwargs are forwarded verbatim by
+            # _split_from_existing_kwargs, so this would otherwise be dropped
+            # in silence and leak the connection it used to control.
+            raise TypeError(
+                "_owns_redis_client is no longer accepted; use owns_client instead"
+            )
+        # Must be assigned before _register_client_finalizer, which gates on
+        # this flag.
+        self._owns_redis_client = (
+            redis_client is None if owns_client is None else bool(owns_client)
+        )
         self._client_finalizer = None
         # Close the owned client when this index is garbage collected. When
         # the client is created lazily, registration happens at creation time
@@ -2191,6 +2243,9 @@ async def from_existing(
                 instantiated redis client.
             redis_url (Optional[str]): The URL of the Redis server to
                 connect to.
+            owns_client (Optional[bool], optional): Whether the index closes
+                the client. Defaults to True when this method created the
+                client from `redis_url`, and False when you supplied one.
         """
         if not redis_url and not redis_client:
             raise ValueError(
@@ -2231,7 +2286,7 @@ async def from_existing(
         schema_dict = convert_index_info_to_schema(index_info)
         schema = IndexSchema.from_dict(schema_dict)
         if created_redis_client:
-            init_kwargs["_owns_redis_client"] = True
+            init_kwargs.setdefault("owns_client", True)
             return cls(
                 schema,
                 redis_client=redis_client,
@@ -3316,8 +3371,14 @@ async def info(self, name: str | None = None) -> dict[str, Any]:
         return await self._info(index_name, client)
 
     async def disconnect(self):
+        """Close the Redis client if this index owns it.
+
+        Always invalidates the cached SQL schema. When the index does not own
+        the client (see ``owns_client``), the client is left open and the
+        index remains usable.
+        """
         self.invalidate_sql_schema_cache()
-        if self._owns_redis_client is False:
+        if not self._owns_redis_client:
             return
         self._detach_client_finalizer()
         if self._redis_client is not None:
@@ -3325,7 +3386,12 @@ async def disconnect(self):
         self._redis_client = None
 
     def disconnect_sync(self):
-        if self._redis_client is None or self._owns_redis_client is False:
+        """Close an owned Redis client from synchronous code.
+
+        For callers outside an event loop, such as ``__del__`` or a shutdown
+        hook. Honours ``owns_client`` exactly as :meth:`disconnect` does.
+        """
+        if self._redis_client is None or not self._owns_redis_client:
             return
         sync_wrapper(self.disconnect)()
 
diff --git a/redisvl/mcp/server.py b/redisvl/mcp/server.py
index 554ccc18e..14097b1f0 100644
--- a/redisvl/mcp/server.py
+++ b/redisvl/mcp/server.py
@@ -643,11 +643,9 @@ async def _load_effective_schema(
     @staticmethod
     def _make_index(schema: IndexSchema, client: Any) -> AsyncSearchIndex:
         """Bind an inspected schema and Redis client into an async index."""
-        index = AsyncSearchIndex(schema=schema, redis_client=client)
         # The server acquired this client explicitly during startup, so hand
         # ownership to the index for a single shutdown path.
-        index._owns_redis_client = True
-        return index
+        return AsyncSearchIndex(schema=schema, redis_client=client, owns_client=True)
 
     async def _initialize_vectorizer(
         self, binding: MCPIndexBindingConfig, schema: IndexSchema, timeout: int
diff --git a/redisvl/migration/async_executor.py b/redisvl/migration/async_executor.py
index 149ae0e9d..87310c61f 100644
--- a/redisvl/migration/async_executor.py
+++ b/redisvl/migration/async_executor.py
@@ -867,8 +867,6 @@ def _notify(step: str, detail: Optional[str] = None) -> None:
 
         try:
             client = await source_index._get_client()
-            if client is None:
-                raise ValueError("Failed to get Redis client from source index")
             aof_enabled = await self._detect_aof_enabled(client)
             disk_estimate = estimate_disk_space(plan, aof_enabled=aof_enabled)
             if disk_estimate.has_quantization:
diff --git a/redisvl/migration/async_planner.py b/redisvl/migration/async_planner.py
index 6c75efda2..99972208e 100644
--- a/redisvl/migration/async_planner.py
+++ b/redisvl/migration/async_planner.py
@@ -235,9 +235,7 @@ async def snapshot_source(
         prefixes = index.schema.index.prefix
         prefix_list = prefixes if isinstance(prefixes, list) else [prefixes]
 
-        client = index.client
-        if client is None:
-            raise ValueError("Failed to get Redis client from index")
+        client = await index._get_client()
 
         return SourceSnapshot(
             index_name=index_name,
diff --git a/redisvl/migration/async_validation.py b/redisvl/migration/async_validation.py
index ce742a3d0..2575daaf6 100644
--- a/redisvl/migration/async_validation.py
+++ b/redisvl/migration/async_validation.py
@@ -74,13 +74,10 @@ async def validate(
         validation.doc_count_match = source_total == target_total
 
         key_sample = plan.source.keyspace.key_sample
-        client = target_index.client
         if not key_sample:
             validation.key_sample_exists = True
-        elif client is None:
-            validation.key_sample_exists = False
-            validation.errors.append("Failed to get Redis client for key sample check")
         else:
+            client = await target_index._get_client()
             # Handle prefix change: transform key_sample to use new prefix.
             # Must match the executor's RENAME logic exactly:
             #   new_key = new_prefix + key[len(old_prefix):]
@@ -140,9 +137,7 @@ async def validate(
 
     async def _count_index_keys(self, index: AsyncSearchIndex) -> int:
         """Count keys matching the target index prefixes with SCAN."""
-        client = index.client
-        if client is None:
-            raise ValueError("Redis client is required to count index keys")
+        client = await index._get_client()
 
         prefixes = index.schema.index.prefix
         prefix_list = prefixes if isinstance(prefixes, list) else [prefixes]
@@ -181,25 +176,16 @@ async def _run_query_checks(
                 )
             )
 
-        client = target_index.client
+        client = await target_index._get_client()
         for key in query_checks.get("keys_exist", []):
-            if client is None:
-                results.append(
-                    QueryCheckResult(
-                        name=f"key:{key}",
-                        passed=False,
-                        details="Failed to get Redis client",
-                    )
-                )
-            else:
-                exists = bool(await client.exists(key))
-                results.append(
-                    QueryCheckResult(
-                        name=f"key:{key}",
-                        passed=exists,
-                        details="Key exists" if exists else "Key not found",
-                    )
+            exists = bool(await client.exists(key))
+            results.append(
+                QueryCheckResult(
+                    name=f"key:{key}",
+                    passed=exists,
+                    details="Key exists" if exists else "Key not found",
                 )
+            )
 
         return results
 
diff --git a/redisvl/migration/planner.py b/redisvl/migration/planner.py
index 4c09fe04c..cd6f711f8 100644
--- a/redisvl/migration/planner.py
+++ b/redisvl/migration/planner.py
@@ -2,7 +2,7 @@
 
 from copy import deepcopy
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple, cast
 
 import yaml
 
@@ -18,6 +18,7 @@
 )
 from redisvl.redis.connection import supports_svs
 from redisvl.schema.schema import IndexSchema
+from redisvl.types import SyncRedisClient
 
 
 class MigrationPlanner:
@@ -226,7 +227,7 @@ def snapshot_source(
                 prefixes=prefix_list,
                 key_separator=index.schema.index.key_separator,
                 key_sample=self._sample_keys(
-                    client=index.client,
+                    client=index._redis_client,
                     prefixes=prefix_list,
                     key_separator=index.schema.index.key_separator,
                 ),
@@ -647,10 +648,10 @@ def write_plan(self, plan: MigrationPlan, plan_out: str) -> None:
             yaml.safe_dump(plan.model_dump(exclude_none=True), f, sort_keys=False)
 
     def _sample_keys(
-        self, *, client: Any, prefixes: List[str], key_separator: str
+        self, *, client: SyncRedisClient, prefixes: List[str], key_separator: str
     ) -> List[str]:
         key_sample: List[str] = []
-        if client is None or self.key_sample_limit <= 0:
+        if self.key_sample_limit <= 0:
             return key_sample
 
         for prefix in prefixes:
@@ -666,10 +667,13 @@ def _sample_keys(
                 match_pattern = f"{prefix}*"
             cursor = 0
             while True:
-                cursor, keys = client.scan(
-                    cursor=cursor,
-                    match=match_pattern,
-                    count=max(self.key_sample_limit, 1000),
+                cursor, keys = cast(
+                    tuple[int, list[Any]],
+                    client.scan(
+                        cursor=cursor,
+                        match=match_pattern,
+                        count=max(self.key_sample_limit, 1000),
+                    ),
                 )
                 for key in keys:
                     decoded_key = key.decode() if isinstance(key, bytes) else str(key)
diff --git a/redisvl/migration/validation.py b/redisvl/migration/validation.py
index f8735a443..fcf14ed8f 100644
--- a/redisvl/migration/validation.py
+++ b/redisvl/migration/validation.py
@@ -12,7 +12,6 @@
     QueryCheckResult,
 )
 from redisvl.migration.utils import build_scan_match_patterns, load_yaml, schemas_equal
-from redisvl.types import SyncRedisClient
 
 
 class MigrationValidator:
@@ -88,9 +87,8 @@ def validate(
                     keys_to_check.append(translated)
             # Check keys one at a time to avoid Redis Cluster cross-slot
             # errors from multi-key EXISTS commands.
-            existing_count = sum(
-                target_index.client.exists(key) for key in keys_to_check
-            )
+            client = target_index._redis_client
+            existing_count = sum(client.exists(key) for key in keys_to_check)
             validation.key_sample_exists = existing_count == len(keys_to_check)
 
         # Run automatic functional checks (always).
@@ -128,10 +126,7 @@ def validate(
 
     def _count_index_keys(self, index: SearchIndex) -> int:
         """Count keys matching the target index prefixes with SCAN."""
-        raw_client = index.client
-        if raw_client is None:
-            raise ValueError("Redis client is required to count index keys")
-        client = cast(SyncRedisClient, raw_client)
+        client = index._redis_client
 
         prefixes = index.schema.index.prefix
         prefix_list = prefixes if isinstance(prefixes, list) else [prefixes]
@@ -173,10 +168,8 @@ def _run_query_checks(
                 )
             )
 
+        client = target_index._redis_client
         for key in query_checks.get("keys_exist", []):
-            client = target_index.client
-            if client is None:
-                raise ValueError("Redis client not connected")
             exists = bool(client.exists(key))
             results.append(
                 QueryCheckResult(
diff --git a/redisvl/redis/connection.py b/redisvl/redis/connection.py
index 53edec80c..d4e44ea8d 100644
--- a/redisvl/redis/connection.py
+++ b/redisvl/redis/connection.py
@@ -35,7 +35,7 @@ def _split_from_existing_kwargs(
     init_kwargs: dict[str, Any] = {}
     connection_kwargs: dict[str, Any] = {}
 
-    for key in ("validate_on_load", "lib_name"):
+    for key in ("validate_on_load", "lib_name", "owns_client"):
         if key in kwargs:
             init_kwargs[key] = kwargs.pop(key)
 
@@ -726,7 +726,7 @@ def get_async_redis_connection(
                 variable is not set.
         """
         warn(
-            "get_async_redis_connection will become async in the next major release.",
+            "get_async_redis_connection will become async in a future release.",
             DeprecationWarning,
         )
         _deprecated_url = kwargs.pop("url", None)
diff --git a/redisvl/utils/utils.py b/redisvl/utils/utils.py
index 85f74397c..bd53af0ae 100644
--- a/redisvl/utils/utils.py
+++ b/redisvl/utils/utils.py
@@ -93,7 +93,9 @@ class MyClass:
             def test_method(cls, old_arg=None, new_arg=None):
                 pass
     """
-    message = f"Argument {argument} is deprecated and will be removed in the next major release."
+    message = (
+        f"Argument {argument} is deprecated and will be removed in a future release."
+    )
     if replacement:
         message += f" Use {replacement} instead."
 
@@ -157,7 +159,7 @@ def decorator(func):
         fn_name = name or func.__name__
         warning_message = (
             f"Function {fn_name} is deprecated and will be "
-            "removed in the next major release. "
+            "removed in a future release. "
         )
         if replacement:
             warning_message += replacement
@@ -194,7 +196,7 @@ def decorator(cls):
         class_name = name or cls.__name__
         warning_message = (
             f"Class {class_name} is deprecated and will be "
-            "removed in the next major release. "
+            "removed in a future release. "
         )
         if replacement:
             warning_message += replacement
diff --git a/tests/unit/test_async_migration_planner.py b/tests/unit/test_async_migration_planner.py
index 93ce3d49d..c4483d974 100644
--- a/tests/unit/test_async_migration_planner.py
+++ b/tests/unit/test_async_migration_planner.py
@@ -35,8 +35,8 @@ def __init__(self, schema, stats, keys):
         self._stats = stats
         self._client = AsyncDummyClient(keys)
 
-    @property
-    def client(self):
+    async def _get_client(self):
+        """Mirrors AsyncSearchIndex._get_client, the lazy async getter."""
         return self._client
 
     async def info(self):
diff --git a/tests/unit/test_connection_normalization.py b/tests/unit/test_connection_normalization.py
index 98622eb1c..28560f79a 100644
--- a/tests/unit/test_connection_normalization.py
+++ b/tests/unit/test_connection_normalization.py
@@ -1,3 +1,4 @@
+import asyncio
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -7,6 +8,7 @@
 from redisvl.extensions.router.semantic import SemanticRouter
 from redisvl.index import AsyncSearchIndex, SearchIndex
 from redisvl.query.sql import SQLQuery
+from redisvl.utils.utils import assert_no_warnings
 
 
 def _schema_dict(name: str = "idx") -> dict:
@@ -105,6 +107,41 @@ def test_search_index_from_existing_owns_factory_created_client():
     created_client.close.assert_called_once_with()
 
 
+def test_search_index_from_existing_honours_explicit_owns_client():
+    """``owns_client`` is an init kwarg, not a connection kwarg.
+
+    Without the allow-list entry in ``_split_from_existing_kwargs`` it would
+    fall through to ``connection_kwargs`` and redis-py would reject it. An
+    explicit value also wins over the ownership ``from_existing`` would
+    otherwise assume for a client it created itself.
+    """
+    created_client = MagicMock()
+
+    with (
+        patch(
+            "redisvl.index.index.RedisConnectionFactory.get_redis_connection",
+            return_value=created_client,
+        ) as mock_get_connection,
+        patch.object(SearchIndex, "_info", return_value={}),
+        patch(
+            "redisvl.index.index.convert_index_info_to_schema",
+            return_value=_schema_dict("search-index"),
+        ),
+    ):
+        index = SearchIndex.from_existing(
+            "search-index",
+            redis_url="redis://localhost:6380",
+            owns_client=False,
+        )
+
+    mock_get_connection.assert_called_once_with(redis_url="redis://localhost:6380")
+    assert index._owns_redis_client is False
+
+    index.disconnect()
+
+    created_client.close.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_async_search_index_from_existing_prefers_provided_client():
     """Use the provided async Redis client instead of constructing a new one."""
@@ -278,7 +315,7 @@ def test_semantic_router_from_existing_rebuilds_from_redis_url():
     assert mock_from_dict.call_args.kwargs["_index_kwargs"] == {
         "_internal_flag": True,
         "_client_validated": True,
-        "_owns_redis_client": True,
+        "owns_client": True,
     }
     assert result is loaded_router
 
@@ -300,6 +337,58 @@ def test_base_cache_sync_client_creation_uses_connection_factory():
     assert client is mock_client
 
 
+@pytest.mark.asyncio
+async def test_base_cache_async_client_creation_emits_no_warning():
+    """Creating a cache's async client must not warn.
+
+    ``_get_async_redis_client`` used to call ``get_async_redis_connection``,
+    which warns unconditionally, so cache users saw a DeprecationWarning for
+    an API they never called. A suite-wide filter in pyproject.toml hid it.
+    This is the guard that replaced that filter.
+    """
+    cache = EmbeddingsCache(redis_url="redis://localhost:6379")
+    mock_client = MagicMock()
+
+    with patch(
+        "redisvl.extensions.cache.base.RedisConnectionFactory._get_aredis_connection",
+        new=AsyncMock(return_value=mock_client),
+    ) as mock_get_connection:
+        with assert_no_warnings():
+            client = await cache._get_async_redis_client()
+
+    mock_get_connection.assert_awaited_once_with(redis_url="redis://localhost:6379")
+    assert client is mock_client
+
+
+@pytest.mark.asyncio
+async def test_base_cache_async_client_creation_is_serialised():
+    """Concurrent callers must share one client, not orphan a connection pool.
+
+    Building the client awaits a CLIENT SETINFO round trip, so a bare
+    check-then-set would let two tasks each create one and leave the first
+    unreachable and never closed.
+    """
+    cache = EmbeddingsCache(redis_url="redis://localhost:6379")
+    created = []
+
+    async def factory(*args, **kwargs):
+        await asyncio.sleep(0)  # the suspension the real factory introduces
+        client = MagicMock(name=f"client{len(created)}")
+        created.append(client)
+        return client
+
+    with patch(
+        "redisvl.extensions.cache.base.RedisConnectionFactory._get_aredis_connection",
+        new=factory,
+    ):
+        first, second = await asyncio.gather(
+            cache._get_async_redis_client(), cache._get_async_redis_client()
+        )
+
+    assert len(created) == 1
+    assert first is second
+
+
 def test_sql_query_uses_connection_factory_for_redis_url():
     """Build SQL query helper connections through the shared connection factory."""
     fake_sql_redis_module = _fake_sql_redis_module()
diff --git a/tests/unit/test_error_handling.py b/tests/unit/test_error_handling.py
index 4dbc47d7c..c6ce31377 100644
--- a/tests/unit/test_error_handling.py
+++ b/tests/unit/test_error_handling.py
@@ -11,7 +11,7 @@
 
 import asyncio
 from collections.abc import Mapping
-from unittest.mock import MagicMock, Mock, patch
+from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 import pytest
 import redis.exceptions
@@ -204,11 +204,11 @@ async def test_connection_kwargs_valid_dict(self):
             "redisvl.extensions.cache.base.RedisConnectionFactory"
         ) as mock_factory:
             mock_client = Mock()
-            mock_factory.get_async_redis_connection.return_value = mock_client
+            mock_factory._get_aredis_connection = AsyncMock(return_value=mock_client)
 
             result = await cache._get_async_redis_client()
             assert result == mock_client
-            mock_factory.get_async_redis_connection.assert_called_once()
+            mock_factory._get_aredis_connection.assert_awaited_once()
 
 
 class TestRouterConfigErrorHandling:
diff --git a/tests/unit/test_index_gc_finalizer.py b/tests/unit/test_index_gc_finalizer.py
index 7a36b5b49..edc0aed05 100644
--- a/tests/unit/test_index_gc_finalizer.py
+++ b/tests/unit/test_index_gc_finalizer.py
@@ -223,3 +223,58 @@ def test_async_injected_client_not_closed_on_collection(self):
         collect()
 
         fake_client.aclose.assert_not_awaited()
+
+
+class TestOwnsClientHandover:
+    """``owns_client`` overrides who closes the client.
+
+    By default an index closes only a client it created itself. These tests
+    cover the two explicit overrides at construction, which is where
+    ownership should be stated: the deprecated ``set_client()`` inherits
+    whatever ownership the index already had, so it can hand the index a
+    caller's client and then close it.
+    """
+
+    def test_sync_injected_client_closed_when_ownership_handed_over(self):
+        schema = IndexSchema.from_dict(SCHEMA_DICT)
+        fake_client = mock.MagicMock(name="handed_over_sync_client")
+        index = SearchIndex(schema, redis_client=fake_client, owns_client=True)
+
+        del index
+        collect()
+
+        fake_client.close.assert_called_once()
+
+    def test_async_injected_client_closed_when_ownership_handed_over(self):
+        schema = IndexSchema.from_dict(SCHEMA_DICT)
+        fake_client = mock.MagicMock(name="handed_over_async_client")
+        fake_client.aclose = mock.AsyncMock()
+        index = AsyncSearchIndex(schema, redis_client=fake_client, owns_client=True)
+
+        del index
+        collect()
+
+        fake_client.aclose.assert_awaited_once()
+
+    def test_sync_lazily_created_client_kept_when_ownership_declined(self):
+        """``owns_client=False`` keeps a client the index would have owned.
+
+        Only covered on the sync class: the flag is read by shared
+        ``__init__`` code, and the per-flavour close paths are covered above.
+        """
+        schema = IndexSchema.from_dict(SCHEMA_DICT)
+        fake_client = mock.MagicMock(name="lazily_created_sync_client")
+
+        with mock.patch(
+            "redisvl.index.index.RedisConnectionFactory.get_redis_connection",
+            return_value=fake_client,
+        ):
+            index = SearchIndex(
+                schema, redis_url="redis://fake:6379", owns_client=False
+            )
+            assert index._redis_client is fake_client
+
+        del index
+        collect()
+
+        fake_client.close.assert_not_called()
diff --git a/tests/unit/test_migration_planner.py b/tests/unit/test_migration_planner.py
index b07f9df93..d8f5df0cc 100644
--- a/tests/unit/test_migration_planner.py
+++ b/tests/unit/test_migration_planner.py
@@ -36,7 +36,8 @@ def __init__(self, schema, stats, keys):
         self._client = DummyClient(keys)
 
     @property
-    def client(self):
+    def _redis_client(self):
+        """Mirrors SearchIndex._redis_client, the lazily creating property."""
         return self._client
 
     def info(self):
@@ -1207,7 +1208,7 @@ def test_exists_called_per_key(self, monkeypatch):
         mock_client.exists.return_value = 1  # Each key exists
 
         mock_index = MagicMock()
-        mock_index.client = mock_client
+        mock_index._redis_client = mock_client
         mock_index.info.return_value = {"num_docs": 3, "hash_indexing_failures": 0}
         mock_index.schema.to_dict.return_value = plan.merged_target_schema
         mock_index.search.return_value = MagicMock(total=3)
@@ -1242,7 +1243,7 @@ def test_multi_prefix_keys_translated(self, monkeypatch):
         mock_client.exists.return_value = 1
 
         mock_index = MagicMock()
-        mock_index.client = mock_client
+        mock_index._redis_client = mock_client
         mock_index.info.return_value = {"num_docs": 3, "hash_indexing_failures": 0}
         mock_index.schema.to_dict.return_value = plan.merged_target_schema
         mock_index.search.return_value = MagicMock(total=3)
@@ -1287,7 +1288,7 @@ def test_expected_source_count_uses_scanned_target_keys(self, monkeypatch):
         }
 
         mock_index = MagicMock()
-        mock_index.client = DummyClient(
+        mock_index._redis_client = DummyClient(
             [b"target:1", b"target:2", b"target:3", b"target:4", b"target:5"]
         )
         mock_index.info.return_value = {"num_docs": 5, "hash_indexing_failures": 0}
diff --git a/tests/unit/test_url_deprecation.py b/tests/unit/test_url_deprecation.py
index d70ff4cb8..7b01cefa5 100644
--- a/tests/unit/test_url_deprecation.py
+++ b/tests/unit/test_url_deprecation.py
@@ -26,7 +26,7 @@ async def test__get_aredis_connection_deprecates_url_kwarg_only():
     assert any(
         str(w.message)
         == (
-            "Argument url is deprecated and will be removed in the next major release. "
+            "Argument url is deprecated and will be removed in a future release. "
             "Use redis_url instead."
         )
         for w in record
diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py
index d7c1e2f0e..e4e2893fd 100644
--- a/tests/unit/test_utils.py
+++ b/tests/unit/test_utils.py
@@ -215,7 +215,7 @@ def test_func(old_arg=None, new_arg=None, neutral_arg=None):
 
         assert len(record) == 1
         assert str(record[0].message) == (
-            "Argument old_arg is deprecated and will be removed in the next major release. Use new_arg instead."
+            "Argument old_arg is deprecated and will be removed in a future release. Use new_arg instead."
         )
 
         # Test that passing the deprecated argument as a positional argument also triggers the warning.
@@ -224,7 +224,7 @@ def test_func(old_arg=None, new_arg=None, neutral_arg=None):
 
         assert len(record) == 1
         assert str(record[0].message) == (
-            "Argument old_arg is deprecated and will be removed in the next major release. Use new_arg instead."
+            "Argument old_arg is deprecated and will be removed in a future release. Use new_arg instead."
         )
 
         with assert_no_warnings():
@@ -242,8 +242,7 @@ def test_func(old_arg=None, neutral_arg=None):
 
         assert len(record) == 1
         assert str(record[0].message) == (
-            "Argument old_arg is deprecated and will be removed"
-            " in the next major release."
+            "Argument old_arg is deprecated and will be removed in a future release."
         )
 
         # As a positional arg
@@ -252,8 +251,7 @@ def test_func(old_arg=None, neutral_arg=None):
 
         assert len(record) == 1
         assert str(record[0].message) == (
-            "Argument old_arg is deprecated and will be removed"
-            " in the next major release."
+            "Argument old_arg is deprecated and will be removed in a future release."
         )
 
         with assert_no_warnings():
@@ -547,7 +545,7 @@ def __init__(self, value):
 
         assert len(record) == 1
         assert str(record[0].message) == (
-            "Class OldClass is deprecated and will be removed in the next major release. "
+            "Class OldClass is deprecated and will be removed in a future release. "
             "Use NewClass instead."
         )
         assert obj.value == 42
@@ -563,7 +561,7 @@ def __init__(self, value):
 
         assert len(record) == 1
         assert str(record[0].message) == (
-            "Class OldClass is deprecated and will be removed in the next major release. "
+            "Class OldClass is deprecated and will be removed in a future release. "
         )
         assert obj.value == 42
 
@@ -577,7 +575,7 @@ class OldClass:
 
         assert len(record) == 1
         assert str(record[0].message) == (
-            "Class CustomOldClass is deprecated and will be removed in the next major release. "
+            "Class CustomOldClass is deprecated and will be removed in a future release. "
             "Use NewClass instead."
         )