From 1cde4ac54dfb9f015d7bf95059caf2c54c3c1e4a Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Thu, 22 Dec 2011 20:23:48 +0000 Subject: [PATCH 1/6] Allow setting compaction strategy via JMX. Patch by brandonwilliams, reviewed by jbellis for CASSANDRA-3452 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-1.0@1222423 13f79535-47bb-0310-9956-ffa450edef68 --- .../org/apache/cassandra/db/ColumnFamilyStore.java | 11 +++++++++++ .../apache/cassandra/db/ColumnFamilyStoreMBean.java | 11 +++++++++++ 2 files changed, 22 insertions(+) diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 50a07b039a..90ca945b99 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -204,6 +204,17 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean } } + public void setCompactionStrategyClass(String compactionStrategyClass) throws ConfigurationException + { + metadata.compactionStrategyClass = CFMetaData.createCompactionStrategy(compactionStrategyClass); + maybeReloadCompactionStrategy(); + } + + public String getCompactionStrategyClass() + { + return metadata.compactionStrategyClass.getName(); + } + private ColumnFamilyStore(Table table, String columnFamilyName, IPartitioner partitioner, int generation, CFMetaData metadata) { assert metadata != null : "null metadata for " + table + ":" + columnFamilyName; diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java b/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java index 4b7c8f5953..b0c49a5214 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java @@ -206,6 +206,17 @@ public interface ColumnFamilyStoreMBean */ public void setMaximumCompactionThreshold(int threshold); + /** + * Sets the compaction strategy by class name + * @param className the name of the compaction strategy class + */ + public void setCompactionStrategyClass(String className) throws ConfigurationException; + + /** + * Gets the compaction strategy class name + */ + public String getCompactionStrategyClass(); + /** * Disable automatic compaction. */ From 364bad8687c9e57224b01238cf0d14cf64f511e3 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Thu, 22 Dec 2011 20:32:31 +0000 Subject: [PATCH 2/6] improve message logged for stream protocol mismatch patch by Peter Schuller; reviewed by jbellis for CASSANDRA-3652 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-1.0@1222427 13f79535-47bb-0310-9956-ffa450edef68 --- src/java/org/apache/cassandra/net/IncomingTcpConnection.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/java/org/apache/cassandra/net/IncomingTcpConnection.java b/src/java/org/apache/cassandra/net/IncomingTcpConnection.java index 8680c4329e..f6768d6213 100644 --- a/src/java/org/apache/cassandra/net/IncomingTcpConnection.java +++ b/src/java/org/apache/cassandra/net/IncomingTcpConnection.java @@ -82,8 +82,9 @@ public class IncomingTcpConnection extends Thread } else { - // streaming connections are per-session and have a fixed version. we can't do anything with a new-version stream connection, so drop it. - logger.error("Received untranslated stream from newer protocol version. Terminating connection!"); + // streaming connections are per-session and have a fixed version. we can't do anything with a wrong-version stream connection, so drop it. + logger.error("Received stream using protocol version {} (my version {}). Terminating connection", + version, MessagingService.version_); } // We are done with this connection.... return; From be53b6de31a535e2f7fb6eb210aa985b0017abf2 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Thu, 22 Dec 2011 20:39:58 +0000 Subject: [PATCH 3/6] throw ConfigurationException if chunk_length_kb would overflow patch by pschuller; reviewed by jbellis for CASSANDRA-3644 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-1.0@1222436 13f79535-47bb-0310-9956-ffa450edef68 --- .../apache/cassandra/io/compress/CompressionParameters.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/java/org/apache/cassandra/io/compress/CompressionParameters.java b/src/java/org/apache/cassandra/io/compress/CompressionParameters.java index eb4a9ec45e..2240af2c67 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressionParameters.java +++ b/src/java/org/apache/cassandra/io/compress/CompressionParameters.java @@ -150,7 +150,10 @@ public class CompressionParameters try { - return 1024 * Integer.parseInt(chLengthKB); + int parsed = Integer.parseInt(chLengthKB); + if (parsed > Integer.MAX_VALUE / 1024) + throw new ConfigurationException("Value of " + CHUNK_LENGTH_KB + " is too large (" + parsed + ")"); + return 1024 * parsed; } catch (NumberFormatException e) { From b41e645ea3c3bfb03a7cc5e138ce4f727d7a3c2e Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Thu, 22 Dec 2011 20:41:30 +0000 Subject: [PATCH 4/6] avoid logging (harmless) exception when GC takes < 1ms patch by jbellis; reviewed by brandonwilliams for CASSANDRA-3656 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-1.0@1222437 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 1 + src/java/org/apache/cassandra/service/GCInspector.java | 3 +-- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 0d153434ed..c72bf69c28 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 1.0.7 + * avoid logging (harmless) exception when GC takes < 1ms (CASSANDRA-3656) * add nodetool setstreamthroughput (CASSANDRA-3571) * fix assertion when dropping a columnfamily with no sstables (CASSANDRA-3614) * more efficient allocation of small bloom filters (CASSANDRA-3618) diff --git a/src/java/org/apache/cassandra/service/GCInspector.java b/src/java/org/apache/cassandra/service/GCInspector.java index 70fd99b440..46357ca1c4 100644 --- a/src/java/org/apache/cassandra/service/GCInspector.java +++ b/src/java/org/apache/cassandra/service/GCInspector.java @@ -99,8 +99,7 @@ public class GCInspector if (previousTotal.equals(total)) continue; gctimes.put(gc.getName(), total); - Long duration = total - previousTotal; - assert duration > 0; + Long duration = total - previousTotal; // may be zero for a really fast collection Long previousCount = gccounts.get(gc.getName()); Long count = gc.getCollectionCount(); From fb2bf7bcd3feab36440255fb3365c4970744cecc Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Thu, 22 Dec 2011 20:43:18 +0000 Subject: [PATCH 5/6] merge from 0.8 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-1.0@1222441 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGES.txt b/CHANGES.txt index c72bf69c28..f4ae790ac1 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,4 @@ 1.0.7 - * avoid logging (harmless) exception when GC takes < 1ms (CASSANDRA-3656) * add nodetool setstreamthroughput (CASSANDRA-3571) * fix assertion when dropping a columnfamily with no sstables (CASSANDRA-3614) * more efficient allocation of small bloom filters (CASSANDRA-3618) @@ -8,6 +7,7 @@ * stop thrift service in shutdown hook so we can quiesce MessagingService (CASSANDRA-3335) Merged from 0.8: + * avoid logging (harmless) exception when GC takes < 1ms (CASSANDRA-3656) * prevent new nodes from thinking down nodes are up forever (CASSANDRA-3626) * Flush non-cfs backed secondary indexes (CASSANDRA-3659) From 23e1aa4c047061da3ab32f222f2172979967fd45 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Thu, 22 Dec 2011 21:34:24 +0000 Subject: [PATCH 6/6] attempt hint delivery every ten minutes; hint handoff throttle delay default changed to 1ms, from 50 patch by jbellis; reviewed by brandonwilliams for CASSANDRA-3554 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-1.0@1222467 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 3 + conf/cassandra.yaml | 4 +- .../cassandra/db/ColumnFamilyStore.java | 1 - .../cassandra/db/HintedHandOffManager.java | 81 ++++++++++++------- .../db/HintedHandOffManagerMBean.java | 8 +- .../org/apache/cassandra/db/RowMutation.java | 2 +- .../cassandra/service/StorageProxy.java | 19 +---- .../cassandra/service/StorageService.java | 15 +--- .../service/StorageServiceMBean.java | 3 - 9 files changed, 72 insertions(+), 64 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index f4ae790ac1..f19f506214 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,7 @@ 1.0.7 + * attempt hint delivery every ten minutes, or when failure detector + notifies us that a node is back up, whichever comes first. hint + handoff throttle delay default changed to 1ms, from 50 (CASSANDRA-3554) * add nodetool setstreamthroughput (CASSANDRA-3571) * fix assertion when dropping a columnfamily with no sstables (CASSANDRA-3614) * more efficient allocation of small bloom filters (CASSANDRA-3618) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 87b0a65eb2..2082db0ca1 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -26,8 +26,8 @@ hinted_handoff_enabled: true # this defines the maximum amount of time a dead host will have hints # generated. After it has been dead this long, hints will be dropped. max_hint_window_in_ms: 3600000 # one hour -# Sleep this long after delivering each row or row fragment -hinted_handoff_throttle_delay_in_ms: 50 +# Sleep this long after delivering each hint +hinted_handoff_throttle_delay_in_ms: 1 # authentication backend, implementing IAuthenticator; used to identify users authenticator: org.apache.cassandra.auth.AllowAllAuthenticator diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 90ca945b99..1b6af6cd87 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -1313,7 +1313,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean * @return true if we found all keys we were looking for, otherwise false */ public List getRangeSlice(ByteBuffer superColumn, final AbstractBounds range, int maxResults, IFilter columnFilter) - throws ExecutionException, InterruptedException { assert range instanceof Bounds || (!((Range)range).isWrapAround() || range.right.equals(StorageService.getPartitioner().getMinimumToken())) diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java index 89c1862318..a028bdf6a4 100644 --- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java +++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java @@ -20,23 +20,26 @@ package org.apache.cassandra.db; import java.io.DataInputStream; import java.io.IOException; -import java.io.InputStream; import java.lang.management.ManagementFactory; import java.net.InetAddress; import java.net.UnknownHostException; import java.nio.ByteBuffer; import java.util.*; import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import javax.management.MBeanServer; import javax.management.ObjectName; +import com.google.common.collect.ImmutableSortedSet; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.compaction.CompactionManager; +import org.apache.cassandra.db.filter.IFilter; +import org.apache.cassandra.db.filter.NamesQueryFilter; import org.apache.cassandra.db.filter.QueryFilter; import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.dht.IPartitioner; @@ -61,7 +64,7 @@ import org.cliffc.high_scale_lib.NonBlockingHashSet; * (We have to use String keys for compatibility with OPP.) * SuperColumns in these rows are the mutations to replay, with uuid names: * - * : { // key + * : { // key * : { // supercolumn * mutation: // subcolumn * version: @@ -96,7 +99,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean private final ExecutorService executor_ = new JMXEnabledThreadPoolExecutor("HintedHandoff", Thread.MIN_PRIORITY); - public HintedHandOffManager() + public void start() { MBeanServer mbs = ManagementFactory.getPlatformMBeanServer(); try @@ -107,25 +110,23 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean { throw new RuntimeException(e); } - } - public void registerMBean() - { logger_.debug("Created HHOM instance, registered MBean."); + + Runnable runnable = new Runnable() + { + public void run() + { + scheduleAllDeliveries(); + } + }; + StorageService.optionalTasks.scheduleWithFixedDelay(runnable, 10, 10, TimeUnit.MINUTES); } - private static boolean sendMutation(InetAddress endpoint, RowMutation mutation) throws IOException + private static void sendMutation(InetAddress endpoint, RowMutation mutation) throws TimeoutException { IWriteResponseHandler responseHandler = WriteResponseHandler.create(endpoint); MessagingService.instance().sendRR(mutation, endpoint, responseHandler); - - try - { - responseHandler.get(); - } - catch (TimeoutException e) - { - return false; - } + responseHandler.get(); try { @@ -135,8 +136,6 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean { throw new AssertionError(e); } - - return true; } private static void deleteHint(ByteBuffer tokenBytes, ByteBuffer hintId, long timestamp) throws IOException @@ -226,7 +225,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean logger_.debug("schema for {} matches local schema", endpoint); return waited; } - + private void deliverHintsToEndpoint(InetAddress endpoint) throws IOException, DigestMismatchException, InvalidRequestException, TimeoutException, InterruptedException { ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(HINTS_CF); @@ -275,12 +274,12 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean while (true) { QueryFilter filter = QueryFilter.getSliceFilter(epkey, new QueryPath(HINTS_CF), startColumn, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, PAGE_SIZE); - ColumnFamily hintColumnFamily = ColumnFamilyStore.removeDeleted(hintStore.getColumnFamily(filter), Integer.MAX_VALUE); - if (pagingFinished(hintColumnFamily, startColumn)) + ColumnFamily hintsPage = ColumnFamilyStore.removeDeleted(hintStore.getColumnFamily(filter), Integer.MAX_VALUE); + if (pagingFinished(hintsPage, startColumn)) break; page: - for (IColumn hint : hintColumnFamily.getSortedColumns()) + for (IColumn hint : hintsPage.getSortedColumns()) { startColumn = hint.name(); for (IColumn subColumn : hint.getSubColumns()) @@ -305,14 +304,15 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean DataInputStream in = new DataInputStream(ByteBufferUtil.inputStream(mutationColumn.value())); RowMutation rm = RowMutation.serializer().deserialize(in, ByteBufferUtil.toInt(versionColumn.value())); - if (sendMutation(endpoint, rm)) + try { + sendMutation(endpoint, rm); deleteHint(tokenBytes, hint.name(), hint.maxTimestamp()); rowsReplayed++; } - else + catch (TimeoutException e) { - logger_.info("Could not complete hinted handoff to " + endpoint); + logger_.info(String.format("Timed out replaying hints to %s; aborting further deliveries", endpoint)); break delivery; } } @@ -335,12 +335,37 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean rowsReplayed, endpoint)); } + /** + * Attempt delivery to any node for which we have hints. Necessary since we can generate hints even for + * nodes which are never officially down/failed. + */ + private void scheduleAllDeliveries() + { + if (logger_.isDebugEnabled()) + logger_.debug("Started scheduleAllDeliveries"); + + ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(HINTS_CF); + IPartitioner p = StorageService.getPartitioner(); + Range range = new Range(p.getMinimumToken(), p.getMinimumToken(), p); + IFilter filter = new NamesQueryFilter(ImmutableSortedSet.of()); + List rows = hintStore.getRangeSlice(null, range, Integer.MAX_VALUE, filter); + for (Row row : rows) + { + Token token = StorageService.getPartitioner().getTokenFactory().fromByteArray(row.key.key); + InetAddress target = StorageService.instance.getTokenMetadata().getEndpoint(token); + scheduleHintDelivery(target); + } + + if (logger_.isDebugEnabled()) + logger_.debug("Finished scheduleAllDeliveries"); + } + /* * This method is used to deliver hints to a particular endpoint. * When we learn that some endpoint is back up we deliver the data * to him via an event driven mechanism. */ - public void deliverHints(final InetAddress to) + public void scheduleHintDelivery(final InetAddress to) { logger_.debug("deliverHints to {}", to); if (!queuedDeliveries.add(to)) @@ -356,9 +381,9 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean executor_.execute(r); } - public void deliverHints(String to) throws UnknownHostException + public void scheduleHintDelivery(String to) throws UnknownHostException { - deliverHints(InetAddress.getByName(to)); + scheduleHintDelivery(InetAddress.getByName(to)); } public List listEndpointsPendingHints() diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManagerMBean.java b/src/java/org/apache/cassandra/db/HintedHandOffManagerMBean.java index 16e914846d..9ac24e8788 100644 --- a/src/java/org/apache/cassandra/db/HintedHandOffManagerMBean.java +++ b/src/java/org/apache/cassandra/db/HintedHandOffManagerMBean.java @@ -18,6 +18,7 @@ package org.apache.cassandra.db; +import java.net.UnknownHostException; import java.util.List; import java.util.Map; @@ -25,9 +26,9 @@ public interface HintedHandOffManagerMBean { /** * Nuke all hints from this node to `ep`. - * @param epaddr String rep. of endpoint address to delete hints for, either ip address ("127.0.0.1") or hostname + * @param host String rep. of endpoint address to delete hints for, either ip address ("127.0.0.1") or hostname */ - public void deleteHintsForEndpoint(final String epaddr); + public void deleteHintsForEndpoint(final String host); /** * List all the endpoints that this node has hints for. @@ -42,5 +43,8 @@ public interface HintedHandOffManagerMBean * @return map of endpoint -> hint count */ public Map countPendingHints(); + + /** force hint delivery to an endpoint **/ + public void scheduleHintDelivery(String host) throws UnknownHostException; } diff --git a/src/java/org/apache/cassandra/db/RowMutation.java b/src/java/org/apache/cassandra/db/RowMutation.java index b7b79ba5aa..7345f722a2 100644 --- a/src/java/org/apache/cassandra/db/RowMutation.java +++ b/src/java/org/apache/cassandra/db/RowMutation.java @@ -112,7 +112,7 @@ public class RowMutation implements IMutation, MessageProducer * The format is the following: * * HintsColumnFamily: { // cf - * : { // key + * : { // key * : { // super-column * table: // columns * key: diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index b3e9894709..9be3ea71fa 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -804,21 +804,10 @@ public class StorageProxy implements StorageProxyMBean if (logger.isDebugEnabled()) logger.debug("local range slice"); ColumnFamilyStore cfs = Table.open(command.keyspace).getColumnFamilyStore(command.column_family); - try - { - rows.addAll(cfs.getRangeSlice(command.super_column, - range, - command.max_keys, - QueryFilter.getFilter(command.predicate, cfs.getComparator()))); - } - catch (ExecutionException e) - { - throw new RuntimeException(e.getCause()); - } - catch (InterruptedException e) - { - throw new AssertionError(e); - } + rows.addAll(cfs.getRangeSlice(command.super_column, + range, + command.max_keys, + QueryFilter.getFilter(command.predicate, cfs.getComparator()))); } else { diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 41bb66caed..e5bff92664 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -510,7 +510,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe MigrationManager.passiveAnnounce(Schema.instance.getVersion()); Gossiper.instance.addLocalApplicationState(ApplicationState.RELEASE_VERSION, valueFactory.releaseVersion()); - HintedHandOffManager.instance.registerMBean(); + HintedHandOffManager.instance.start(); if (DatabaseDescriptor.isAutoBootstrap() && DatabaseDescriptor.getSeeds().contains(FBUtilities.getBroadcastAddress()) @@ -1465,7 +1465,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe public void onAlive(InetAddress endpoint, EndpointState state) { if (!isClientMode && getTokenMetadata().isMember(endpoint)) - deliverHints(endpoint); + HintedHandOffManager.instance.scheduleHintDelivery(endpoint); } public void onRemove(InetAddress endpoint) @@ -1516,18 +1516,9 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe return map; } - /** - * Deliver hints to the specified node when it has crashed - * and come back up/ marked as alive after a network partition - */ - public final void deliverHints(InetAddress endpoint) - { - HintedHandOffManager.instance.deliverHints(endpoint); - } - public final void deliverHints(String host) throws UnknownHostException { - HintedHandOffManager.instance.deliverHints(host); + HintedHandOffManager.instance.scheduleHintDelivery(host); } public Token getLocalToken() diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index fd2804fa5e..656edcfd2e 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -300,9 +300,6 @@ public interface StorageServiceMBean */ public void truncate(String keyspace, String columnFamily) throws UnavailableException, TimeoutException, IOException; - /** force hint delivery to an endpoint **/ - public void deliverHints(String host) throws UnknownHostException; - /** save row and key caches */ public void saveCaches() throws ExecutionException, InterruptedException;