diff --git a/.rat-excludes b/.rat-excludes index 994c1b6215..37e68337de 100644 --- a/.rat-excludes +++ b/.rat-excludes @@ -13,7 +13,10 @@ build/** lib/licenses/*.txt .settings/** contrib/pig/example-script.pig -contrib/redhat/cassandra **/cassandra.yaml **/*.db redhat/apache-cassandra.spec +redhat/cassandra +redhat/cassandra.conf +redhat/cassandra.in.sh +redhat/default diff --git a/CHANGES.txt b/CHANGES.txt index 3b0827fd12..4b22549a84 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -4,6 +4,13 @@ 0.7-dev + * fix cli crash after backgrounding (CASSANDRA-1875) + * count timeouts in storageproxy latencies, and include latency + histograms in StorageProxyMBean (CASSANDRA-1893) + * check log4j configuration for changes every 10s (CASSANDRA-1525) + + +0.7.0-rc3 * expose getNaturalEndpoints in StorageServiceMBean taking byte[] key; RMI cannot serialize ByteBuffer (CASSANDRA-1833) * infer org.apache.cassandra.locator for replication strategy classes diff --git a/bin/cassandra b/bin/cassandra index 85fe9540ff..1563cf532f 100755 --- a/bin/cassandra +++ b/bin/cassandra @@ -108,7 +108,7 @@ launch_service() foreground=$2 props=$3 class=$4 - cassandra_parms="-Dlog4j.configuration=log4j-server.properties" + cassandra_parms="-Dlog4j.configuration=log4j-server.properties -Dlog4j.defaultInitOverride=true" if [ "x$pidpath" != "x" ]; then cassandra_parms="$cassandra_parms -Dcassandra-pidfile=$pidpath" diff --git a/bin/cassandra.bat b/bin/cassandra.bat index 985781750d..d016460a8e 100644 --- a/bin/cassandra.bat +++ b/bin/cassandra.bat @@ -37,7 +37,8 @@ set JAVA_OPTS=^ -Dcom.sun.management.jmxremote.port=8080^ -Dcom.sun.management.jmxremote.ssl=false^ -Dcom.sun.management.jmxremote.authenticate=false^ - -Dlog4j.configuration=log4j-server.properties + -Dlog4j.configuration=log4j-server.properties^ + -Dlog4j.defaultInitOverride=true REM ***** CLASSPATH library setting ***** diff --git a/build.xml b/build.xml index 4a832f5350..448edc139b 100644 --- a/build.xml +++ b/build.xml @@ -47,7 +47,7 @@ - + diff --git a/conf/log4j-server.properties b/conf/log4j-server.properties index 5499c3e6dd..cde670729f 100644 --- a/conf/log4j-server.properties +++ b/conf/log4j-server.properties @@ -35,6 +35,7 @@ log4j.appender.R.layout.ConversionPattern=%5p [%t] %d{ISO8601} %F (line %L) %m%n log4j.appender.R.File=/var/log/cassandra/system.log # Application logging options -#log4j.logger.com.facebook=DEBUG -#log4j.logger.com.facebook.infrastructure.gms=DEBUG -#log4j.logger.com.facebook.infrastructure.db=DEBUG +#log4j.logger.org.apache.cassandra=DEBUG +#log4j.logger.org.apache.cassandra.db=DEBUG +#log4j.logger.org.apache.cassandra.service.StorageProxy=DEBUG + diff --git a/debian/changelog b/debian/changelog index 765ccda88f..6d8218934d 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,9 @@ +cassandra (0.7.0~rc3) unstable; urgency=low + + * Release candidate release. + + -- Eric Evans Tue, 21 Dec 2010 17:36:31 -0600 + cassandra (0.7.0~rc2) unstable; urgency=low * Release candidate release. diff --git a/redhat/apache-cassandra.spec b/redhat/apache-cassandra.spec index a215b7905a..e2f477efc5 100644 --- a/redhat/apache-cassandra.spec +++ b/redhat/apache-cassandra.spec @@ -2,22 +2,27 @@ %global username cassandra +%define relname %{name}-%{version}-%{release} + Name: apache-cassandra Version: 0.7.0 -Release: 1%{?dist} +Release: rc3 Summary: Cassandra is a highly scalable, eventually consistent, distributed, structured key-value store. Group: Development/Libraries -License: BSD +License: Apache Software License URL: http://cassandra.apache.org/ -Source0: http://www.ibiblio.org/pub/mirrors/apache/%{username}/%{version}/%{name}-%{version}-src.tar.gz -BuildRoot: %{_tmppath}/%{name}-%{version}-%{release}-root-%(%{__id_u} -n) +Source0: http://www.ibiblio.org/pub/mirrors/apache/%{username}/%{version}/%{relname}-src.tar.gz +BuildRoot: %{_tmppath}/%{relname}-root-%(%{__id_u} -n) BuildRequires: java-devel BuildRequires: jpackage-utils BuildRequires: ant BuildRequires: ant-nodeps +Conflicts: cassandra +Obsoletes: cassandra07 + Requires: java >= 1.6.0 Requires: jna >= 3.2.7 Requires: jpackage-utils @@ -35,10 +40,10 @@ and the data model from Google's BigTable. Like Dynamo, Cassandra is eventually consistent. Like BigTable, Cassandra provides a ColumnFamily-based data model richer than typical key/value systems. -For more information see http://incubator.apache.org/cassandra +For more information see http://cassandra.apache.org/ %prep -%setup -q -n %{name}-%{version}-src +%setup -q -n %{relname}-src %build ant clean jar -Drelease=true @@ -66,7 +71,8 @@ rm bin/cassandra.in.sh mv bin/cassandra %{buildroot}/usr/sbin rm bin/*.bat cp -p bin/* %{buildroot}/usr/bin -cp build/%{name}-%{version}.jar %{buildroot}/usr/share/%{username}/lib +# Handle the case of interim SNAPHOST builds +cp build/%{relname}*jar %{buildroot}/usr/share/%{username}/lib mkdir -p %{buildroot}/var/lib/%{username}/commitlog mkdir -p %{buildroot}/var/lib/%{username}/data mkdir -p %{buildroot}/var/lib/%{username}/saved_caches diff --git a/src/java/org/apache/cassandra/auth/SimpleAuthenticator.java b/src/java/org/apache/cassandra/auth/SimpleAuthenticator.java index 66c236c20a..a8bfc5f32b 100644 --- a/src/java/org/apache/cassandra/auth/SimpleAuthenticator.java +++ b/src/java/org/apache/cassandra/auth/SimpleAuthenticator.java @@ -28,6 +28,7 @@ import java.util.Properties; import java.util.Map; import org.apache.cassandra.config.ConfigurationException; +import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.thrift.AuthenticationException; import org.apache.cassandra.utils.FBUtilities; @@ -92,12 +93,12 @@ public class SimpleAuthenticator implements IAuthenticator boolean authenticated = false; + InputStream in = null; try { - FileInputStream in = new FileInputStream(pfilename); + in = new BufferedInputStream(new FileInputStream(pfilename)); Properties props = new Properties(); props.load(in); - in.close(); // note we keep the message here and for the wrong password exactly the same to prevent attackers from guessing what users are valid if (null == props.getProperty(username)) throw new AuthenticationException(authenticationErrorMessage(mode, username)); @@ -125,6 +126,10 @@ public class SimpleAuthenticator implements IAuthenticator { throw new RuntimeException("Unexpected authentication problem", e); } + finally + { + FileUtils.closeQuietly(in); + } if (!authenticated) throw new AuthenticationException(authenticationErrorMessage(mode, username)); diff --git a/src/java/org/apache/cassandra/cli/CliMain.java b/src/java/org/apache/cassandra/cli/CliMain.java index 40ac620204..1ca095e3d7 100644 --- a/src/java/org/apache/cassandra/cli/CliMain.java +++ b/src/java/org/apache/cassandra/cli/CliMain.java @@ -325,7 +325,14 @@ public class CliMain { prompt = (inCompoundStatement) ? "...\t" : getPrompt(cliClient); - line = reader.readLine(prompt); + try + { + line = reader.readLine(prompt); + } + catch (IOException e) + { + // retry on I/O Exception + } if (line == null) return; diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 03455c7cce..286dec2e0a 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -267,12 +267,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean Set keys = new TreeSet(); if (path.exists()) { + ObjectInputStream in = null; try { long start = System.currentTimeMillis(); logger.info(String.format("reading saved cache %s", path)); - ObjectInputStream in = new ObjectInputStream(new BufferedInputStream(new FileInputStream(path))); + in = new ObjectInputStream(new BufferedInputStream(new FileInputStream(path))); while (in.available() > 0) { int size = in.readInt(); @@ -280,7 +281,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean in.readFully(bytes); keys.add(StorageService.getPartitioner().decorateKey(ByteBuffer.wrap(bytes))); } - in.close(); if (logger.isDebugEnabled()) logger.debug(String.format("completed reading (%d ms; %d keys) saved cache %s", System.currentTimeMillis() - start, keys.size(), path)); @@ -289,6 +289,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean { logger.warn(String.format("error reading saved cache %s", path.getAbsolutePath()), ioe); } + finally + { + FileUtils.closeQuietly(in); + } } return keys; } diff --git a/src/java/org/apache/cassandra/db/ReadResponse.java b/src/java/org/apache/cassandra/db/ReadResponse.java index 079f583589..c7a0ca5313 100644 --- a/src/java/org/apache/cassandra/db/ReadResponse.java +++ b/src/java/org/apache/cassandra/db/ReadResponse.java @@ -47,20 +47,21 @@ private static ICompactSerializer serializer_; return serializer_; } - private Row row_; - private ByteBuffer digest_ = FBUtilities.EMPTY_BYTE_BUFFER; - private boolean isDigestQuery_ = false; + private final Row row_; + private final ByteBuffer digest_; public ReadResponse(ByteBuffer digest ) { assert digest != null; digest_= digest; + row_ = null; } public ReadResponse(Row row) { assert row != null; row_ = row; + digest_ = null; } public Row row() @@ -75,12 +76,7 @@ private static ICompactSerializer serializer_; public boolean isDigestQuery() { - return isDigestQuery_; - } - - public void setIsDigestQuery(boolean isDigestQuery) - { - isDigestQuery_ = isDigestQuery; + return digest_ != null; } } @@ -88,31 +84,35 @@ class ReadResponseSerializer implements ICompactSerializer { public void serialize(ReadResponse rm, DataOutputStream dos) throws IOException { - dos.writeInt(rm.digest().remaining()); - dos.write(rm.digest().array(), rm.digest().position() + rm.digest().arrayOffset(), rm.digest().remaining()); + dos.writeInt(rm.isDigestQuery() ? rm.digest().remaining() : 0); + ByteBuffer buffer = rm.isDigestQuery() ? rm.digest() : FBUtilities.EMPTY_BYTE_BUFFER; + dos.write(buffer.array(), buffer.position() + buffer.arrayOffset(), buffer.remaining()); dos.writeBoolean(rm.isDigestQuery()); - - if( !rm.isDigestQuery() && rm.row() != null ) - { + + if (!rm.isDigestQuery()) + { Row.serializer().serialize(rm.row(), dos); - } - } + } + } public ReadResponse deserialize(DataInputStream dis) throws IOException { + byte[] digest = null; int digestSize = dis.readInt(); - byte[] digest = new byte[digestSize]; - dis.read(digest, 0 , digestSize); + if (digestSize > 0) + { + digest = new byte[digestSize]; + dis.readFully(digest, 0, digestSize); + } boolean isDigest = dis.readBoolean(); - + assert isDigest == digestSize > 0; + Row row = null; if (!isDigest) { row = Row.serializer().deserialize(dis); } - ReadResponse rmsg = isDigest ? new ReadResponse(ByteBuffer.wrap(digest)) : new ReadResponse(row); - rmsg.setIsDigestQuery(isDigest); - return rmsg; + return isDigest ? new ReadResponse(ByteBuffer.wrap(digest)) : new ReadResponse(row); } } diff --git a/src/java/org/apache/cassandra/db/ReadVerbHandler.java b/src/java/org/apache/cassandra/db/ReadVerbHandler.java index 5f6ffd1271..864c40429f 100644 --- a/src/java/org/apache/cassandra/db/ReadVerbHandler.java +++ b/src/java/org/apache/cassandra/db/ReadVerbHandler.java @@ -82,7 +82,6 @@ public class ReadVerbHandler implements IVerbHandler { readResponse = new ReadResponse(row); } - readResponse.setIsDigestQuery(command.isDigestQuery()); /* serialize the ReadResponseMessage. */ readCtx.bufOut_.reset(); diff --git a/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java b/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java index 8e6ced5b4c..1e4e4039f6 100644 --- a/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java +++ b/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java @@ -39,6 +39,7 @@ import org.apache.cassandra.io.sstable.IndexHelper; import org.apache.cassandra.io.sstable.SSTableReader; import org.apache.cassandra.io.util.FileDataInput; import org.apache.cassandra.io.util.FileMark; +import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.utils.BloomFilter; import org.apache.cassandra.utils.FBUtilities; import org.slf4j.Logger; @@ -78,14 +79,7 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement } finally { - try - { - file.close(); - } - catch (IOException ioe) - { - logger.warn("error closing " + file.getPath()); - } + FileUtils.closeQuietly(file); } } diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java index b0e54a6469..af00d86a0d 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java @@ -339,7 +339,7 @@ public class CommitLog } finally { - reader.close(); + FileUtils.closeQuietly(reader); logger.info("Finished reading " + file); } } diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java index 3efdc99094..538cbb9264 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java @@ -25,8 +25,8 @@ import java.util.Map; import java.util.zip.CRC32; import java.util.zip.Checksum; -import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.io.ICompactSerializer2; +import org.apache.cassandra.io.util.FileUtils; public class CommitLogHeader { @@ -132,13 +132,12 @@ public class CommitLogHeader DataInputStream reader = null; try { - reader = new DataInputStream(new FileInputStream(headerFile)); + reader = new DataInputStream(new BufferedInputStream(new FileInputStream(headerFile))); return serializer.deserialize(reader); } finally { - if (reader != null) - reader.close(); + FileUtils.closeQuietly(reader); } } diff --git a/src/java/org/apache/cassandra/gms/FailureDetector.java b/src/java/org/apache/cassandra/gms/FailureDetector.java index e1212b4cf8..1c7d4dc283 100644 --- a/src/java/org/apache/cassandra/gms/FailureDetector.java +++ b/src/java/org/apache/cassandra/gms/FailureDetector.java @@ -18,10 +18,12 @@ package org.apache.cassandra.gms; +import java.io.BufferedOutputStream; import java.io.File; import java.io.FileOutputStream; import java.io.IOException; import java.io.IOError; +import java.io.OutputStream; import java.lang.management.ManagementFactory; import java.util.*; import javax.management.MBeanServer; @@ -31,6 +33,7 @@ import org.apache.commons.lang.StringUtils; import java.net.InetAddress; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.BoundedStatsDeque; import org.slf4j.Logger; @@ -83,17 +86,21 @@ public class FailureDetector implements IFailureDetector, FailureDetectorMBean */ public void dumpInterArrivalTimes() { + OutputStream os = null; try { File file = File.createTempFile("failuredetector-", ".dat"); - FileOutputStream fos = new FileOutputStream(file, true); - fos.write(toString().getBytes()); - fos.close(); + os = new BufferedOutputStream(new FileOutputStream(file, true)); + os.write(toString().getBytes()); } catch (IOException e) { throw new IOError(e); } + finally + { + FileUtils.closeQuietly(os); + } } public void setPhiConvictThreshold(int phi) diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java index a697b89964..d3d3ac0264 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java @@ -19,6 +19,7 @@ package org.apache.cassandra.io.sstable; +import java.io.BufferedInputStream; import java.io.DataInput; import java.io.DataInputStream; import java.io.File; @@ -53,6 +54,7 @@ import org.apache.cassandra.dht.Range; import org.apache.cassandra.io.ICompactSerializer2; import org.apache.cassandra.io.util.BufferedRandomAccessFile; import org.apache.cassandra.io.util.FileDataInput; +import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.io.util.SegmentedFile; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.*; @@ -174,11 +176,18 @@ public class SSTableReader extends SSTable implements Comparable File statsFile = new File(descriptor.filenameFor(SSTable.COMPONENT_STATS)); if (statsFile.exists()) { - logger.debug("Load statistics for {}", descriptor); - DataInputStream dis = new DataInputStream(new FileInputStream(statsFile)); - rowSizes = EstimatedHistogram.serializer.deserialize(dis); - columnCounts = EstimatedHistogram.serializer.deserialize(dis); - dis.close(); + DataInputStream dis = null; + try + { + logger.debug("Load statistics for {}", descriptor); + dis = new DataInputStream(new BufferedInputStream(new FileInputStream(statsFile))); + rowSizes = EstimatedHistogram.serializer.deserialize(dis); + columnCounts = EstimatedHistogram.serializer.deserialize(dis); + } + finally + { + FileUtils.closeQuietly(dis); + } } else { @@ -253,14 +262,15 @@ public class SSTableReader extends SSTable implements Comparable void loadBloomFilter() throws IOException { - DataInputStream stream = new DataInputStream(new FileInputStream(descriptor.filenameFor(Component.FILTER))); + DataInputStream stream = null; try { + stream = new DataInputStream(new BufferedInputStream(new FileInputStream(descriptor.filenameFor(Component.FILTER)))); bf = BloomFilter.serializer().deserialize(stream); } finally { - stream.close(); + FileUtils.closeQuietly(stream); } } @@ -316,7 +326,7 @@ public class SSTableReader extends SSTable implements Comparable } finally { - input.close(); + FileUtils.closeQuietly(input); } // finalize the state of the reader @@ -497,14 +507,7 @@ public class SSTableReader extends SSTable implements Comparable } finally { - try - { - input.close(); - } - catch (IOException e) - { - logger.error("error closing file", e); - } + FileUtils.closeQuietly(input); } } diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableTracker.java b/src/java/org/apache/cassandra/io/sstable/SSTableTracker.java index 60900307e0..cb04c3191f 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableTracker.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableTracker.java @@ -37,7 +37,9 @@ import org.apache.cassandra.cache.JMXInstrumentedCache; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamily; import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.utils.Pair; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -75,19 +77,29 @@ public class SSTableTracker implements Iterable logger.info("saving " + msgSuffix); int count = 0; File tmpFile = File.createTempFile(savedCachePath.getName(), null, savedCachePath.getParentFile()); - FileOutputStream fout = new FileOutputStream(tmpFile); - ObjectOutputStream out = new ObjectOutputStream(new BufferedOutputStream(fout)); - FileDescriptor fd = fout.getFD(); - for (K key : cache.getKeySet()) + + FileOutputStream fout = null; + ObjectOutputStream out = null; + try { - ByteBuffer bytes = converter.apply(key); - out.writeInt(bytes.remaining()); - out.write(bytes.array(),bytes.position()+bytes.arrayOffset(),bytes.remaining()); - ++count; + fout = new FileOutputStream(tmpFile); + out = new ObjectOutputStream(new BufferedOutputStream(fout)); + FileDescriptor fd = fout.getFD(); + for (K key : cache.getKeySet()) + { + ByteBuffer bytes = converter.apply(key); + out.writeInt(bytes.remaining()); + out.write(bytes.array(),bytes.position()+bytes.arrayOffset(),bytes.remaining()); + ++count; + } + out.flush(); + fd.sync(); + } + finally + { + FileUtils.closeQuietly(out); + FileUtils.closeQuietly(fout); } - out.flush(); - fd.sync(); - out.close(); if (!tmpFile.renameTo(savedCachePath)) throw new IOException("Unable to rename cache to " + savedCachePath); if (logger.isDebugEnabled()) diff --git a/src/java/org/apache/cassandra/io/util/FileUtils.java b/src/java/org/apache/cassandra/io/util/FileUtils.java index 42bc06a0ad..6bd8537d13 100644 --- a/src/java/org/apache/cassandra/io/util/FileUtils.java +++ b/src/java/org/apache/cassandra/io/util/FileUtils.java @@ -18,14 +18,12 @@ package org.apache.cassandra.io.util; -import java.io.File; -import java.io.FileNotFoundException; -import java.io.IOException; -import java.io.RandomAccessFile; +import java.io.*; import java.text.DecimalFormat; import java.util.Comparator; import java.util.List; +import org.apache.cassandra.utils.FBUtilities; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -85,6 +83,19 @@ public class FileUtils } } + public static void closeQuietly(Closeable c) + { + try + { + if (c != null) + c.close(); + } + catch (Exception e) + { + logger_.warn("Failed closing stream", e); + } + } + public static class FileComparator implements Comparator { public int compare(File f, File f2) diff --git a/src/java/org/apache/cassandra/io/util/MmappedSegmentedFile.java b/src/java/org/apache/cassandra/io/util/MmappedSegmentedFile.java index 693d850384..7b5880268c 100644 --- a/src/java/org/apache/cassandra/io/util/MmappedSegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/MmappedSegmentedFile.java @@ -173,14 +173,7 @@ public class MmappedSegmentedFile extends SegmentedFile } finally { - try - { - if (raf != null) raf.close(); - } - catch (IOException e) - { - throw new IOError(e); - } + FileUtils.closeQuietly(raf); } return segments; } diff --git a/src/java/org/apache/cassandra/locator/PropertyFileSnitch.java b/src/java/org/apache/cassandra/locator/PropertyFileSnitch.java index cedb5fa35a..9982e54ccc 100644 --- a/src/java/org/apache/cassandra/locator/PropertyFileSnitch.java +++ b/src/java/org/apache/cassandra/locator/PropertyFileSnitch.java @@ -18,22 +18,24 @@ package org.apache.cassandra.locator; +import java.io.BufferedReader; import java.io.FileReader; import java.io.IOException; +import java.io.Reader; import java.net.InetAddress; import java.net.UnknownHostException; import java.util.HashMap; import java.util.Map; import java.util.Properties; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import org.apache.cassandra.config.ConfigurationException; +import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.ResourceWatcher; import org.apache.cassandra.utils.WrappedRunnable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Used to determine if two IP's are in the same datacenter or on the same rack. @@ -112,14 +114,20 @@ public class PropertyFileSnitch extends AbstractNetworkTopologySnitch String rackPropertyFilename = FBUtilities.resourceToFile(RACK_PROPERTY_FILENAME); Properties properties = new Properties(); + Reader reader = null; try { - properties.load(new FileReader(rackPropertyFilename)); + reader = new BufferedReader(new FileReader(rackPropertyFilename)); + properties.load(reader); } catch (IOException e) { throw new ConfigurationException("Unable to read " + RACK_PROPERTY_FILENAME, e); } + finally + { + FileUtils.closeQuietly(reader); + } for (Map.Entry entry : properties.entrySet()) { diff --git a/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java b/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java index 62f20202ff..5b76e98b9d 100644 --- a/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java @@ -42,6 +42,7 @@ import org.apache.cassandra.db.migration.Migration; import org.apache.cassandra.utils.CLibrary; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.Mx4jTool; +import org.apache.log4j.PropertyConfigurator; import org.mortbay.thread.ThreadPool; /** @@ -53,8 +54,16 @@ import org.mortbay.thread.ThreadPool; */ public abstract class AbstractCassandraDaemon implements CassandraDaemon { - private static Logger logger = LoggerFactory - .getLogger(AbstractCassandraDaemon.class); + + //Initialize logging in such a way that it checks for config changes every 10 seconds. + static + { + String config = System.getProperty("log4j.configuration", "log4j-server.properties"); + PropertyConfigurator.configureAndWatch(ClassLoader.getSystemResource(config).getFile(), 10000); + org.apache.log4j.Logger.getLogger(AbstractCassandraDaemon.class).info("Logging initialized"); + } + + private static Logger logger = LoggerFactory.getLogger(AbstractCassandraDaemon.class); protected InetAddress listenAddr; protected int listenPort; diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 595d8ea5c2..91552fc9ba 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -186,7 +186,6 @@ public class StorageProxy implements StorageProxyMBean { writeStats.addNano(System.nanoTime() - startTime); } - } /** @@ -265,19 +264,23 @@ public class StorageProxy implements StorageProxyMBean if (StorageService.instance.isBootstrapMode()) throw new UnavailableException(); long startTime = System.nanoTime(); - List rows; - if (consistency_level == ConsistencyLevel.ONE) + try { - rows = weakRead(commands); + if (consistency_level == ConsistencyLevel.ONE) + { + rows = weakRead(commands); + } + else + { + assert consistency_level.getValue() >= ConsistencyLevel.QUORUM.getValue(); + rows = strongRead(commands, consistency_level); + } } - else + finally { - assert consistency_level.getValue() >= ConsistencyLevel.QUORUM.getValue(); - rows = strongRead(commands, consistency_level); + readStats.addNano(System.nanoTime() - startTime); } - - readStats.addNano(System.nanoTime() - startTime); return rows; } @@ -463,77 +466,82 @@ public class StorageProxy implements StorageProxyMBean if (logger.isDebugEnabled()) logger.debug(command.toString()); long startTime = System.nanoTime(); - - List ranges = getRestrictedRanges(command.range); + List rows; // now scan until we have enough results - List rows = new ArrayList(command.max_keys); - for (AbstractBounds range : ranges) + try { - List liveEndpoints = StorageService.instance.getLiveNaturalEndpoints(command.keyspace, range.right); - - if (consistency_level == ConsistencyLevel.ONE && liveEndpoints.contains(FBUtilities.getLocalAddress())) + rows = new ArrayList(command.max_keys); + List ranges = getRestrictedRanges(command.range); + for (AbstractBounds range : ranges) { - 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); - } - } - else - { - DatabaseDescriptor.getEndpointSnitch().sortByProximity(FBUtilities.getLocalAddress(), liveEndpoints); - RangeSliceCommand c2 = new RangeSliceCommand(command.keyspace, command.column_family, command.super_column, command.predicate, range, command.max_keys); - Message message = c2.getMessage(); + List liveEndpoints = StorageService.instance.getLiveNaturalEndpoints(command.keyspace, range.right); - // collect replies and resolve according to consistency level - RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, liveEndpoints); - AbstractReplicationStrategy rs = Table.open(command.keyspace).getReplicationStrategy(); - QuorumResponseHandler> handler = rs.getQuorumResponseHandler(resolver, consistency_level); - // TODO bail early if live endpoints can't satisfy requested consistency level - for (InetAddress endpoint : liveEndpoints) + if (consistency_level == ConsistencyLevel.ONE && liveEndpoints.contains(FBUtilities.getLocalAddress())) { - MessagingService.instance.sendRR(message, endpoint, handler); if (logger.isDebugEnabled()) - logger.debug("reading " + c2 + " from " + message.getMessageId() + "@" + endpoint); - } - // TODO read repair on remaining replicas? - - // if we're done, great, otherwise, move to the next range - try - { - if (logger.isDebugEnabled()) + logger.debug("local range slice"); + ColumnFamilyStore cfs = Table.open(command.keyspace).getColumnFamilyStore(command.column_family); + try { - for (Row row : handler.get()) - { - logger.debug("range slices read " + row.key); - } - } - rows.addAll(handler.get()); - } - catch (DigestMismatchException e) - { - throw new AssertionError(e); // no digests in range slices yet + 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); + } } - } - - if (rows.size() >= command.max_keys) - break; - } + else + { + DatabaseDescriptor.getEndpointSnitch().sortByProximity(FBUtilities.getLocalAddress(), liveEndpoints); + RangeSliceCommand c2 = new RangeSliceCommand(command.keyspace, command.column_family, command.super_column, command.predicate, range, command.max_keys); + Message message = c2.getMessage(); - rangeStats.addNano(System.nanoTime() - startTime); + // collect replies and resolve according to consistency level + RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, liveEndpoints); + AbstractReplicationStrategy rs = Table.open(command.keyspace).getReplicationStrategy(); + QuorumResponseHandler> handler = rs.getQuorumResponseHandler(resolver, consistency_level); + // TODO bail early if live endpoints can't satisfy requested consistency level + for (InetAddress endpoint : liveEndpoints) + { + MessagingService.instance.sendRR(message, endpoint, handler); + if (logger.isDebugEnabled()) + logger.debug("reading " + c2 + " from " + message.getMessageId() + "@" + endpoint); + } + // TODO read repair on remaining replicas? + + // if we're done, great, otherwise, move to the next range + try + { + if (logger.isDebugEnabled()) + { + for (Row row : handler.get()) + { + logger.debug("range slices read " + row.key); + } + } + rows.addAll(handler.get()); + } + catch (DigestMismatchException e) + { + throw new AssertionError(e); // no digests in range slices yet + } + } + + if (rows.size() >= command.max_keys) + break; + } + } + finally + { + rangeStats.addNano(System.nanoTime() - startTime); + } return rows.size() > command.max_keys ? rows.subList(0, command.max_keys) : rows; } @@ -666,6 +674,16 @@ public class StorageProxy implements StorageProxyMBean return readStats.getRecentLatencyMicros(); } + public long[] getTotalReadLatencyHistogramMicros() + { + return readStats.getTotalLatencyHistogramMicros(); + } + + public long[] getRecentReadLatencyHistogramMicros() + { + return readStats.getRecentLatencyHistogramMicros(); + } + public long getRangeOperations() { return rangeStats.getOpCount(); @@ -681,6 +699,16 @@ public class StorageProxy implements StorageProxyMBean return rangeStats.getRecentLatencyMicros(); } + public long[] getTotalRangeLatencyHistogramMicros() + { + return rangeStats.getTotalLatencyHistogramMicros(); + } + + public long[] getRecentRangeLatencyHistogramMicros() + { + return rangeStats.getRecentLatencyHistogramMicros(); + } + public long getWriteOperations() { return writeStats.getOpCount(); @@ -696,6 +724,16 @@ public class StorageProxy implements StorageProxyMBean return writeStats.getRecentLatencyMicros(); } + public long[] getTotalWriteLatencyHistogramMicros() + { + return writeStats.getTotalLatencyHistogramMicros(); + } + + public long[] getRecentWriteLatencyHistogramMicros() + { + return writeStats.getRecentLatencyHistogramMicros(); + } + public static List scan(String keyspace, String column_family, IndexClause index_clause, SlicePredicate column_predicate, ConsistencyLevel consistency_level) throws IOException, TimeoutException, UnavailableException { diff --git a/src/java/org/apache/cassandra/service/StorageProxyMBean.java b/src/java/org/apache/cassandra/service/StorageProxyMBean.java index 8c2711df65..0c63cf2ba5 100644 --- a/src/java/org/apache/cassandra/service/StorageProxyMBean.java +++ b/src/java/org/apache/cassandra/service/StorageProxyMBean.java @@ -23,14 +23,20 @@ public interface StorageProxyMBean public long getReadOperations(); public long getTotalReadLatencyMicros(); public double getRecentReadLatencyMicros(); + public long[] getTotalReadLatencyHistogramMicros(); + public long[] getRecentReadLatencyHistogramMicros(); public long getRangeOperations(); public long getTotalRangeLatencyMicros(); public double getRecentRangeLatencyMicros(); + public long[] getTotalRangeLatencyHistogramMicros(); + public long[] getRecentRangeLatencyHistogramMicros(); public long getWriteOperations(); public long getTotalWriteLatencyMicros(); public double getRecentWriteLatencyMicros(); + public long[] getTotalWriteLatencyHistogramMicros(); + public long[] getRecentWriteLatencyHistogramMicros(); public boolean getHintedHandoffEnabled(); public void setHintedHandoffEnabled(boolean b); diff --git a/src/java/org/apache/cassandra/streaming/FileStreamTask.java b/src/java/org/apache/cassandra/streaming/FileStreamTask.java index 647fd4aa7b..e4d32c56a4 100644 --- a/src/java/org/apache/cassandra/streaming/FileStreamTask.java +++ b/src/java/org/apache/cassandra/streaming/FileStreamTask.java @@ -25,7 +25,7 @@ import java.nio.ByteBuffer; import java.nio.channels.FileChannel; import java.nio.channels.SocketChannel; -import org.apache.cassandra.streaming.StreamHeader; +import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.utils.FBUtilities; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -35,6 +35,7 @@ import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.utils.Pair; import org.apache.cassandra.utils.WrappedRunnable; + public class FileStreamTask extends WrappedRunnable { private static Logger logger = LoggerFactory.getLogger( FileStreamTask.class ); @@ -109,14 +110,7 @@ public class FileStreamTask extends WrappedRunnable } finally { - try - { - raf.close(); - } - catch (IOException e) - { - throw new AssertionError(e); - } + FileUtils.closeQuietly(raf); } }