Bug 36409014 - Build: Intermittent failure in queues.GrpcClientDequeTests.shouldGetNewInstanceOfDestroyedQueue

(auto-submit 108107 after successfully running remote remote.full)
Job ID: job.9.20240403092259.17510

[git-p4: depot-paths = "//dev/coherence-ce/main/": change = 108190]

Author: thegridman

prj/coherence-grpc-proxy-common/src/main/java/com/oracle/coherence/grpc/proxy/common/BaseNamedCacheServiceImpl.java

@@ -13,6 +13,7 @@
 import com.google.protobuf.Empty;
 import com.oracle.coherence.common.base.Classes;
 import com.oracle.coherence.common.base.Exceptions;
+import com.oracle.coherence.common.base.Logger;
 import com.oracle.coherence.grpc.AddIndexRequest;
 import com.oracle.coherence.grpc.AggregateRequest;
 import com.oracle.coherence.grpc.BinaryHelper;
@@ -232,24 +233,14 @@ public void destroy(DestroyRequest request, StreamObserver<Empty> observer)
                         .asRuntimeException();
                 }
 
-            ConfigurableCacheFactory   ccf           = getCCF(request.getScope());
-            NamedCache<Binary, Binary> cachePassThru = ccf.ensureCache(sCacheName, NullImplementation.getClassLoader());
-            NamedCache<Binary, Binary> cache         = ccf.ensureCache(sCacheName, Classes.getContextClassLoader());
-            // we get caches via the CCF, so we must destroy them that way too
-            ccf.destroyCache(cachePassThru);
-            try
-                {
-                ccf.destroyCache(cache);
-                }
-            catch (Exception ignored)
-                {
-                // We may get an exception if destroying the first pass-thru cache also destroys the plain cache.
-                // We can just ignore it.
-                }
+            ConfigurableCacheFactory   ccf   = getCCF(request.getScope());
+            NamedCache<Binary, Binary> cache = ccf.ensureCache(sCacheName, null);
+            cache.destroy();
             handleUnary(Empty.getDefaultInstance(), null, safeObserver);
             }
         catch (Throwable t)
             {
+            Logger.err(t);
             handleUnary(Empty.getDefaultInstance(), t, safeObserver);
             }
         }

prj/coherence-grpc-proxy-common/src/main/java/com/oracle/coherence/grpc/proxy/common/MapListenerProxy.java

@@ -93,9 +93,21 @@ public MapListenerProxy(NamedCacheService service, StreamObserver<MapListenerRes
     @Override
     public void onNext(MapListenerRequest request)
         {
+        if (m_fCompleted)
+            {
+            throw Status.FAILED_PRECONDITION
+                    .withDescription("Received message after completion")
+                    .asRuntimeException();
+            }
         f_lock.lock();
         try
             {
+            if (m_fCompleted)
+                {
+                throw Status.FAILED_PRECONDITION
+                        .withDescription("Received message after completion")
+                        .asRuntimeException();
+                }
             if (m_holder == null)
                 {
                 m_holder = f_service.createRequestHolder(request,
@@ -189,6 +201,8 @@ public void onError(Throwable throwable)
                 {
                 // ignored - we already have errors.
                 }
+            m_fCompleted = true;
+            m_holder     = null;
             }
         finally
             {

prj/test/functional/grpc-proxy-netty/src/main/java/netty/grpc/proxy/NamedCacheServiceImplTest.java

@@ -170,6 +170,7 @@ void setupEach()
         m_testAsyncCache = testCache.async();
 
         m_testCCF = mock(ConfigurableCacheFactory.class);
+        when(m_testCCF.ensureCache(eq(TEST_CACHE_NAME), any())).thenReturn(testCache);
         when(m_testCCF.ensureCache(eq(TEST_CACHE_NAME), any(ClassLoader.class))).thenReturn(testCache);
         when(m_testCCF.getScopeName()).thenReturn(GrpcDependencies.DEFAULT_SCOPE);
 
@@ -758,7 +759,10 @@ public void shouldNotExecuteDestroyWithoutCacheName() throws Exception
     @Test
     public void shouldHandleDestroyError() throws Exception
         {
+        NamedCache cache = mock(NamedCache.class);
+        when(m_testAsyncCache.getNamedCache()).thenReturn(cache);
         doThrow(ERROR).when(m_testCCF).destroyCache(any(NamedCache.class));
+        doThrow(ERROR).when(cache).destroy();
 
         TestStreamObserver<Empty>  observer = new TestStreamObserver<>();
         NamedCacheService service = s_serviceProvider.getService(m_dependencies);

prj/test/functional/queues/src/main/java/queues/AbstractCollectionTests.java

@@ -7,6 +7,7 @@
 
 package queues;
 
+import com.oracle.bedrock.testsupport.deferred.Eventually;
 import com.tangosol.io.DefaultSerializer;
 import com.tangosol.io.Serializer;
 import com.tangosol.io.pof.ConfigurablePofContext;
@@ -147,7 +148,9 @@ public void shouldGetNewInstanceOfDestroyedQueue()
         {
         String sName = getNewName();
         NC     col1  = getNamedCollection(sName);
-        col1.destroy();
+        assertThat(col1.isActive(), is(true));
+        getSession().destroy(col1);
+        assertThat(col1.isDestroyed(), is(true));
         NC col2 = getNamedCollection(sName);
         assertThat(col2, is(not(sameInstance(col1))));
         }