Repository: cassandra Updated Branches: refs/heads/trunk ba56b8221 -> a1f331f17
Add new JMX methods to change local compaction strategy Patch by marcuse; reviewed by iamaleksey for CASSANDRA-9965 Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/5aca7d79 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/5aca7d79 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/5aca7d79 Branch: refs/heads/trunk Commit: 5aca7d79aaf88f9c34dcae52f24bb62a28add91e Parents: c8d163f Author: Marcus Eriksson <marc...@apache.org> Authored: Tue Aug 4 20:31:25 2015 +0200 Committer: Marcus Eriksson <marc...@apache.org> Committed: Mon Aug 10 09:02:47 2015 +0200 ---------------------------------------------------------------------- CHANGES.txt | 1 + NEWS.txt | 3 +- .../org/apache/cassandra/config/CFMetaData.java | 10 ++- .../apache/cassandra/db/ColumnFamilyStore.java | 35 +++++++++ .../cassandra/db/ColumnFamilyStoreMBean.java | 21 +++++ .../compaction/AbstractCompactionStrategy.java | 2 +- .../compaction/WrappingCompactionStrategy.java | 51 +++++++++--- .../db/compaction/CompactionsCQLTest.java | 82 +++++++++++++++++++- 8 files changed, 190 insertions(+), 15 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/5aca7d79/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 7151883..462de44 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 2.1.9 + * Add new JMX methods to change local compaction strategy (CASSANDRA-9965) * Write hints for paxos commits (CASSANDRA-7342) * (cqlsh) Fix timestamps before 1970 on Windows, always use UTC for timestamp display (CASSANDRA-10000) http://git-wip-us.apache.org/repos/asf/cassandra/blob/5aca7d79/NEWS.txt ---------------------------------------------------------------------- diff --git a/NEWS.txt b/NEWS.txt index 0b64e31..f6e2665 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -24,7 +24,8 @@ Upgrading - Commit log files are no longer recycled by default, due to negative performance implications. This can be enabled again with the commitlog_segment_recycling option in your cassandra.yaml - + - JMX methods set/getCompactionStrategyClass have been deprecated, use + set/getLocalCompactionStrategy/set/getLocalCompactionStrategyJson instead 2.1.8 ===== http://git-wip-us.apache.org/repos/asf/cassandra/blob/5aca7d79/src/java/org/apache/cassandra/config/CFMetaData.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java index 4bc5f1b..2c6a30c 100644 --- a/src/java/org/apache/cassandra/config/CFMetaData.java +++ b/src/java/org/apache/cassandra/config/CFMetaData.java @@ -1283,7 +1283,9 @@ public final class CFMetaData return strategyClass; } - public AbstractCompactionStrategy createCompactionStrategyInstance(ColumnFamilyStore cfs) + public static AbstractCompactionStrategy createCompactionStrategyInstance(Class<? extends AbstractCompactionStrategy> compactionStrategyClass, + ColumnFamilyStore cfs, + Map<String, String> compactionStrategyOptions) { try { @@ -1297,6 +1299,12 @@ public final class CFMetaData } } + @Deprecated + public AbstractCompactionStrategy createCompactionStrategyInstance(ColumnFamilyStore cfs) + { + return createCompactionStrategyInstance(compactionStrategyClass, cfs, compactionStrategyOptions); + } + // converts CFM to thrift CfDef public org.apache.cassandra.thrift.CfDef toThrift() { http://git-wip-us.apache.org/repos/asf/cassandra/blob/5aca7d79/src/java/org/apache/cassandra/db/ColumnFamilyStore.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 6777e7a..f8d796e 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -252,6 +252,41 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean }; } + public void setLocalCompactionStrategyJson(String options) + { + setLocalCompactionStrategy(FBUtilities.fromJsonMap(options)); + } + + public String getLocalCompactionStrategyJson() + { + return FBUtilities.json(getLocalCompactionStrategy()); + } + + public void setLocalCompactionStrategy(Map<String, String> options) + { + try + { + Map<String, String> optionsCopy = new HashMap<>(options); + Class<? extends AbstractCompactionStrategy> compactionStrategyClass = CFMetaData.createCompactionStrategy(optionsCopy.get("class")); + optionsCopy.remove("class"); + CFMetaData.validateCompactionOptions(compactionStrategyClass, optionsCopy); + compactionStrategyWrapper.setNewLocalCompactionStrategy(compactionStrategyClass, optionsCopy); + } + catch (Throwable t) + { + logger.error("Could not set new local compaction strategy", t); + // dont propagate the ConfigurationException over jmx, user will only see a ClassNotFoundException + throw new IllegalArgumentException("Could not set new local compaction strategy: "+t.getMessage()); + } + } + + public Map<String, String> getLocalCompactionStrategy() + { + Map<String, String> options = new HashMap<>(compactionStrategyWrapper.options); + options.put("class", compactionStrategyWrapper.getName()); + return options; + } + public void setCompactionStrategyClass(String compactionStrategyClass) { try http://git-wip-us.apache.org/repos/asf/cassandra/blob/5aca7d79/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java b/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java index 4df593b..e292be4 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java @@ -311,14 +311,35 @@ public interface ColumnFamilyStoreMBean public void setMaximumCompactionThreshold(int threshold); /** + * Sets the compaction strategy locally for this node + * + * Note that this will be set until an ALTER with compaction = {..} is executed or the node is restarted + * + * @param options compaction options with the same syntax as when doing ALTER ... WITH compaction = {..} + */ + public void setLocalCompactionStrategyJson(String options); + public String getLocalCompactionStrategyJson(); + + /** + * Sets the compaction strategy locally for this node + * + * Note that this will be set until an ALTER with compaction = {..} is executed or the node is restarted + * + * @param options compaction options map + */ + public void setLocalCompactionStrategy(Map<String, String> options); + public Map<String, String> getLocalCompactionStrategy(); + /** * Sets the compaction strategy by class name * @param className the name of the compaction strategy class */ + @Deprecated public void setCompactionStrategyClass(String className); /** * Gets the compaction strategy class name */ + @Deprecated public String getCompactionStrategyClass(); /** http://git-wip-us.apache.org/repos/asf/cassandra/blob/5aca7d79/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java index 73cda77..77ca404 100644 --- a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java @@ -60,7 +60,7 @@ public abstract class AbstractCompactionStrategy protected static final String UNCHECKED_TOMBSTONE_COMPACTION_OPTION = "unchecked_tombstone_compaction"; protected static final String COMPACTION_ENABLED = "enabled"; - protected Map<String, String> options; + public Map<String, String> options; protected final ColumnFamilyStore cfs; protected float tombstoneThreshold; http://git-wip-us.apache.org/repos/asf/cassandra/blob/5aca7d79/src/java/org/apache/cassandra/db/compaction/WrappingCompactionStrategy.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/compaction/WrappingCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/WrappingCompactionStrategy.java index 0fed733..ae67599 100644 --- a/src/java/org/apache/cassandra/db/compaction/WrappingCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/WrappingCompactionStrategy.java @@ -22,6 +22,7 @@ import java.util.Arrays; import java.util.Collection; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.Callable; @@ -47,6 +48,16 @@ public final class WrappingCompactionStrategy extends AbstractCompactionStrategy private static final Logger logger = LoggerFactory.getLogger(WrappingCompactionStrategy.class); private volatile AbstractCompactionStrategy repaired; private volatile AbstractCompactionStrategy unrepaired; + /* + We keep a copy of the schema compaction options and class here to be able to decide if we + should update the compaction strategy in maybeReloadCompactionStrategy() due to an ALTER. + + If a user changes the local compaction strategy and then later ALTERs a compaction option, + we will use the new compaction options. + */ + private Map<String, String> schemaCompactionOptions; + private Class<?> schemaCompactionStrategyClass; + public WrappingCompactionStrategy(ColumnFamilyStore cfs) { super(cfs, cfs.metadata.compactionStrategyOptions); @@ -146,10 +157,9 @@ public final class WrappingCompactionStrategy extends AbstractCompactionStrategy public synchronized void maybeReloadCompactionStrategy(CFMetaData metadata) { - if (repaired != null && repaired.getClass().equals(metadata.compactionStrategyClass) - && unrepaired != null && unrepaired.getClass().equals(metadata.compactionStrategyClass) - && repaired.options.equals(metadata.compactionStrategyOptions) - && unrepaired.options.equals(metadata.compactionStrategyOptions)) + // compare the old schema configuration to the new one, ignore any locally set changes. + if (metadata.compactionStrategyClass.equals(schemaCompactionStrategyClass) && + metadata.compactionStrategyOptions.equals(schemaCompactionOptions)) return; reloadCompactionStrategy(metadata); } @@ -157,13 +167,10 @@ public final class WrappingCompactionStrategy extends AbstractCompactionStrategy public synchronized void reloadCompactionStrategy(CFMetaData metadata) { boolean disabledWithJMX = !enabled && shouldBeEnabled(); - if (repaired != null) - repaired.shutdown(); - if (unrepaired != null) - unrepaired.shutdown(); - repaired = metadata.createCompactionStrategyInstance(cfs); - unrepaired = metadata.createCompactionStrategyInstance(cfs); - options = ImmutableMap.copyOf(metadata.compactionStrategyOptions); + setStrategy(metadata.compactionStrategyClass, metadata.compactionStrategyOptions); + schemaCompactionOptions = ImmutableMap.copyOf(metadata.compactionStrategyOptions); + schemaCompactionStrategyClass = repaired.getClass(); + if (disabledWithJMX || !shouldBeEnabled()) disable(); else @@ -393,4 +400,26 @@ public final class WrappingCompactionStrategy extends AbstractCompactionStrategy { return Arrays.asList(repaired, unrepaired); } + + public synchronized void setNewLocalCompactionStrategy(Class<? extends AbstractCompactionStrategy> compactionStrategyClass, Map<String, String> options) + { + logger.info("Switching local compaction strategy from {} to {} with options={}", repaired == null ? "null" : repaired.getClass(), compactionStrategyClass, options); + setStrategy(compactionStrategyClass, options); + if (shouldBeEnabled()) + enable(); + else + disable(); + startup(); + } + + private void setStrategy(Class<? extends AbstractCompactionStrategy> compactionStrategyClass, Map<String, String> options) + { + if (repaired != null) + repaired.shutdown(); + if (unrepaired != null) + unrepaired.shutdown(); + repaired = CFMetaData.createCompactionStrategyInstance(compactionStrategyClass, cfs, options); + unrepaired = CFMetaData.createCompactionStrategyInstance(compactionStrategyClass, cfs, options); + this.options = ImmutableMap.copyOf(options); + } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/5aca7d79/test/unit/org/apache/cassandra/db/compaction/CompactionsCQLTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionsCQLTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionsCQLTest.java index 58fc062..2798689 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionsCQLTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionsCQLTest.java @@ -17,12 +17,16 @@ */ package org.apache.cassandra.db.compaction; +import java.util.HashMap; +import java.util.Map; + import org.junit.Test; import org.apache.cassandra.cql3.CQLTester; import org.apache.cassandra.cql3.UntypedResultSet; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Keyspace; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; @@ -141,12 +145,88 @@ public class CompactionsCQLTest extends CQLTester assertTrue(minorWasTriggered(KEYSPACE, currentTable())); } + @Test + public void testSetLocalCompactionStrategy() throws Throwable + { + createTable("CREATE TABLE %s (id text PRIMARY KEY)"); + Map<String, String> localOptions = new HashMap<>(); + localOptions.put("class", "DateTieredCompactionStrategy"); + getCurrentColumnFamilyStore().setLocalCompactionStrategy(localOptions); + WrappingCompactionStrategy wrappingCompactionStrategy = (WrappingCompactionStrategy) getCurrentColumnFamilyStore().getCompactionStrategy(); + assertTrue(verifyStrategies(wrappingCompactionStrategy, DateTieredCompactionStrategy.class)); + // altering something non-compaction related + execute("ALTER TABLE %s WITH gc_grace_seconds = 1000"); + // should keep the local compaction strat + assertTrue(verifyStrategies(wrappingCompactionStrategy, DateTieredCompactionStrategy.class)); + // altering a compaction option + execute("ALTER TABLE %s WITH compaction = {'class':'SizeTieredCompactionStrategy', 'min_threshold':3}"); + // will use the new option + assertTrue(verifyStrategies(wrappingCompactionStrategy, SizeTieredCompactionStrategy.class)); + } + + + @Test + public void testSetLocalCompactionStrategyDisable() throws Throwable + { + createTable("CREATE TABLE %s (id text PRIMARY KEY)"); + Map<String, String> localOptions = new HashMap<>(); + localOptions.put("class", "DateTieredCompactionStrategy"); + localOptions.put("enabled", "false"); + getCurrentColumnFamilyStore().setLocalCompactionStrategy(localOptions); + assertFalse(getCurrentColumnFamilyStore().getCompactionStrategy().isEnabled()); + localOptions.clear(); + localOptions.put("class", "DateTieredCompactionStrategy"); + // localOptions.put("enabled", "true"); - this is default! + getCurrentColumnFamilyStore().setLocalCompactionStrategy(localOptions); + assertTrue(getCurrentColumnFamilyStore().getCompactionStrategy().isEnabled()); + } + + + @Test + public void testSetLocalCompactionStrategyEnable() throws Throwable + { + createTable("CREATE TABLE %s (id text PRIMARY KEY)"); + Map<String, String> localOptions = new HashMap<>(); + localOptions.put("class", "DateTieredCompactionStrategy"); + + getCurrentColumnFamilyStore().disableAutoCompaction(); + assertFalse(getCurrentColumnFamilyStore().getCompactionStrategy().isEnabled()); + + getCurrentColumnFamilyStore().setLocalCompactionStrategy(localOptions); + assertTrue(getCurrentColumnFamilyStore().getCompactionStrategy().isEnabled()); + + } + + + + @Test(expected = IllegalArgumentException.class) + public void testBadLocalCompactionStrategyOptions() + { + createTable("CREATE TABLE %s (id text PRIMARY KEY)"); + Map<String, String> localOptions = new HashMap<>(); + localOptions.put("class","SizeTieredCompactionStrategy"); + localOptions.put("sstable_size_in_mb","1234"); // not for STCS + getCurrentColumnFamilyStore().setLocalCompactionStrategy(localOptions); + } + + public boolean verifyStrategies(WrappingCompactionStrategy wrappingStrategy, Class<? extends AbstractCompactionStrategy> expected) + { + boolean found = false; + for (AbstractCompactionStrategy actualStrategy : wrappingStrategy.getWrappedStrategies()) + { + if (!actualStrategy.getClass().equals(expected)) + return false; + found = true; + } + return found; + } + private ColumnFamilyStore getCurrentColumnFamilyStore() { return Keyspace.open(KEYSPACE).getColumnFamilyStore(currentTable()); } - public boolean minorWasTriggered(String keyspace, String cf) throws Throwable + private boolean minorWasTriggered(String keyspace, String cf) throws Throwable { UntypedResultSet res = execute("SELECT * FROM system.compaction_history"); boolean minorWasTriggered = false;
