mhansonp commented on code in PR #7608: URL: https://github.com/apache/geode/pull/7608#discussion_r853584296
########## geode-cq/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/DurableClientTestCase.java: ########## @@ -590,73 +616,92 @@ private void durableFailoverAfterReconnect(int redundancyLevel) { } durableClientVM.invoke(CacheServerTestUtil::disableShufflingOfEndpoints); - durableClientVM.invoke(() -> CacheServerTestUtil.createCacheClient(clientPool, regionName, + // Create the durable client cache (type = cache) + durableClientVM.invoke(() -> createCacheClient(clientPool, regionName, getClientDistributedSystemProperties(durableClientId, VERY_LONG_DURABLE_TIMEOUT_SECONDS), - Boolean.TRUE)); + true)); + + // Register interest in all entries in the region + registerInterest(durableClientVM, regionName, true, InterestResultPolicy.NONE); // Send clientReady message - durableClientVM.invoke(new CacheSerializableRunnable("Send clientReady") { + durableClientVM.invoke("Send clientReady", new CacheSerializableRunnable() { @Override public void run2() throws CacheException { - CacheServerTestUtil.getCache().readyForEvents(); + getCache().readyForEvents(); } }); - registerInterest(durableClientVM, regionName, true, InterestResultPolicy.NONE); - + // Verify the durable client is connected to the servers. + verifyDurableClientPresent(VERY_LONG_DURABLE_TIMEOUT_SECONDS, durableClientId, server1VM); + if (redundancyLevel == 1) { + verifyDurableClientPresent(VERY_LONG_DURABLE_TIMEOUT_SECONDS, durableClientId, server2VM); + } // Start normal publisher client - publisherClientVM.invoke(() -> CacheServerTestUtil.createCacheClient( + publisherClientVM.invoke(() -> createCacheClient( getClientPool(NetworkUtils.getServerHostName(), server1Port, server2Port, false), regionName)); - // Publish some entries - publishEntries(0, 1); + // put key 0 + publishEntries(publisherClientVM, 0, 1); - // Verify the durable client received the updates + // Verify the durable client received key 0 checkListenerEvents(1, 1, -1, durableClientVM); - verifyDurableClientPresent(VERY_LONG_DURABLE_TIMEOUT_SECONDS, durableClientId, server1VM); + // Wait to until the HARegionQueue has emptied before disconnecting the Durable client + // to avoid having left over messages that processed, in particular, the key 0 message + waitUntilHARegionQueueSizeIsZero(server1VM); + if (redundancyLevel == 1) { + waitUntilHARegionQueueSizeIsZero(server2VM); + } // Stop the durable client disconnectDurableClient(true); - // Stop server 1 - publisher will put 10 entries during shutdown/primary identification + // Stop server 1 - publisher will put 1 entries during shutdown/primary identification server1VM.invoke((SerializableRunnableIF) CacheServerTestUtil::closeCache); - // Publish updates during client downtime - publishEntries(1, 1); + // Put key 1 during client downtime + publishEntries(publisherClientVM, 1, 1); // Re-start the durable client that is kept alive on the server - restartDurableClient(VERY_LONG_DURABLE_TIMEOUT_SECONDS, clientPool, Boolean.TRUE); + restartDurableClient(VERY_LONG_DURABLE_TIMEOUT_SECONDS, clientPool, true); + // Re-register interest in all entries in the region registerInterest(durableClientVM, regionName, true, InterestResultPolicy.NONE); - publishEntries(2, 2); + // Send clientReady message + sendClientReady(durableClientVM); + + // Put keys 2 and 3 + publishEntries(publisherClientVM, 2, 2); // Verify the durable client received the updates before failover if (redundancyLevel == 1) { - checkListenerEvents(4, 1, -1, durableClientVM); + checkListenerEvents(3, 1, -1, durableClientVM); Review Comment: These numbers were contradictory to the required record count because 0 should not have been there to be counted. ########## geode-cq/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/DurableClientTestCase.java: ########## @@ -590,73 +616,92 @@ private void durableFailoverAfterReconnect(int redundancyLevel) { } durableClientVM.invoke(CacheServerTestUtil::disableShufflingOfEndpoints); - durableClientVM.invoke(() -> CacheServerTestUtil.createCacheClient(clientPool, regionName, + // Create the durable client cache (type = cache) + durableClientVM.invoke(() -> createCacheClient(clientPool, regionName, getClientDistributedSystemProperties(durableClientId, VERY_LONG_DURABLE_TIMEOUT_SECONDS), - Boolean.TRUE)); + true)); + + // Register interest in all entries in the region + registerInterest(durableClientVM, regionName, true, InterestResultPolicy.NONE); // Send clientReady message - durableClientVM.invoke(new CacheSerializableRunnable("Send clientReady") { + durableClientVM.invoke("Send clientReady", new CacheSerializableRunnable() { @Override public void run2() throws CacheException { - CacheServerTestUtil.getCache().readyForEvents(); + getCache().readyForEvents(); } }); - registerInterest(durableClientVM, regionName, true, InterestResultPolicy.NONE); - + // Verify the durable client is connected to the servers. + verifyDurableClientPresent(VERY_LONG_DURABLE_TIMEOUT_SECONDS, durableClientId, server1VM); + if (redundancyLevel == 1) { + verifyDurableClientPresent(VERY_LONG_DURABLE_TIMEOUT_SECONDS, durableClientId, server2VM); + } // Start normal publisher client - publisherClientVM.invoke(() -> CacheServerTestUtil.createCacheClient( + publisherClientVM.invoke(() -> createCacheClient( getClientPool(NetworkUtils.getServerHostName(), server1Port, server2Port, false), regionName)); - // Publish some entries - publishEntries(0, 1); + // put key 0 + publishEntries(publisherClientVM, 0, 1); - // Verify the durable client received the updates + // Verify the durable client received key 0 checkListenerEvents(1, 1, -1, durableClientVM); - verifyDurableClientPresent(VERY_LONG_DURABLE_TIMEOUT_SECONDS, durableClientId, server1VM); + // Wait to until the HARegionQueue has emptied before disconnecting the Durable client + // to avoid having left over messages that processed, in particular, the key 0 message + waitUntilHARegionQueueSizeIsZero(server1VM); + if (redundancyLevel == 1) { + waitUntilHARegionQueueSizeIsZero(server2VM); + } // Stop the durable client disconnectDurableClient(true); - // Stop server 1 - publisher will put 10 entries during shutdown/primary identification + // Stop server 1 - publisher will put 1 entries during shutdown/primary identification server1VM.invoke((SerializableRunnableIF) CacheServerTestUtil::closeCache); - // Publish updates during client downtime - publishEntries(1, 1); + // Put key 1 during client downtime + publishEntries(publisherClientVM, 1, 1); // Re-start the durable client that is kept alive on the server - restartDurableClient(VERY_LONG_DURABLE_TIMEOUT_SECONDS, clientPool, Boolean.TRUE); + restartDurableClient(VERY_LONG_DURABLE_TIMEOUT_SECONDS, clientPool, true); + // Re-register interest in all entries in the region registerInterest(durableClientVM, regionName, true, InterestResultPolicy.NONE); - publishEntries(2, 2); + // Send clientReady message + sendClientReady(durableClientVM); + + // Put keys 2 and 3 + publishEntries(publisherClientVM, 2, 2); // Verify the durable client received the updates before failover if (redundancyLevel == 1) { - checkListenerEvents(4, 1, -1, durableClientVM); + checkListenerEvents(3, 1, -1, durableClientVM); Review Comment: These numbers were contradictory to the required record count because key 0 should not have been there to be counted. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: notifications-unsubscr...@geode.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org