probe test servers

Author: Myron Scott

azure-pipelines.yml

@@ -28,41 +28,41 @@ jobs:
   parameters:
     jdkVersion: '1.8'
     jobName: 'LinuxJava8'
-    gradleTasks: ':clustered:integration-test:test --tests=org.ehcache.clustered.replication.BasicClusteredCacheOpsReplicationMultiThreadedTest'
+    gradleTasks: ':clustered:integration-test:test --info --tests=org.ehcache.clustered.replication.BasicClusteredCacheOpsReplicationMultiThreadedTest'
 
 - template: build-templates/gradle-common.yml@templates
   parameters:
     jdkVersion: '1.8'
     options: '-PtestVM=java11Home'
     jobName: 'LinuxJava11'
-    gradleTasks: ':clustered:integration-test:test --tests=org.ehcache.clustered.replication.BasicClusteredCacheOpsReplicationMultiThreadedTest'
+    gradleTasks: ':clustered:integration-test:test --info --tests=org.ehcache.clustered.replication.BasicClusteredCacheOpsReplicationMultiThreadedTest'
 
 - template: build-templates/gradle-common.yml@templates
   parameters:
     jdkVersion: '1.8'
     options: '-PtestVM=java17Home'
     jobName: 'LinuxJava17'
-    gradleTasks: ':clustered:integration-test:test --tests=org.ehcache.clustered.replication.BasicClusteredCacheOpsReplicationMultiThreadedTest'
+    gradleTasks: ':clustered:integration-test:test --info --tests=org.ehcache.clustered.replication.BasicClusteredCacheOpsReplicationMultiThreadedTest'
 
 - template: build-templates/gradle-common.yml@templates
   parameters:
     vmImage: 'windows-latest'
     jdkVersion: '1.8'
     jobName: 'WindowsJava8'
-    gradleTasks: ':clustered:integration-test:test --tests=org.ehcache.clustered.replication.BasicClusteredCacheOpsReplicationMultiThreadedTest'
+    gradleTasks: ':clustered:integration-test:test --info --tests=org.ehcache.clustered.replication.BasicClusteredCacheOpsReplicationMultiThreadedTest'
 
 - template: build-templates/gradle-common.yml@templates
   parameters:
     vmImage: 'windows-latest'
     jdkVersion: '1.8'
     options: '-PtestVM=java11Home'
     jobName: 'WindowsJava11'
-    gradleTasks: ':clustered:integration-test:test --tests=org.ehcache.clustered.replication.BasicClusteredCacheOpsReplicationMultiThreadedTest'
+    gradleTasks: ':clustered:integration-test:test --info --tests=org.ehcache.clustered.replication.BasicClusteredCacheOpsReplicationMultiThreadedTest'
 
 - template: build-templates/gradle-common.yml@templates
   parameters:
     vmImage: 'windows-latest'
     jdkVersion: '1.8'
     options: '-PtestVM=java17Home'
     jobName: 'WindowsJava17'
-    gradleTasks: ':clustered:integration-test:test --tests=org.ehcache.clustered.replication.BasicClusteredCacheOpsReplicationMultiThreadedTest'
+    gradleTasks: ':clustered:integration-test:test --info --tests=org.ehcache.clustered.replication.BasicClusteredCacheOpsReplicationMultiThreadedTest'

clustered/integration-test/src/test/java/org/ehcache/clustered/replication/BasicClusteredCacheOpsReplicationMultiThreadedTest.java

@@ -45,13 +45,15 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.Serializable;
+import java.net.InetSocketAddress;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
@@ -69,6 +71,9 @@
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
 import static org.junit.Assert.fail;
+import org.terracotta.connection.ConnectionException;
+import org.terracotta.connection.Diagnostics;
+import org.terracotta.connection.DiagnosticsFactory;
 
 
 /**
@@ -98,7 +103,7 @@ public static Consistency[] data() {
 
   @ClassRule @Rule
   public static final ParallelTestCluster CLUSTER = new ParallelTestCluster(newCluster(2).in(clusterPath())
-    .withServiceFragment(offheapResource("primary-server-resource", 24)).build());
+    .withServiceFragment(offheapResource("primary-server-resource", 24)).withTcProperty("tc.messages.grouping.enabled", "false").build());
 
   @Rule
   public final TestName testName = new TestName();
@@ -110,6 +115,8 @@ public static Consistency[] data() {
   private final ThreadLocalRandom random = ThreadLocalRandom.current();
 
   private final ExecutorService executorService = Executors.newWorkStealingPool(NUM_OF_THREADS);
+  
+  private final Probe probe = new Probe();
 
   @Before
   public void startServers() throws Exception {
@@ -134,6 +141,7 @@ public void startServers() throws Exception {
     cache2 = cacheManager2.createCache(testName.getMethodName(), config);
 
     caches = Arrays.asList(cache1, cache2);
+    probe.loop();
   }
 
   @After
@@ -148,9 +156,10 @@ public void tearDown() throws Exception {
     if(cacheManager2 != null && cacheManager2.getStatus() != Status.UNINITIALIZED) {
       cacheManager2.close();
     }
+    
   }
 
-  @Test(timeout=180000)
+  @Test(timeout=360000)
   public void testCRUD() throws Exception {
     Set<Long> universalSet = ConcurrentHashMap.newKeySet();
     List<Future<?>> futures = new ArrayList<>();
@@ -192,7 +201,7 @@ public void testCRUD() throws Exception {
 
   }
 
-  @Test(timeout=180000)
+  @Test(timeout=360000)
   public void testBulkOps() throws Exception {
     Set<Long> universalSet = ConcurrentHashMap.newKeySet();
     List<Future<?>> futures = new ArrayList<>();
@@ -237,7 +246,7 @@ public void testBulkOps() throws Exception {
 
   }
 
-  @Test(timeout=180000)
+  @Test(timeout=360000)
   public void testClear() throws Exception {
     List<Future<?>> futures = new ArrayList<>();
     Set<Long> universalSet = ConcurrentHashMap.newKeySet();
@@ -287,4 +296,44 @@ private static class BlobValue implements Serializable {
     private final byte[] data = new byte[10 * 1024];
   }
 
+  public class Probe {
+  
+  /**
+   * @param args the command line arguments
+   */
+  public void loop() {
+    String[] servers = CLUSTER.getClusterHostPorts();
+    for (String hostPort : servers) {
+      String[] hp = hostPort.split("[:]");
+      InetSocketAddress inet = InetSocketAddress.createUnresolved(hp[0], Integer.parseInt(hp[1]));
+      log.info("starting probe for " + hostPort);
+      new Thread(()->{
+        while (!executorService.isShutdown()) {
+
+        try (Diagnostics d = DiagnosticsFactory.connect(inet, new Properties())) {
+          while (!executorService.isShutdown()) {
+            try {
+              probe(d);
+              log.info("sleeping for 10 sec.");
+              Thread.sleep(10_000L);
+            } catch (InterruptedException ie) {
+              ie.printStackTrace();
+            }
+          }
+        } catch (ConnectionException e) {
+          e.printStackTrace();
+        }
+      }
+      
+      }).start();
+    }
+  }
+  
+  private void probe(Diagnostics d) {
+    log.info("===== PROBE =====");
+    log.info(d.getClusterState());
+    log.info(d.getThreadDump());
+  }
+  }
+  
 }