natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-1.0' of https://git-wip-us.apache.org/repos/asf/cassandra into cassandra-1.0

This commit is contained in:
Sylvain Lebresne 2012-01-04 17:43:01 +01:00
parent 9f320143da de333791e8
commit 0f2121e310
4 changed files with 26 additions and 200 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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

18
examples/hadoop_word_count/src/WordCountSetup.java
View File

@ -106,16 +106,24 @@ public class WordCountSetup
private static void addToMutationMap(Map<ByteBuffer,Map<String,List<Mutation>>> mutationMap, ByteBuffer key, String cf, Column c)
{
Map<String,List<Mutation>> cfMutation = new HashMap<String,List<Mutation>>();
List<Mutation> mList = new ArrayList<Mutation>();
Map<String, List<Mutation>> cfMutation = mutationMap.get(key);
if (cfMutation == null) {
cfMutation = new HashMap<String, List<Mutation>>();
mutationMap.put(key, cfMutation);
}
List<Mutation> mutationList = cfMutation.get(cf);
if (mutationList == null) {
mutationList = new ArrayList<Mutation>();
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 {

14
src/java/org/apache/cassandra/db/HintedHandOffManager.java
View File

@ -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;

193
test/unit/org/apache/cassandra/service/ConsistencyLevelTest.java
View File

@ -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<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
List<InetAddress> hostsInUse = new ArrayList<InetAddress>();
List<InetAddress> hosts = new ArrayList<InetAddress>();
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<Row> 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);
}
}