test: fix testSendReceiveTwo and testBatchNack

multicloudj-common/src/test/java/com/salesforce/multicloudj/common/util/common/TestsUtil.java

@@ -143,4 +143,14 @@ public static void stopWireMockRecording() {
             wireMockServer.stopRecording();
         }
     }
+
+    /**
+     * Resets all WireMock scenario states to their initial "Started" state.
+     * ensuring each test starts with a clean scenario state.
+     */
+    public static void resetWireMockScenarios() {
+        if (wireMockServer != null) {
+            wireMockServer.resetScenarios();
+        }
+    }
 }

pubsub/pubsub-client/src/test/java/com/salesforce/multicloudj/pubsub/client/AbstractPubsubIT.java

@@ -80,10 +80,12 @@ public void shutdownWireMockServer() throws Exception {
     }
 
     /**
-     * Initialize the harness and start recording
+     * Initialize the harness and start recording.
+     * Resets WireMock scenario states to prevent state pollution between tests.
      */
     @BeforeEach
     public void setupTestEnvironment() {
+        TestsUtil.resetWireMockScenarios();
         TestsUtil.startWireMockRecording(harness.getPubsubEndpoint());
     }
 
@@ -234,8 +236,7 @@ public void testBatchAck() throws Exception {
             subscription.sendAcks(ackIDs).join();
         }
     }
-
-    @Disabled
+
     @Test
     @Timeout(120) // Integration test with batch operations - allow time for message delivery
     public void testBatchNack() throws Exception {
@@ -251,23 +252,10 @@ public void testBatchNack() throws Exception {
 
             TimeUnit.MILLISECONDS.sleep(500);
 
-            List<AckID> ackIDs = new java.util.ArrayList<>();
-            boolean isRecording = System.getProperty("record") != null;
-            long timeoutSeconds = isRecording ? 120 : 60; // Increased timeout for integration tests
-            long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSeconds);
-
-            while (ackIDs.size() < toSend.size() && System.nanoTime() < deadline) {
-                try {
-                    Message r = subscription.receive();
-                    if (r != null && r.getAckID() != null) {
-                        ackIDs.add(r.getAckID());
-                    } else {
-                        TimeUnit.MILLISECONDS.sleep(100);
-                    }
-                } catch (Exception e) {
-                    TimeUnit.MILLISECONDS.sleep(100);
-                }
-            }
+            List<Message> received = receiveMessages(subscription, toSend.size());
+            List<AckID> ackIDs = received.stream()
+                    .map(Message::getAckID)
+                    .collect(java.util.stream.Collectors.toList());
 
             Assertions.assertEquals(toSend.size(), ackIDs.size(),
                     "Should collect all AckIDs. Expected: " + toSend.size() + ", Got: " + ackIDs.size());
@@ -396,7 +384,6 @@ public void testMultipleSendReceiveWithoutBatch() throws Exception {
      */
     @Test
     @Timeout(120) // Integration test with multiple subscriptions - allow time for message delivery
-    @Disabled
     public void testSendReceiveTwo() throws Exception {
         // Create two subscriptions to the same topic
         AbstractSubscription subscription1 = harness.createSubscriptionDriverWithIndex(1);
@@ -443,22 +430,28 @@ public void testSendReceiveTwo() throws Exception {
      * Helper function: Receives messages from a subscription until the expected count is reached.
      */
     private List<Message> receiveMessages(AbstractSubscription subscription, int expectedCount) throws InterruptedException {
-        boolean isRecording = System.getProperty("record") != null;
-        long timeoutSeconds = isRecording ? 120 : 60;
+        long timeoutSeconds = 60;
         long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSeconds);
 
         List<Message> received = new ArrayList<>();
-        while (received.size() < expectedCount && System.nanoTime() < deadline) {
-            try {
+        try {
+            while (received.size() < expectedCount && System.nanoTime() < deadline) {
                 Message r = subscription.receive();
-                if (r != null && r.getAckID() != null) {
-                    received.add(r);
-                } else {
-                    TimeUnit.MILLISECONDS.sleep(100);
-                }
-            } catch (Exception e) {
-                TimeUnit.MILLISECONDS.sleep(100);
+                // receive() either returns a Message or throws an exception
+                received.add(r);
             }
+        } catch (Exception e) {
+            String errorMsg = String.format(
+                "Failed to receive messages: Got exception after receiving %d/%d messages.",
+                received.size(), expectedCount);
+            Assertions.fail(errorMsg, e);
+        }
+        // If the loop exits but received count is less than expected, it indicates a timeout occurred.
+        if (received.size() < expectedCount) {
+            String errorMsg = String.format(
+                "Timeout waiting for messages: Received %d/%d messages.",
+                received.size(), expectedCount);
+            Assertions.fail(errorMsg);
         }
         return received;
     }