From 1a3a52642c6e72666beb20346093cc3cf2e374f5 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Tue, 3 Jan 2012 21:12:33 -0600 Subject: [PATCH 1/3] remove ConsistencyLevelTest.java patch by slebresne; reviewed by jbellis for CASSANDRA-3531 --- .../service/ConsistencyLevelTest.java | 193 ------------------ 1 file changed, 193 deletions(-) delete mode 100644 test/unit/org/apache/cassandra/service/ConsistencyLevelTest.java diff --git a/test/unit/org/apache/cassandra/service/ConsistencyLevelTest.java b/test/unit/org/apache/cassandra/service/ConsistencyLevelTest.java deleted file mode 100644 index af65779ad5..0000000000 --- a/test/unit/org/apache/cassandra/service/ConsistencyLevelTest.java +++ /dev/null @@ -1,193 +0,0 @@ -package org.apache.cassandra.service; -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - * - */ - - -import java.net.InetAddress; -import java.util.ArrayList; -import java.util.List; - -import org.apache.cassandra.config.Schema; -import org.junit.Test; - -import org.apache.cassandra.CleanupHelper; -import org.apache.cassandra.Util; -import org.apache.cassandra.config.ConfigurationException; -import org.apache.cassandra.config.KSMetaData; -import org.apache.cassandra.db.Row; -import org.apache.cassandra.dht.IPartitioner; -import org.apache.cassandra.dht.RandomPartitioner; -import org.apache.cassandra.dht.Token; -import org.apache.cassandra.locator.AbstractReplicationStrategy; -import org.apache.cassandra.locator.SimpleSnitch; -import org.apache.cassandra.locator.TokenMetadata; -import org.apache.cassandra.thrift.ConsistencyLevel; -import org.apache.cassandra.thrift.UnavailableException; -import org.apache.cassandra.utils.ByteBufferUtil; - -import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; - -public class ConsistencyLevelTest extends CleanupHelper -{ - @Test - public void testReadWriteConsistencyChecks() throws Exception - { - StorageService ss = StorageService.instance; - final int RING_SIZE = 3; - - TokenMetadata tmd = ss.getTokenMetadata(); - tmd.clearUnsafe(); - IPartitioner partitioner = new RandomPartitioner(); - - ss.setPartitionerUnsafe(partitioner); - - ArrayList endpointTokens = new ArrayList(); - ArrayList keyTokens = new ArrayList(); - List hostsInUse = new ArrayList(); - List hosts = new ArrayList(); - - Util.createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, RING_SIZE); - - AbstractReplicationStrategy strategy; - - for (final String table : Schema.instance.getNonSystemTables()) - { - strategy = getStrategy(table, tmd); - StorageService.calculatePendingRanges(strategy, table); - int replicationFactor = strategy.getReplicationFactor(); - if (replicationFactor < 2) - continue; - - for (ConsistencyLevel c : ConsistencyLevel.values()) - { - - if (c == ConsistencyLevel.EACH_QUORUM || c == ConsistencyLevel.LOCAL_QUORUM) - continue; - - for (int i = 0; i < replicationFactor; i++) - { - hostsInUse.clear(); - for (int j = 0 ; j < i ; j++) - { - hostsInUse.add(hosts.get(j)); - } - - if (hostsInUse.isEmpty()) - { - // We skip this case as it means RF = 0 in this simulation. - continue; - } - - IWriteResponseHandler writeHandler = strategy.getWriteResponseHandler(hostsInUse, c); - - IReadCommand command = new IReadCommand() - { - public String getKeyspace() - { - return table; - } - }; - RowRepairResolver resolver = new RowRepairResolver(table, ByteBufferUtil.bytes("foo")); - ReadCallback readHandler = StorageProxy.getReadCallback(resolver, command, c, hostsInUse); - - boolean isWriteUnavailable = false; - boolean isReadUnavailable = false; - try - { - writeHandler.assureSufficientLiveNodes(); - } - catch (UnavailableException e) - { - isWriteUnavailable = true; - } - - try - { - readHandler.assureSufficientLiveNodes(); - } - catch (UnavailableException e) - { - isReadUnavailable = true; - } - - //these should always match (in this kind of test) - assertTrue(String.format("Node Alive: %d - CL: %s - isWriteUnavailable: %b - isReadUnavailable: %b", hostsInUse.size(), c, isWriteUnavailable, isReadUnavailable), - isWriteUnavailable == isReadUnavailable); - - switch (c) - { - case ALL: - if (isWriteUnavailable) - assertTrue(hostsInUse.size() < replicationFactor); - else - assertTrue(hostsInUse.size() >= replicationFactor); - - break; - case ONE: - case ANY: - if (isWriteUnavailable) - assertTrue(hostsInUse.size() == 0); - else - assertTrue(hostsInUse.size() > 0); - break; - case TWO: - if (isWriteUnavailable) - assertTrue(hostsInUse.size() < 2); - else - assertTrue(hostsInUse.size() >= 2); - break; - case THREE: - if (isWriteUnavailable) - assertTrue(hostsInUse.size() < 3); - else - assertTrue(hostsInUse.size() >= 3); - break; - case QUORUM: - if (isWriteUnavailable) - assertTrue(hostsInUse.size() < (replicationFactor / 2 + 1)); - else - assertTrue(hostsInUse.size() >= (replicationFactor / 2 + 1)); - break; - default: - fail("Unhandled CL: " + c); - - } - } - } - return; - } - - fail("Test requires at least one table with RF > 1"); - } - - private AbstractReplicationStrategy getStrategy(String table, TokenMetadata tmd) throws ConfigurationException - { - KSMetaData ksmd = Schema.instance.getKSMetaData(table); - return AbstractReplicationStrategy.createReplicationStrategy( - table, - ksmd.strategyClass, - tmd, - new SimpleSnitch(), - ksmd.strategyOptions); - } - -} From 190fb8c8eaf8127228118ab5559e21c4d694f21e Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 28 Dec 2011 20:58:22 -0600 Subject: [PATCH 2/3] dynamic HH page size patch by jbellis; reviewed by brandonwilliams for CASSANDRA-3624 --- CHANGES.txt | 1 + .../apache/cassandra/db/HintedHandOffManager.java | 14 ++++++++++++-- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 287fed8987..7bb9f6d889 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 1.0.7 + * fix regression in HH page size calculation (CASSANDRA-3624) * retry failed stream on IOException (CASSANDRA-3686) * allow configuring bloom_filter_fp_chance (CASSANDRA-3497) * attempt hint delivery every ten minutes, or when failure detector diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java index a028bdf6a4..6661ee3d28 100644 --- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java +++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java @@ -87,7 +87,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean public static final String HINTS_CF = "HintsColumnFamily"; private static final Logger logger_ = LoggerFactory.getLogger(HintedHandOffManager.class); - private static final int PAGE_SIZE = 1024; + private static final int PAGE_SIZE = 128; private static final int LARGE_NUMBER = 65536; // 64k nodes ought to be enough for anybody. // in 0.8, subcolumns were KS-CF bytestrings, and the data was stored in the "normal" storage there. @@ -270,10 +270,20 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean int rowsReplayed = 0; ByteBuffer startColumn = ByteBufferUtil.EMPTY_BYTE_BUFFER; + int pageSize = PAGE_SIZE; + // read less columns (mutations) per page if they are very large + if (hintStore.getMeanColumns() > 0) + { + int averageColumnSize = (int) (hintStore.getMeanRowSize() / hintStore.getMeanColumns()); + pageSize = Math.min(PAGE_SIZE, DatabaseDescriptor.getInMemoryCompactionLimit() / averageColumnSize); + pageSize = Math.max(2, pageSize); // page size of 1 does not allow actual paging b/c of >= behavior on startColumn + logger_.debug("average hinted-row column size is {}; using pageSize of {}", averageColumnSize, pageSize); + } + delivery: while (true) { - QueryFilter filter = QueryFilter.getSliceFilter(epkey, new QueryPath(HINTS_CF), startColumn, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, PAGE_SIZE); + QueryFilter filter = QueryFilter.getSliceFilter(epkey, new QueryPath(HINTS_CF), startColumn, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, pageSize); ColumnFamily hintsPage = ColumnFamilyStore.removeDeleted(hintStore.getColumnFamily(filter), Integer.MAX_VALUE); if (pagingFinished(hintsPage, startColumn)) break; From de333791e81ae5364e2454af1e2fbf674bacb761 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Wed, 4 Jan 2012 10:33:57 -0600 Subject: [PATCH 3/3] Fix flawed addToMutationMap in word count example. Patch by Dave Brosius, reviewed by Brandon Williams for CASSANDRA-3669 --- .../hadoop_word_count/src/WordCountSetup.java | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/examples/hadoop_word_count/src/WordCountSetup.java b/examples/hadoop_word_count/src/WordCountSetup.java index 4e9fddccd7..7857f023d4 100644 --- a/examples/hadoop_word_count/src/WordCountSetup.java +++ b/examples/hadoop_word_count/src/WordCountSetup.java @@ -106,16 +106,24 @@ public class WordCountSetup private static void addToMutationMap(Map>> mutationMap, ByteBuffer key, String cf, Column c) { - Map> cfMutation = new HashMap>(); - List mList = new ArrayList(); + Map> cfMutation = mutationMap.get(key); + if (cfMutation == null) { + cfMutation = new HashMap>(); + mutationMap.put(key, cfMutation); + } + + List mutationList = cfMutation.get(cf); + if (mutationList == null) { + mutationList = new ArrayList(); + cfMutation.put(cf, mutationList); + } + ColumnOrSuperColumn cc = new ColumnOrSuperColumn(); Mutation m = new Mutation(); cc.setColumn(c); m.setColumn_or_supercolumn(cc); - mList.add(m); - cfMutation.put(cf, mList); - mutationMap.put(key, cfMutation); + mutationList.add(m); } private static void setupKeyspace(Cassandra.Iface client) throws TException, InvalidRequestException, SchemaDisagreementException {