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

…rdening

(merge ce/main -> 24.09 113664)

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

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

@@ -1,5 +1,5 @@
 /*
- * 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.
@@ -655,10 +655,12 @@ private String getCacheName(String sPrefix)
 
     private void restartService(NamedTopic<?> topic)
         {
-        Service service     = topic.getService();
-        String  serviceName = service.getInfo().getServiceName();
+        PagedTopicService service     = (PagedTopicService) topic.getService();
+        int               cMember     = service.getInfo().getServiceMembers().size();
+        String            serviceName = service.getInfo().getServiceName();
 
-        System.err.println("Stopping topics cache service " + serviceName);
+
+        System.err.println("Stopping topics service " + serviceName);
 
         Service serviceFinal = service instanceof SafeCacheService
                 ? ((SafeCacheService) service).getRunningCacheService()
@@ -667,8 +669,10 @@ private void restartService(NamedTopic<?> topic)
         serviceFinal.stop();
         // wait for DCS to restart the service
         Eventually.assertDeferred("Failed to restart service " + service, service::isRunning, is(true));
+        Eventually.assertDeferred("Failed to restart service waiting for membership count" + service,
+                () -> service.getInfo().getServiceMembers().size(), is(cMember));
 
-        System.err.println("Restarted topics cache service " + serviceName);
+        System.err.println("Restarted topics service " + serviceName);
         }
 
     // ----- inner class: Message -------------------------------------------