Enh 37387065 - [37381796->25.03] Topics: general refactoring and hardening

(merge main -> ce/main 113487)

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

Author: thegridman

prj/coherence-core/src/main/java/com/tangosol/internal/net/topic/impl/paged/PagedTopicSubscriberConnector.java

@@ -1,12 +1,13 @@
 /*
- * Copyright (c) 2000, 2024, Oracle and/or its affiliates.
+ * Copyright (c) 2000, 2025, Oracle and/or its affiliates.
  *
  * Licensed under the Universal Permissive License v 1.0 as shown at
  * https://oss.oracle.com/licenses/upl.
  */
 
 package com.tangosol.internal.net.topic.impl.paged;
 
+import com.oracle.coherence.common.base.Blocking;
 import com.oracle.coherence.common.base.Exceptions;
 import com.oracle.coherence.common.base.Logger;
 
@@ -261,6 +262,25 @@ public void closeSubscription(ConnectedSubscriber<V> subscriber, boolean fDestro
             // finalizers in the JVM are not reliable that is probably not such a good idea.
             destroy(f_caches, f_subscriberGroupId, m_subscriptionId);
             }
+
+        // We need to ensure that the subscription has really gone.
+        // During a fail-over situation the subscriber may still exist in the configmap
+        // so we  need to repeat the closure notification
+        TopicSubscription subscription = getSubscription(subscriber, m_subscriptionId);
+        while (subscription != null && subscription.getSubscriberTimestamp(f_subscriberId) != Long.MAX_VALUE)
+            {
+            try
+                {
+                Blocking.sleep(100);
+                }
+            catch (InterruptedException e)
+                {
+                break;
+                }
+            Logger.fine("Repeating subscriber closed notification for topic subscriber: " + subscriber);
+            PagedTopicSubscription.notifyClosed(f_caches.Subscriptions, f_subscriberGroupId, m_subscriptionId, f_subscriberId);
+            subscription = getSubscription(subscriber, m_subscriptionId);
+            }
         }
 
     @Override

prj/test/functional/topics/src/main/java/topics/AbstractTopicsStorageRecoveryTests.java

@@ -237,7 +237,7 @@ public void shouldRecoverAfterCleanStorageRestart() throws Exception
                 catch (Throwable t)
                     {
                     Logger.err("Error in publish loop");
-                    t.printStackTrace();
+                    Logger.err(t);
                     }
                 fPublished.set(true);
                 };
@@ -296,7 +296,24 @@ optComplete, withIdentifyingName(sName)))
                                     {
                                     if (future != null && !future.isDone())
                                         {
-                                        future.cancel(true);
+                                        if (!future.cancel(true))
+                                            {
+                                            // the future has actually completed
+                                            if (future.isDone() && !future.isCompletedExceptionally())
+                                                {
+                                                Subscriber.Element<Message> element = future.get();
+                                                if (element != null)
+                                                    {
+                                                    mapReceived.put(element.getValue(), element);
+                                                    element.commit();
+                                                    cReceived.incrementAndGet();
+                                                    if (i >= 5)
+                                                        {
+                                                        fSubscribed.set(true);
+                                                        }
+                                                    }
+                                                }
+                                            }
                                         }
                                     if (!(t instanceof TimeoutException))
                                         {