diff --git a/redis/cache.py b/redis/cache.py
index cb29ffe785..041818e700 100644
--- a/redis/cache.py
+++ b/redis/cache.py
@@ -199,15 +199,25 @@ def delete_by_cache_keys(self, cache_keys: List[CacheKey]) -> List[bool]:
 
         return response
 
-    def delete_by_redis_keys(self, redis_keys: List[bytes]) -> List[bool]:
+    def delete_by_redis_keys(
+        self, redis_keys: Union[List[bytes], List[str]]
+    ) -> List[bool]:
         response = []
         keys_to_delete = []
 
         for redis_key in redis_keys:
-            if isinstance(redis_key, bytes):
-                redis_key = redis_key.decode()
+            # Prepare both versions for lookup
+            candidates = [redis_key]
+            if isinstance(redis_key, str):
+                candidates.append(redis_key.encode("utf-8"))
+            elif isinstance(redis_key, bytes):
+                try:
+                    candidates.append(redis_key.decode("utf-8"))
+                except UnicodeDecodeError:
+                    pass  # Non-UTF-8 bytes, skip str version
+
             for cache_key in self._cache:
-                if redis_key in cache_key.redis_keys:
+                if any(candidate in cache_key.redis_keys for candidate in candidates):
                     keys_to_delete.append(cache_key)
                     response.append(True)
 
diff --git a/tests/test_cache.py b/tests/test_cache.py
index 1f3193c49d..93b5870bb6 100644
--- a/tests/test_cache.py
+++ b/tests/test_cache.py
@@ -1038,6 +1038,40 @@ def test_delete_by_redis_keys_removes_associated_entries(self, mock_connection):
         assert len(cache.collection) == 1
         assert cache.get(cache_key4).cache_value == b"bar3"
 
+    def test_delete_by_redis_keys_with_non_utf8_bytes_key(self, mock_connection):
+        """cache fails to invalidate entries when redis_keys contain non-UTF-8 bytes."""
+        cache = DefaultCache(CacheConfig(max_size=5))
+
+        # Valid UTF-8 key works
+        utf8_key = b"foo"
+        utf8_cache_key = CacheKey(command="GET", redis_keys=(utf8_key,))
+        assert cache.set(
+            CacheEntry(
+                cache_key=utf8_cache_key,
+                cache_value=b"bar",
+                status=CacheEntryStatus.VALID,
+                connection_ref=mock_connection,
+            )
+        )
+
+        # Non-UTF-8 bytes key
+        bad_key = b"f\xffoo"
+        bad_cache_key = CacheKey(command="GET", redis_keys=(bad_key,))
+        assert cache.set(
+            CacheEntry(
+                cache_key=bad_cache_key,
+                cache_value=b"bar2",
+                status=CacheEntryStatus.VALID,
+                connection_ref=mock_connection,
+            )
+        )
+
+        # Delete both keys: utf8 should succeed, non-utf8 exposes bug
+        results = cache.delete_by_redis_keys([utf8_key, bad_key])
+
+        assert results[0] is True
+        assert results[1] is True, "Cache did not remove entry for non-UTF8 bytes key"
+
     def test_flush(self, mock_connection):
         cache = DefaultCache(CacheConfig(max_size=5))