cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject [07/10] cassandra git commit: Merge branch 'cassandra-2.2' into cassandra-3.0
Date Wed, 22 Mar 2017 20:59:32 GMT
Merge branch 'cassandra-2.2' into cassandra-3.0


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

Branch: refs/heads/trunk
Commit: 2836a644a357c0992ba89622f04668422ce2761a
Parents: f4ba908 bf0906b
Author: Joel Knighton <jkni@apache.org>
Authored: Wed Mar 22 13:13:44 2017 -0500
Committer: Joel Knighton <jkni@apache.org>
Committed: Wed Mar 22 13:18:59 2017 -0500

----------------------------------------------------------------------
 CHANGES.txt                                     |  1 +
 .../gms/GossipDigestAckVerbHandler.java         | 26 ++++++---
 src/java/org/apache/cassandra/gms/Gossiper.java | 56 ++++++++++++++------
 .../apache/cassandra/service/MigrationTask.java | 12 ++---
 .../cassandra/service/StorageService.java       | 17 +++---
 5 files changed, 73 insertions(+), 39 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/2836a644/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 6021315,df2421d..9140c73
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,27 -1,9 +1,28 @@@
 -2.2.10
 +3.0.13
 + * Fix CONTAINS filtering for null collections (CASSANDRA-13246)
 + * Applying: Use a unique metric reservoir per test run when using Cassandra-wide metrics
residing in MBeans (CASSANDRA-13216)
 + * Propagate row deletions in 2i tables on upgrade (CASSANDRA-13320)
 + * Slice.isEmpty() returns false for some empty slices (CASSANDRA-13305)
 + * Add formatted row output to assertEmpty in CQL Tester (CASSANDRA-13238)
 +Merged from 2.2:
+  * Discard in-flight shadow round responses (CASSANDRA-12653)
   * Don't anti-compact repaired data to avoid inconsistencies (CASSANDRA-13153)
   * Wrong logger name in AnticompactionTask (CASSANDRA-13343)
 + * Commitlog replay may fail if last mutation is within 4 bytes of end of segment (CASSANDRA-13282)
   * Fix queries updating multiple time the same list (CASSANDRA-13130)
   * Fix GRANT/REVOKE when keyspace isn't specified (CASSANDRA-13053)
 +
 +
 +3.0.12
 + * Prevent data loss on upgrade 2.1 - 3.0 by adding component separator to LogRecord absolute
path (CASSANDRA-13294)
 + * Improve testing on macOS by eliminating sigar logging (CASSANDRA-13233)
 + * Cqlsh copy-from should error out when csv contains invalid data for collections (CASSANDRA-13071)
 + * Update c.yaml doc for offheap memtables (CASSANDRA-13179)
 + * Faster StreamingHistogram (CASSANDRA-13038)
 + * Legacy deserializer can create unexpected boundary range tombstones (CASSANDRA-13237)
 + * Remove unnecessary assertion from AntiCompactionTest (CASSANDRA-13070)
 + * Fix cqlsh COPY for dates before 1900 (CASSANDRA-13185)
 +Merged from 2.2:
   * Avoid race on receiver by starting streaming sender thread after sending init message
(CASSANDRA-12886)
   * Fix "multiple versions of ant detected..." when running ant test (CASSANDRA-13232)
   * Coalescing strategy sleeps too much (CASSANDRA-13090)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/2836a644/src/java/org/apache/cassandra/gms/Gossiper.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/gms/Gossiper.java
index cbfa750,c2eccba..802ff9c
--- a/src/java/org/apache/cassandra/gms/Gossiper.java
+++ b/src/java/org/apache/cassandra/gms/Gossiper.java
@@@ -124,6 -128,9 +128,8 @@@ public class Gossiper implements IFailu
      private final Map<InetAddress, Long> expireTimeEndpointMap = new ConcurrentHashMap<InetAddress,
Long>();
  
      private volatile boolean inShadowRound = false;
 -
+     // endpoint states as gathered during shadow round
+     private final Map<InetAddress, EndpointState> endpointShadowStateMap = new ConcurrentHashMap<>();
  
      private volatile long lastProcessedMessageAt = System.currentTimeMillis();
  
@@@ -818,28 -826,6 +827,20 @@@
          return endpointStateMap.get(ep);
      }
  
 +    public boolean valuesEqual(InetAddress ep1, InetAddress ep2, ApplicationState as)
 +    {
 +        EndpointState state1 = getEndpointStateForEndpoint(ep1);
 +        EndpointState state2 = getEndpointStateForEndpoint(ep2);
 +
 +        if (state1 == null || state2 == null)
 +            return false;
 +
 +        VersionedValue value1 = state1.getApplicationState(as);
 +        VersionedValue value2 = state2.getApplicationState(as);
 +
 +        return !(value1 == null || value2 == null) && value1.value.equals(value2.value);
 +    }
 +
-     // removes ALL endpoint states; should only be called after shadow gossip
-     public void resetEndpointStateMap()
-     {
-         endpointStateMap.clear();
-         unreachableEndpoints.clear();
-         liveEndpoints.clear();
-     }
- 
      public Set<Entry<InetAddress, EndpointState>> getEndpointStates()
      {
          return endpointStateMap.entrySet();
@@@ -1321,12 -1312,32 +1327,27 @@@
      }
  
      /**
-      *  Do a single 'shadow' round of gossip, where we do not modify any state
-      *  Only used when replacing a node, to get and assume its states
+      * Do a single 'shadow' round of gossip by retrieving endpoint states that will be stored
exclusively in the
+      * map return value, instead of endpointStateMap.
+      *
+      * Used when preparing to join the ring:
+      * <ul>
+      *     <li>when replacing a node, to get and assume its tokens</li>
+      *     <li>when joining, to check that the local host id matches any previous
id for the endpoint address</li>
+      * </ul>
+      *
+      * Method is synchronized, as we use an in-progress flag to indicate that shadow round
must be cleared
+      * again by calling {@link Gossiper#finishShadowRound(Map)}. This will update
+      * {@link Gossiper#endpointShadowStateMap} with received values, in order to return
an immutable copy to the
+      * caller of {@link Gossiper#doShadowRound()}. Therefor only a single shadow round execution
is permitted at
+      * the same time.
+      *
+      * @return endpoint states gathered during shadow round or empty map
       */
-     public void doShadowRound()
+     public synchronized Map<InetAddress, EndpointState> doShadowRound()
      {
          buildSeedsList();
 -        // it may be that the local address is the only entry in the seed
 -        // list in which case, attempting a shadow round is pointless
 -        if (seeds.isEmpty())
 -            return endpointShadowStateMap;
 -
+         endpointShadowStateMap.clear();
          // send a completely empty syn
          List<GossipDigest> gDigests = new ArrayList<GossipDigest>();
          GossipDigestSyn digestSynMessage = new GossipDigestSyn(DatabaseDescriptor.getClusterName(),

http://git-wip-us.apache.org/repos/asf/cassandra/blob/2836a644/src/java/org/apache/cassandra/service/MigrationTask.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/service/MigrationTask.java
index 39a5a11,b065d90..6b04756
--- a/src/java/org/apache/cassandra/service/MigrationTask.java
+++ b/src/java/org/apache/cassandra/service/MigrationTask.java
@@@ -56,13 -46,14 +56,19 @@@ class MigrationTask extends WrappedRunn
          this.endpoint = endpoint;
      }
  
 +    public static ConcurrentLinkedQueue<CountDownLatch> getInflightTasks()
 +    {
 +        return inflightTasks;
 +    }
 +
      public void runMayThrow() throws Exception
      {
+         if (!FailureDetector.instance.isAlive(endpoint))
+         {
+             logger.warn("Can't send schema pull request: node {} is down.", endpoint);
+             return;
+         }
+ 
          // There is a chance that quite some time could have passed between now and the
MM#maybeScheduleSchemaPull(),
          // potentially enough for the endpoint node to restart - which is an issue if it
does restart upgraded, with
          // a higher major.
@@@ -72,16 -63,8 +78,10 @@@
              return;
          }
  
-         if (!FailureDetector.instance.isAlive(endpoint))
-         {
-             logger.debug("Can't send schema pull request: node {} is down.", endpoint);
-             return;
-         }
- 
          MessageOut message = new MessageOut<>(MessagingService.Verb.MIGRATION_REQUEST,
null, MigrationManager.MigrationsSerializer.instance);
  
 +        final CountDownLatch completionLatch = new CountDownLatch(1);
 +
          IAsyncCallback<Collection<Mutation>> cb = new IAsyncCallback<Collection<Mutation>>()
          {
              @Override

http://git-wip-us.apache.org/repos/asf/cassandra/blob/2836a644/src/java/org/apache/cassandra/service/StorageService.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/service/StorageService.java
index 35b2423,65f536b..6760040
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@@ -481,18 -443,18 +481,17 @@@ public class StorageService extends Not
              MessagingService.instance().listen();
  
          // make magic happen
-         Gossiper.instance.doShadowRound();
- 
+         Map<InetAddress, EndpointState> epStates = Gossiper.instance.doShadowRound();
 -
          // now that we've gossiped at least once, we should be able to find the node we're
replacing
-         if (Gossiper.instance.getEndpointStateForEndpoint(DatabaseDescriptor.getReplaceAddress())==
null)
+         if (epStates.get(DatabaseDescriptor.getReplaceAddress())== null)
              throw new RuntimeException("Cannot replace_address " + DatabaseDescriptor.getReplaceAddress()
+ " because it doesn't exist in gossip");
-         replacingId = Gossiper.instance.getHostId(DatabaseDescriptor.getReplaceAddress());
+         replacingId = Gossiper.instance.getHostId(DatabaseDescriptor.getReplaceAddress(),
epStates);
          try
          {
-             VersionedValue tokensVersionedValue = Gossiper.instance.getEndpointStateForEndpoint(DatabaseDescriptor.getReplaceAddress()).getApplicationState(ApplicationState.TOKENS);
+             VersionedValue tokensVersionedValue = epStates.get(DatabaseDescriptor.getReplaceAddress()).getApplicationState(ApplicationState.TOKENS);
              if (tokensVersionedValue == null)
                  throw new RuntimeException("Could not find tokens for " + DatabaseDescriptor.getReplaceAddress()
+ " to replace");
 -            Collection<Token> tokens = TokenSerializer.deserialize(getPartitioner(),
new DataInputStream(new ByteArrayInputStream(tokensVersionedValue.toBytes())));
 +            Collection<Token> tokens = TokenSerializer.deserialize(tokenMetadata.partitioner,
new DataInputStream(new ByteArrayInputStream(tokensVersionedValue.toBytes())));
  
              if (isReplacingSameAddress())
              {


Mime
View raw message