You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by sl...@apache.org on 2013/07/22 15:47:43 UTC
[1/3] git commit: Fix ConcurrentModificationException during streaming
Updated Branches:
refs/heads/trunk abea32e19 -> b7a016d99
Fix ConcurrentModificationException during streaming
patch by slebresne; reviewed by yukim for CASSANDRA-5782
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/c7220e3a
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/c7220e3a
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/c7220e3a
Branch: refs/heads/trunk
Commit: c7220e3a7b8debd8fa348ad258bc0ae8a3cd8509
Parents: abea32e
Author: Sylvain Lebresne <sy...@datastax.com>
Authored: Mon Jul 22 15:43:08 2013 +0200
Committer: Sylvain Lebresne <sy...@datastax.com>
Committed: Mon Jul 22 15:43:08 2013 +0200
----------------------------------------------------------------------
CHANGES.txt | 1 +
src/java/org/apache/cassandra/streaming/StreamSession.java | 5 +++--
src/java/org/apache/cassandra/streaming/StreamTransferTask.java | 4 +++-
3 files changed, 7 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7220e3a/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index bbf2773..4d56c98 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -13,6 +13,7 @@
* CAS on 'primary key only' table (CASSANDRA-5715)
* Support streaming SSTables of old versions (CASSANDRA-5772)
* Always respect protocol version in native protocol (CASSANDRA-5778)
+ * Fix ConcurrentModificationException during streaming (CASSANDRA-5782)
2.0.0-beta1
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7220e3a/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 c61b404..aeb4419 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -620,8 +620,9 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
state(State.STREAMING);
for (StreamTransferTask task : transfers.values())
{
- if (task.getFileMessages().size() > 0)
- handler.sendMessages(task.getFileMessages());
+ Collection<FileMessage> messages = task.getFileMessages();
+ if (messages.size() > 0)
+ handler.sendMessages(messages);
else
taskCompleted(task); // there is no file to send
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7220e3a/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 ba2df03..956692d 100644
--- a/src/java/org/apache/cassandra/streaming/StreamTransferTask.java
+++ b/src/java/org/apache/cassandra/streaming/StreamTransferTask.java
@@ -73,7 +73,9 @@ public class StreamTransferTask extends StreamTask
public Collection<FileMessage> getFileMessages()
{
- return files.values();
+ // 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)
[3/3] git commit: Thrift cas() method crashes if input columns are
not sorted
Posted by sl...@apache.org.
Thrift cas() method crashes if input columns are not sorted
patch by slebresne; reviewed by jbellis for CASSANDRA-5786
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/b7a016d9
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/b7a016d9
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/b7a016d9
Branch: refs/heads/trunk
Commit: b7a016d992254701c5ad8d8d0ec038df4eda5026
Parents: 37130b5
Author: Sylvain Lebresne <sy...@datastax.com>
Authored: Mon Jul 22 15:46:25 2013 +0200
Committer: Sylvain Lebresne <sy...@datastax.com>
Committed: Mon Jul 22 15:46:25 2013 +0200
----------------------------------------------------------------------
CHANGES.txt | 1 +
src/java/org/apache/cassandra/db/ColumnFamily.java | 3 ++-
src/java/org/apache/cassandra/service/paxos/Commit.java | 4 +++-
.../org/apache/cassandra/service/paxos/PrepareResponse.java | 6 ++++--
4 files changed, 10 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b7a016d9/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 20a6373..dec32d7 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -15,6 +15,7 @@
* Always respect protocol version in native protocol (CASSANDRA-5778)
* Fix ConcurrentModificationException during streaming (CASSANDRA-5782)
* Update deletion timestamp in Commit#updatesWithPaxosTime (CASSANDRA-5787)
+ * Thrift cas() method crashes if input columns are not sorted (CASSANDRA-5786)
2.0.0-beta1
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b7a016d9/src/java/org/apache/cassandra/db/ColumnFamily.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamily.java b/src/java/org/apache/cassandra/db/ColumnFamily.java
index 5715058..3bbf533 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamily.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamily.java
@@ -471,6 +471,7 @@ public abstract class ColumnFamily implements Iterable<Column>, IRowCacheEntry
return builder.build();
}
+ // Note: the returned ColumnFamily will be an UnsortedColumns.
public static ColumnFamily fromBytes(ByteBuffer bytes)
{
if (bytes == null)
@@ -478,7 +479,7 @@ public abstract class ColumnFamily implements Iterable<Column>, IRowCacheEntry
try
{
- return serializer.deserialize(new DataInputStream(ByteBufferUtil.inputStream(bytes)), MessagingService.current_version);
+ return serializer.deserialize(new DataInputStream(ByteBufferUtil.inputStream(bytes)), UnsortedColumns.factory, ColumnSerializer.Flag.LOCAL, MessagingService.current_version);
}
catch (IOException e)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b7a016d9/src/java/org/apache/cassandra/service/paxos/Commit.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/paxos/Commit.java b/src/java/org/apache/cassandra/service/paxos/Commit.java
index 04ddca8..84589cc 100644
--- a/src/java/org/apache/cassandra/service/paxos/Commit.java
+++ b/src/java/org/apache/cassandra/service/paxos/Commit.java
@@ -32,8 +32,10 @@ import com.google.common.base.Objects;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.Column;
import org.apache.cassandra.db.ColumnFamily;
+import org.apache.cassandra.db.ColumnSerializer;
import org.apache.cassandra.db.EmptyColumns;
import org.apache.cassandra.db.RowMutation;
+import org.apache.cassandra.db.UnsortedColumns;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.UUIDGen;
@@ -139,7 +141,7 @@ public class Commit
{
return new Commit(ByteBufferUtil.readWithShortLength(in),
UUIDSerializer.serializer.deserialize(in, version),
- ColumnFamily.serializer.deserialize(in, version));
+ ColumnFamily.serializer.deserialize(in, UnsortedColumns.factory, ColumnSerializer.Flag.LOCAL, version));
}
public long serializedSize(Commit commit, int version)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b7a016d9/src/java/org/apache/cassandra/service/paxos/PrepareResponse.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/paxos/PrepareResponse.java b/src/java/org/apache/cassandra/service/paxos/PrepareResponse.java
index 246e09a..9f5fda6 100644
--- a/src/java/org/apache/cassandra/service/paxos/PrepareResponse.java
+++ b/src/java/org/apache/cassandra/service/paxos/PrepareResponse.java
@@ -27,6 +27,8 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import org.apache.cassandra.db.ColumnFamily;
+import org.apache.cassandra.db.ColumnSerializer;
+import org.apache.cassandra.db.UnsortedColumns;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.UUIDSerializer;
@@ -74,10 +76,10 @@ public class PrepareResponse
return new PrepareResponse(success,
new Commit(key,
UUIDSerializer.serializer.deserialize(in, version),
- ColumnFamily.serializer.deserialize(in, version)),
+ ColumnFamily.serializer.deserialize(in, UnsortedColumns.factory, ColumnSerializer.Flag.LOCAL, version)),
new Commit(key,
UUIDSerializer.serializer.deserialize(in, version),
- ColumnFamily.serializer.deserialize(in, version)));
+ ColumnFamily.serializer.deserialize(in, UnsortedColumns.factory, ColumnSerializer.Flag.LOCAL, version)));
}
public long serializedSize(PrepareResponse response, int version)
[2/3] git commit: Update deletion timestamps in
Commit#updatesWithPaxosTime
Posted by sl...@apache.org.
Update deletion timestamps in Commit#updatesWithPaxosTime
patch by slebresne; reviewed by jbellis for CASSANDRA-5787
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/37130b59
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/37130b59
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/37130b59
Branch: refs/heads/trunk
Commit: 37130b5998e060b52c379fbd726d61e3e7926a3c
Parents: c7220e3
Author: Sylvain Lebresne <sy...@datastax.com>
Authored: Mon Jul 22 15:44:55 2013 +0200
Committer: Sylvain Lebresne <sy...@datastax.com>
Committed: Mon Jul 22 15:44:55 2013 +0200
----------------------------------------------------------------------
CHANGES.txt | 1 +
src/java/org/apache/cassandra/db/DeletionInfo.java | 10 ++++++++++
src/java/org/apache/cassandra/db/RangeTombstoneList.java | 6 ++++++
src/java/org/apache/cassandra/service/paxos/Commit.java | 1 +
4 files changed, 18 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/37130b59/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 4d56c98..20a6373 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -14,6 +14,7 @@
* Support streaming SSTables of old versions (CASSANDRA-5772)
* Always respect protocol version in native protocol (CASSANDRA-5778)
* Fix ConcurrentModificationException during streaming (CASSANDRA-5782)
+ * Update deletion timestamp in Commit#updatesWithPaxosTime (CASSANDRA-5787)
2.0.0-beta1
http://git-wip-us.apache.org/repos/asf/cassandra/blob/37130b59/src/java/org/apache/cassandra/db/DeletionInfo.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/DeletionInfo.java b/src/java/org/apache/cassandra/db/DeletionInfo.java
index 49aa427..32cc2a7 100644
--- a/src/java/org/apache/cassandra/db/DeletionInfo.java
+++ b/src/java/org/apache/cassandra/db/DeletionInfo.java
@@ -265,6 +265,16 @@ public class DeletionInfo
return sb.toString();
}
+ // Updates all the timestamp of the deletion contained in this DeletionInfo to be {@code timestamp}.
+ public void updateAllTimestamp(long timestamp)
+ {
+ if (topLevel.markedForDeleteAt != Long.MIN_VALUE)
+ topLevel = new DeletionTime(timestamp, topLevel.localDeletionTime);
+
+ if (ranges != null)
+ ranges.updateAllTimestamp(timestamp);
+ }
+
@Override
public boolean equals(Object o)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/37130b59/src/java/org/apache/cassandra/db/RangeTombstoneList.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/RangeTombstoneList.java b/src/java/org/apache/cassandra/db/RangeTombstoneList.java
index 771b1a8..e99ed9f 100644
--- a/src/java/org/apache/cassandra/db/RangeTombstoneList.java
+++ b/src/java/org/apache/cassandra/db/RangeTombstoneList.java
@@ -288,6 +288,12 @@ public class RangeTombstoneList implements Iterable<RangeTombstone>
return max;
}
+ public void updateAllTimestamp(long timestamp)
+ {
+ for (int i = 0; i < size; i++)
+ markedAts[i] = timestamp;
+ }
+
/**
* Removes all range tombstones whose local deletion time is older than gcBefore.
*/
http://git-wip-us.apache.org/repos/asf/cassandra/blob/37130b59/src/java/org/apache/cassandra/service/paxos/Commit.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/paxos/Commit.java b/src/java/org/apache/cassandra/service/paxos/Commit.java
index fdc707b..04ddca8 100644
--- a/src/java/org/apache/cassandra/service/paxos/Commit.java
+++ b/src/java/org/apache/cassandra/service/paxos/Commit.java
@@ -114,6 +114,7 @@ public class Commit
{
ColumnFamily cf = updates.cloneMeShallow();
long t = UUIDGen.microsTimestamp(ballot);
+ cf.deletionInfo().updateAllTimestamp(t);
for (Column column : updates)
cf.addAtom(column.withUpdatedTimestamp(t));
return cf;