bigdata-commit Mailing List for Blazegraph (powered by bigdata) (Page 59)

SourceForge Headquarters 225 Broadway Suite 1600 San Diego, CA 92101 +1 (858) 422-6466

Revision: 7276
          http://bigdata.svn.sourceforge.net/bigdata/?rev=7276&view=rev
Author:   thompsonbry
Date:     2013-08-09 16:01:16 +0000 (Fri, 09 Aug 2013)
Log Message:
-----------
The leader variable was not set by the time we tested the quorum token
and HAReadyToken. This made it impossible to report the error back to
the quorum leader.  The leader UUID is now passed as part of the
gather protocol message and the follower discovered the leader
synchronously when it receives that message and then passes the leader
proxy into the gather task.  This ensures that we can report the error
back to the service that initiated the gather.

@see #720

Modified Paths:
--------------
    branches/READ_CACHE2/bigdata/src/java/com/bigdata/ha/msg/HAGatherReleaseTimeRequest.java
    branches/READ_CACHE2/bigdata/src/java/com/bigdata/ha/msg/IHAGatherReleaseTimeRequest.java
    branches/READ_CACHE2/bigdata/src/java/com/bigdata/journal/AbstractJournal.java
    branches/READ_CACHE2/bigdata/src/java/com/bigdata/journal/Journal.java
    branches/READ_CACHE2/bigdata/src/java/com/bigdata/service/AbstractHATransactionService.java

Modified: branches/READ_CACHE2/bigdata/src/java/com/bigdata/ha/msg/HAGatherReleaseTimeRequest.java
===================================================================

--- branches/READ_CACHE2/bigdata/src/java/com/bigdata/ha/msg/HAGatherReleaseTimeRequest.java	2013-08-09 15:37:16 UTC (rev 7275)
+++ branches/READ_CACHE2/bigdata/src/java/com/bigdata/ha/msg/HAGatherReleaseTimeRequest.java	2013-08-09 16:01:16 UTC (rev 7276)
@@ -23,6 +23,8 @@
 */
 package com.bigdata.ha.msg;
 
+import java.util.UUID;
+
 public class HAGatherReleaseTimeRequest implements
         IHAGatherReleaseTimeRequest {
 
@@ -33,17 +35,19 @@
 
     private final long token;
     private final long timestampOnLeader;
+    private final UUID leaderId;
 
     public HAGatherReleaseTimeRequest(final long token,
-            final long timestampOnLeader) {
+            final long timestampOnLeader, final UUID leaderId) {
         this.token = token;
         this.timestampOnLeader = timestampOnLeader;
+        this.leaderId = leaderId;
     }
 
     @Override
     public String toString() {
         return super.toString() + "{token=" + token + ",timestampOnLeader="
-                + timestampOnLeader + "}";
+                + timestampOnLeader + ", leaderId=" + leaderId + "}";
     }
     
     @Override
@@ -55,5 +59,10 @@
     public long getTimestampOnLeader() {
         return timestampOnLeader;
     }
+    
+    @Override
+    public UUID getLeaderId() {
+        return leaderId;
+    }
 
 }

Modified: branches/READ_CACHE2/bigdata/src/java/com/bigdata/ha/msg/IHAGatherReleaseTimeRequest.java
===================================================================
--- branches/READ_CACHE2/bigdata/src/java/com/bigdata/ha/msg/IHAGatherReleaseTimeRequest.java	2013-08-09 15:37:16 UTC (rev 7275)
+++ branches/READ_CACHE2/bigdata/src/java/com/bigdata/ha/msg/IHAGatherReleaseTimeRequest.java	2013-08-09 16:01:16 UTC (rev 7276)
@@ -23,6 +23,8 @@
 */
 package com.bigdata.ha.msg;
 
+import java.util.UUID;
+
 /**
  * Message used to request information about the earliest commit point that is
  * pinned on a follower. This is used by the leader to make a decision about the
@@ -48,5 +50,11 @@
      * where the clocks are not synchronized on the services.
      */
     public long getTimestampOnLeader();
+
+    /**
+     * The UUID of the leader. This may be used to discover the service that
+     * is (or was) the leader even if the token has been invalidated.
+     */
+    public UUID getLeaderId();
     
 }

Modified: branches/READ_CACHE2/bigdata/src/java/com/bigdata/journal/AbstractJournal.java
===================================================================
--- branches/READ_CACHE2/bigdata/src/java/com/bigdata/journal/AbstractJournal.java	2013-08-09 15:37:16 UTC (rev 7275)
+++ branches/READ_CACHE2/bigdata/src/java/com/bigdata/journal/AbstractJournal.java	2013-08-09 16:01:16 UTC (rev 7276)
@@ -7691,10 +7691,28 @@
 
             }
             
+            /*
+             * Lookup the leader using its UUID.
+             * 
+             * Note: We do not use the token to find the leader. If the token is
+             * invalid, then we will handle that once we are in the GatherTask.
+             * 
+             * Note: We do this early and pass it into the GatherTask. We can
+             * not send back an RMI response unless we know the leader's proxy.
+             */
+            
+            final UUID leaderId = req.getLeaderId();
+
+            final HAGlue leader = getQuorum().getClient().getService(leaderId);
+
+            if (leader == null)
+                throw new RuntimeException(
+                        "Could not discover the quorum leader.");
+            
             final Callable<IHANotifyReleaseTimeResponse> task = ((AbstractHATransactionService) AbstractJournal.this
                     .getLocalTransactionManager()
                     .getTransactionService())
-                    .newGatherMinimumVisibleCommitTimeTask(req);
+                    .newGatherMinimumVisibleCommitTimeTask(leader, req);
 
             final FutureTask<IHANotifyReleaseTimeResponse> ft = new FutureTask<IHANotifyReleaseTimeResponse>(task);
 

Modified: branches/READ_CACHE2/bigdata/src/java/com/bigdata/journal/Journal.java
===================================================================
--- branches/READ_CACHE2/bigdata/src/java/com/bigdata/journal/Journal.java	2013-08-09 15:37:16 UTC (rev 7275)
+++ branches/READ_CACHE2/bigdata/src/java/com/bigdata/journal/Journal.java	2013-08-09 16:01:16 UTC (rev 7276)
@@ -598,7 +598,7 @@
             try {
 
                 final IHAGatherReleaseTimeRequest msg = new HAGatherReleaseTimeRequest(
-                        token, timestampOnLeader);
+                        token, timestampOnLeader, leaderId);
                 
                 // Do not send message to self (leader is at index 0).
                 for (int i = 1; i < joinedServiceIds.length; i++) {
@@ -1376,9 +1376,9 @@
 
         @Override
         public Callable<IHANotifyReleaseTimeResponse> newGatherMinimumVisibleCommitTimeTask(
-                final IHAGatherReleaseTimeRequest req) {
+                final HAGlue leader, final IHAGatherReleaseTimeRequest req) {
 
-            return new GatherTask(req);
+            return new GatherTask(leader, req);
 
         }
 
@@ -1415,13 +1415,19 @@
          */
         private class GatherTask implements Callable<IHANotifyReleaseTimeResponse> {
 
+            private final HAGlue leader;
             private final IHAGatherReleaseTimeRequest req;
 
-            public GatherTask(final IHAGatherReleaseTimeRequest req) {
+            public GatherTask(final HAGlue leader, final IHAGatherReleaseTimeRequest req) {
 
+                if (leader == null)
+                    throw new IllegalArgumentException();
+
                 if (req == null)
                     throw new IllegalArgumentException();
 
+                this.leader = leader;
+                
                 this.req = req;
 
             }
@@ -1445,7 +1451,6 @@
                  */
                 long now = 0L;
                 UUID serviceId = null;
-                HAGlue leader = null;
                 
                 boolean didNotifyLeader = false;
                 
@@ -1461,8 +1466,10 @@
                      * this case.
                      */
 
+                    // Verify quorum valid for token (implies leader valid)
                     getQuorum().assertQuorum(token);
 
+                    // Verify this service is HAReady for token.
                     assertHAReady(token);
 
                     /*
@@ -1481,12 +1488,13 @@
                      */
                     now = newConsensusProtocolTimestamp();
 
-                    /*
-                     * If the token is invalid, making it impossible for us to
-                     * discover and message the leader, then then leader will
-                     * reset() the CyclicBarrier.
-                     */
-                    leader = quorumService.getLeader(token);
+                    // The leader is obtained by its serviceId above.
+//                    /*
+//                     * If the token is invalid, making it impossible for us to
+//                     * discover and message the leader, then the leader will
+//                     * reset() the CyclicBarrier.
+//                     */
+//                    leader = quorumService.getLeader(token);
 
                     /*
                      * Note: At this point we have everything we need to form up
@@ -1588,7 +1596,7 @@
 
                     log.error(t, t);
                     
-                    if (!didNotifyLeader && leader != null) {
+                    if (!didNotifyLeader) {
 
                         /*
                          * Send mock response to the leader so it does not block

Modified: branches/READ_CACHE2/bigdata/src/java/com/bigdata/service/AbstractHATransactionService.java
===================================================================
--- branches/READ_CACHE2/bigdata/src/java/com/bigdata/service/AbstractHATransactionService.java	2013-08-09 15:37:16 UTC (rev 7275)
+++ branches/READ_CACHE2/bigdata/src/java/com/bigdata/service/AbstractHATransactionService.java	2013-08-09 16:01:16 UTC (rev 7276)
@@ -34,6 +34,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import com.bigdata.ha.HAGlue;
 import com.bigdata.ha.HATXSGlue;
 import com.bigdata.ha.msg.IHAGatherReleaseTimeRequest;
 import com.bigdata.ha.msg.IHANotifyReleaseTimeResponse;
@@ -66,7 +67,7 @@
      *      Native thread leak in HAJournalServer process </a>
      */
     abstract public Callable<IHANotifyReleaseTimeResponse> newGatherMinimumVisibleCommitTimeTask(
-            final IHAGatherReleaseTimeRequest req);
+            final HAGlue leader, final IHAGatherReleaseTimeRequest req);
 
     /**
      * Coordinate the update of the <i>releaseTime</i> on each service that is

This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.





2010	Jan	Feb	Mar	Apr	May	Jun	Jul (139)	Aug (94)	Sep (232)	Oct (143)	Nov (138)	Dec (55)
2011	Jan (127)	Feb (90)	Mar (101)	Apr (74)	May (148)	Jun (241)	Jul (169)	Aug (121)	Sep (157)	Oct (199)	Nov (281)	Dec (75)
2012	Jan (107)	Feb (122)	Mar (184)	Apr (73)	May (14)	Jun (49)	Jul (26)	Aug (103)	Sep (133)	Oct (61)	Nov (51)	Dec (55)
2013	Jan (59)	Feb (72)	Mar (99)	Apr (62)	May (92)	Jun (19)	Jul (31)	Aug (138)	Sep (47)	Oct (83)	Nov (95)	Dec (111)
2014	Jan (125)	Feb (60)	Mar (119)	Apr (136)	May (270)	Jun (83)	Jul (88)	Aug (30)	Sep (47)	Oct (27)	Nov (23)	Dec
2015	Jan	Feb	Mar	Apr	May	Jun	Jul	Aug	Sep (3)	Oct	Nov	Dec
2016	Jan	Feb	Mar (4)	Apr (1)	May	Jun	Jul	Aug	Sep	Oct	Nov	Dec

bigdata-commit Mailing List for Blazegraph (powered by bigdata) (Page 59)

Fast, scalable, robust graph database platform

bigdata-commit — Commit traffic.