geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kl...@apache.org
Subject [01/18] incubator-geode git commit: GEODE-1178 Unexpected DistributedSystemDisconnectedException caused by RejectedExecutionException
Date Thu, 14 Apr 2016 20:30:11 GMT
Repository: incubator-geode
Updated Branches:
  refs/heads/feature/GEODE-1162 d9344e0a7 -> e2813de40


GEODE-1178 Unexpected DistributedSystemDisconnectedException caused by RejectedExecutionException

This has been reported to JGroups.  While they're deciding what to do about
it I have coded a workaround in our StatRecorder class.  StatRecorder sits
in the JGroups stack just above the transport protocol that is throwing this
exception from its down() method.  StatRecorder will now catch the exception
and, after sleeping a short amount of time (10ms) it will retry as long as
the Manager is not shutting down.


Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/39e94bc8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/39e94bc8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/39e94bc8

Branch: refs/heads/feature/GEODE-1162
Commit: 39e94bc8beb22b62ed727640bbad3511affc9923
Parents: 86ab7cd
Author: Bruce Schuchardt <bschuchardt@pivotal.io>
Authored: Tue Apr 12 10:43:26 2016 -0700
Committer: Bruce Schuchardt <bschuchardt@pivotal.io>
Committed: Tue Apr 12 10:45:40 2016 -0700

----------------------------------------------------------------------
 .../internal/membership/gms/Services.java       | 23 +++++--
 .../membership/gms/fd/GMSHealthMonitor.java     |  1 -
 .../gms/messenger/JGroupsMessenger.java         |  2 +-
 .../membership/gms/messenger/StatRecorder.java  | 30 +++++++--
 .../gms/membership/StatRecorderJUnitTest.java   | 67 ++++++++++++++------
 5 files changed, 93 insertions(+), 30 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/39e94bc8/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/Services.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/Services.java
b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/Services.java
index 4484c00..4f5a1a4 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/Services.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/Services.java
@@ -102,6 +102,19 @@ public class Services {
   }
   
 
+  /**
+   * for testing only - create a non-functional Services object with a Stopper
+   */
+  public Services() {
+    this.cancelCriterion = new Stopper();
+    this.stats = null;
+    this.config = null;
+    this.manager = null;
+    this.joinLeave = null;
+    this.healthMon = null;
+    this.messenger = null;
+    this.auth = null;
+  }
 
   public Services(
       DistributedMembershipListener listener, DistributionConfig config,
@@ -348,10 +361,10 @@ public class Services {
   public boolean isAutoReconnectEnabled() {
     return !getConfig().getDistributionConfig().getDisableAutoReconnect();
   }
-  
+
   public class Stopper extends CancelCriterion {
     volatile String reasonForStopping = null;
-    
+
     public void cancel(String reason) {
       this.reasonForStopping = reason;
     }
@@ -362,7 +375,7 @@ public class Services {
         return Services.this.shutdownCause.toString();
       return reasonForStopping;
     }
-    
+
     public boolean isCancelInProgress() {
       return cancelInProgress() != null;
     }
@@ -381,7 +394,7 @@ public class Services {
         }
       }
     }
-    
+
   }
-  
+
 }

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/39e94bc8/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/fd/GMSHealthMonitor.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/fd/GMSHealthMonitor.java
b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/fd/GMSHealthMonitor.java
index 510c5a8..5427d77 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/fd/GMSHealthMonitor.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/fd/GMSHealthMonitor.java
@@ -1190,7 +1190,6 @@ public class GMSHealthMonitor implements HealthMonitor, MessageHandler
{
    * @param initiator
    * @param sMembers
    * @param cv
-   * @param initiateRemoval
    */
   private void checkIfAvailable(final InternalDistributedMember initiator,
       List<SuspectRequest> sMembers, final NetView cv) {

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/39e94bc8/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessenger.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessenger.java
b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessenger.java
index 2dfeeaa..a9abea3 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessenger.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessenger.java
@@ -293,7 +293,7 @@ public class JGroupsMessenger implements Messenger {
     // give the stats to the jchannel statistics recorder
     StatRecorder sr = (StatRecorder)myChannel.getProtocolStack().findProtocol(StatRecorder.class);
     if (sr != null) {
-      sr.setDMStats(services.getStatistics());
+      sr.setServices(services);
     }
     
     Transport transport = (Transport)myChannel.getProtocolStack().getTransport();

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/39e94bc8/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/StatRecorder.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/StatRecorder.java
b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/StatRecorder.java
index e29d71e..e013de6 100755
--- a/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/StatRecorder.java
+++ b/geode-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/StatRecorder.java
@@ -32,6 +32,8 @@ import com.gemstone.gemfire.distributed.internal.DMStats;
 import com.gemstone.gemfire.distributed.internal.membership.gms.GMSUtil;
 import com.gemstone.gemfire.distributed.internal.membership.gms.Services;
 
+import java.util.concurrent.RejectedExecutionException;
+
 /**
  * JGroups doesn't capture quite the stats we want so this protocol is
  * inserted into the stack to gather the missing ones.
@@ -46,17 +48,19 @@ public class StatRecorder extends Protocol {
   private static final int INCOMING = 1;
   
   DMStats stats;
+  Services services;
   
   private final short nakackHeaderId = ClassConfigurator.getProtocolId(NAKACK2.class);
   private final short unicastHeaderId = ClassConfigurator.getProtocolId(UNICAST3.class);
   private final short frag2HeaderId = ClassConfigurator.getProtocolId(FRAG2.class);
   
   /**
-   * set the statistics object to modify when events are detected
-   * @param stats
+   * sets the services object of the GMS that is using this recorder
+   * @param services the Services collective of the GMS
    */
-  public void setDMStats(DMStats stats) {
-    this.stats = stats;
+  public void setServices(Services services) {
+    this.services = services;
+    this.stats = services.getStatistics();
   }
   
   @Override
@@ -81,7 +85,23 @@ public class StatRecorder extends Protocol {
       filter(msg, OUTGOING);
       break;
     }
-    return down_prot.down(evt);
+    do {
+      try {
+        return down_prot.down(evt);
+      } catch (RejectedExecutionException e) {
+        logger.debug("retrying JGroups message transmission due to rejected execution (GEODE-1178)");
+        try {
+          Thread.sleep(10);
+        } catch (InterruptedException ie) {
+          // down() does not throw InterruptedException so we can only set the interrupt
flag and return
+          Thread.currentThread().interrupt();
+          return null;
+        }
+      }
+    } while (services != null
+      && !services.getManager().shutdownInProgress()
+      && !services.getCancelCriterion().isCancelInProgress());
+    return null;
   }
   
 

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/39e94bc8/geode-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/StatRecorderJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/StatRecorderJUnitTest.java
b/geode-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/StatRecorderJUnitTest.java
index 91d4a57..b7b80ac 100755
--- a/geode-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/StatRecorderJUnitTest.java
+++ b/geode-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/membership/StatRecorderJUnitTest.java
@@ -16,33 +16,34 @@
  */
 package com.gemstone.gemfire.distributed.internal.membership.gms.membership;
 
-import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.assertEquals;
-
-import java.util.Properties;
-
-import org.jgroups.Event;
-import org.jgroups.Message;
-import org.jgroups.protocols.UNICAST3.Header;
-import org.jgroups.protocols.pbcast.NakAckHeader2;
-import org.jgroups.stack.Protocol;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-
 import com.gemstone.gemfire.distributed.internal.DistributionConfig;
 import com.gemstone.gemfire.distributed.internal.DistributionConfigImpl;
 import com.gemstone.gemfire.distributed.internal.DistributionManager;
 import com.gemstone.gemfire.distributed.internal.LonerDistributionManager.DummyDMStats;
 import com.gemstone.gemfire.distributed.internal.membership.gms.ServiceConfig;
 import com.gemstone.gemfire.distributed.internal.membership.gms.Services;
+import com.gemstone.gemfire.distributed.internal.membership.gms.interfaces.Manager;
 import com.gemstone.gemfire.distributed.internal.membership.gms.messenger.JGroupsMessenger;
 import com.gemstone.gemfire.distributed.internal.membership.gms.messenger.StatRecorder;
 import com.gemstone.gemfire.internal.admin.remote.RemoteTransportConfig;
 import com.gemstone.gemfire.test.junit.categories.UnitTest;
+import org.jgroups.Event;
+import org.jgroups.Message;
+import org.jgroups.protocols.UNICAST3.Header;
+import org.jgroups.protocols.pbcast.NakAckHeader2;
+import org.jgroups.stack.Protocol;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.util.Properties;
+import java.util.concurrent.RejectedExecutionException;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.isA;
+import static org.mockito.Mockito.*;
 
 /**
  * This class tests the GMS StatRecorder class, which records JGroups
@@ -53,14 +54,17 @@ public class StatRecorderJUnitTest {
   Protocol mockDownProtocol, mockUpProtocol;
   StatRecorder recorder;
   MyStats stats = new MyStats();
+  Services services;
   
   @Before
   public void initMocks() throws Exception {
     // create a StatRecorder that has mock up/down protocols and stats
     mockDownProtocol = mock(Protocol.class);
     mockUpProtocol = mock(Protocol.class);
+    services = mock(Services.class);
+    when(services.getStatistics()).thenReturn(stats);
     recorder = new StatRecorder();
-    recorder.setDMStats(stats);
+    recorder.setServices(services);
     recorder.setUpProtocol(mockUpProtocol);
     recorder.setDownProtocol(mockDownProtocol);
   }
@@ -94,6 +98,33 @@ public class StatRecorderJUnitTest {
         stats.ucastRetransmits == 1);
   }
 
+
+  @Test
+  public void recorderHandlesRejectedExecution() throws Exception {
+    Message msg = mock(Message.class);
+    when(msg.getHeader(any(Short.class))).thenReturn(Header.createDataHeader(1L, (short)1,
true));
+    when(msg.size()).thenReturn(150L);
+
+
+    // GEODE-1178, the TP protocol may throw a RejectedExecutionException & StatRecorder
should retry
+    when(mockDownProtocol.down(any(Event.class))).thenThrow(new RejectedExecutionException());
+
+    // after the first down() throws an exception we want StatRecorder to retry, so
+    // we set the Manager to say no shutdown is in progress the first time and then say
+    // one IS in progress so we can break out of the StatRecorder exception handling loop
+    when(services.getCancelCriterion()).thenReturn(new Services().getCancelCriterion());
+    Manager manager = mock(Manager.class);
+    when(services.getManager()).thenReturn(manager);
+    when(manager.shutdownInProgress()).thenReturn(Boolean.FALSE, Boolean.TRUE);
+
+    verify(mockDownProtocol, never()).down(isA(Event.class));
+
+    Event evt = new Event(Event.MSG, msg);
+    recorder.down(evt);
+
+    verify(mockDownProtocol, times(2)).down(isA(Event.class));
+  }
+
   /**
    * ensure that multicast events are recorded in DMStats
    */


Mime
View raw message