geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From zho...@apache.org
Subject [geode] 01/01: GEODE-5249: add test code to display serial gateway sender queue's content
Date Thu, 24 May 2018 18:29:48 GMT
This is an automated email from the ASF dual-hosted git repository.

zhouxj pushed a commit to branch feature/GEODE-5249
in repository https://gitbox.apache.org/repos/asf/geode.git

commit 40f51112fb8759c4063f650b3bb7b7a90cac59dc
Author: zhouxh <gzhou@pivotal.io>
AuthorDate: Wed May 23 17:50:28 2018 -0700

    GEODE-5249: add test code to display serial gateway sender queue's content
---
 .../wan/AbstractGatewaySenderEventProcessor.java   |  9 +++++
 ...oncurrentSerialGatewaySenderEventProcessor.java | 16 +++++++++
 .../serial/SerialGatewaySenderEventProcessor.java  | 17 +++++++++
 .../cache/wan/serial/SerialGatewaySenderQueue.java |  4 +++
 ...SerialGatewaySenderEventProcessorJUnitTest.java | 33 ++++++++++++++++++
 .../geode/internal/cache/wan/WANTestBase.java      | 40 ++++++++++++++++++++++
 .../SerialGatewaySenderOperationsDUnitTest.java    | 34 ++++++++++++++++++
 7 files changed, 153 insertions(+)

diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
index 89fa586..0004789 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
@@ -1380,4 +1380,13 @@ public abstract class AbstractGatewaySenderEventProcessor extends Thread
{
       return true;
     }
   }
+
+  public String printUnprocessedEvents() {
+    return null;
+  }
+
+  public String printUnprocessedTokens() {
+    return null;
+  }
+
 }
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/ConcurrentSerialGatewaySenderEventProcessor.java
b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/ConcurrentSerialGatewaySenderEventProcessor.java
index 8ec6ce1..195e1a7 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/ConcurrentSerialGatewaySenderEventProcessor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/ConcurrentSerialGatewaySenderEventProcessor.java
@@ -396,4 +396,20 @@ public class ConcurrentSerialGatewaySenderEventProcessor
       serialProcessor.enqueueEvent(event);
     }
   }
+
+  public String printUnprocessedEvents() {
+    StringBuffer sb = new StringBuffer();
+    for (SerialGatewaySenderEventProcessor serialProcessor : this.processors) {
+      sb.append(serialProcessor.printUnprocessedEvents() + "\n");
+    }
+    return sb.toString();
+  }
+
+  public String printUnprocessedTokens() {
+    StringBuffer sb = new StringBuffer();
+    for (SerialGatewaySenderEventProcessor serialProcessor : this.processors) {
+      sb.append(serialProcessor.printUnprocessedTokens() + "\n");
+    }
+    return sb.toString();
+  }
 }
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderEventProcessor.java
b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderEventProcessor.java
index 39609c7..121469f 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderEventProcessor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderEventProcessor.java
@@ -20,12 +20,14 @@ import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Random;
+import java.util.Set;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 import org.apache.logging.log4j.Logger;
 
@@ -902,4 +904,19 @@ public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEven
     this.getSender().setModifiedEventId(droppedEvent);
     sendBatchDestroyOperationForDroppedEvent(droppedEvent, -1);
   }
+
+  private String printEventIdList(Set<EventID> eventIds) {
+    StringBuffer sb = new StringBuffer().append("[").append(
+        eventIds.stream().map(entry -> entry.expensiveToString()).collect(Collectors.joining(",
")))
+        .append("]");
+    return sb.toString();
+  }
+
+  public String printUnprocessedEvents() {
+    return printEventIdList(this.unprocessedEvents.keySet());
+  }
+
+  public String printUnprocessedTokens() {
+    return printEventIdList(this.unprocessedTokens.keySet());
+  }
 }
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderQueue.java
b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderQueue.java
index 64e3a6a..6cfe7f4 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderQueue.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderQueue.java
@@ -1294,4 +1294,8 @@ public class SerialGatewaySenderQueue implements RegionQueue {
       }
     }
   }
+
+  public String displayContent() {
+    return this.region.keySet().toString();
+  }
 }
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderEventProcessorJUnitTest.java
b/geode-core/src/test/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderEventProcessorJUnitTest.java
index f21634e..6cb0fdb 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderEventProcessorJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderEventProcessorJUnitTest.java
@@ -14,23 +14,30 @@
  */
 package org.apache.geode.internal.cache.wan.serial;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import java.util.Map;
+
+import org.apache.logging.log4j.Logger;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
+import org.springframework.test.util.ReflectionTestUtils;
 
 import org.apache.geode.cache.Operation;
 import org.apache.geode.internal.cache.EntryEventImpl;
 import org.apache.geode.internal.cache.EventID;
 import org.apache.geode.internal.cache.LocalRegion;
+import org.apache.geode.internal.cache.ha.ThreadIdentifier;
 import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
 import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
 import org.apache.geode.internal.cache.wan.GatewaySenderStats;
+import org.apache.geode.internal.logging.LogService;
 import org.apache.geode.test.junit.categories.UnitTest;
 
 @Category(UnitTest.class)
@@ -40,6 +47,8 @@ public class SerialGatewaySenderEventProcessorJUnitTest {
 
   private TestSerialGatewaySenderEventProcessor processor;
 
+  private static final Logger logger = LogService.getLogger();
+
   @Before
   public void setUp() throws Exception {
     this.sender = mock(AbstractGatewaySender.class);
@@ -104,6 +113,30 @@ public class SerialGatewaySenderEventProcessorJUnitTest {
     }
   }
 
+  @Test
+  public void validateUnProcessedEventsList() {
+    Map<EventID, AbstractGatewaySender.EventWrapper> unprocessedEvents =
+        (Map<EventID, AbstractGatewaySender.EventWrapper>) ReflectionTestUtils.getField(processor,
+            "unprocessedEvents");
+
+    long complexThreadId1 = ThreadIdentifier.createFakeThreadIDForParallelGSPrimaryBucket(0,
1, 1);
+    long complexThreadId3 = ThreadIdentifier.createFakeThreadIDForParallelGSPrimaryBucket(0,
3, 3);
+    unprocessedEvents.put(new EventID("mem1".getBytes(), complexThreadId1, 1L), null);
+    unprocessedEvents.put(new EventID("mem2".getBytes(), 2L, 2L), null);
+
+    String unProcessedEvents = this.processor.printUnprocessedEvents();
+    logger.info("UnprocessedEvents: " + unProcessedEvents);
+    assertThat(unProcessedEvents).contains("threadID=0x1010000|1;sequenceID=1");
+    assertThat(unProcessedEvents).contains("threadID=2;sequenceID=2");
+
+    processor.unprocessedTokens.put(new EventID("mem3".getBytes(), complexThreadId3, 3L),
3L);
+    processor.unprocessedTokens.put(new EventID("mem4".getBytes(), 4L, 4L), 4L);
+    String unProcessedTokens = this.processor.printUnprocessedTokens();
+    logger.info("UnprocessedTokens: " + unProcessedTokens);
+    assertThat(unProcessedTokens).contains("threadID=0x3010000|3;sequenceID=3");
+    assertThat(unProcessedTokens).contains("threadID=4;sequenceID=4");
+  }
+
   private EventID handlePrimaryEvent() {
     GatewaySenderEventImpl gsei = mock(GatewaySenderEventImpl.class);
     EventID id = mock(EventID.class);
diff --git a/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/WANTestBase.java
b/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/WANTestBase.java
index a7a281a..6b11349 100644
--- a/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/WANTestBase.java
+++ b/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/WANTestBase.java
@@ -37,6 +37,7 @@ import static org.apache.geode.distributed.ConfigurationProperties.REMOTE_LOCATO
 import static org.apache.geode.distributed.ConfigurationProperties.START_LOCATOR;
 import static org.apache.geode.test.dunit.Host.getHost;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.hamcrest.Matchers.greaterThan;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -71,6 +72,7 @@ import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import javax.management.ObjectName;
 
@@ -1249,6 +1251,15 @@ public class WANTestBase extends DistributedTestCase {
     assert (statistics.getEventsDistributed() >= eventsDistributed);
   }
 
+  public static void validateGatewaySenderQueueHasContent(String senderId, VM... targetVms)
{
+    Awaitility.await().atMost(60, TimeUnit.SECONDS)
+        .until(() -> Arrays.stream(targetVms).map(vm -> vm.invoke(() -> {
+          AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender(senderId);
+          logger.info(displaySerialQueueContent(sender));
+          return sender.getEventQueueSize();
+        })).mapToInt(i -> i).sum(), greaterThan(0));
+  }
+
   public static void checkGatewayReceiverStats(int processBatches, int eventsReceived,
       int creates) {
     Set<GatewayReceiver> gatewayReceivers = cache.getGatewayReceivers();
@@ -3171,6 +3182,35 @@ public class WANTestBase extends DistributedTestCase {
     return null;
   }
 
+  public static String displaySerialQueueContent(final AbstractGatewaySender sender) {
+    StringBuilder message = new StringBuilder();
+    message.append("Is Primary: ").append(sender.isPrimary()).append(", ").append("Queue
Size: ")
+        .append(sender.getEventQueueSize());
+
+    if (sender.getQueues() != null) {
+      message.append(", ").append("Queue Count: ").append(sender.getQueues().size());
+      Stream<Object> stream = sender.getQueues().stream()
+          .map(regionQueue -> ((SerialGatewaySenderQueue) regionQueue).displayContent());
+
+      List<Object> list = stream.collect(Collectors.toList());
+      message.append(", ").append("Keys: ").append(list.toString());
+    }
+
+    AbstractGatewaySenderEventProcessor abstractProcessor = sender.getEventProcessor();
+    if (abstractProcessor == null) {
+      message.append(", ").append("Null Event Processor: ");
+    }
+    if (sender.isPrimary()) {
+      message.append(", ").append("Unprocessed Tokens: ")
+          .append(abstractProcessor.printUnprocessedTokens());
+    } else {
+      message.append(", ").append("Unprocessed Events: ")
+          .append(abstractProcessor.printUnprocessedEvents());
+    }
+
+    return message.toString();
+  }
+
   public static Integer getQueueContentSize(final String senderId) {
     return getQueueContentSize(senderId, false);
   }
diff --git a/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderOperationsDUnitTest.java
b/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderOperationsDUnitTest.java
index 4993f24..a729541 100644
--- a/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderOperationsDUnitTest.java
+++ b/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderOperationsDUnitTest.java
@@ -313,6 +313,40 @@ public class SerialGatewaySenderOperationsDUnitTest extends WANTestBase
{
   }
 
   @Test
+  public void testSerialGatewaySendersPrintQueueContents() throws Throwable {
+    Integer lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
+    Integer nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));
+
+    createCacheInVMs(nyPort, vm2, vm3);
+    createReceiverInVMs(vm2, vm3);
+
+    createSenderCaches(lnPort);
+
+    createSenderVM4();
+    createSenderVM5();
+
+    createReceiverRegions();
+
+    createSenderRegions();
+
+    startSenderInVMs("ln", vm4, vm5);
+    vm4.invoke(() -> pauseSender("ln"));
+    vm5.invoke(() -> pauseSender("ln"));
+
+    vm7.invoke(() -> doPuts(getTestMethodName() + "_RR", 20));
+
+    validateGatewaySenderQueueHasContent("ln", vm4, vm5);
+
+    vm4.invokeAsync(() -> resumeSender("ln"));
+    vm5.invokeAsync(() -> resumeSender("ln"));
+
+    vm4.invoke(() -> validateQueueSizeStat("ln", 0));
+    vm5.invoke(() -> validateQueueSizeStat("ln", 0));
+    vm4.invoke(() -> validateSecondaryQueueSizeStat("ln", 0));
+    vm5.invoke(() -> validateSecondaryQueueSizeStat("ln", 0));
+  }
+
+  @Test
   public void testStopOneSerialGatewaySenderBothPrimary() throws Throwable {
     Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
     Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2,
lnPort));

-- 
To stop receiving notification emails like this one, please contact
zhouxj@apache.org.

Mime
View raw message