You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ja...@apache.org on 2014/01/31 15:12:44 UTC
[01/11] git commit: sstables from stalled repair sessions become live
after a reboot and can resurrect deleted data patch by jasobrown,
reviewed by yukim for CASSANDRA-6503
Updated Branches:
refs/heads/trunk 4b54b8acd -> 811de14ec
sstables from stalled repair sessions become live after a reboot and can resurrect deleted data
patch by jasobrown, reviewed by yukim for CASSANDRA-6503
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/36af4092
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/36af4092
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/36af4092
Branch: refs/heads/trunk
Commit: 36af40925b3e8e01ede5ff6d7ed9a16046409fe3
Parents: b5a2b65
Author: Jason Brown <ja...@gmail.com>
Authored: Thu Jan 30 16:17:01 2014 -0800
Committer: Jason Brown <ja...@gmail.com>
Committed: Thu Jan 30 16:17:01 2014 -0800
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../apache/cassandra/db/ColumnFamilyStore.java | 28 ++++-
.../apache/cassandra/io/sstable/Descriptor.java | 20 ++-
.../apache/cassandra/io/sstable/SSTable.java | 2 +-
.../org/apache/cassandra/io/util/FileUtils.java | 2 +-
.../cassandra/streaming/StreamLockfile.java | 121 +++++++++++++++++++
.../cassandra/streaming/StreamReader.java | 5 +-
.../cassandra/streaming/StreamReceiveTask.java | 72 ++++++++---
.../cassandra/streaming/StreamSession.java | 16 ++-
.../cassandra/streaming/StreamTransferTask.java | 10 +-
.../compress/CompressedStreamReader.java | 4 +-
.../streaming/messages/FileMessage.java | 112 -----------------
.../streaming/messages/IncomingFileMessage.java | 78 ++++++++++++
.../streaming/messages/OutgoingFileMessage.java | 92 ++++++++++++++
.../streaming/messages/StreamMessage.java | 18 ++-
15 files changed, 429 insertions(+), 152 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/36af4092/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 531ac15..56a72ef 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -16,6 +16,7 @@
* Release sstables upon rebuilding 2i (CASSANDRA-6635)
* Add AbstractCompactionStrategy.startup() method (CASSANDRA-6637)
* SSTableScanner may skip rows during cleanup (CASSANDRA-6638)
+ * sstables from stalled repair sessions can resurrect deleted data (CASSANDRA-6503)
Merged from 1.2:
* fsync compression metadata (CASSANDRA-6531)
* Validate CF existence on execution for prepared statement (CASSANDRA-6535)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/36af4092/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 8d09453..8750026 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -18,6 +18,7 @@
package org.apache.cassandra.db;
import java.io.File;
+import java.io.FileFilter;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.nio.ByteBuffer;
@@ -66,6 +67,7 @@ import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.metrics.ColumnFamilyMetrics;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.streaming.StreamLockfile;
import org.apache.cassandra.thrift.IndexExpression;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.*;
@@ -424,9 +426,33 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
*/
public static void scrubDataDirectories(String keyspaceName, String columnFamily)
{
+ Directories directories = Directories.create(keyspaceName, columnFamily);
+
+ // remove any left-behind SSTables from failed/stalled streaming
+ FileFilter filter = new FileFilter()
+ {
+ public boolean accept(File pathname)
+ {
+ return pathname.toString().endsWith(StreamLockfile.FILE_EXT);
+ }
+ };
+ for (File dir : directories.getCFDirectories())
+ {
+ File[] lockfiles = dir.listFiles(filter);
+ if (lockfiles.length == 0)
+ continue;
+ logger.info("Removing SSTables from failed streaming session. Found {} files to cleanup.", lockfiles.length);
+
+ for (File lockfile : lockfiles)
+ {
+ StreamLockfile streamLockfile = new StreamLockfile(lockfile);
+ streamLockfile.cleanup();
+ streamLockfile.delete();
+ }
+ }
+
logger.debug("Removing compacted SSTable files from {} (see http://wiki.apache.org/cassandra/MemtableSSTable)", columnFamily);
- Directories directories = Directories.create(keyspaceName, columnFamily);
for (Map.Entry<Descriptor,Set<Component>> sstableFiles : directories.sstableLister().list().entrySet())
{
Descriptor desc = sstableFiles.getKey();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/36af4092/src/java/org/apache/cassandra/io/sstable/Descriptor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/Descriptor.java b/src/java/org/apache/cassandra/io/sstable/Descriptor.java
index 1b29c1c..d65da45 100644
--- a/src/java/org/apache/cassandra/io/sstable/Descriptor.java
+++ b/src/java/org/apache/cassandra/io/sstable/Descriptor.java
@@ -201,7 +201,18 @@ public class Descriptor
public static Descriptor fromFilename(String filename)
{
File file = new File(filename);
- return fromFilename(file.getParentFile(), file.getName()).left;
+ return fromFilename(file.getParentFile(), file.getName(), false).left;
+ }
+
+ public static Descriptor fromFilename(String filename, boolean skipComponent)
+ {
+ File file = new File(filename);
+ return fromFilename(file.getParentFile(), file.getName(), skipComponent).left;
+ }
+
+ public static Pair<Descriptor,String> fromFilename(File directory, String name)
+ {
+ return fromFilename(directory, name, false);
}
/**
@@ -209,10 +220,11 @@ public class Descriptor
*
* @param directory The directory of the SSTable files
* @param name The name of the SSTable file
+ * @param skipComponent true if the name param should not be parsed for a component tag
*
* @return A Descriptor for the SSTable, and the Component remainder.
*/
- public static Pair<Descriptor,String> fromFilename(File directory, String name)
+ public static Pair<Descriptor,String> fromFilename(File directory, String name, boolean skipComponent)
{
// tokenize the filename
StringTokenizer st = new StringTokenizer(name, String.valueOf(separator));
@@ -239,7 +251,9 @@ public class Descriptor
int generation = Integer.parseInt(nexttok);
// component suffix
- String component = st.nextToken();
+ String component = null;
+ if (!skipComponent)
+ component = st.nextToken();
directory = directory != null ? directory : new File(".");
return Pair.create(new Descriptor(version, directory, ksname, cfname, generation, temporary), component);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/36af4092/src/java/org/apache/cassandra/io/sstable/SSTable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTable.java b/src/java/org/apache/cassandra/io/sstable/SSTable.java
index eaa4522..69c6521 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTable.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTable.java
@@ -205,7 +205,7 @@ public abstract class SSTable
/**
* Discovers existing components for the descriptor. Slow: only intended for use outside the critical path.
*/
- static Set<Component> componentsFor(final Descriptor desc)
+ public static Set<Component> componentsFor(final Descriptor desc)
{
try
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/36af4092/src/java/org/apache/cassandra/io/util/FileUtils.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/util/FileUtils.java b/src/java/org/apache/cassandra/io/util/FileUtils.java
index 6b91bd3..0d8538e 100644
--- a/src/java/org/apache/cassandra/io/util/FileUtils.java
+++ b/src/java/org/apache/cassandra/io/util/FileUtils.java
@@ -310,7 +310,7 @@ public class FileUtils
return f.delete();
}
- public static void delete(File[] files)
+ public static void delete(File... files)
{
for ( File file : files )
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/36af4092/src/java/org/apache/cassandra/streaming/StreamLockfile.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamLockfile.java b/src/java/org/apache/cassandra/streaming/StreamLockfile.java
new file mode 100644
index 0000000..0eb01c5
--- /dev/null
+++ b/src/java/org/apache/cassandra/streaming/StreamLockfile.java
@@ -0,0 +1,121 @@
+/*
+ * 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.streaming;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.StandardOpenOption;
+import java.util.*;
+
+import com.google.common.base.Charsets;
+import org.apache.cassandra.io.sstable.Descriptor;
+import org.apache.cassandra.io.sstable.SSTable;
+import org.apache.cassandra.io.sstable.SSTableWriter;
+import org.apache.cassandra.io.util.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Encapsulates the behavior for 'locking' any streamed sttables to a node.
+ * If a process crashes while converting a set of SSTableWriters to SSTReaders
+ * (meaning, some subset of SSTWs were converted, but not the entire set), we want
+ * to disregard the entire set as we will surely have missing data (by definition).
+ *
+ * Basic behavior id to write out the names of all SSTWs to a file, one SSTW per line,
+ * and then delete the file when complete (normal behavior). This should happen before
+ * converting any SSTWs. Thus, the lockfile is created, some SSTWs are converted,
+ * and if the process crashes, on restart, we look for any existing lockfile, and delete
+ * any referenced SSTRs.
+ */
+public class StreamLockfile
+{
+ public static final String FILE_EXT = ".lockfile";
+ private static final Logger logger = LoggerFactory.getLogger(StreamLockfile.class);
+
+ private final File lockfile;
+
+ public StreamLockfile(File directory, UUID uuid)
+ {
+ lockfile = new File(directory, uuid.toString() + FILE_EXT);
+ }
+
+ public StreamLockfile(File lockfile)
+ {
+ assert lockfile != null;
+ this.lockfile = lockfile;
+ }
+
+ public void create(Collection<SSTableWriter> sstables)
+ {
+ List<String> sstablePaths = new ArrayList<>(sstables.size());
+ for (SSTableWriter writer : sstables)
+ {
+ /* write out the file names *without* the 'tmp-file' flag in the file name.
+ this class will not need to clean up tmp files (on restart), CassandraDaemon does that already,
+ just make sure we delete the fully-formed SSTRs. */
+ sstablePaths.add(writer.descriptor.asTemporary(false).baseFilename());
+ }
+
+ try
+ {
+ Files.write(lockfile.toPath(), sstablePaths, Charsets.UTF_8,
+ StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE, StandardOpenOption.DSYNC);
+ }
+ catch (IOException e)
+ {
+ logger.warn(String.format("Could not create lockfile %s for stream session, nothing to worry too much about", lockfile), e);
+ }
+ }
+
+ public void delete()
+ {
+ FileUtils.delete(lockfile);
+ }
+
+ public void cleanup()
+ {
+ List<String> files = readLockfile(lockfile);
+ for (String file : files)
+ {
+ try
+ {
+ Descriptor desc = Descriptor.fromFilename(file, true);
+ SSTable.delete(desc, SSTable.componentsFor(desc));
+ }
+ catch (Exception e)
+ {
+ logger.warn("failed to delete a potentially stale sstable {}", file);
+ }
+ }
+ }
+
+ private List<String> readLockfile(File lockfile)
+ {
+ try
+ {
+ return Files.readAllLines(lockfile.toPath(), Charsets.UTF_8);
+ }
+ catch (IOException e)
+ {
+ logger.info("couldn't read lockfile {}, ignoring", lockfile.getAbsolutePath());
+ return Collections.emptyList();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/36af4092/src/java/org/apache/cassandra/streaming/StreamReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamReader.java b/src/java/org/apache/cassandra/streaming/StreamReader.java
index d72cb5e..72c239c 100644
--- a/src/java/org/apache/cassandra/streaming/StreamReader.java
+++ b/src/java/org/apache/cassandra/streaming/StreamReader.java
@@ -36,7 +36,6 @@ import org.apache.cassandra.db.Directories;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.Descriptor;
-import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.io.sstable.SSTableWriter;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.messages.FileMessageHeader;
@@ -71,7 +70,7 @@ public class StreamReader
* @return SSTable transferred
* @throws IOException if reading the remote sstable fails. Will throw an RTE if local write fails.
*/
- public SSTableReader read(ReadableByteChannel channel) throws IOException
+ public SSTableWriter read(ReadableByteChannel channel) throws IOException
{
long totalSize = totalSize();
@@ -89,7 +88,7 @@ public class StreamReader
// TODO move this to BytesReadTracker
session.progress(desc, ProgressInfo.Direction.IN, in.getBytesRead(), totalSize);
}
- return writer.closeAndOpenReader();
+ return writer;
}
catch (Throwable e)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/36af4092/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java b/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java
index ac21352..9a2568d 100644
--- a/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java
+++ b/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java
@@ -19,12 +19,15 @@ package org.apache.cassandra.streaming;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.List;
import java.util.UUID;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.io.sstable.SSTableReader;
+import org.apache.cassandra.io.sstable.SSTableWriter;
+import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.Pair;
/**
@@ -36,16 +39,17 @@ public class StreamReceiveTask extends StreamTask
private final int totalFiles;
// total size of files to receive
private final long totalSize;
+ private volatile boolean aborted;
// holds references to SSTables received
- protected Collection<SSTableReader> sstables;
+ protected Collection<SSTableWriter> sstables;
public StreamReceiveTask(StreamSession session, UUID cfId, int totalFiles, long totalSize)
{
super(session, cfId);
this.totalFiles = totalFiles;
this.totalSize = totalSize;
- this.sstables = new ArrayList<>(totalFiles);
+ this.sstables = new ArrayList<>(totalFiles);
}
/**
@@ -53,9 +57,10 @@ public class StreamReceiveTask extends StreamTask
*
* @param sstable SSTable file received.
*/
- public void received(SSTableReader sstable)
+ public void received(SSTableWriter sstable)
{
assert cfId.equals(sstable.metadata.cfId);
+ assert !aborted;
sstables.add(sstable);
if (sstables.size() == totalFiles)
@@ -72,24 +77,61 @@ public class StreamReceiveTask extends StreamTask
return totalSize;
}
- // TODO should be run in background so that this does not block streaming
private void complete()
{
- if (!SSTableReader.acquireReferences(sstables))
- throw new AssertionError("We shouldn't fail acquiring a reference on a sstable that has just been transferred");
- try
+ if (!sstables.isEmpty())
+ StorageService.tasks.submit(new OnCompletionRunnable(this));
+ }
+
+ private static class OnCompletionRunnable implements Runnable
+ {
+ private final StreamReceiveTask task;
+
+ public OnCompletionRunnable(StreamReceiveTask task)
{
- Pair<String, String> kscf = Schema.instance.getCF(cfId);
- ColumnFamilyStore cfs = Keyspace.open(kscf.left).getColumnFamilyStore(kscf.right);
- // add sstables and build secondary indexes
- cfs.addSSTables(sstables);
- cfs.indexManager.maybeBuildSecondaryIndexes(sstables, cfs.indexManager.allIndexesNames());
+ this.task = task;
}
- finally
+
+ public void run()
{
- SSTableReader.releaseReferences(sstables);
+ Pair<String, String> kscf = Schema.instance.getCF(task.cfId);
+ ColumnFamilyStore cfs = Keyspace.open(kscf.left).getColumnFamilyStore(kscf.right);
+
+ StreamLockfile lockfile = new StreamLockfile(cfs.directories.getWriteableLocationAsFile(), UUID.randomUUID());
+ lockfile.create(task.sstables);
+ List<SSTableReader> readers = new ArrayList<>();
+ for (SSTableWriter writer : task.sstables)
+ readers.add(writer.closeAndOpenReader());
+ lockfile.delete();
+
+ if (!SSTableReader.acquireReferences(readers))
+ throw new AssertionError("We shouldn't fail acquiring a reference on a sstable that has just been transferred");
+ try
+ {
+ // add sstables and build secondary indexes
+ cfs.addSSTables(readers);
+ cfs.indexManager.maybeBuildSecondaryIndexes(readers, cfs.indexManager.allIndexesNames());
+ }
+ finally
+ {
+ SSTableReader.releaseReferences(readers);
+ }
+
+ task.session.taskCompleted(task);
}
+ }
- session.taskCompleted(this);
+ public void abort()
+ {
+ aborted = true;
+ Runnable r = new Runnable()
+ {
+ public void run()
+ {
+ for (SSTableWriter writer : sstables)
+ writer.abort();
+ }
+ };
+ StorageService.tasks.submit(r);
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/36af4092/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamSession.java b/src/java/org/apache/cassandra/streaming/StreamSession.java
index 98a76fc..4777995 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -23,6 +23,7 @@ import java.util.*;
import java.util.concurrent.Future;
import com.google.common.collect.Lists;
+import org.apache.cassandra.io.sstable.SSTableWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -312,6 +313,12 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
{
state(finalState);
+ if (finalState == State.FAILED)
+ {
+ for (StreamReceiveTask srt : receivers.values())
+ srt.abort();
+ }
+
// Note that we shouldn't block on this close because this method is called on the handler
// incoming thread (so we would deadlock).
handler.close();
@@ -359,7 +366,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
break;
case FILE:
- receive((FileMessage) message);
+ receive((IncomingFileMessage) message);
break;
case RECEIVED:
@@ -458,7 +465,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
*
* @param message received file
*/
- public void receive(FileMessage message)
+ public void receive(IncomingFileMessage message)
{
long headerSize = message.header.size();
StreamingMetrics.totalIncomingBytes.inc(headerSize);
@@ -487,7 +494,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
*/
public void retry(UUID cfId, int sequenceNumber)
{
- FileMessage message = transfers.get(cfId).createMessageForRetry(sequenceNumber);
+ OutgoingFileMessage message = transfers.get(cfId).createMessageForRetry(sequenceNumber);
handler.sendMessage(message);
}
@@ -502,6 +509,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
}
else
{
+ handler.sendMessage(new CompleteMessage());
state(State.WAIT_COMPLETE);
}
}
@@ -623,7 +631,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
state(State.STREAMING);
for (StreamTransferTask task : transfers.values())
{
- Collection<FileMessage> messages = task.getFileMessages();
+ Collection<OutgoingFileMessage> messages = task.getFileMessages();
if (messages.size() > 0)
handler.sendMessages(messages);
else
http://git-wip-us.apache.org/repos/asf/cassandra/blob/36af4092/src/java/org/apache/cassandra/streaming/StreamTransferTask.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamTransferTask.java b/src/java/org/apache/cassandra/streaming/StreamTransferTask.java
index 61ad058..8e461cc 100644
--- a/src/java/org/apache/cassandra/streaming/StreamTransferTask.java
+++ b/src/java/org/apache/cassandra/streaming/StreamTransferTask.java
@@ -21,7 +21,7 @@ import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.cassandra.io.sstable.SSTableReader;
-import org.apache.cassandra.streaming.messages.FileMessage;
+import org.apache.cassandra.streaming.messages.OutgoingFileMessage;
import org.apache.cassandra.utils.Pair;
/**
@@ -31,7 +31,7 @@ public class StreamTransferTask extends StreamTask
{
private final AtomicInteger sequenceNumber = new AtomicInteger(0);
- private final Map<Integer, FileMessage> files = new HashMap<>();
+ private final Map<Integer, OutgoingFileMessage> files = new HashMap<>();
private long totalSize;
@@ -43,7 +43,7 @@ public class StreamTransferTask extends StreamTask
public void addTransferFile(SSTableReader sstable, long estimatedKeys, List<Pair<Long, Long>> sections)
{
assert sstable != null && cfId.equals(sstable.metadata.cfId);
- FileMessage message = new FileMessage(sstable, sequenceNumber.getAndIncrement(), estimatedKeys, sections);
+ OutgoingFileMessage message = new OutgoingFileMessage(sstable, sequenceNumber.getAndIncrement(), estimatedKeys, sections);
files.put(message.header.sequenceNumber, message);
totalSize += message.header.size();
}
@@ -71,14 +71,14 @@ public class StreamTransferTask extends StreamTask
return totalSize;
}
- public Collection<FileMessage> getFileMessages()
+ public Collection<OutgoingFileMessage> getFileMessages()
{
// We may race between queuing all those messages and the completion of the completion of
// the first ones. So copy the values to avoid a ConcurrentModificationException
return new ArrayList<>(files.values());
}
- public FileMessage createMessageForRetry(int sequenceNumber)
+ public OutgoingFileMessage createMessageForRetry(int sequenceNumber)
{
return files.get(sequenceNumber);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/36af4092/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java b/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java
index 6f5d0f5..4aac941 100644
--- a/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java
+++ b/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java
@@ -55,7 +55,7 @@ public class CompressedStreamReader extends StreamReader
* @throws java.io.IOException if reading the remote sstable fails. Will throw an RTE if local write fails.
*/
@Override
- public SSTableReader read(ReadableByteChannel channel) throws IOException
+ public SSTableWriter read(ReadableByteChannel channel) throws IOException
{
long totalSize = totalSize();
@@ -81,7 +81,7 @@ public class CompressedStreamReader extends StreamReader
session.progress(desc, ProgressInfo.Direction.IN, cis.getTotalCompressedBytesRead(), totalSize);
}
}
- return writer.closeAndOpenReader();
+ return writer;
}
catch (Throwable e)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/36af4092/src/java/org/apache/cassandra/streaming/messages/FileMessage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/messages/FileMessage.java b/src/java/org/apache/cassandra/streaming/messages/FileMessage.java
deleted file mode 100644
index a0543c0..0000000
--- a/src/java/org/apache/cassandra/streaming/messages/FileMessage.java
+++ /dev/null
@@ -1,112 +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.streaming.messages;
-
-import java.io.DataInputStream;
-import java.io.DataOutput;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.nio.channels.Channels;
-import java.nio.channels.ReadableByteChannel;
-import java.nio.channels.WritableByteChannel;
-import java.util.List;
-
-import org.apache.cassandra.io.compress.CompressionMetadata;
-import org.apache.cassandra.io.sstable.SSTableReader;
-import org.apache.cassandra.streaming.StreamReader;
-import org.apache.cassandra.streaming.StreamSession;
-import org.apache.cassandra.streaming.StreamWriter;
-import org.apache.cassandra.streaming.compress.CompressedStreamReader;
-import org.apache.cassandra.streaming.compress.CompressedStreamWriter;
-import org.apache.cassandra.streaming.compress.CompressionInfo;
-import org.apache.cassandra.utils.Pair;
-
-/**
- * FileMessage is used to transfer/receive the part(or whole) of a SSTable data file.
- */
-public class FileMessage extends StreamMessage
-{
- public static Serializer<FileMessage> serializer = new Serializer<FileMessage>()
- {
- public FileMessage deserialize(ReadableByteChannel in, int version, StreamSession session) throws IOException
- {
- DataInputStream input = new DataInputStream(Channels.newInputStream(in));
- FileMessageHeader header = FileMessageHeader.serializer.deserialize(input, version);
- StreamReader reader = header.compressionInfo == null ? new StreamReader(header, session)
- : new CompressedStreamReader(header, session);
-
- try
- {
- return new FileMessage(reader.read(in), header);
- }
- catch (Throwable e)
- {
- session.doRetry(header, e);
- return null;
- }
- }
-
- public void serialize(FileMessage message, WritableByteChannel out, int version, StreamSession session) throws IOException
- {
- DataOutput output = new DataOutputStream(Channels.newOutputStream(out));
- FileMessageHeader.serializer.serialize(message.header, output, version);
- StreamWriter writer = message.header.compressionInfo == null ?
- new StreamWriter(message.sstable, message.header.sections, session) :
- new CompressedStreamWriter(message.sstable,
- message.header.sections,
- message.header.compressionInfo, session);
- writer.write(out);
- session.fileSent(message.header);
- }
- };
-
- public final FileMessageHeader header;
- public final SSTableReader sstable;
-
- public FileMessage(SSTableReader sstable, FileMessageHeader header)
- {
- super(Type.FILE);
- this.header = header;
- this.sstable = sstable;
- }
-
- public FileMessage(SSTableReader sstable, int sequenceNumber, long estimatedKeys, List<Pair<Long, Long>> sections)
- {
- super(Type.FILE);
- this.sstable = sstable;
-
- CompressionInfo compressionInfo = null;
- if (sstable.compression)
- {
- CompressionMetadata meta = sstable.getCompressionMetadata();
- compressionInfo = new CompressionInfo(meta.getChunksForSections(sections), meta.parameters);
- }
- this.header = new FileMessageHeader(sstable.metadata.cfId,
- sequenceNumber,
- sstable.descriptor.version.toString(),
- estimatedKeys,
- sections,
- compressionInfo);
- }
-
- @Override
- public String toString()
- {
- return "File (" + header + ", file: " + sstable.getFilename() + ")";
- }
-}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/36af4092/src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java b/src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java
new file mode 100644
index 0000000..a403390
--- /dev/null
+++ b/src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java
@@ -0,0 +1,78 @@
+/*
+ * 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.streaming.messages;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.WritableByteChannel;
+
+import org.apache.cassandra.io.sstable.SSTableWriter;
+import org.apache.cassandra.streaming.StreamReader;
+import org.apache.cassandra.streaming.StreamSession;
+import org.apache.cassandra.streaming.compress.CompressedStreamReader;
+
+/**
+ * IncomingFileMessage is used to receive the part(or whole) of a SSTable data file.
+ */
+public class IncomingFileMessage extends StreamMessage
+{
+ public static Serializer<IncomingFileMessage> serializer = new Serializer<IncomingFileMessage>()
+ {
+ public IncomingFileMessage deserialize(ReadableByteChannel in, int version, StreamSession session) throws IOException
+ {
+ DataInputStream input = new DataInputStream(Channels.newInputStream(in));
+ FileMessageHeader header = FileMessageHeader.serializer.deserialize(input, version);
+ StreamReader reader = header.compressionInfo == null ? new StreamReader(header, session)
+ : new CompressedStreamReader(header, session);
+
+ try
+ {
+ return new IncomingFileMessage(reader.read(in), header);
+ }
+ catch (Throwable e)
+ {
+ session.doRetry(header, e);
+ return null;
+ }
+ }
+
+ public void serialize(IncomingFileMessage message, WritableByteChannel out, int version, StreamSession session) throws IOException
+ {
+ throw new UnsupportedOperationException("Not allowed to call serialize on an incoming file");
+ }
+ };
+
+ public FileMessageHeader header;
+ public SSTableWriter sstable;
+
+ public IncomingFileMessage(SSTableWriter sstable, FileMessageHeader header)
+ {
+ super(Type.FILE);
+ this.header = header;
+ this.sstable = sstable;
+ }
+
+ @Override
+ public String toString()
+ {
+ return "File (" + header + ", file: " + sstable.getFilename() + ")";
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/cassandra/blob/36af4092/src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java b/src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java
new file mode 100644
index 0000000..1fa115f
--- /dev/null
+++ b/src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java
@@ -0,0 +1,92 @@
+/*
+ * 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.streaming.messages;
+
+import java.io.DataOutput;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.WritableByteChannel;
+import java.util.List;
+
+import org.apache.cassandra.io.compress.CompressionMetadata;
+import org.apache.cassandra.io.sstable.SSTableReader;
+import org.apache.cassandra.streaming.StreamSession;
+import org.apache.cassandra.streaming.StreamWriter;
+import org.apache.cassandra.streaming.compress.CompressedStreamWriter;
+import org.apache.cassandra.streaming.compress.CompressionInfo;
+import org.apache.cassandra.utils.Pair;
+
+/**
+ * OutgoingFileMessage is used to transfer the part(or whole) of a SSTable data file.
+ */
+public class OutgoingFileMessage extends StreamMessage
+{
+ public static Serializer<OutgoingFileMessage> serializer = new Serializer<OutgoingFileMessage>()
+ {
+ public OutgoingFileMessage deserialize(ReadableByteChannel in, int version, StreamSession session) throws IOException
+ {
+ throw new UnsupportedOperationException("Not allowed to call deserialize on an outgoing file");
+ }
+
+ public void serialize(OutgoingFileMessage message, WritableByteChannel out, int version, StreamSession session) throws IOException
+ {
+ DataOutput output = new DataOutputStream(Channels.newOutputStream(out));
+ FileMessageHeader.serializer.serialize(message.header, output, version);
+
+ final SSTableReader reader = message.sstable;
+ StreamWriter writer = message.header.compressionInfo == null ?
+ new StreamWriter(reader, message.header.sections, session) :
+ new CompressedStreamWriter(reader,
+ message.header.sections,
+ message.header.compressionInfo, session);
+ writer.write(out);
+ session.fileSent(message.header);
+ }
+ };
+
+ public FileMessageHeader header;
+ public SSTableReader sstable;
+
+ public OutgoingFileMessage(SSTableReader sstable, int sequenceNumber, long estimatedKeys, List<Pair<Long, Long>> sections)
+ {
+ super(Type.FILE);
+ this.sstable = sstable;
+
+ CompressionInfo compressionInfo = null;
+ if (sstable.compression)
+ {
+ CompressionMetadata meta = sstable.getCompressionMetadata();
+ compressionInfo = new CompressionInfo(meta.getChunksForSections(sections), meta.parameters);
+ }
+ this.header = new FileMessageHeader(sstable.metadata.cfId,
+ sequenceNumber,
+ sstable.descriptor.version.toString(),
+ estimatedKeys,
+ sections,
+ compressionInfo);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "File (" + header + ", file: " + sstable.getFilename() + ")";
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/cassandra/blob/36af4092/src/java/org/apache/cassandra/streaming/messages/StreamMessage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/messages/StreamMessage.java b/src/java/org/apache/cassandra/streaming/messages/StreamMessage.java
index 9e146e8..7010c95 100644
--- a/src/java/org/apache/cassandra/streaming/messages/StreamMessage.java
+++ b/src/java/org/apache/cassandra/streaming/messages/StreamMessage.java
@@ -42,7 +42,7 @@ public abstract class StreamMessage
buff.flip();
while (buff.hasRemaining())
out.write(buff);
- message.type.serializer.serialize(message, out, version, session);
+ message.type.outSerializer.serialize(message, out, version, session);
}
public static StreamMessage deserialize(ReadableByteChannel in, int version, StreamSession session) throws IOException
@@ -52,7 +52,7 @@ public abstract class StreamMessage
{
buff.flip();
Type type = Type.get(buff.get());
- return type.serializer.deserialize(in, version, session);
+ return type.inSerializer.deserialize(in, version, session);
}
else
{
@@ -73,7 +73,7 @@ public abstract class StreamMessage
public static enum Type
{
PREPARE(1, 5, PrepareMessage.serializer),
- FILE(2, 0, FileMessage.serializer),
+ FILE(2, 0, IncomingFileMessage.serializer, OutgoingFileMessage.serializer),
RECEIVED(3, 4, ReceivedMessage.serializer),
RETRY(4, 4, RetryMessage.serializer),
COMPLETE(5, 1, CompleteMessage.serializer),
@@ -91,14 +91,22 @@ public abstract class StreamMessage
private final byte type;
public final int priority;
- public final Serializer<StreamMessage> serializer;
+ public final Serializer<StreamMessage> inSerializer;
+ public final Serializer<StreamMessage> outSerializer;
@SuppressWarnings("unchecked")
private Type(int type, int priority, Serializer serializer)
{
+ this(type, priority, serializer, serializer);
+ }
+
+ @SuppressWarnings("unchecked")
+ private Type(int type, int priority, Serializer inSerializer, Serializer outSerializer)
+ {
this.type = (byte) type;
this.priority = priority;
- this.serializer = serializer;
+ this.inSerializer = inSerializer;
+ this.outSerializer = outSerializer;
}
}
[04/11] git commit: Update versions and licenses for 1.2.14 release
Posted by ja...@apache.org.
Update versions and licenses for 1.2.14 release
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/6a931440
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/6a931440
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/6a931440
Branch: refs/heads/trunk
Commit: 6a9314408cbd69ce26c2ba04bf49df7809a911bf
Parents: 6ad995e
Author: Sylvain Lebresne <sy...@datastax.com>
Authored: Fri Jan 31 10:45:03 2014 +0100
Committer: Sylvain Lebresne <sy...@datastax.com>
Committed: Fri Jan 31 10:45:03 2014 +0100
----------------------------------------------------------------------
.rat-excludes | 1 +
build.xml | 2 +-
debian/changelog | 6 ++++++
.../apache/cassandra/db/BatchlogManagerTest.java | 17 +++++++++++++++++
4 files changed, 25 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/6a931440/.rat-excludes
----------------------------------------------------------------------
diff --git a/.rat-excludes b/.rat-excludes
index f7f9637..112aec0 100644
--- a/.rat-excludes
+++ b/.rat-excludes
@@ -32,3 +32,4 @@ test/data/legacy-sstables/**
examples/pig/**
examples/hadoop_word_count/conf/log4j.properties
pylib/cqlshlib/test/**
+src/resources/org/apache/cassandra/config/version.properties
http://git-wip-us.apache.org/repos/asf/cassandra/blob/6a931440/build.xml
----------------------------------------------------------------------
diff --git a/build.xml b/build.xml
index 23f8c71..2a6c524 100644
--- a/build.xml
+++ b/build.xml
@@ -25,7 +25,7 @@
<property name="debuglevel" value="source,lines,vars"/>
<!-- default version and SCM information -->
- <property name="base.version" value="1.2.13"/>
+ <property name="base.version" value="1.2.14"/>
<property name="scm.connection" value="scm:git://git.apache.org/cassandra.git"/>
<property name="scm.developerConnection" value="scm:git://git.apache.org/cassandra.git"/>
<property name="scm.url" value="http://git-wip-us.apache.org/repos/asf?p=cassandra.git;a=tree"/>
http://git-wip-us.apache.org/repos/asf/cassandra/blob/6a931440/debian/changelog
----------------------------------------------------------------------
diff --git a/debian/changelog b/debian/changelog
index b419cb7..472063e 100644
--- a/debian/changelog
+++ b/debian/changelog
@@ -1,3 +1,9 @@
+cassandra (1.2.14) unstable; urgency=low
+
+ * New release
+
+ -- Sylvain Lebresne <sl...@apache.org> Fri, 31 Jan 2014 10:29:13 +0100
+
cassandra (1.2.13) unstable; urgency=low
* New release
http://git-wip-us.apache.org/repos/asf/cassandra/blob/6a931440/test/unit/org/apache/cassandra/db/BatchlogManagerTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/BatchlogManagerTest.java b/test/unit/org/apache/cassandra/db/BatchlogManagerTest.java
index 637815b..08728fc 100644
--- a/test/unit/org/apache/cassandra/db/BatchlogManagerTest.java
+++ b/test/unit/org/apache/cassandra/db/BatchlogManagerTest.java
@@ -1,3 +1,20 @@
+/*
+ * 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.db;
import java.net.InetAddress;
[09/11] git commit: Merge branch 'cassandra-1.2' into cassandra-2.0
Posted by ja...@apache.org.
Merge branch 'cassandra-1.2' into cassandra-2.0
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/f6a7a5be
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/f6a7a5be
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/f6a7a5be
Branch: refs/heads/trunk
Commit: f6a7a5be533ad39fe5af72fb2f1287b9c9d73bad
Parents: 9d70cf3 df86e8f
Author: Sylvain Lebresne <sy...@datastax.com>
Authored: Fri Jan 31 13:28:26 2014 +0100
Committer: Sylvain Lebresne <sy...@datastax.com>
Committed: Fri Jan 31 13:28:26 2014 +0100
----------------------------------------------------------------------
CHANGES.txt | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f6a7a5be/CHANGES.txt
----------------------------------------------------------------------
[07/11] git commit: Merge branch 'cassandra-1.2' into cassandra-2.0
Posted by ja...@apache.org.
Merge branch 'cassandra-1.2' into cassandra-2.0
Conflicts:
.rat-excludes
build.xml
debian/changelog
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/9d70cf3b
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/9d70cf3b
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/9d70cf3b
Branch: refs/heads/trunk
Commit: 9d70cf3b90e703865b4015bb19d171349953d5c7
Parents: 1141cdb 4cb045a
Author: Sylvain Lebresne <sy...@datastax.com>
Authored: Fri Jan 31 13:25:45 2014 +0100
Committer: Sylvain Lebresne <sy...@datastax.com>
Committed: Fri Jan 31 13:25:45 2014 +0100
----------------------------------------------------------------------
.rat-excludes | 2 +-
doc/cql3/CQL.textile | 10 +++++-----
.../apache/cassandra/db/BatchlogManagerTest.java | 17 +++++++++++++++++
3 files changed, 23 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/9d70cf3b/.rat-excludes
----------------------------------------------------------------------
diff --cc .rat-excludes
index cf618c0,112aec0..871f5df
--- a/.rat-excludes
+++ b/.rat-excludes
@@@ -20,12 -18,18 +20,12 @@@ lib/licenses/*.tx
.externalToolBuilders/**
test/data/serialization/*/*
**/*.wpr
-conf/schema-sample.txt
-drivers/txpy/txcql/cassandra/*
-drivers/py/cql/cassandra/*
doc/**
build.properties.default
-test/data/legacy-sstables/**
+test/data/**
examples/pig/**
+examples/triggers/build.xml
+examples/triggers/conf/*
examples/hadoop_word_count/conf/log4j.properties
pylib/cqlshlib/test/**
- src/resources/org/apache/cassandra/config/version.properties
+ src/resources/org/apache/cassandra/config/version.properties
http://git-wip-us.apache.org/repos/asf/cassandra/blob/9d70cf3b/doc/cql3/CQL.textile
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/9d70cf3b/test/unit/org/apache/cassandra/db/BatchlogManagerTest.java
----------------------------------------------------------------------
[02/11] git commit: Switch stress to use ITransportFactory
Posted by ja...@apache.org.
Switch stress to use ITransportFactory
patch by Sam Tunnicliffe; reviewed by Aleksey Yeschenko for
CASSANDRA-6641
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/e7b23cc7
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/e7b23cc7
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/e7b23cc7
Branch: refs/heads/trunk
Commit: e7b23cc741622ccf8b1487d8622cca47dcb9cc34
Parents: 36af409
Author: Aleksey Yeschenko <al...@apache.org>
Authored: Fri Jan 31 01:38:54 2014 -0500
Committer: Aleksey Yeschenko <al...@apache.org>
Committed: Fri Jan 31 01:38:54 2014 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../org/apache/cassandra/cli/CliOptions.java | 45 +++++++++++--
.../org/apache/cassandra/stress/Session.java | 69 +++++++++++++-------
3 files changed, 88 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e7b23cc7/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 56a72ef..94b21d4 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -17,6 +17,7 @@
* Add AbstractCompactionStrategy.startup() method (CASSANDRA-6637)
* SSTableScanner may skip rows during cleanup (CASSANDRA-6638)
* sstables from stalled repair sessions can resurrect deleted data (CASSANDRA-6503)
+ * Switch stress to use ITransportFactory (CASSANDRA-6641)
Merged from 1.2:
* fsync compression metadata (CASSANDRA-6531)
* Validate CF existence on execution for prepared statement (CASSANDRA-6535)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e7b23cc7/src/java/org/apache/cassandra/cli/CliOptions.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cli/CliOptions.java b/src/java/org/apache/cassandra/cli/CliOptions.java
index 68f17c9..7894bf9 100644
--- a/src/java/org/apache/cassandra/cli/CliOptions.java
+++ b/src/java/org/apache/cassandra/cli/CliOptions.java
@@ -17,9 +17,15 @@
*/
package org.apache.cassandra.cli;
+import java.util.HashMap;
+import java.util.Map;
+
+import com.google.common.base.Joiner;
import org.apache.commons.cli.*;
+import org.apache.cassandra.config.EncryptionOptions;
import org.apache.cassandra.thrift.ITransportFactory;
+import org.apache.cassandra.thrift.SSLTransportFactory;
/**
*
@@ -114,11 +120,6 @@ public class CliOptions
css.hostName = DEFAULT_HOST;
}
- if (cmd.hasOption(TRANSPORT_FACTORY))
- {
- css.transportFactory = validateAndSetTransportFactory(cmd.getOptionValue(TRANSPORT_FACTORY));
- }
-
if (cmd.hasOption(DEBUG_OPTION))
{
css.debug = true;
@@ -217,6 +218,12 @@ public class CliOptions
css.encOptions.cipher_suites = cmd.getOptionValue(SSL_CIPHER_SUITES).split(",");
}
+ if (cmd.hasOption(TRANSPORT_FACTORY))
+ {
+ css.transportFactory = validateAndSetTransportFactory(cmd.getOptionValue(TRANSPORT_FACTORY));
+ configureTransportFactory(css.transportFactory, css.encOptions);
+ }
+
// Abort if there are any unrecognized arguments left
if (cmd.getArgs().length > 0)
{
@@ -281,4 +288,32 @@ public class CliOptions
throw new IllegalArgumentException(String.format("Cannot create a transport factory '%s'.", transportFactory), e);
}
}
+
+ private static void configureTransportFactory(ITransportFactory transportFactory, EncryptionOptions encOptions)
+ {
+ Map<String, String> options = new HashMap<>();
+ // If the supplied factory supports the same set of options as our SSL impl, set those
+ if (transportFactory.supportedOptions().contains(SSLTransportFactory.TRUSTSTORE))
+ options.put(SSLTransportFactory.TRUSTSTORE, encOptions.truststore);
+ if (transportFactory.supportedOptions().contains(SSLTransportFactory.TRUSTSTORE_PASSWORD))
+ options.put(SSLTransportFactory.TRUSTSTORE_PASSWORD, encOptions.truststore_password);
+ if (transportFactory.supportedOptions().contains(SSLTransportFactory.PROTOCOL))
+ options.put(SSLTransportFactory.PROTOCOL, encOptions.protocol);
+ if (transportFactory.supportedOptions().contains(SSLTransportFactory.CIPHER_SUITES))
+ options.put(SSLTransportFactory.CIPHER_SUITES, Joiner.on(',').join(encOptions.cipher_suites));
+
+ if (transportFactory.supportedOptions().contains(SSLTransportFactory.KEYSTORE)
+ && encOptions.require_client_auth)
+ options.put(SSLTransportFactory.KEYSTORE, encOptions.keystore);
+ if (transportFactory.supportedOptions().contains(SSLTransportFactory.KEYSTORE_PASSWORD)
+ && encOptions.require_client_auth)
+ options.put(SSLTransportFactory.KEYSTORE_PASSWORD, encOptions.keystore_password);
+
+ // Now check if any of the factory's supported options are set as system properties
+ for (String optionKey : transportFactory.supportedOptions())
+ if (System.getProperty(optionKey) != null)
+ options.put(optionKey, System.getProperty(optionKey));
+
+ transportFactory.setOptions(options);
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e7b23cc7/tools/stress/src/org/apache/cassandra/stress/Session.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/Session.java b/tools/stress/src/org/apache/cassandra/stress/Session.java
index 242fa14..9ac865d 100644
--- a/tools/stress/src/org/apache/cassandra/stress/Session.java
+++ b/tools/stress/src/org/apache/cassandra/stress/Session.java
@@ -24,29 +24,25 @@ import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
+import com.google.common.base.Joiner;
+import org.apache.commons.cli.*;
+import org.apache.commons.lang3.StringUtils;
import com.yammer.metrics.Metrics;
import org.apache.cassandra.auth.IAuthenticator;
-import org.apache.cassandra.cli.transport.FramedTransportFactory;
import org.apache.cassandra.config.CFMetaData;
-import org.apache.cassandra.config.EncryptionOptions;
import org.apache.cassandra.config.EncryptionOptions.ClientEncryptionOptions;
+import org.apache.cassandra.config.EncryptionOptions;
+import org.apache.cassandra.db.ColumnFamilyType;
+import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.exceptions.SyntaxException;
-import org.apache.cassandra.db.marshal.*;
-import org.apache.commons.cli.*;
-
-import org.apache.cassandra.db.ColumnFamilyType;
import org.apache.cassandra.stress.util.CassandraClient;
-import org.apache.cassandra.transport.SimpleClient;
import org.apache.cassandra.thrift.*;
-import org.apache.commons.lang3.StringUtils;
-
+import org.apache.cassandra.transport.SimpleClient;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.thrift.protocol.TBinaryProtocol;
-import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
-import org.apache.thrift.transport.TTransportFactory;
public class Session implements Serializable
{
@@ -175,7 +171,7 @@ public class Session implements Serializable
public final boolean timeUUIDComparator;
public double traceProbability = 0.0;
public EncryptionOptions encOptions = new ClientEncryptionOptions();
- public TTransportFactory transportFactory = new FramedTransportFactory();
+ public ITransportFactory transportFactory = new TFramedTransportFactory();
public Session(String[] arguments) throws IllegalArgumentException, SyntaxException
{
@@ -455,7 +451,10 @@ public class Session implements Serializable
encOptions.cipher_suites = cmd.getOptionValue(SSL_CIPHER_SUITES).split(",");
if (cmd.hasOption("tf"))
+ {
transportFactory = validateAndSetTransportFactory(cmd.getOptionValue("tf"));
+ configureTransportFactory(transportFactory, encOptions);
+ }
if (cmd.hasOption("un"))
username = cmd.getOptionValue("un");
@@ -476,17 +475,17 @@ public class Session implements Serializable
sigma = numDifferentKeys * STDev;
}
- private TTransportFactory validateAndSetTransportFactory(String transportFactory)
+ private ITransportFactory validateAndSetTransportFactory(String transportFactory)
{
try
{
Class factory = Class.forName(transportFactory);
- if(!TTransportFactory.class.isAssignableFrom(factory))
+ if(!ITransportFactory.class.isAssignableFrom(factory))
throw new IllegalArgumentException(String.format("transport factory '%s' " +
- "not derived from TTransportFactory", transportFactory));
+ "not derived from ITransportFactory", transportFactory));
- return (TTransportFactory) factory.newInstance();
+ return (ITransportFactory) factory.newInstance();
}
catch (Exception e)
{
@@ -494,6 +493,34 @@ public class Session implements Serializable
}
}
+ private void configureTransportFactory(ITransportFactory transportFactory, EncryptionOptions encOptions)
+ {
+ Map<String, String> options = new HashMap<>();
+ // If the supplied factory supports the same set of options as our SSL impl, set those
+ if (transportFactory.supportedOptions().contains(SSLTransportFactory.TRUSTSTORE))
+ options.put(SSLTransportFactory.TRUSTSTORE, encOptions.truststore);
+ if (transportFactory.supportedOptions().contains(SSLTransportFactory.TRUSTSTORE_PASSWORD))
+ options.put(SSLTransportFactory.TRUSTSTORE_PASSWORD, encOptions.truststore_password);
+ if (transportFactory.supportedOptions().contains(SSLTransportFactory.PROTOCOL))
+ options.put(SSLTransportFactory.PROTOCOL, encOptions.protocol);
+ if (transportFactory.supportedOptions().contains(SSLTransportFactory.CIPHER_SUITES))
+ options.put(SSLTransportFactory.CIPHER_SUITES, Joiner.on(',').join(encOptions.cipher_suites));
+
+ if (transportFactory.supportedOptions().contains(SSLTransportFactory.KEYSTORE)
+ && encOptions.require_client_auth)
+ options.put(SSLTransportFactory.KEYSTORE, encOptions.keystore);
+ if (transportFactory.supportedOptions().contains(SSLTransportFactory.KEYSTORE_PASSWORD)
+ && encOptions.require_client_auth)
+ options.put(SSLTransportFactory.KEYSTORE_PASSWORD, encOptions.keystore_password);
+
+ // Now check if any of the factory's supported options are set as system properties
+ for (String optionKey : transportFactory.supportedOptions())
+ if (System.getProperty(optionKey) != null)
+ options.put(optionKey, System.getProperty(optionKey));
+
+ transportFactory.setOptions(options);
+ }
+
public int getCardinality()
{
return cardinality;
@@ -748,12 +775,11 @@ public class Session implements Serializable
// random node selection for fake load balancing
String currentNode = nodes[Stress.randomizer.nextInt(nodes.length)];
- TSocket socket = new TSocket(currentNode, port);
- TTransport transport = transportFactory.getTransport(socket);
- CassandraClient client = new CassandraClient(new TBinaryProtocol(transport));
-
try
{
+ TTransport transport = transportFactory.openTransport(currentNode, port);
+ CassandraClient client = new CassandraClient(new TBinaryProtocol(transport));
+
if (!transport.isOpen())
transport.open();
@@ -771,6 +797,7 @@ public class Session implements Serializable
AuthenticationRequest authenticationRequest = new AuthenticationRequest(credentials);
client.login(authenticationRequest);
}
+ return client;
}
catch (AuthenticationException e)
{
@@ -788,8 +815,6 @@ public class Session implements Serializable
{
throw new RuntimeException(e.getMessage());
}
-
- return client;
}
public SimpleClient getNativeClient()
[11/11] git commit: fix minor compilation problem on CFS
Posted by ja...@apache.org.
fix minor compilation problem on CFS
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/811de14e
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/811de14e
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/811de14e
Branch: refs/heads/trunk
Commit: 811de14eccbcadffe3dba740d8747fd8eac0248f
Parents: d8e9bf1
Author: Jason Brown <ja...@gmail.com>
Authored: Fri Jan 31 06:09:20 2014 -0800
Committer: Jason Brown <ja...@gmail.com>
Committed: Fri Jan 31 06:11:32 2014 -0800
----------------------------------------------------------------------
src/java/org/apache/cassandra/db/ColumnFamilyStore.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/811de14e/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 76772f4..ec902cf 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -491,7 +491,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
}
- logger.debug("Removing compacted SSTable files from {} (see http://wiki.apache.org/cassandra/MemtableSSTable)", columnFamily);
+ logger.debug("Removing compacted SSTable files from {} (see http://wiki.apache.org/cassandra/MemtableSSTable)", metadata.cfName);
for (Map.Entry<Descriptor,Set<Component>> sstableFiles : directories.sstableLister().list().entrySet())
{
[08/11] git commit: Fix typo in changelog
Posted by ja...@apache.org.
Fix typo in changelog
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/df86e8fb
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/df86e8fb
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/df86e8fb
Branch: refs/heads/trunk
Commit: df86e8fb10d191280ece762bbe6294ccef6ad48d
Parents: 4cb045a
Author: Sylvain Lebresne <sy...@datastax.com>
Authored: Fri Jan 31 13:28:08 2014 +0100
Committer: Sylvain Lebresne <sy...@datastax.com>
Committed: Fri Jan 31 13:28:08 2014 +0100
----------------------------------------------------------------------
CHANGES.txt | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/df86e8fb/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index d85d3a4..68bed3b 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -13,7 +13,7 @@
* Add ability to throttle batchlog replay (CASSANDRA-6550)
* Fix executing LOCAL_QUORUM with SimpleStrategy (CASSANDRA-6545)
* Avoid StackOverflow when using large IN queries (CASSANDRA-6567)
- * Nodetool upgradesstables includes secondary indexes (CASSANDRA-6589)
+ * Nodetool upgradesstables includes secondary indexes (CASSANDRA-6598)
* Paginate batchlog replay (CASSANDRA-6569)
* skip blocking on streaming during drain (CASSANDRA-6603)
* Improve error message when schema doesn't match loaded sstable (CASSANDRA-6262)
[06/11] git commit: Improve CQL3 documentation section on COMPACT
STORAGE (CASSANDRA-6642)
Posted by ja...@apache.org.
Improve CQL3 documentation section on COMPACT STORAGE (CASSANDRA-6642)
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/4cb045a5
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/4cb045a5
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/4cb045a5
Branch: refs/heads/trunk
Commit: 4cb045a589e5807818a20e2b264a266f0cccccd7
Parents: 6a93144
Author: Sylvain Lebresne <sy...@datastax.com>
Authored: Fri Jan 31 12:58:16 2014 +0100
Committer: Sylvain Lebresne <sy...@datastax.com>
Committed: Fri Jan 31 12:58:16 2014 +0100
----------------------------------------------------------------------
doc/cql3/CQL.textile | 10 +++++-----
1 file changed, 5 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/4cb045a5/doc/cql3/CQL.textile
----------------------------------------------------------------------
diff --git a/doc/cql3/CQL.textile b/doc/cql3/CQL.textile
index b02d24a..8b6cb08 100644
--- a/doc/cql3/CQL.textile
+++ b/doc/cql3/CQL.textile
@@ -277,7 +277,7 @@ CREATE TABLE t (
Moreover, a table must define at least one column that is not part of the PRIMARY KEY as a row exists in Cassandra only if it contains at least one value for one such column.
-h4(#createTablepartitionClustering). Partition key and clustering
+h4(#createTablepartitionClustering). Partition key and clustering columns
In CQL, the order in which columns are defined for the @PRIMARY KEY@ matters. The first column of the key is called the __partition key__. It has the property that all the rows sharing the same partition key (even across table in fact) are stored on the same physical node. Also, insertion/update/deletion on rows sharing the same partition key for a given table are performed __atomically__ and in __isolation__. Note that it is possible to have a composite partition key, i.e. a partition key formed of multiple columns, using an extra set of parentheses to define which columns forms the partition key.
@@ -287,9 +287,9 @@ h4(#createTableOptions). @<option>@
The @CREATE TABLE@ statement supports a number of options that controls the configuration of a new table. These options can be specified after the @WITH@ keyword.
-The first of these option is @COMPACT STORAGE@. This option is meanly targeted towards backward compatibility with some table definition created before CQL3. But it also provides a slightly more compact layout of data on disk, though at the price of flexibility and extensibility, and for that reason is not recommended unless for the backward compatibility reason. The restriction for table with @COMPACT STORAGE@ is that they support one and only one column outside of the ones part of the @PRIMARY KEY@. It also follows that columns cannot be added nor removed after creation. A table with @COMPACT STORAGE@ must also define at least one "clustering key":createTablepartitionClustering.
+The first of these option is @COMPACT STORAGE@. This option is mainly targeted towards backward compatibility for definitions created before CQL3 (see "www.datastax.com/dev/blog/thrift-to-cql3":http://www.datastax.com/dev/blog/thrift-to-cql3 for more details). The option also provides a slightly more compact layout of data on disk but at the price of diminished flexibility and extensibility for the table. Most notably, @COMPACT STORAGE@ tables cannot have collections and a @COMPACT STORAGE@ table with at least one clustering column supports exactly one (as in not 0 nor more than 1) column not part of the @PRIMARY KEY@ definition (which imply in particular that you cannot add nor remove columns after creation). For those reasons, @COMPACT STORAGE@ is not recommended outside of the backward compatibility reason evoked above.
-Another option is @CLUSTERING ORDER@. It allows to define the ordering of rows on disk. It takes the list of the clustering key names with, for each of them, the on-disk order (Ascending or descending). Note that this option affects "what @ORDER BY@ are allowed during @SELECT@":#selectOrderBy.
+Another option is @CLUSTERING ORDER@. It allows to define the ordering of rows on disk. It takes the list of the clustering column names with, for each of them, the on-disk order (Ascending or descending). Note that this option affects "what @ORDER BY@ are allowed during @SELECT@":#selectOrderBy.
Table creation supports the following other @<property>@:
@@ -668,7 +668,7 @@ Moreover, the @IN@ relation is only allowed on the last column of the partition
h4(#selectOrderBy). @<order-by>@
The @ORDER BY@ option allows to select the order of the returned results. It takes as argument a list of column names along with the order for the column (@ASC@ for ascendant and @DESC@ for descendant, omitting the order being equivalent to @ASC@). Currently the possible orderings are limited (which depends on the table "@CLUSTERING ORDER@":#createTableOptions):
-* if the table has been defined without any specific @CLUSTERING ORDER@, then then allowed orderings are the order induced by the clustering key and the reverse of that one.
+* if the table has been defined without any specific @CLUSTERING ORDER@, then then allowed orderings are the order induced by the clustering columns and the reverse of that one.
* otherwise, the orderings allowed are the order of the @CLUSTERING ORDER@ option and the reversed one.
h4(#selectLimit). @LIMIT@
@@ -942,7 +942,7 @@ CREATE TABLE users (
...
)
-then the @token@ function will take a single argument of type @text@ (in that case, the partition key is @userid@ (there is no clustering key so the partition key is the same than the primary key)), and the return type will be @bigint@.
+then the @token@ function will take a single argument of type @text@ (in that case, the partition key is @userid@ (there is no clustering columns so the partition key is the same than the primary key)), and the return type will be @bigint@.
h3(#timeuuidFun). Timeuuid functions
[10/11] git commit: Merge branch 'cassandra-2.0' into trunk
Posted by ja...@apache.org.
Merge branch 'cassandra-2.0' into trunk
Conflicts:
src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
tools/stress/src/org/apache/cassandra/stress/Session.java
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/d8e9bf17
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/d8e9bf17
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/d8e9bf17
Branch: refs/heads/trunk
Commit: d8e9bf172f459caed01fadd90be40a9f0857d60b
Parents: 4b54b8a f6a7a5b
Author: Jason Brown <ja...@gmail.com>
Authored: Fri Jan 31 05:46:23 2014 -0800
Committer: Jason Brown <ja...@gmail.com>
Committed: Fri Jan 31 05:46:23 2014 -0800
----------------------------------------------------------------------
.rat-excludes | 2 +-
CHANGES.txt | 5 +-
doc/cql3/CQL.textile | 10 +-
.../org/apache/cassandra/cli/CliOptions.java | 45 ++++++-
.../cql3/statements/BatchStatement.java | 5 +-
.../cql3/statements/ModificationStatement.java | 6 +-
.../cql3/statements/SelectStatement.java | 11 +-
.../apache/cassandra/db/ColumnFamilyStore.java | 32 ++++-
.../apache/cassandra/io/sstable/Descriptor.java | 20 ++-
.../apache/cassandra/io/sstable/SSTable.java | 2 +-
.../org/apache/cassandra/io/util/FileUtils.java | 2 +-
.../cassandra/streaming/StreamLockfile.java | 121 +++++++++++++++++++
.../cassandra/streaming/StreamReader.java | 5 +-
.../cassandra/streaming/StreamReceiveTask.java | 72 ++++++++---
.../cassandra/streaming/StreamSession.java | 27 ++++-
.../cassandra/streaming/StreamTransferTask.java | 10 +-
.../compress/CompressedStreamReader.java | 4 +-
.../streaming/messages/FileMessage.java | 112 -----------------
.../streaming/messages/IncomingFileMessage.java | 78 ++++++++++++
.../streaming/messages/OutgoingFileMessage.java | 92 ++++++++++++++
.../streaming/messages/StreamMessage.java | 18 ++-
.../cassandra/db/BatchlogManagerTest.java | 17 +++
22 files changed, 528 insertions(+), 168 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d8e9bf17/CHANGES.txt
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d8e9bf17/doc/cql3/CQL.textile
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d8e9bf17/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d8e9bf17/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d8e9bf17/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d8e9bf17/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 540f1ce,8750026..76772f4
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@@ -74,6 -67,8 +75,9 @@@ import org.apache.cassandra.io.util.Fil
import org.apache.cassandra.metrics.ColumnFamilyMetrics;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.service.StorageService;
++
+ import org.apache.cassandra.streaming.StreamLockfile;
+ import org.apache.cassandra.thrift.IndexExpression;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.*;
@@@ -460,11 -424,35 +464,35 @@@ public class ColumnFamilyStore implemen
* Removes unnecessary files from the cf directory at startup: these include temp files, orphans, zero-length files
* and compacted sstables. Files that cannot be recognized will be ignored.
*/
- public static void scrubDataDirectories(String keyspaceName, String columnFamily)
+ public static void scrubDataDirectories(CFMetaData metadata)
{
- logger.debug("Removing compacted SSTable files from {} (see http://wiki.apache.org/cassandra/MemtableSSTable)", metadata.cfName);
-
- Directories directories = Directories.create(keyspaceName, columnFamily);
+ Directories directories = new Directories(metadata);
+
+ // remove any left-behind SSTables from failed/stalled streaming
+ FileFilter filter = new FileFilter()
+ {
+ public boolean accept(File pathname)
+ {
+ return pathname.toString().endsWith(StreamLockfile.FILE_EXT);
+ }
+ };
+ for (File dir : directories.getCFDirectories())
+ {
+ File[] lockfiles = dir.listFiles(filter);
+ if (lockfiles.length == 0)
+ continue;
+ logger.info("Removing SSTables from failed streaming session. Found {} files to cleanup.", lockfiles.length);
+
+ for (File lockfile : lockfiles)
+ {
+ StreamLockfile streamLockfile = new StreamLockfile(lockfile);
+ streamLockfile.cleanup();
+ streamLockfile.delete();
+ }
+ }
+
+ logger.debug("Removing compacted SSTable files from {} (see http://wiki.apache.org/cassandra/MemtableSSTable)", columnFamily);
+
for (Map.Entry<Descriptor,Set<Component>> sstableFiles : directories.sstableLister().list().entrySet())
{
Descriptor desc = sstableFiles.getKey();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d8e9bf17/src/java/org/apache/cassandra/io/sstable/Descriptor.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d8e9bf17/src/java/org/apache/cassandra/io/sstable/SSTable.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d8e9bf17/src/java/org/apache/cassandra/io/util/FileUtils.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d8e9bf17/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d8e9bf17/test/unit/org/apache/cassandra/db/BatchlogManagerTest.java
----------------------------------------------------------------------
[05/11] git commit: 6503 followup,
make sure we send CompleteMessage when streaming is done.
Posted by ja...@apache.org.
6503 followup, make sure we send CompleteMessage when streaming is done.
Patch by yukim, reviewed by marcuse for CASSANDRA-6503
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/1141cdb0
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/1141cdb0
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/1141cdb0
Branch: refs/heads/trunk
Commit: 1141cdb0c8122d3e990f29dc82576bededcbfd40
Parents: a8a12d9
Author: Marcus Eriksson <ma...@apache.org>
Authored: Fri Jan 31 11:55:02 2014 +0100
Committer: Marcus Eriksson <ma...@apache.org>
Committed: Fri Jan 31 11:55:02 2014 +0100
----------------------------------------------------------------------
.../org/apache/cassandra/streaming/StreamSession.java | 13 ++++++++++++-
1 file changed, 12 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/1141cdb0/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamSession.java b/src/java/org/apache/cassandra/streaming/StreamSession.java
index 4777995..7972183 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -144,6 +144,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
}
private volatile State state = State.INITIALIZED;
+ private volatile boolean completeSent = false;
/**
* Create new streaming session with the peer.
@@ -505,11 +506,15 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
{
if (state == State.WAIT_COMPLETE)
{
+ if (!completeSent)
+ {
+ handler.sendMessage(new CompleteMessage());
+ completeSent = true;
+ }
closeSession(State.COMPLETE);
}
else
{
- handler.sendMessage(new CompleteMessage());
state(State.WAIT_COMPLETE);
}
}
@@ -594,12 +599,18 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
{
if (state == State.WAIT_COMPLETE)
{
+ if (!completeSent)
+ {
+ handler.sendMessage(new CompleteMessage());
+ completeSent = true;
+ }
closeSession(State.COMPLETE);
}
else
{
// notify peer that this session is completed
handler.sendMessage(new CompleteMessage());
+ completeSent = true;
state(State.WAIT_COMPLETE);
}
}
[03/11] git commit: Fix IllegalArgumentException during statement
preparation
Posted by ja...@apache.org.
Fix IllegalArgumentException during statement preparation
patch by slebresne; reviewed by thobbs for CASSANDRA-6592
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/a8a12d9e
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/a8a12d9e
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/a8a12d9e
Branch: refs/heads/trunk
Commit: a8a12d9e0adee65bad19dcfd86e7b0cf6b1ed89a
Parents: e7b23cc
Author: Sylvain Lebresne <sy...@datastax.com>
Authored: Fri Jan 31 10:09:32 2014 +0100
Committer: Sylvain Lebresne <sy...@datastax.com>
Committed: Fri Jan 31 10:09:32 2014 +0100
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../apache/cassandra/cql3/statements/BatchStatement.java | 5 ++++-
.../cassandra/cql3/statements/ModificationStatement.java | 6 +++++-
.../cassandra/cql3/statements/SelectStatement.java | 11 ++++++++++-
4 files changed, 20 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/a8a12d9e/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 94b21d4..a29f212 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -18,6 +18,7 @@
* SSTableScanner may skip rows during cleanup (CASSANDRA-6638)
* sstables from stalled repair sessions can resurrect deleted data (CASSANDRA-6503)
* Switch stress to use ITransportFactory (CASSANDRA-6641)
+ * Fix IllegalArgumentException during prepare (CASSANDRA-6592)
Merged from 1.2:
* fsync compression metadata (CASSANDRA-6531)
* Validate CF existence on execution for prepared statement (CASSANDRA-6535)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/a8a12d9e/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
index 25f61fb..6151490 100644
--- a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
@@ -66,7 +66,10 @@ public class BatchStatement implements CQLStatement, MeasurableForPreparedCache
public long measureForPreparedCache(MemoryMeter meter)
{
- long size = meter.measure(this) + meter.measure(statements) + meter.measureDeep(attrs);
+ long size = meter.measure(this)
+ + meter.measureDeep(type)
+ + meter.measure(statements)
+ + meter.measureDeep(attrs);
for (ModificationStatement stmt : statements)
size += stmt.measureForPreparedCache(meter);
return size;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/a8a12d9e/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
index 9e0fd62..c0bf428 100644
--- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
@@ -66,7 +66,11 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF
public long measureForPreparedCache(MemoryMeter meter)
{
- return meter.measureDeep(this) - meter.measureDeep(cfm);
+ return meter.measure(this)
+ + meter.measureDeep(attrs)
+ + meter.measureDeep(processedKeys)
+ + meter.measureDeep(columnOperations)
+ + (columnConditions == null ? 0 : meter.measureDeep(columnConditions));
}
public abstract boolean requireFullClusteringKey();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/a8a12d9e/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
index c3671be..29cafa2 100644
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@ -112,7 +112,16 @@ public class SelectStatement implements CQLStatement, MeasurableForPreparedCache
public long measureForPreparedCache(MemoryMeter meter)
{
- return meter.measureDeep(this) - meter.measureDeep(cfDef);
+ return meter.measure(this)
+ + meter.measureDeep(parameters)
+ + meter.measureDeep(selection)
+ + (limit == null ? 0 : meter.measureDeep(limit))
+ + meter.measureDeep(keyRestrictions)
+ + meter.measureDeep(columnRestrictions)
+ + meter.measureDeep(metadataRestrictions)
+ + meter.measureDeep(restrictedNames)
+ + (sliceRestriction == null ? 0 : meter.measureDeep(sliceRestriction))
+ + (orderingIndexes == null ? 0 : meter.measureDeep(orderingIndexes));
}
public int getBoundTerms()