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

Merge branch 'cassandra-3.5' into trunk

This commit is contained in:
Josh McKenzie 2016-04-01 11:54:51 -04:00
parent bdae4520c5 af9b9cd5ae
commit 6d1254ac23
7 changed files with 282 additions and 127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -144,6 +144,7 @@ Merged from 2.2:
* (cqlsh) Support timezone conversion using pytz (CASSANDRA-10397)
* cqlsh: change default encoding to UTF-8 (CASSANDRA-11124)
Merged from 2.1:
* Fix out-of-space error treatment in memtable flushing (CASSANDRA-11448).
* Don't do defragmentation if reading from repaired sstables (CASSANDRA-10342)
* Fix streaming_socket_timeout_in_ms not enforced (CASSANDRA-11286)
* Avoid dropping message too quickly due to missing unit conversion (CASSANDRA-11302)

146
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -921,6 +921,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
final OpOrder.Barrier writeBarrier;
final CountDownLatch latch = new CountDownLatch(1);
final ReplayPosition lastReplayPosition;
volatile Throwable flushFailure = null;
private PostFlush(boolean flushSecondaryIndexes, OpOrder.Barrier writeBarrier, ReplayPosition lastReplayPosition)
{
@ -956,12 +957,16 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
// must check lastReplayPosition != null because Flush may find that all memtables are clean
// and so not set a lastReplayPosition
if (lastReplayPosition != null)
// If a flush errored out but the error was ignored, make sure we don't discard the commit log.
if (lastReplayPosition != null && flushFailure == null)
{
CommitLog.instance.discardCompletedSegments(metadata.cfId, lastReplayPosition);
}
metric.pendingFlushes.dec();
if (flushFailure != null)
throw Throwables.propagate(flushFailure);
}
}
@ -1061,84 +1066,109 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
metric.memtableSwitchCount.inc();
for (Memtable memtable : memtables)
try
{
List<Future<SSTableMultiWriter>> futures = new ArrayList<>();
long totalBytesOnDisk = 0;
long maxBytesOnDisk = 0;
long minBytesOnDisk = Long.MAX_VALUE;
List<SSTableReader> sstables = new ArrayList<>();
try (LifecycleTransaction txn = LifecycleTransaction.offline(OperationType.FLUSH))
for (Memtable memtable : memtables)
{
flushMemtable(memtable);
}
}
catch (Throwable t)
{
JVMStabilityInspector.inspectThrowable(t);
postFlush.flushFailure = t;
}
// signal the post-flush we've done our work
postFlush.latch.countDown();
}
public void flushMemtable(Memtable memtable)
{
List<Future<SSTableMultiWriter>> futures = new ArrayList<>();
long totalBytesOnDisk = 0;
long maxBytesOnDisk = 0;
long minBytesOnDisk = Long.MAX_VALUE;
List<SSTableReader> sstables = new ArrayList<>();
try (LifecycleTransaction txn = LifecycleTransaction.offline(OperationType.FLUSH))
{
List<Memtable.FlushRunnable> flushRunnables = null;
List<SSTableMultiWriter> flushResults = null;
try
{
// flush the memtable
List<Memtable.FlushRunnable> flushRunnables = memtable.flushRunnables(txn);
flushRunnables = memtable.flushRunnables(txn);
for (int i = 0; i < flushRunnables.size(); i++)
futures.add(perDiskflushExecutors[i].submit(flushRunnables.get(i)));
List<SSTableMultiWriter> flushResults = Lists.newArrayList(FBUtilities.waitOnFutures(futures));
flushResults = Lists.newArrayList(FBUtilities.waitOnFutures(futures));
}
catch (Throwable t)
{
t = memtable.abortRunnables(flushRunnables, t);
t = txn.abort(t);
throw Throwables.propagate(t);
}
try
try
{
Iterator<SSTableMultiWriter> writerIterator = flushResults.iterator();
while (writerIterator.hasNext())
{
Iterator<SSTableMultiWriter> writerIterator = flushResults.iterator();
while (writerIterator.hasNext())
@SuppressWarnings("resource")
SSTableMultiWriter writer = writerIterator.next();
if (writer.getFilePointer() > 0)
{
@SuppressWarnings("resource")
SSTableMultiWriter writer = writerIterator.next();
if (writer.getFilePointer() > 0)
{
writer.setOpenResult(true).prepareToCommit();
}
else
{
maybeFail(writer.abort(null));
writerIterator.remove();
}
writer.setOpenResult(true).prepareToCommit();
}
}
catch (Throwable t)
{
for (SSTableMultiWriter writer : flushResults)
t = writer.abort(t);
t = txn.abort(t);
Throwables.propagate(t);
}
txn.prepareToCommit();
Throwable accumulate = null;
for (SSTableMultiWriter writer : flushResults)
accumulate = writer.commit(accumulate);
maybeFail(txn.commit(accumulate));
for (SSTableMultiWriter writer : flushResults)
{
Collection<SSTableReader> flushedSSTables = writer.finished();
for (SSTableReader sstable : flushedSSTables)
else
{
if (sstable != null)
{
sstables.add(sstable);
long size = sstable.bytesOnDisk();
totalBytesOnDisk += size;
maxBytesOnDisk = Math.max(maxBytesOnDisk, size);
minBytesOnDisk = Math.min(minBytesOnDisk, size);
}
maybeFail(writer.abort(null));
writerIterator.remove();
}
}
}
memtable.cfs.replaceFlushed(memtable, sstables);
reclaim(memtable);
catch (Throwable t)
{
for (SSTableMultiWriter writer : flushResults)
t = writer.abort(t);
t = txn.abort(t);
Throwables.propagate(t);
}
txn.prepareToCommit();
Throwable accumulate = null;
for (SSTableMultiWriter writer : flushResults)
accumulate = writer.commit(accumulate);
maybeFail(txn.commit(accumulate));
for (SSTableMultiWriter writer : flushResults)
{
Collection<SSTableReader> flushedSSTables = writer.finished();
for (SSTableReader sstable : flushedSSTables)
{
if (sstable != null)
{
sstables.add(sstable);
long size = sstable.bytesOnDisk();
totalBytesOnDisk += size;
maxBytesOnDisk = Math.max(maxBytesOnDisk, size);
minBytesOnDisk = Math.min(minBytesOnDisk, size);
}
}
}
}
memtable.cfs.replaceFlushed(memtable, sstables);
reclaim(memtable);
logger.debug("Flushed to {} ({} sstables, {}), biggest {}, smallest {}",
sstables,
sstables.size(),
FBUtilities.prettyPrintMemory(totalBytesOnDisk),
FBUtilities.prettyPrintMemory(maxBytesOnDisk),
FBUtilities.prettyPrintMemory(minBytesOnDisk));
}
// signal the post-flush we've done our work
postFlush.latch.countDown();
}
private void reclaim(final Memtable memtable)

16
src/java/org/apache/cassandra/db/Directories.java
View File

@ -320,7 +320,7 @@ public class Directories
* which may return any non-blacklisted directory - even a data directory that has no usable space.
* Do not use this method in production code.
*
* @throws IOError if all directories are blacklisted.
* @throws FSWriteError if all directories are blacklisted.
*/
public File getDirectoryForNewSSTables()
{
@ -330,11 +330,14 @@ public class Directories
/**
* Returns a non-blacklisted data directory that _currently_ has {@code writeSize} bytes as usable space.
*
* @throws IOError if all directories are blacklisted.
* @throws FSWriteError if all directories are blacklisted.
*/
public File getWriteableLocationAsFile(long writeSize)
{
return getLocationForDisk(getWriteableLocation(writeSize));
File location = getLocationForDisk(getWriteableLocation(writeSize));
if (location == null)
throw new FSWriteError(new IOException("No configured data directory contains enough space to write " + writeSize + " bytes"), "");
return location;
}
/**
@ -366,9 +369,10 @@ public class Directories
}
/**
* Returns a non-blacklisted data directory that _currently_ has {@code writeSize} bytes as usable space.
* Returns a non-blacklisted data directory that _currently_ has {@code writeSize} bytes as usable space, null if
* there is not enough space left in all directories.
*
* @throws IOError if all directories are blacklisted.
* @throws FSWriteError if all directories are blacklisted.
*/
public DataDirectory getWriteableLocation(long writeSize)
{
@ -401,7 +405,7 @@ public class Directories
if (tooBig)
return null;
else
throw new IOError(new IOException("All configured data directories have been blacklisted as unwritable for erroring out"));
throw new FSWriteError(new IOException("All configured data directories have been blacklisted as unwritable for erroring out"), "");
// shortcut for single data directory systems
if (candidates.size() == 1)

29
src/java/org/apache/cassandra/db/Memtable.java
View File

@ -24,6 +24,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Throwables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -44,6 +45,7 @@ import org.apache.cassandra.db.rows.UnfilteredRowIterators;
import org.apache.cassandra.dht.*;
import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
import org.apache.cassandra.index.transactions.UpdateTransaction;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.SSTableMultiWriter;
import org.apache.cassandra.io.sstable.SSTableTxnWriter;
@ -274,13 +276,28 @@ public class Memtable implements Comparable<Memtable>
List<FlushRunnable> runnables = new ArrayList<>(boundaries.size());
PartitionPosition rangeStart = cfs.getPartitioner().getMinimumToken().minKeyBound();
ReplayPosition context = lastReplayPosition.get();
for (int i = 0; i < boundaries.size(); i++)
try
{
PartitionPosition t = boundaries.get(i);
runnables.add(new FlushRunnable(context, rangeStart, t, locations[i], txn));
rangeStart = t;
for (int i = 0; i < boundaries.size(); i++)
{
PartitionPosition t = boundaries.get(i);
runnables.add(new FlushRunnable(context, rangeStart, t, locations[i], txn));
rangeStart = t;
}
return runnables;
}
return runnables;
catch (Throwable e)
{
throw Throwables.propagate(abortRunnables(runnables, e));
}
}
public Throwable abortRunnables(List<FlushRunnable> runnables, Throwable t)
{
if (runnables != null)
for (FlushRunnable runnable : runnables)
t = runnable.writer.abort(t);
return t;
}
public String toString()
@ -389,7 +406,7 @@ public class Memtable implements Comparable<Memtable>
this.isBatchLogTable = cfs.name.equals(SystemKeyspace.BATCHES) && cfs.keyspace.getName().equals(SystemKeyspace.NAME);
if (flushLocation == null)
writer = createFlushWriter(txn, cfs.getSSTablePath(getDirectories().getLocationForDisk(getDirectories().getWriteableLocation(estimatedSize))), columnsCollector.get(), statsCollector.get());
writer = createFlushWriter(txn, cfs.getSSTablePath(getDirectories().getWriteableLocationAsFile(estimatedSize)), columnsCollector.get(), statsCollector.get());
else
writer = createFlushWriter(txn, cfs.getSSTablePath(getDirectories().getLocationForDisk(flushLocation)), columnsCollector.get(), statsCollector.get());

4
src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManager.java
View File

@ -34,6 +34,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Iterables;
import com.google.common.util.concurrent.*;
import org.slf4j.Logger;
@ -555,7 +556,8 @@ public class CommitLogSegmentManager
/**
* @return a read-only collection of the active commit log segments
*/
Collection<CommitLogSegment> getActiveSegments()
@VisibleForTesting
public Collection<CommitLogSegment> getActiveSegments()
{
return Collections.unmodifiableCollection(activeSegments);
}

56
src/java/org/apache/cassandra/io/util/DiskAwareRunnable.java
View File

@ -1,56 +0,0 @@
/*
* 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.
*/
package org.apache.cassandra.io.util;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.WrappedRunnable;
public abstract class DiskAwareRunnable extends WrappedRunnable
{
protected Directories.DataDirectory getWriteDirectory(long writeSize)
{
Directories.DataDirectory directory;
directory = getDirectory();
if (directory == null) // ok panic - write anywhere
directory = getDirectories().getWriteableLocation(writeSize);
if (directory == null)
throw new RuntimeException(String.format("Insufficient disk space to write %s",
FBUtilities.prettyPrintMemory(writeSize)));
return directory;
}
/**
* Get sstable directories for the CF.
* @return Directories instance for the CF.
*/
protected abstract Directories getDirectories();
protected abstract Directories.DataDirectory getDirectory();
/**
* Called if no disk is available with free space for the full write size.
* @return true if the scope of the task was successfully reduced.
*/
public boolean reduceScopeForLimitedSpace()
{
return false;
}
}

157
test/unit/org/apache/cassandra/cql3/OutOfSpaceTest.java Normal file
View File

@ -0,0 +1,157 @@
/*
* 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.
*/
package org.apache.cassandra.cql3;
import static junit.framework.Assert.fail;
import java.io.IOError;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import org.junit.Assert;
import org.junit.Test;
import org.apache.cassandra.config.Config.DiskFailurePolicy;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.BlacklistedDirectories;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Directories.DataDirectory;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.commitlog.CommitLogSegment;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.KillerForTests;
/**
* Test that TombstoneOverwhelmingException gets thrown when it should be and doesn't when it shouldn't be.
*/
public class OutOfSpaceTest extends CQLTester
{
@Test
public void testFlushUnwriteableDie() throws Throwable
{
makeTable();
markDirectoriesUnwriteable();
KillerForTests killerForTests = new KillerForTests();
JVMStabilityInspector.Killer originalKiller = JVMStabilityInspector.replaceKiller(killerForTests);
DiskFailurePolicy oldPolicy = DatabaseDescriptor.getDiskFailurePolicy();
try
{
DatabaseDescriptor.setDiskFailurePolicy(DiskFailurePolicy.die);
flushAndExpectError();
Assert.assertTrue(killerForTests.wasKilled());
Assert.assertFalse(killerForTests.wasKilledQuietly()); //only killed quietly on startup failure
}
finally
{
DatabaseDescriptor.setDiskFailurePolicy(oldPolicy);
JVMStabilityInspector.replaceKiller(originalKiller);
}
}
@Test
public void testFlushUnwriteableStop() throws Throwable
{
makeTable();
markDirectoriesUnwriteable();
DiskFailurePolicy oldPolicy = DatabaseDescriptor.getDiskFailurePolicy();
try
{
DatabaseDescriptor.setDiskFailurePolicy(DiskFailurePolicy.stop);
flushAndExpectError();
Assert.assertFalse(Gossiper.instance.isEnabled());
}
finally
{
DatabaseDescriptor.setDiskFailurePolicy(oldPolicy);
}
}
@Test
public void testFlushUnwriteableIgnore() throws Throwable
{
makeTable();
markDirectoriesUnwriteable();
DiskFailurePolicy oldPolicy = DatabaseDescriptor.getDiskFailurePolicy();
try
{
DatabaseDescriptor.setDiskFailurePolicy(DiskFailurePolicy.ignore);
flushAndExpectError();
}
finally
{
DatabaseDescriptor.setDiskFailurePolicy(oldPolicy);
}
// Next flush should succeed.
makeTable();
flush();
}
public void makeTable() throws Throwable
{
createTable("CREATE TABLE %s (a text, b text, c text, PRIMARY KEY (a, b));");
// insert exactly the amount of tombstones that shouldn't trigger an exception
for (int i = 0; i < 10; i++)
execute("INSERT INTO %s (a, b, c) VALUES ('key', 'column" + i + "', null);");
}
public void markDirectoriesUnwriteable()
{
ColumnFamilyStore cfs = Keyspace.open(keyspace()).getColumnFamilyStore(currentTable());
try
{
for ( ; ; )
{
DataDirectory dir = cfs.getDirectories().getWriteableLocation(1);
BlacklistedDirectories.maybeMarkUnwritable(cfs.getDirectories().getLocationForDisk(dir));
}
}
catch (IOError e)
{
// Expected -- marked all directories as unwritable
}
}
public void flushAndExpectError() throws InterruptedException, ExecutionException
{
try
{
Keyspace.open(KEYSPACE).getColumnFamilyStore(currentTable()).forceFlush().get();
fail("FSWriteError expected.");
}
catch (ExecutionException e)
{
// Correct path.
Assert.assertTrue(e.getCause() instanceof FSWriteError);
}
// Make sure commit log wasn't discarded.
UUID cfid = currentTableMetadata().cfId;
for (CommitLogSegment segment : CommitLog.instance.allocator.getActiveSegments())
if (segment.getDirtyCFIDs().contains(cfid))
return;
fail("Expected commit log to remain dirty for the affected table.");
}
}