cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From marc...@apache.org
Subject [2/2] cassandra git commit: Merge branch 'cassandra-2.1' into trunk
Date Wed, 21 Jan 2015 09:38:43 GMT
Merge branch 'cassandra-2.1' into trunk

Conflicts:
	src/java/org/apache/cassandra/db/compaction/CompactionManager.java
	src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java
	src/java/org/apache/cassandra/service/ActiveRepairService.java


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

Branch: refs/heads/trunk
Commit: 6d65d264eed638e86d19c102a2095e3e42f9deb4
Parents: b66a547 6d9d175
Author: Marcus Eriksson <marcuse@apache.org>
Authored: Wed Jan 21 10:24:02 2015 +0100
Committer: Marcus Eriksson <marcuse@apache.org>
Committed: Wed Jan 21 10:24:02 2015 +0100

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 .../db/compaction/CompactionManager.java        |  42 ++++-
 .../repair/RepairMessageVerbHandler.java        | 157 ++++++++++---------
 .../cassandra/service/ActiveRepairService.java  |  42 +++--
 .../cassandra/streaming/StreamSession.java      |   1 -
 5 files changed, 151 insertions(+), 92 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/6d65d264/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index b601bb9,ea2ecc0..6d60699
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,58 -1,5 +1,59 @@@
 +3.0
 + * Make CassandraException unchecked, extend RuntimeException (CASSANDRA-8560)
 + * Support direct buffer decompression for reads (CASSANDRA-8464)
 + * DirectByteBuffer compatible LZ4 methods (CASSANDRA-7039)
 + * Add role based access control (CASSANDRA-7653)
 + * Group sstables for anticompaction correctly (CASSANDRA-8578)
 + * Add ReadFailureException to native protocol, respond
 +   immediately when replicas encounter errors while handling
 +   a read request (CASSANDRA-7886)
 + * Switch CommitLogSegment from RandomAccessFile to nio (CASSANDRA-8308)
 + * Allow mixing token and partition key restrictions (CASSANDRA-7016)
 + * Support index key/value entries on map collections (CASSANDRA-8473)
 + * Modernize schema tables (CASSANDRA-8261)
 + * Support for user-defined aggregation functions (CASSANDRA-8053)
 + * Fix NPE in SelectStatement with empty IN values (CASSANDRA-8419)
 + * Refactor SelectStatement, return IN results in natural order instead
 +   of IN value list order (CASSANDRA-7981)
 + * Support UDTs, tuples, and collections in user-defined
 +   functions (CASSANDRA-7563)
 + * Fix aggregate fn results on empty selection, result column name,
 +   and cqlsh parsing (CASSANDRA-8229)
 + * Mark sstables as repaired after full repair (CASSANDRA-7586)
 + * Extend Descriptor to include a format value and refactor reader/writer
 +   APIs (CASSANDRA-7443)
 + * Integrate JMH for microbenchmarks (CASSANDRA-8151)
 + * Keep sstable levels when bootstrapping (CASSANDRA-7460)
 + * Add Sigar library and perform basic OS settings check on startup (CASSANDRA-7838)
 + * Support for aggregation functions (CASSANDRA-4914)
 + * Remove cassandra-cli (CASSANDRA-7920)
 + * Accept dollar quoted strings in CQL (CASSANDRA-7769)
 + * Make assassinate a first class command (CASSANDRA-7935)
 + * Support IN clause on any clustering column (CASSANDRA-4762)
 + * Improve compaction logging (CASSANDRA-7818)
 + * Remove YamlFileNetworkTopologySnitch (CASSANDRA-7917)
 + * Do anticompaction in groups (CASSANDRA-6851)
 + * Support user-defined functions (CASSANDRA-7395, 7526, 7562, 7740, 7781, 7929,
 +   7924, 7812, 8063, 7813, 7708)
 + * Permit configurable timestamps with cassandra-stress (CASSANDRA-7416)
 + * Move sstable RandomAccessReader to nio2, which allows using the
 +   FILE_SHARE_DELETE flag on Windows (CASSANDRA-4050)
 + * Remove CQL2 (CASSANDRA-5918)
 + * Add Thrift get_multi_slice call (CASSANDRA-6757)
 + * Optimize fetching multiple cells by name (CASSANDRA-6933)
 + * Allow compilation in java 8 (CASSANDRA-7028)
 + * Make incremental repair default (CASSANDRA-7250)
 + * Enable code coverage thru JaCoCo (CASSANDRA-7226)
 + * Switch external naming of 'column families' to 'tables' (CASSANDRA-4369) 
 + * Shorten SSTable path (CASSANDRA-6962)
 + * Use unsafe mutations for most unit tests (CASSANDRA-6969)
 + * Fix race condition during calculation of pending ranges (CASSANDRA-7390)
 + * Fail on very large batch sizes (CASSANDRA-8011)
 + * Improve concurrency of repair (CASSANDRA-6455, 8208)
 +
 +
  2.1.3
+  * Don't allow starting multiple inc repairs on the same sstables (CASSANDRA-8316)
   * Invalidate prepared BATCH statements when related tables
     or keyspaces are dropped (CASSANDRA-8652)
   * Fix missing results in secondary index queries on collections

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6d65d264/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index 2bba863,02f5e81..1e06f5e
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@@ -391,9 -414,9 +416,9 @@@ public class CompactionManager implemen
      public void performAnticompaction(ColumnFamilyStore cfs,
                                        Collection<Range<Token>> ranges,
                                        Collection<SSTableReader> validatedForRepair,
 -                                      long repairedAt) throws InterruptedException, ExecutionException,
IOException
 +                                      long repairedAt) throws InterruptedException, IOException
      {
-         logger.info("Starting anticompaction for {}/{}", cfs.keyspace.getName(), cfs.getColumnFamilyName());
+         logger.info("Starting anticompaction for {}.{} on {}/{} sstables", cfs.keyspace.getName(),
cfs.getColumnFamilyName(), validatedForRepair.size(), cfs.getSSTables().size());
          logger.debug("Starting anticompaction for ranges {}", ranges);
          Set<SSTableReader> sstables = new HashSet<>(validatedForRepair);
          Set<SSTableReader> mutatedRepairStatuses = new HashSet<>();
@@@ -928,21 -921,12 +953,28 @@@
              {
                  // flush first so everyone is validating data that is as similar as possible
                  StorageService.instance.forceKeyspaceFlush(cfs.keyspace.getName(), cfs.name);
 -                // we don't mark validating sstables as compacting in DataTracker, so we
have to mark them referenced
 -                // instead so they won't be cleaned up if they do get compacted during the
validation
 -                if (validator.desc.parentSessionId == null || ActiveRepairService.instance.getParentRepairSession(validator.desc.parentSessionId)
== null)
 -                    sstables = cfs.markCurrentSSTablesReferenced();
 -                else
 -                    sstables = ActiveRepairService.instance.getParentRepairSession(validator.desc.parentSessionId).getAndReferenceSSTables(cfs.metadata.cfId);
 +                ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(validator.desc.parentSessionId);
 +                Set<SSTableReader> sstablesToValidate = new HashSet<>();
 +                for (SSTableReader sstable : cfs.getSSTables())
 +                {
 +                    if (new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(Arrays.asList(validator.desc.range)))
 +                    {
 +                        if (!prs.isIncremental || !sstable.isRepaired())
 +                        {
 +                            sstablesToValidate.add(sstable);
 +                        }
 +                    }
 +                }
++                Set<SSTableReader> currentlyRepairing = ActiveRepairService.instance.currentlyRepairing(cfs.metadata.cfId,
validator.desc.parentSessionId);
++
++                if (!Sets.intersection(currentlyRepairing, sstablesToValidate).isEmpty())
++                {
++                    logger.error("Cannot start multiple repair sessions over the same sstables");
++                    throw new RuntimeException("Cannot start multiple repair sessions over
the same sstables");
++                }
 +                prs.addSSTables(cfs.metadata.cfId, sstablesToValidate);
 +
 +                sstables = prs.getAndReferenceSSTablesInRange(cfs.metadata.cfId, validator.desc.range);
  
                  if (validator.gcBefore > 0)
                      gcBefore = validator.gcBefore;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6d65d264/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java
index 1880e8e,c7cf4c8..be685cd
--- a/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java
+++ b/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java
@@@ -26,11 -24,6 +26,12 @@@ import java.util.UUID
  import java.util.concurrent.Future;
  
  import com.google.common.base.Predicate;
++import com.google.common.collect.Sets;
 +
 +import org.apache.cassandra.dht.Bounds;
 +import org.apache.cassandra.dht.Range;
 +import org.apache.cassandra.dht.Token;
 +import org.apache.cassandra.io.sstable.format.SSTableReader;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
@@@ -60,87 -57,88 +61,103 @@@ public class RepairMessageVerbHandler i
      {
          // TODO add cancel/interrupt message
          RepairJobDesc desc = message.payload.desc;
-         switch (message.payload.messageType)
+         try
          {
-             case PREPARE_MESSAGE:
-                 PrepareMessage prepareMessage = (PrepareMessage) message.payload;
-                 logger.debug("Preparing, {}", prepareMessage);
-                 List<ColumnFamilyStore> columnFamilyStores = new ArrayList<>(prepareMessage.cfIds.size());
-                 for (UUID cfId : prepareMessage.cfIds)
-                 {
-                     Pair<String, String> kscf = Schema.instance.getCF(cfId);
-                     ColumnFamilyStore columnFamilyStore = Keyspace.open(kscf.left).getColumnFamilyStore(kscf.right);
-                     columnFamilyStores.add(columnFamilyStore);
-                 }
-                 ActiveRepairService.instance.registerParentRepairSession(prepareMessage.parentRepairSession,
-                                                                          columnFamilyStores,
-                                                                          prepareMessage.ranges,
-                                                                          prepareMessage.isIncremental);
-                 MessagingService.instance().sendReply(new MessageOut(MessagingService.Verb.INTERNAL_RESPONSE),
id, message.from);
-                 break;
+             switch (message.payload.messageType)
+             {
+                 case PREPARE_MESSAGE:
+                     PrepareMessage prepareMessage = (PrepareMessage) message.payload;
++                    logger.debug("Preparing, {}", prepareMessage);
+                     List<ColumnFamilyStore> columnFamilyStores = new ArrayList<>(prepareMessage.cfIds.size());
+                     for (UUID cfId : prepareMessage.cfIds)
+                     {
+                         Pair<String, String> kscf = Schema.instance.getCF(cfId);
+                         ColumnFamilyStore columnFamilyStore = Keyspace.open(kscf.left).getColumnFamilyStore(kscf.right);
+                         columnFamilyStores.add(columnFamilyStore);
+                     }
+                     ActiveRepairService.instance.registerParentRepairSession(prepareMessage.parentRepairSession,
+                             columnFamilyStores,
 -                            prepareMessage.ranges);
++                            prepareMessage.ranges,
++                            prepareMessage.isIncremental);
+                     MessagingService.instance().sendReply(new MessageOut(MessagingService.Verb.INTERNAL_RESPONSE),
id, message.from);
+                     break;
  
-             case SNAPSHOT:
-                 logger.debug("Snapshotting {}", desc);
-                 ColumnFamilyStore cfs = Keyspace.open(desc.keyspace).getColumnFamilyStore(desc.columnFamily);
-                 final Range<Token> repairingRange = desc.range;
-                 Set<SSTableReader> snapshottedSSSTables = cfs.snapshot(desc.sessionId.toString(),
new Predicate<SSTableReader>()
-                 {
-                     public boolean apply(SSTableReader sstable)
+                 case SNAPSHOT:
++                    logger.debug("Snapshotting {}", desc);
+                     ColumnFamilyStore cfs = Keyspace.open(desc.keyspace).getColumnFamilyStore(desc.columnFamily);
+                     final Range<Token> repairingRange = desc.range;
 -                    cfs.snapshot(desc.sessionId.toString(), new Predicate<SSTableReader>()
++                    Set<SSTableReader> snapshottedSSSTables = cfs.snapshot(desc.sessionId.toString(),
new Predicate<SSTableReader>()
+                     {
+                         public boolean apply(SSTableReader sstable)
+                         {
+                             return sstable != null &&
+                                     !(sstable.partitioner instanceof LocalPartitioner) &&
// exclude SSTables from 2i
+                                     new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(Collections.singleton(repairingRange));
+                         }
+                     });
 -
++                    Set<SSTableReader> currentlyRepairing = ActiveRepairService.instance.currentlyRepairing(cfs.metadata.cfId,
desc.parentSessionId);
++                    if (!Sets.intersection(currentlyRepairing, snapshottedSSSTables).isEmpty())
 +                    {
-                         return sstable != null &&
-                                !(sstable.partitioner instanceof LocalPartitioner) &&
// exclude SSTables from 2i
-                                new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(Collections.singleton(repairingRange));
++                        logger.error("Cannot start multiple repair sessions over the same
sstables");
++                        throw new RuntimeException("Cannot start multiple repair sessions
over the same sstables");
 +                    }
-                 });
-                 ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId).addSSTables(cfs.metadata.cfId,
snapshottedSSSTables);
-                 logger.debug("Enqueuing response to snapshot request {} to {}", desc.sessionId,
message.from);
-                 MessagingService.instance().sendReply(new MessageOut(MessagingService.Verb.INTERNAL_RESPONSE),
id, message.from);
-                 break;
++                    ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId).addSSTables(cfs.metadata.cfId,
snapshottedSSSTables);
+                     logger.debug("Enqueuing response to snapshot request {} to {}", desc.sessionId,
message.from);
+                     MessagingService.instance().sendReply(new MessageOut(MessagingService.Verb.INTERNAL_RESPONSE),
id, message.from);
+                     break;
  
-             case VALIDATION_REQUEST:
-                 ValidationRequest validationRequest = (ValidationRequest) message.payload;
-                 logger.debug("Validating {}", validationRequest);
-                 // trigger read-only compaction
-                 ColumnFamilyStore store = Keyspace.open(desc.keyspace).getColumnFamilyStore(desc.columnFamily);
+                 case VALIDATION_REQUEST:
+                     ValidationRequest validationRequest = (ValidationRequest) message.payload;
++                    logger.debug("Validating {}", validationRequest);
+                     // trigger read-only compaction
+                     ColumnFamilyStore store = Keyspace.open(desc.keyspace).getColumnFamilyStore(desc.columnFamily);
  
-                 Validator validator = new Validator(desc, message.from, validationRequest.gcBefore);
-                 CompactionManager.instance.submitValidation(store, validator);
-                 break;
+                     Validator validator = new Validator(desc, message.from, validationRequest.gcBefore);
+                     CompactionManager.instance.submitValidation(store, validator);
+                     break;
  
-             case SYNC_REQUEST:
-                 // forwarded sync request
-                 SyncRequest request = (SyncRequest) message.payload;
-                 logger.debug("Syncing {}", request);
-                 long repairedAt = ActiveRepairService.UNREPAIRED_SSTABLE;
-                 if (desc.parentSessionId != null && ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId)
!= null)
-                     repairedAt = ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId).repairedAt;
+                 case SYNC_REQUEST:
+                     // forwarded sync request
+                     SyncRequest request = (SyncRequest) message.payload;
 -                    StreamingRepairTask task = new StreamingRepairTask(desc, request);
++                    logger.debug("Syncing {}", request);
++                    long repairedAt = ActiveRepairService.UNREPAIRED_SSTABLE;
++                    if (desc.parentSessionId != null && ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId)
!= null)
++                        repairedAt = ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId).repairedAt;
 +
-                 StreamingRepairTask task = new StreamingRepairTask(desc, request, repairedAt);
-                 task.run();
-                 break;
++                    StreamingRepairTask task = new StreamingRepairTask(desc, request, repairedAt);
+                     task.run();
+                     break;
  
-             case ANTICOMPACTION_REQUEST:
-                 AnticompactionRequest anticompactionRequest = (AnticompactionRequest) message.payload;
-                 logger.debug("Got anticompaction request {}", anticompactionRequest);
-                 try
-                 {
-                     List<Future<?>> futures = ActiveRepairService.instance.doAntiCompaction(anticompactionRequest.parentRepairSession,
anticompactionRequest.successfulRanges);
-                     FBUtilities.waitOnFutures(futures);
-                 }
-                 catch (Exception e)
-                 {
-                     throw new RuntimeException(e);
-                 }
-                 finally
-                 {
-                     ActiveRepairService.instance.removeParentRepairSession(anticompactionRequest.parentRepairSession);
-                 }
+                 case ANTICOMPACTION_REQUEST:
 -                    logger.debug("Got anticompaction request");
+                     AnticompactionRequest anticompactionRequest = (AnticompactionRequest)
message.payload;
++                    logger.debug("Got anticompaction request {}", anticompactionRequest);
+                     try
+                     {
 -                        List<Future<?>> futures = ActiveRepairService.instance.doAntiCompaction(anticompactionRequest.parentRepairSession);
++                        List<Future<?>> futures = ActiveRepairService.instance.doAntiCompaction(anticompactionRequest.parentRepairSession,
anticompactionRequest.successfulRanges);
+                         FBUtilities.waitOnFutures(futures);
+                     }
+                     catch (Exception e)
+                     {
+                         throw new RuntimeException(e);
+                     }
+                     finally
+                     {
+                         ActiveRepairService.instance.removeParentRepairSession(anticompactionRequest.parentRepairSession);
+                     }
  
-                 break;
+                     break;
  
-             default:
-                 ActiveRepairService.instance.handleMessage(message.from, message.payload);
-                 break;
+                 default:
+                     ActiveRepairService.instance.handleMessage(message.from, message.payload);
+                     break;
+             }
+         }
+         catch (Exception e)
+         {
+             logger.error("Got error, removing parent repair session");
 -            if (desc!=null && desc.parentSessionId != null)
++            if (desc != null && desc.parentSessionId != null)
+                 ActiveRepairService.instance.removeParentRepairSession(desc.parentSessionId);
+             throw new RuntimeException(e);
          }
      }
  }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6d65d264/src/java/org/apache/cassandra/service/ActiveRepairService.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/service/ActiveRepairService.java
index e4b7fff,36f7c5c..35ddeef
--- a/src/java/org/apache/cassandra/service/ActiveRepairService.java
+++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java
@@@ -103,48 -116,34 +103,48 @@@ public class ActiveRepairServic
       *
       * @return Future for asynchronous call or null if there is no need to repair
       */
 -    public RepairFuture submitRepairSession(UUID parentRepairSession, Range<Token>
range, String keyspace, RepairParallelism parallelismDegree, Set<InetAddress> endpoints,
String... cfnames)
 +    public RepairSession submitRepairSession(UUID parentRepairSession,
 +                                             Range<Token> range,
 +                                             String keyspace,
 +                                             RepairParallelism parallelismDegree,
 +                                             Set<InetAddress> endpoints,
 +                                             long repairedAt,
 +                                             ListeningExecutorService executor,
 +                                             String... cfnames)
      {
 -        RepairSession session = new RepairSession(parentRepairSession, range, keyspace,
parallelismDegree, endpoints, cfnames);
 -        if (session.endpoints.isEmpty())
 +        if (endpoints.isEmpty())
              return null;
 -        RepairFuture futureTask = new RepairFuture(session);
 -        executor.execute(futureTask);
 -        return futureTask;
 -    }
  
 -    public void addToActiveSessions(RepairSession session)
 -    {
 +        final RepairSession session = new RepairSession(parentRepairSession, UUIDGen.getTimeUUID(),
range, keyspace, parallelismDegree, endpoints, repairedAt, cfnames);
 +
          sessions.put(session.getId(), session);
 -        Gossiper.instance.register(session);
 -        FailureDetector.instance.registerFailureDetectionEventListener(session);
 -    }
 +        // register listeners
 +        gossiper.register(session);
 +        failureDetector.registerFailureDetectionEventListener(session);
  
 -    public void removeFromActiveSessions(RepairSession session)
 -    {
 -        Gossiper.instance.unregister(session);
 -        sessions.remove(session.getId());
 +        // unregister listeners at completion
 +        session.addListener(new Runnable()
 +        {
 +            /**
 +             * When repair finished, do clean up
 +             */
 +            public void run()
 +            {
 +                failureDetector.unregisterFailureDetectionEventListener(session);
 +                gossiper.unregister(session);
 +                sessions.remove(session.getId());
 +            }
 +        }, MoreExecutors.sameThreadExecutor());
 +        session.start(executor);
 +        return session;
      }
  
-     public void terminateSessions()
+     public synchronized void terminateSessions()
      {
 +        Throwable cause = new IOException("Terminate session is called");
          for (RepairSession session : sessions.values())
          {
 -            session.forceShutdown();
 +            session.forceShutdown(cause);
          }
          parentRepairSessions.clear();
      }
@@@ -229,10 -241,10 +229,10 @@@
          return neighbors;
      }
  
-     public UUID prepareForRepair(Set<InetAddress> endpoints, RepairOption options,
List<ColumnFamilyStore> columnFamilyStores)
 -    public synchronized UUID prepareForRepair(Set<InetAddress> endpoints, Collection<Range<Token>>
ranges, List<ColumnFamilyStore> columnFamilyStores)
++    public synchronized UUID prepareForRepair(Set<InetAddress> endpoints, RepairOption
options, List<ColumnFamilyStore> columnFamilyStores)
      {
          UUID parentRepairSession = UUIDGen.getTimeUUID();
 -        registerParentRepairSession(parentRepairSession, columnFamilyStores, ranges);
 +        registerParentRepairSession(parentRepairSession, columnFamilyStores, options.getRanges(),
options.isIncremental());
          final CountDownLatch prepareLatch = new CountDownLatch(endpoints.size());
          final AtomicBoolean status = new AtomicBoolean(true);
          final Set<String> failedNodes = Collections.synchronizedSet(new HashSet<String>());
@@@ -285,12 -297,46 +285,24 @@@
          return parentRepairSession;
      }
  
 -    public synchronized void registerParentRepairSession(UUID parentRepairSession, List<ColumnFamilyStore>
columnFamilyStores, Collection<Range<Token>> ranges)
 +    public void registerParentRepairSession(UUID parentRepairSession, List<ColumnFamilyStore>
columnFamilyStores, Collection<Range<Token>> ranges, boolean isIncremental)
      {
 -        Map<UUID, Set<SSTableReader>> sstablesToRepair = new HashMap<>();
 -        for (ColumnFamilyStore cfs : columnFamilyStores)
 -        {
 -            Set<SSTableReader> sstables = new HashSet<>();
 -            Set<SSTableReader> currentlyRepairing = currentlyRepairing(cfs.metadata.cfId);
 -            for (SSTableReader sstable : cfs.getSSTables())
 -            {
 -                if (new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(ranges))
 -                {
 -                    if (!sstable.isRepaired())
 -                    {
 -                        if (currentlyRepairing.contains(sstable))
 -                        {
 -                            logger.error("Already repairing "+sstable+", can not continue.");
 -                            throw new RuntimeException("Already repairing "+sstable+", can
not continue.");
 -                        }
 -                        sstables.add(sstable);
 -                    }
 -                }
 -            }
 -            sstablesToRepair.put(cfs.metadata.cfId, sstables);
 -        }
 -        parentRepairSessions.put(parentRepairSession, new ParentRepairSession(columnFamilyStores,
ranges, sstablesToRepair, System.currentTimeMillis()));
 +        parentRepairSessions.put(parentRepairSession, new ParentRepairSession(columnFamilyStores,
ranges, isIncremental, System.currentTimeMillis()));
      }
  
 -    private Set<SSTableReader> currentlyRepairing(UUID cfId)
++    public Set<SSTableReader> currentlyRepairing(UUID cfId, UUID parentRepairSession)
+     {
+         Set<SSTableReader> repairing = new HashSet<>();
+         for (Map.Entry<UUID, ParentRepairSession> entry : parentRepairSessions.entrySet())
+         {
+             Collection<SSTableReader> sstables = entry.getValue().sstableMap.get(cfId);
 -            if (sstables != null)
++            if (sstables != null && !entry.getKey().equals(parentRepairSession))
+                 repairing.addAll(sstables);
+         }
+         return repairing;
+     }
+ 
 -    public synchronized void finishParentSession(UUID parentSession, Set<InetAddress>
neighbors, boolean doAntiCompaction) throws InterruptedException, ExecutionException, IOException
 +    public void finishParentSession(UUID parentSession, Set<InetAddress> neighbors,
Collection<Range<Token>> successfulRanges)
      {
          try
          {
@@@ -323,29 -372,13 +335,17 @@@
      {
          assert parentRepairSession != null;
          ParentRepairSession prs = getParentRepairSession(parentRepairSession);
 +        assert prs.ranges.containsAll(successfulRanges) : "Trying to perform anticompaction
on unknown ranges";
  
          List<Future<?>> futures = new ArrayList<>();
 +        // if we don't have successful repair ranges, then just skip anticompaction
 +        if (successfulRanges.isEmpty())
 +            return futures;
          for (Map.Entry<UUID, ColumnFamilyStore> columnFamilyStoreEntry : prs.columnFamilyStores.entrySet())
          {
- 
              Collection<SSTableReader> sstables = new HashSet<>(prs.getAndReferenceSSTables(columnFamilyStoreEntry.getKey()));
              ColumnFamilyStore cfs = columnFamilyStoreEntry.getValue();
-             boolean success = false;
-             while (!success)
-             {
-                 for (SSTableReader compactingSSTable : cfs.getDataTracker().getCompacting())
-                 {
-                     if (sstables.remove(compactingSSTable))
-                         SSTableReader.releaseReferences(Arrays.asList(compactingSSTable));
-                 }
-                 success = sstables.isEmpty() || cfs.getDataTracker().markCompacting(sstables);
-             }
- 
 -            futures.add(CompactionManager.instance.submitAntiCompaction(cfs, prs.ranges,
sstables, prs.repairedAt));
 +            futures.add(CompactionManager.instance.submitAntiCompaction(cfs, successfulRanges,
sstables, prs.repairedAt));
          }
  
          return futures;
@@@ -419,19 -442,15 +419,29 @@@
              return sstables;
          }
  
 +        public synchronized Set<SSTableReader> getAndReferenceSSTablesInRange(UUID
cfId, Range<Token> range)
 +        {
-             Collection<SSTableReader> allSSTables= getAndReferenceSSTables(cfId);
++            Collection<SSTableReader> allSSTables = getAndReferenceSSTables(cfId);
 +            Set<SSTableReader> sstables = new HashSet<>();
 +            for (SSTableReader sstable : allSSTables)
 +            {
 +                if (new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(Arrays.asList(range)))
 +                    sstables.add(sstable);
 +                else
 +                    sstable.releaseReference();
 +            }
 +            return sstables;
++        }
 +
+         @Override
+         public String toString()
+         {
+             return "ParentRepairSession{" +
+                     "columnFamilyStores=" + columnFamilyStores +
+                     ", ranges=" + ranges +
+                     ", sstableMap=" + sstableMap +
+                     ", repairedAt=" + repairedAt +
+                     '}';
          }
      }
  }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6d65d264/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------


Mime
View raw message