COH-21923 Build: intermittent failure in LocalNamedTopicTests.should…

…HandleErrorWhenPublishing (main->ce)

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

prj/test/functional/topics/src/test/java/topics/LocalNamedTopicTests.java

@@ -55,6 +55,8 @@
 
 import common.SystemPropertyIsolation;
 
+import org.hamcrest.Matchers;
+
 import org.junit.After;
 import org.junit.Assume;
 import org.junit.BeforeClass;
@@ -225,18 +227,41 @@ public void shouldHandleErrorWhenPublishing() throws Exception
         int                 nError   = 1;
         CompletableFuture[] aFutures = new CompletableFuture[cValues];
 
+        final AtomicBoolean fPublisherClosed  = new AtomicBoolean(false);
+
+        // introduce an error after first asynchronous send succeeds
         addErrorInterceptor(topic, nError);
 
         String sPrefix = "Element-";
 
         try (Publisher<String>  publisher = topic.createPublisher();
              Subscriber<String> subscriber = topic.createSubscriber(Subscriber.CompleteOnEmpty.enabled()))
             {
-            for (int i=0; i<cValues; i++)
+            publisher.onClose(() -> fPublisherClosed.set(true));
+
+            System.out.print("Publishing");
+
+            for (int i=0; i<cValues && !fPublisherClosed.get(); i++)
                 {
-                aFutures[i] = publisher.send(sPrefix + i);
+                try
+                    {
+                    // validate that exception introduced by addErrorInterceptor call is not thrown at site of async send call
+                    aFutures[i] = publisher.send(sPrefix + i);
+                    System.out.print(".");
+                    }
+                catch (IllegalStateException e)
+                    {
+                    // ignore exception This publisher is no longer active
+                    // all it means is that second asynchronous send occurred before loop completed and resulted in publisher closing
+                    assertThat("should not fail on first publish", i, Matchers.greaterThan(0));
+                    assertThat(e.getMessage().contains("This publisher is no longer active"), is(true));
+                    System.out.println("handled IllegalStateException: " + e.getMessage());
+                    System.out.println("Publisher Closed: " + fPublisherClosed.get());
+                    break;
+                    }
                 }
 
+            System.out.println("");
             publisher.flush().join();
 
             // topic should contain just the first value
@@ -247,10 +272,13 @@ public void shouldHandleErrorWhenPublishing() throws Exception
         // First add completes as normal
         assertCompletedNormally(aFutures[0]);
 
-        // All other adds should have failed
+        // All other adds should have not occurred at all or be cancelled
         for (int i=1; i<cValues; i++)
             {
-            assertCancelled(aFutures[i]);
+            if (aFutures[i] != null)
+                {
+                assertCancelled(aFutures[i]);
+                }
             }
         }