cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bdeggles...@apache.org
Subject cassandra git commit: Send IR coordinator messages synchronously
Date Thu, 06 Jul 2017 17:33:28 GMT
Repository: cassandra
Updated Branches:
  refs/heads/trunk 7df240e74 -> 3234c0704


Send IR coordinator messages synchronously

Patch by Blake Eggleston; Reviewed by Marcus Eriksson for CASSANDRA-13673


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/3234c070
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/3234c070
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/3234c070

Branch: refs/heads/trunk
Commit: 3234c0704a4fef08dedc4ff78f4ded3b9226fe80
Parents: 7df240e
Author: Blake Eggleston <bdeggleston@gmail.com>
Authored: Wed Jul 5 13:20:32 2017 -0700
Committer: Blake Eggleston <bdeggleston@gmail.com>
Committed: Thu Jul 6 10:31:37 2017 -0700

----------------------------------------------------------------------
 CHANGES.txt                                     |  1 +
 .../apache/cassandra/repair/RepairRunnable.java |  3 +-
 .../repair/consistent/ConsistentSession.java    |  5 ++-
 .../repair/consistent/CoordinatorSession.java   | 35 ++++++++------------
 .../consistent/CoordinatorSessionTest.java      | 22 +++++-------
 .../consistent/CoordinatorSessionsTest.java     |  3 +-
 6 files changed, 27 insertions(+), 42 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/3234c070/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 22045e8..6cd8bc5 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 4.0
+ * Send IR coordinator messages synchronously (CASSANDRA-13673)
  * Flush system.repair table before IR finalize promise (CASSANDRA-13660)
  * Fix column filter creation for wildcard queries (CASSANDRA-13650)
  * Add 'nodetool getbatchlogreplaythrottle' and 'nodetool setbatchlogreplaythrottle' (CASSANDRA-13614)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/3234c070/src/java/org/apache/cassandra/repair/RepairRunnable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/repair/RepairRunnable.java b/src/java/org/apache/cassandra/repair/RepairRunnable.java
index 29347a4..3f761ee 100644
--- a/src/java/org/apache/cassandra/repair/RepairRunnable.java
+++ b/src/java/org/apache/cassandra/repair/RepairRunnable.java
@@ -329,8 +329,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
         CoordinatorSession coordinatorSession = ActiveRepairService.instance.consistent.coordinated.registerSession(parentSession,
allParticipants);
         ListeningExecutorService executor = createExecutor();
         AtomicBoolean hasFailure = new AtomicBoolean(false);
-        ListenableFuture repairResult = coordinatorSession.execute(executor,
-                                                                   () -> submitRepairSessions(parentSession,
true, executor, commonRanges, cfnames),
+        ListenableFuture repairResult = coordinatorSession.execute(() -> submitRepairSessions(parentSession,
true, executor, commonRanges, cfnames),
                                                                    hasFailure);
         Collection<Range<Token>> ranges = new HashSet<>();
         for (Collection<Range<Token>> range : Iterables.transform(commonRanges,
cr -> cr.right))

http://git-wip-us.apache.org/repos/asf/cassandra/blob/3234c070/src/java/org/apache/cassandra/repair/consistent/ConsistentSession.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/repair/consistent/ConsistentSession.java b/src/java/org/apache/cassandra/repair/consistent/ConsistentSession.java
index af0a0dd..803a1f8 100644
--- a/src/java/org/apache/cassandra/repair/consistent/ConsistentSession.java
+++ b/src/java/org/apache/cassandra/repair/consistent/ConsistentSession.java
@@ -25,7 +25,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
-import java.util.concurrent.Executor;
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableSet;
@@ -96,8 +95,8 @@ import org.apache.cassandra.tools.nodetool.RepairAdmin;
  *  conflicts with in progress compactions. The sstables will be marked repaired as part
of the normal compaction process.
  *  <p/>
  *
- *  On the coordinator side, see {@link CoordinatorSession#finalizePropose(Executor)}, {@link
CoordinatorSession#handleFinalizePromise(InetAddress, boolean)},
- *  & {@link CoordinatorSession#finalizeCommit(Executor)}
+ *  On the coordinator side, see {@link CoordinatorSession#finalizePropose()}, {@link CoordinatorSession#handleFinalizePromise(InetAddress,
boolean)},
+ *  & {@link CoordinatorSession#finalizeCommit()}
  *  <p/>
  *
  *  On the local session side, see {@link LocalSessions#handleFinalizeProposeMessage(InetAddress,
FinalizePropose)}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/3234c070/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java b/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java
index bac749e..830ed2c 100644
--- a/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java
+++ b/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java
@@ -22,7 +22,6 @@ import java.net.InetAddress;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
 
@@ -34,7 +33,6 @@ import com.google.common.util.concurrent.AsyncFunction;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
 import org.apache.commons.lang3.time.DurationFormatUtils;
 import org.slf4j.Logger;
@@ -139,7 +137,7 @@ public class CoordinatorSession extends ConsistentSession
         MessagingService.instance().sendOneWay(messageOut, destination);
     }
 
-    public ListenableFuture<Boolean> prepare(Executor executor)
+    public ListenableFuture<Boolean> prepare()
     {
         Preconditions.checkArgument(allStates(State.PREPARING));
 
@@ -147,7 +145,7 @@ public class CoordinatorSession extends ConsistentSession
         PrepareConsistentRequest message = new PrepareConsistentRequest(sessionID, coordinator,
participants);
         for (final InetAddress participant : participants)
         {
-            executor.execute(() -> sendMessage(participant, message));
+            sendMessage(participant, message);
         }
         return prepareFuture;
     }
@@ -181,14 +179,14 @@ public class CoordinatorSession extends ConsistentSession
         setAll(State.REPAIRING);
     }
 
-    public synchronized ListenableFuture<Boolean> finalizePropose(Executor executor)
+    public synchronized ListenableFuture<Boolean> finalizePropose()
     {
         Preconditions.checkArgument(allStates(State.REPAIRING));
         logger.debug("Proposing finalization of repair session {}", sessionID);
         FinalizePropose message = new FinalizePropose(sessionID);
         for (final InetAddress participant : participants)
         {
-            executor.execute(() -> sendMessage(participant, message));
+            sendMessage(participant, message);
         }
         return finalizeProposeFuture;
     }
@@ -217,25 +215,20 @@ public class CoordinatorSession extends ConsistentSession
         }
     }
 
-    public synchronized void finalizeCommit(Executor executor)
+    public synchronized void finalizeCommit()
     {
         Preconditions.checkArgument(allStates(State.FINALIZE_PROMISED));
         logger.debug("Committing finalization of repair session {}", sessionID);
         FinalizeCommit message = new FinalizeCommit(sessionID);
         for (final InetAddress participant : participants)
         {
-            executor.execute(() -> sendMessage(participant, message));
+            sendMessage(participant, message);
         }
         setAll(State.FINALIZED);
         logger.info("Incremental repair session {} completed", sessionID);
     }
 
-    public void fail()
-    {
-        fail(MoreExecutors.directExecutor());
-    }
-
-    public synchronized void fail(Executor executor)
+    public synchronized void fail()
     {
         logger.info("Incremental repair session {} failed", sessionID);
         FailSession message = new FailSession(sessionID);
@@ -243,7 +236,7 @@ public class CoordinatorSession extends ConsistentSession
         {
             if (participantStates.get(participant) != State.FAILED)
             {
-                executor.execute(() -> sendMessage(participant, message));
+                sendMessage(participant, message);
             }
         }
         setAll(State.FAILED);
@@ -262,12 +255,12 @@ public class CoordinatorSession extends ConsistentSession
     /**
      * Runs the asynchronous consistent repair session. Actual repair sessions are scheduled
via a submitter to make unit testing easier
      */
-    public ListenableFuture execute(Executor executor, Supplier<ListenableFuture<List<RepairSessionResult>>>
sessionSubmitter, AtomicBoolean hasFailure)
+    public ListenableFuture execute(Supplier<ListenableFuture<List<RepairSessionResult>>>
sessionSubmitter, AtomicBoolean hasFailure)
     {
         logger.info("Beginning coordination of incremental repair session {}", sessionID);
 
         sessionStart = System.currentTimeMillis();
-        ListenableFuture<Boolean> prepareResult = prepare(executor);
+        ListenableFuture<Boolean> prepareResult = prepare();
 
         // run repair sessions normally
         ListenableFuture<List<RepairSessionResult>> repairSessionResults = Futures.transform(prepareResult,
new AsyncFunction<Boolean, List<RepairSessionResult>>()
@@ -309,7 +302,7 @@ public class CoordinatorSession extends ConsistentSession
                 }
                 else
                 {
-                    return finalizePropose(executor);
+                    return finalizePropose();
                 }
             }
         });
@@ -325,7 +318,7 @@ public class CoordinatorSession extends ConsistentSession
                     {
                         logger.debug("Incremental repair {} finalization phase completed
in {}", sessionID, formatDuration(finalizeStart, System.currentTimeMillis()));
                     }
-                    finalizeCommit(executor);
+                    finalizeCommit();
                     if (logger.isDebugEnabled())
                     {
                         logger.debug("Incremental repair {} phase completed in {}", sessionID,
formatDuration(sessionStart, System.currentTimeMillis()));
@@ -334,7 +327,7 @@ public class CoordinatorSession extends ConsistentSession
                 else
                 {
                     hasFailure.set(true);
-                    fail(executor);
+                    fail();
                 }
             }
 
@@ -345,7 +338,7 @@ public class CoordinatorSession extends ConsistentSession
                     logger.debug("Incremental repair {} phase failed in {}", sessionID, formatDuration(sessionStart,
System.currentTimeMillis()));
                 }
                 hasFailure.set(true);
-                fail(executor);
+                fail();
             }
         });
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/3234c070/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionTest.java
b/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionTest.java
index 3c27b5e..fb312c3 100644
--- a/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionTest.java
+++ b/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionTest.java
@@ -24,14 +24,12 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
 
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
 
 import org.junit.Assert;
@@ -118,26 +116,26 @@ public class CoordinatorSessionTest extends AbstractRepairTest
 
         Runnable onFinalizeCommit = null;
         boolean finalizeCommitCalled = false;
-        public synchronized void finalizeCommit(Executor executor)
+        public synchronized void finalizeCommit()
         {
             finalizeCommitCalled = true;
             if (onFinalizeCommit != null)
             {
                 onFinalizeCommit.run();
             }
-            super.finalizeCommit(executor);
+            super.finalizeCommit();
         }
 
         Runnable onFail = null;
         boolean failCalled = false;
-        public synchronized void fail(Executor executor)
+        public synchronized void fail()
         {
             failCalled = true;
             if (onFail != null)
             {
                 onFail.run();
             }
-            super.fail(executor);
+            super.fail();
         }
     }
 
@@ -209,7 +207,6 @@ public class CoordinatorSessionTest extends AbstractRepairTest
     public void successCase()
     {
         InstrumentedCoordinatorSession coordinator = createInstrumentedSession();
-        Executor executor = MoreExecutors.directExecutor();
         AtomicBoolean repairSubmitted = new AtomicBoolean(false);
         SettableFuture<List<RepairSessionResult>> repairFuture = SettableFuture.create();
         Supplier<ListenableFuture<List<RepairSessionResult>>> sessionSupplier
= () ->
@@ -222,7 +219,7 @@ public class CoordinatorSessionTest extends AbstractRepairTest
         AtomicBoolean hasFailures = new AtomicBoolean(false);
         Assert.assertFalse(repairSubmitted.get());
         Assert.assertTrue(coordinator.sentMessages.isEmpty());
-        ListenableFuture sessionResult = coordinator.execute(executor, sessionSupplier, hasFailures);
+        ListenableFuture sessionResult = coordinator.execute(sessionSupplier, hasFailures);
 
         for (InetAddress participant : PARTICIPANTS)
         {
@@ -294,7 +291,6 @@ public class CoordinatorSessionTest extends AbstractRepairTest
     public void failedRepairs()
     {
         InstrumentedCoordinatorSession coordinator = createInstrumentedSession();
-        Executor executor = MoreExecutors.directExecutor();
         AtomicBoolean repairSubmitted = new AtomicBoolean(false);
         SettableFuture<List<RepairSessionResult>> repairFuture = SettableFuture.create();
         Supplier<ListenableFuture<List<RepairSessionResult>>> sessionSupplier
= () ->
@@ -307,7 +303,7 @@ public class CoordinatorSessionTest extends AbstractRepairTest
         AtomicBoolean hasFailures = new AtomicBoolean(false);
         Assert.assertFalse(repairSubmitted.get());
         Assert.assertTrue(coordinator.sentMessages.isEmpty());
-        ListenableFuture sessionResult = coordinator.execute(executor, sessionSupplier, hasFailures);
+        ListenableFuture sessionResult = coordinator.execute(sessionSupplier, hasFailures);
         for (InetAddress participant : PARTICIPANTS)
         {
             PrepareConsistentRequest expected = new PrepareConsistentRequest(coordinator.sessionID,
COORDINATOR, new HashSet<>(PARTICIPANTS));
@@ -357,7 +353,6 @@ public class CoordinatorSessionTest extends AbstractRepairTest
     public void failedPrepare()
     {
         InstrumentedCoordinatorSession coordinator = createInstrumentedSession();
-        Executor executor = MoreExecutors.directExecutor();
         AtomicBoolean repairSubmitted = new AtomicBoolean(false);
         SettableFuture<List<RepairSessionResult>> repairFuture = SettableFuture.create();
         Supplier<ListenableFuture<List<RepairSessionResult>>> sessionSupplier
= () ->
@@ -370,7 +365,7 @@ public class CoordinatorSessionTest extends AbstractRepairTest
         AtomicBoolean hasFailures = new AtomicBoolean(false);
         Assert.assertFalse(repairSubmitted.get());
         Assert.assertTrue(coordinator.sentMessages.isEmpty());
-        ListenableFuture sessionResult = coordinator.execute(executor, sessionSupplier, hasFailures);
+        ListenableFuture sessionResult = coordinator.execute(sessionSupplier, hasFailures);
         for (InetAddress participant : PARTICIPANTS)
         {
             PrepareConsistentRequest expected = new PrepareConsistentRequest(coordinator.sessionID,
COORDINATOR, new HashSet<>(PARTICIPANTS));
@@ -413,7 +408,6 @@ public class CoordinatorSessionTest extends AbstractRepairTest
     public void failedPropose()
     {
         InstrumentedCoordinatorSession coordinator = createInstrumentedSession();
-        Executor executor = MoreExecutors.directExecutor();
         AtomicBoolean repairSubmitted = new AtomicBoolean(false);
         SettableFuture<List<RepairSessionResult>> repairFuture = SettableFuture.create();
         Supplier<ListenableFuture<List<RepairSessionResult>>> sessionSupplier
= () ->
@@ -426,7 +420,7 @@ public class CoordinatorSessionTest extends AbstractRepairTest
         AtomicBoolean hasFailures = new AtomicBoolean(false);
         Assert.assertFalse(repairSubmitted.get());
         Assert.assertTrue(coordinator.sentMessages.isEmpty());
-        ListenableFuture sessionResult = coordinator.execute(executor, sessionSupplier, hasFailures);
+        ListenableFuture sessionResult = coordinator.execute(sessionSupplier, hasFailures);
 
         for (InetAddress participant : PARTICIPANTS)
         {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/3234c070/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionsTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionsTest.java
b/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionsTest.java
index b7adb27..b40e185 100644
--- a/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionsTest.java
+++ b/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionsTest.java
@@ -21,7 +21,6 @@ package org.apache.cassandra.repair.consistent;
 import java.net.InetAddress;
 import java.util.Set;
 import java.util.UUID;
-import java.util.concurrent.Executor;
 
 import com.google.common.collect.Sets;
 import org.junit.Assert;
@@ -76,7 +75,7 @@ public class CoordinatorSessionsTest extends AbstractRepairTest
         }
 
         int failCalls = 0;
-        public synchronized void fail(Executor executor)
+        public synchronized void fail()
         {
             failCalls++;
         }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org


Mime
View raw message