Enh 37387064 - [37381796->24.09.1] Topics: general refactoring and hardening

(merge ce/main -> ce/24.09 113488)

[git-p4: depot-paths = "//dev/coherence-ce/release/coherence-ce-v24.09/": change = 113492]

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