@@ -17,6 +17,7 @@
 import abc
 import collections
 import os
+import threading
 import time
 import uuid
 
@@ -203,7 +204,14 @@ def from_environment(cls):
 
     def __init__(self, redis):
         self.redis = redis
-        self.pipes = {}
+        self._pipes = threading.local()
+
+    @property
+    def pipes(self):
+        local = self._pipes
+        if not hasattr(local, "pipes"):
+            local.pipes = {}
+        return local.pipes
 
     def get(self, keys):
         """Implements :meth:`GlobalCache.get`."""
 
@@ -17,8 +17,12 @@
 """
 import os
 import pickle
+import threading
+import time
 import traceback
 
+import redis
+
 try:
     from unittest import mock
 except ImportError:  # pragma: NO PY3 COVER
@@ -346,3 +350,54 @@ def callback():
     error = error_info.value
     message = "".join(traceback.format_exception_only(type(error), error))
     assert "hello dad" not in message
+
+
+@pytest.mark.skipif(not USE_REDIS_CACHE, reason="Redis is not configured")
+@pytest.mark.usefixtures("client_context")
+def test_parallel_threads_lookup_w_redis_cache(namespace, dispose_of):
+    """Regression test for #496
+
+    https://github.com/googleapis/python-ndb/issues/496
+    """
+
+    class MonkeyPipeline(redis.client.Pipeline):
+        def mset(self, mapping):
+            """Force a delay here to expose concurrency error."""
+            time.sleep(0.05)
+            return super(MonkeyPipeline, self).mset(mapping)
+
+    with mock.patch("redis.client.Pipeline", MonkeyPipeline):
+        client = ndb.Client()
+        global_cache = ndb.RedisCache.from_environment()
+        activity = {"calls": 0}
+
+        class SomeKind(ndb.Model):
+            foo = ndb.IntegerProperty()
+
+        class LookupThread(threading.Thread):
+            def __init__(self, id):
+                super(LookupThread, self).__init__()
+                self.id = id
+
+            def run(self):
+                context = client.context(
+                    cache_policy=False,
+                    global_cache=global_cache,
+                    namespace=namespace,
+                )
+                with context:
+                    entity = SomeKind.get_by_id(self.id)
+                    assert entity.foo == 42
+                    activity["calls"] += 1
+
+        key = SomeKind(foo=42).put()
+        dispose_of(key._key)
+        id = key.id()
+
+        thread1, thread2 = LookupThread(id), LookupThread(id)
+        thread1.start()
+        thread2.start()
+        thread1.join()
+        thread2.join()
+
+        assert activity["calls"] == 2
@@ -235,7 +235,7 @@ def test_compare_and_swap():
         cache = global_cache.RedisCache(redis)
         pipe1 = mock.Mock(spec=("multi", "mset", "execute", "reset"))
         pipe2 = mock.Mock(spec=("multi", "mset", "execute", "reset"))
-        cache.pipes = {
+        cache._pipes.pipes = {
             "ay": global_cache._Pipeline(pipe1, "abc123"),
             "be": global_cache._Pipeline(pipe1, "abc123"),
             "see": global_cache._Pipeline(pipe2, "def456"),
@@ -277,7 +277,7 @@ def mock_expire(key, expires):
             expire=mock_expire,
             spec=("multi", "mset", "execute", "expire", "reset"),
         )
-        cache.pipes = {
+        cache._pipes.pipes = {
             "ay": global_cache._Pipeline(pipe1, "abc123"),
             "be": global_cache._Pipeline(pipe1, "abc123"),
             "see": global_cache._Pipeline(pipe2, "def456"),