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/4354fb27 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/4354fb27 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/4354fb27 Branch: refs/heads/trunk Commit: 4354fb27dcb706cdb3e7cb611d3ff1b9b53f04e9 Parents: c64d3a5 37680ee Author: Yuki Morishita <[email protected]> Authored: Thu Feb 18 11:23:02 2016 -0600 Committer: Yuki Morishita <[email protected]> Committed: Thu Feb 18 11:23:02 2016 -0600 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../apache/cassandra/db/ColumnFamilyStore.java | 36 ++++++++++++++++ .../repair/RepairMessageVerbHandler.java | 43 ++++++++++++++------ 3 files changed, 68 insertions(+), 12 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/4354fb27/CHANGES.txt ---------------------------------------------------------------------- diff --cc CHANGES.txt index a79a411,53fc168..1e69cd0 --- a/CHANGES.txt +++ b/CHANGES.txt @@@ -1,18 -1,5 +1,19 @@@ -2.2.6 +3.0.4 + * Disallow drop/alter operations of UDTs used by UDAs (CASSANDRA-10721) + * Add query time validation method on Index (CASSANDRA-11043) + * Avoid potential AssertionError in mixed version cluster (CASSANDRA-11128) + * Properly handle hinted handoff after topology changes (CASSANDRA-5902) + * AssertionError when listing sstable files on inconsistent disk state (CASSANDRA-11156) + * Fix wrong rack counting and invalid conditions check for TokenAllocation + (CASSANDRA-11139) + * Avoid creating empty hint files (CASSANDRA-11090) + * Fix leak detection strong reference loop using weak reference (CASSANDRA-11120) + * Configurie BatchlogManager to stop delayed tasks on shutdown (CASSANDRA-11062) + * Hadoop integration is incompatible with Cassandra Driver 3.0.0 (CASSANDRA-11001) + * Add dropped_columns to the list of schema table so it gets handled + properly (CASSANDRA-11050) +Merged from 2.2: + * Protect from keyspace dropped during repair (CASSANDRA-11065) * Handle adding fields to a UDT in SELECT JSON and toJson() (CASSANDRA-11146) * Better error message for cleanup (CASSANDRA-10991) * cqlsh pg-style-strings broken if line ends with ';' (CASSANDRA-11123) http://git-wip-us.apache.org/repos/asf/cassandra/blob/4354fb27/src/java/org/apache/cassandra/db/ColumnFamilyStore.java ---------------------------------------------------------------------- diff --cc src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 3d7be4e,da4a84a..c564d8d --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@@ -2322,4 -3043,72 +2322,40 @@@ public class ColumnFamilyStore implemen fileIndexGenerator.set(0); } - // returns the "canonical" version of any current sstable, i.e. if an sstable is being replaced and is only partially - // visible to reads, this sstable will be returned as its original entirety, and its replacement will not be returned - // (even if it completely replaces it) - public static final Function<View, List<SSTableReader>> CANONICAL_SSTABLES = new Function<View, List<SSTableReader>>() - { - public List<SSTableReader> apply(View view) - { - List<SSTableReader> sstables = new ArrayList<>(); - for (SSTableReader sstable : view.compacting) - if (sstable.openReason != SSTableReader.OpenReason.EARLY) - sstables.add(sstable); - for (SSTableReader sstable : view.sstables) - if (!view.compacting.contains(sstable) && sstable.openReason != SSTableReader.OpenReason.EARLY) - sstables.add(sstable); - return sstables; - } - }; - - public static final Function<View, List<SSTableReader>> UNREPAIRED_SSTABLES = new Function<View, List<SSTableReader>>() - { - public List<SSTableReader> apply(View view) - { - List<SSTableReader> sstables = new ArrayList<>(); - for (SSTableReader sstable : CANONICAL_SSTABLES.apply(view)) - { - if (!sstable.isRepaired()) - sstables.add(sstable); - } - return sstables; - } - }; - + /** + * Returns a ColumnFamilyStore by cfId if it exists, null otherwise + * Differently from others, this method does not throw exception if the table does not exist. + */ + public static ColumnFamilyStore getIfExists(UUID cfId) + { + Pair<String, String> kscf = Schema.instance.getCF(cfId); + if (kscf == null) + return null; + + Keyspace keyspace = Keyspace.open(kscf.left); + if (keyspace == null) + return null; + + return keyspace.getColumnFamilyStore(cfId); + } + + /** + * Returns a ColumnFamilyStore by ksname and cfname if it exists, null otherwise + * Differently from others, this method does not throw exception if the keyspace or table does not exist. + */ + public static ColumnFamilyStore getIfExists(String ksName, String cfName) + { + if (ksName == null || cfName == null) + return null; + + Keyspace keyspace = Keyspace.open(ksName); + if (keyspace == null) + return null; + + UUID id = Schema.instance.getId(ksName, cfName); + if (id == null) + return null; + + return keyspace.getColumnFamilyStore(id); + } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/4354fb27/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java ---------------------------------------------------------------------- diff --cc src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java index 9ac859f,b8f8b65..703d187 --- a/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java +++ b/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java @@@ -26,12 -27,11 +27,9 @@@ import com.google.common.util.concurren import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import org.apache.cassandra.config.Schema; import org.apache.cassandra.db.ColumnFamilyStore; - import org.apache.cassandra.db.Keyspace; -import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.db.compaction.CompactionManager; import org.apache.cassandra.dht.Bounds; --import org.apache.cassandra.dht.LocalPartitioner; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.sstable.format.SSTableReader; @@@ -41,7 -41,7 +39,6 @@@ import org.apache.cassandra.net.Message import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.repair.messages.*; import org.apache.cassandra.service.ActiveRepairService; - import org.apache.cassandra.utils.Pair; -import org.apache.cassandra.utils.CassandraVersion; /** * Handles all repair related message. @@@ -65,10 -66,21 +62,15 @@@ public class RepairMessageVerbHandler i 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); + ColumnFamilyStore columnFamilyStore = ColumnFamilyStore.getIfExists(cfId); + if (columnFamilyStore == null) + { + logErrorAndSendFailureResponse(String.format("Table with id %s was dropped during prepare phase of repair", + cfId.toString()), message.from, id); + return; + } columnFamilyStores.add(columnFamilyStore); } - CassandraVersion peerVersion = SystemKeyspace.getReleaseVersion(message.from); - // note that we default isGlobal to true since old version always default to true: - boolean isGlobal = peerVersion == null || - peerVersion.compareTo(ActiveRepairService.SUPPORTS_GLOBAL_PREPARE_FLAG_VERSION) < 0 || - message.payload.messageType.equals(RepairMessage.Type.PREPARE_GLOBAL_MESSAGE); - logger.debug("Received prepare message: global message = {}, peerVersion = {},", message.payload.messageType.equals(RepairMessage.Type.PREPARE_GLOBAL_MESSAGE), peerVersion); ActiveRepairService.instance.registerParentRepairSession(prepareMessage.parentRepairSession, columnFamilyStores, prepareMessage.ranges, @@@ -80,8 -91,14 +82,14 @@@ case SNAPSHOT: logger.debug("Snapshotting {}", desc); - ColumnFamilyStore cfs = Keyspace.open(desc.keyspace).getColumnFamilyStore(desc.columnFamily); + final ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(desc.keyspace, desc.columnFamily); + if (cfs == null) + { + logErrorAndSendFailureResponse(String.format("Table %s.%s was dropped during snapshot phase of repair", + desc.keyspace, desc.columnFamily), message.from, id); + return; + } - final Range<Token> repairingRange = desc.range; + final Collection<Range<Token>> repairingRange = desc.ranges; Set<SSTableReader> snapshottedSSSTables = cfs.snapshot(desc.sessionId.toString(), new Predicate<SSTableReader>() { public boolean apply(SSTableReader sstable)
