You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jb...@apache.org on 2013/04/26 21:07:04 UTC
[3/4] Removed compatibility with pre-1.2.5 sstables and network
messages patch by jbellis; reviewed by marcuse for CASSANDRA-5511
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java b/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java
index e19ddcb..4f93e34 100644
--- a/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java
@@ -38,7 +38,6 @@ import org.apache.cassandra.db.marshal.CompositeType;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.io.util.FileDataInput;
-import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.tracing.Tracing;
public class SliceQueryFilter implements IDiskAtomFilter
@@ -50,8 +49,6 @@ public class SliceQueryFilter implements IDiskAtomFilter
public final boolean reversed;
public volatile int count;
public final int compositesToGroup;
- // This is a hack to allow rolling upgrade with pre-1.2 nodes
- private final int countMutliplierForCompatibility;
// Not serialized, just a ack for range slices to find the number of live column counted, even when we group
private ColumnCounter columnCounter;
@@ -63,7 +60,7 @@ public class SliceQueryFilter implements IDiskAtomFilter
public SliceQueryFilter(ByteBuffer start, ByteBuffer finish, boolean reversed, int count, int compositesToGroup)
{
- this(new ColumnSlice[] { new ColumnSlice(start, finish) }, reversed, count, compositesToGroup, 1);
+ this(new ColumnSlice[] { new ColumnSlice(start, finish) }, reversed, count, compositesToGroup);
}
/**
@@ -72,36 +69,35 @@ public class SliceQueryFilter implements IDiskAtomFilter
*/
public SliceQueryFilter(ColumnSlice[] slices, boolean reversed, int count)
{
- this(slices, reversed, count, -1, 1);
+ this(slices, reversed, count, -1);
}
- public SliceQueryFilter(ColumnSlice[] slices, boolean reversed, int count, int compositesToGroup, int countMutliplierForCompatibility)
+ public SliceQueryFilter(ColumnSlice[] slices, boolean reversed, int count, int compositesToGroup)
{
this.slices = slices;
this.reversed = reversed;
this.count = count;
this.compositesToGroup = compositesToGroup;
- this.countMutliplierForCompatibility = countMutliplierForCompatibility;
}
public SliceQueryFilter cloneShallow()
{
- return new SliceQueryFilter(slices, reversed, count, compositesToGroup, countMutliplierForCompatibility);
+ return new SliceQueryFilter(slices, reversed, count, compositesToGroup);
}
public SliceQueryFilter withUpdatedCount(int newCount)
{
- return new SliceQueryFilter(slices, reversed, newCount, compositesToGroup, countMutliplierForCompatibility);
+ return new SliceQueryFilter(slices, reversed, newCount, compositesToGroup);
}
public SliceQueryFilter withUpdatedSlices(ColumnSlice[] newSlices)
{
- return new SliceQueryFilter(newSlices, reversed, count, compositesToGroup, countMutliplierForCompatibility);
+ return new SliceQueryFilter(newSlices, reversed, count, compositesToGroup);
}
public SliceQueryFilter withUpdatedSlice(ByteBuffer start, ByteBuffer finish)
{
- return new SliceQueryFilter(new ColumnSlice[]{ new ColumnSlice(start, finish) }, reversed, count, compositesToGroup, countMutliplierForCompatibility);
+ return new SliceQueryFilter(new ColumnSlice[]{ new ColumnSlice(start, finish) }, reversed, count, compositesToGroup);
}
public OnDiskAtomIterator getMemtableColumnIterator(ColumnFamily cf, DecoratedKey key)
@@ -241,50 +237,28 @@ public class SliceQueryFilter implements IDiskAtomFilter
{
public void serialize(SliceQueryFilter f, DataOutput out, int version) throws IOException
{
- if (version < MessagingService.VERSION_12)
- {
- // It's kind of lame, but probably better than throwing an exception
- ColumnSlice slice = new ColumnSlice(f.start(), f.finish());
+ out.writeInt(f.slices.length);
+ for (ColumnSlice slice : f.slices)
ColumnSlice.serializer.serialize(slice, out, version);
- }
- else
- {
- out.writeInt(f.slices.length);
- for (ColumnSlice slice : f.slices)
- ColumnSlice.serializer.serialize(slice, out, version);
- }
out.writeBoolean(f.reversed);
int count = f.count;
- if (f.compositesToGroup > 0 && version < MessagingService.VERSION_12)
- count *= f.countMutliplierForCompatibility;
out.writeInt(count);
- if (version < MessagingService.VERSION_12)
- return;
-
out.writeInt(f.compositesToGroup);
}
public SliceQueryFilter deserialize(DataInput in, int version) throws IOException
{
ColumnSlice[] slices;
- if (version < MessagingService.VERSION_12)
- {
- slices = new ColumnSlice[]{ ColumnSlice.serializer.deserialize(in, version) };
- }
- else
- {
- slices = new ColumnSlice[in.readInt()];
- for (int i = 0; i < slices.length; i++)
- slices[i] = ColumnSlice.serializer.deserialize(in, version);
- }
+ slices = new ColumnSlice[in.readInt()];
+ for (int i = 0; i < slices.length; i++)
+ slices[i] = ColumnSlice.serializer.deserialize(in, version);
boolean reversed = in.readBoolean();
int count = in.readInt();
int compositesToGroup = -1;
- if (version >= MessagingService.VERSION_12)
- compositesToGroup = in.readInt();
+ compositesToGroup = in.readInt();
- return new SliceQueryFilter(slices, reversed, count, compositesToGroup, 1);
+ return new SliceQueryFilter(slices, reversed, count, compositesToGroup);
}
public long serializedSize(SliceQueryFilter f, int version)
@@ -292,21 +266,13 @@ public class SliceQueryFilter implements IDiskAtomFilter
TypeSizes sizes = TypeSizes.NATIVE;
int size = 0;
- if (version < MessagingService.VERSION_12)
- {
- size += ColumnSlice.serializer.serializedSize(new ColumnSlice(f.start(), f.finish()), version);
- }
- else
- {
- size += sizes.sizeof(f.slices.length);
- for (ColumnSlice slice : f.slices)
- size += ColumnSlice.serializer.serializedSize(slice, version);
- }
+ size += sizes.sizeof(f.slices.length);
+ for (ColumnSlice slice : f.slices)
+ size += ColumnSlice.serializer.serializedSize(slice, version);
size += sizes.sizeof(f.reversed);
size += sizes.sizeof(f.count);
- if (version >= MessagingService.VERSION_12)
- size += sizes.sizeof(f.compositesToGroup);
+ size += sizes.sizeof(f.compositesToGroup);
return size;
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/dht/AbstractBounds.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/dht/AbstractBounds.java b/src/java/org/apache/cassandra/dht/AbstractBounds.java
index efa886c..5e0e619 100644
--- a/src/java/org/apache/cassandra/dht/AbstractBounds.java
+++ b/src/java/org/apache/cassandra/dht/AbstractBounds.java
@@ -129,11 +129,6 @@ public abstract class AbstractBounds<T extends RingPosition> implements Serializ
{
public void serialize(AbstractBounds<?> range, DataOutput out, int version) throws IOException
{
- // Older version don't know how to handle abstract bounds of keys
- // However, the serialization has been designed so that token bounds are serialized the same way that before 1.1
- if (version < MessagingService.VERSION_11)
- range = range.toTokenBounds();
-
/*
* The first int tells us if it's a range or bounds (depending on the value) _and_ if it's tokens or keys (depending on the
* sign). We use negative kind for keys so as to preserve the serialization of token from older version.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/dht/AbstractByteOrderedPartitioner.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/dht/AbstractByteOrderedPartitioner.java b/src/java/org/apache/cassandra/dht/AbstractByteOrderedPartitioner.java
index d24977f..d037518 100644
--- a/src/java/org/apache/cassandra/dht/AbstractByteOrderedPartitioner.java
+++ b/src/java/org/apache/cassandra/dht/AbstractByteOrderedPartitioner.java
@@ -45,11 +45,6 @@ public abstract class AbstractByteOrderedPartitioner extends AbstractPartitioner
return new DecoratedKey(getToken(key), key);
}
- public DecoratedKey convertFromDiskFormat(ByteBuffer key)
- {
- return new DecoratedKey(getToken(key), key);
- }
-
public BytesToken midpoint(Token ltoken, Token rtoken)
{
int ll,rl;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/dht/IPartitioner.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/dht/IPartitioner.java b/src/java/org/apache/cassandra/dht/IPartitioner.java
index 053a95f..084a1e5 100644
--- a/src/java/org/apache/cassandra/dht/IPartitioner.java
+++ b/src/java/org/apache/cassandra/dht/IPartitioner.java
@@ -27,15 +27,6 @@ import org.apache.cassandra.db.marshal.AbstractType;
public interface IPartitioner<T extends Token>
{
/**
- * @deprecated Used by SSTables before version 'e'.
- *
- * Convert the on disk representation to a DecoratedKey object
- * @param key On disk representation
- * @return DecoratedKey object
- */
- public DecoratedKey convertFromDiskFormat(ByteBuffer key);
-
- /**
* Transform key to object representation of the on-disk format.
*
* @param key the raw, client-facing key
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/dht/LocalPartitioner.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/dht/LocalPartitioner.java b/src/java/org/apache/cassandra/dht/LocalPartitioner.java
index 51e8f74..b24eede 100644
--- a/src/java/org/apache/cassandra/dht/LocalPartitioner.java
+++ b/src/java/org/apache/cassandra/dht/LocalPartitioner.java
@@ -35,11 +35,6 @@ public class LocalPartitioner extends AbstractPartitioner<LocalToken>
this.comparator = comparator;
}
- public DecoratedKey convertFromDiskFormat(ByteBuffer key)
- {
- return decorateKey(key);
- }
-
public DecoratedKey decorateKey(ByteBuffer key)
{
return new DecoratedKey(getToken(key), key);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/dht/Murmur3Partitioner.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/dht/Murmur3Partitioner.java b/src/java/org/apache/cassandra/dht/Murmur3Partitioner.java
index a969320..f1411a4 100644
--- a/src/java/org/apache/cassandra/dht/Murmur3Partitioner.java
+++ b/src/java/org/apache/cassandra/dht/Murmur3Partitioner.java
@@ -41,11 +41,6 @@ public class Murmur3Partitioner extends AbstractPartitioner<LongToken>
public static final LongToken MINIMUM = new LongToken(Long.MIN_VALUE);
public static final long MAXIMUM = Long.MAX_VALUE;
- public DecoratedKey convertFromDiskFormat(ByteBuffer key)
- {
- throw new UnsupportedOperationException();
- }
-
public DecoratedKey decorateKey(ByteBuffer key)
{
return new DecoratedKey(getToken(key), key);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java b/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java
index a5c194b..57c7297 100644
--- a/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java
+++ b/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java
@@ -44,11 +44,6 @@ public class OrderPreservingPartitioner extends AbstractPartitioner<StringToken>
return new DecoratedKey(getToken(key), key);
}
- public DecoratedKey convertFromDiskFormat(ByteBuffer key)
- {
- return new DecoratedKey(getToken(key), key);
- }
-
public StringToken midpoint(Token ltoken, Token rtoken)
{
int sigchars = Math.max(((StringToken)ltoken).token.length(), ((StringToken)rtoken).token.length());
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/dht/RandomPartitioner.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/dht/RandomPartitioner.java b/src/java/org/apache/cassandra/dht/RandomPartitioner.java
index 26b399e..194f12e 100644
--- a/src/java/org/apache/cassandra/dht/RandomPartitioner.java
+++ b/src/java/org/apache/cassandra/dht/RandomPartitioner.java
@@ -20,7 +20,6 @@ package org.apache.cassandra.dht;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.ByteBuffer;
-import java.nio.charset.CharacterCodingException;
import java.util.*;
import org.apache.cassandra.exceptions.ConfigurationException;
@@ -48,35 +47,6 @@ public class RandomPartitioner extends AbstractPartitioner<BigIntegerToken>
return new DecoratedKey(getToken(key), key);
}
- public DecoratedKey convertFromDiskFormat(ByteBuffer fromdisk)
- {
- // find the delimiter position
- int splitPoint = -1;
- for (int i = fromdisk.position(); i < fromdisk.limit(); i++)
- {
- if (fromdisk.get(i) == DELIMITER_BYTE)
- {
- splitPoint = i;
- break;
- }
- }
- assert splitPoint != -1;
-
- // and decode the token and key
- String token = null;
- try
- {
- token = ByteBufferUtil.string(fromdisk, fromdisk.position(), splitPoint - fromdisk.position());
- }
- catch (CharacterCodingException e)
- {
- throw new RuntimeException(e);
- }
- ByteBuffer key = fromdisk.duplicate();
- key.position(splitPoint + 1);
- return new DecoratedKey(new BigIntegerToken(token), key);
- }
-
public Token midpoint(Token ltoken, Token rtoken)
{
// the symbolic MINIMUM token should act as ZERO: the empty bit array
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/gms/GossipDigestAck.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/gms/GossipDigestAck.java b/src/java/org/apache/cassandra/gms/GossipDigestAck.java
index 61244c8..e7a0906 100644
--- a/src/java/org/apache/cassandra/gms/GossipDigestAck.java
+++ b/src/java/org/apache/cassandra/gms/GossipDigestAck.java
@@ -63,8 +63,6 @@ class GossipDigestAckSerializer implements IVersionedSerializer<GossipDigestAck>
public void serialize(GossipDigestAck gDigestAckMessage, DataOutput out, int version) throws IOException
{
GossipDigestSerializationHelper.serialize(gDigestAckMessage.gDigestList, out, version);
- if (version < MessagingService.VERSION_12)
- out.writeBoolean(true); // 0.6 compatibility
out.writeInt(gDigestAckMessage.epStateMap.size());
for (Map.Entry<InetAddress, EndpointState> entry : gDigestAckMessage.epStateMap.entrySet())
{
@@ -77,8 +75,6 @@ class GossipDigestAckSerializer implements IVersionedSerializer<GossipDigestAck>
public GossipDigestAck deserialize(DataInput in, int version) throws IOException
{
List<GossipDigest> gDigestList = GossipDigestSerializationHelper.deserialize(in, version);
- if (version < MessagingService.VERSION_12)
- in.readBoolean(); // 0.6 compatibility
int size = in.readInt();
Map<InetAddress, EndpointState> epStateMap = new HashMap<InetAddress, EndpointState>(size);
@@ -94,8 +90,6 @@ class GossipDigestAckSerializer implements IVersionedSerializer<GossipDigestAck>
public long serializedSize(GossipDigestAck ack, int version)
{
int size = GossipDigestSerializationHelper.serializedSize(ack.gDigestList, version);
- if (version < MessagingService.VERSION_12)
- size += TypeSizes.NATIVE.sizeof(true);
size += TypeSizes.NATIVE.sizeof(ack.epStateMap.size());
for (Map.Entry<InetAddress, EndpointState> entry : ack.epStateMap.entrySet())
size += CompactEndpointSerializationHelper.serializedSize(entry.getKey())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/gms/GossipDigestSyn.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/gms/GossipDigestSyn.java b/src/java/org/apache/cassandra/gms/GossipDigestSyn.java
index fce764b..cf196ff 100644
--- a/src/java/org/apache/cassandra/gms/GossipDigestSyn.java
+++ b/src/java/org/apache/cassandra/gms/GossipDigestSyn.java
@@ -82,8 +82,7 @@ class GossipDigestSynSerializer implements IVersionedSerializer<GossipDigestSyn>
public void serialize(GossipDigestSyn gDigestSynMessage, DataOutput out, int version) throws IOException
{
out.writeUTF(gDigestSynMessage.clusterId);
- if (version >= MessagingService.VERSION_12)
- out.writeUTF(gDigestSynMessage.partioner);
+ out.writeUTF(gDigestSynMessage.partioner);
GossipDigestSerializationHelper.serialize(gDigestSynMessage.gDigests, out, version);
}
@@ -91,8 +90,7 @@ class GossipDigestSynSerializer implements IVersionedSerializer<GossipDigestSyn>
{
String clusterId = in.readUTF();
String partioner = null;
- if (version >= MessagingService.VERSION_12)
- partioner = in.readUTF();
+ partioner = in.readUTF();
List<GossipDigest> gDigests = GossipDigestSerializationHelper.deserialize(in, version);
return new GossipDigestSyn(clusterId, partioner, gDigests);
}
@@ -100,8 +98,7 @@ class GossipDigestSynSerializer implements IVersionedSerializer<GossipDigestSyn>
public long serializedSize(GossipDigestSyn syn, int version)
{
long size = TypeSizes.NATIVE.sizeof(syn.clusterId);
- if (version >= MessagingService.VERSION_12)
- size += TypeSizes.NATIVE.sizeof(syn.partioner);
+ size += TypeSizes.NATIVE.sizeof(syn.partioner);
size += GossipDigestSerializationHelper.serializedSize(syn.gDigests, version);
return size;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/gms/Gossiper.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java
index 88cd76c..7314429 100644
--- a/src/java/org/apache/cassandra/gms/Gossiper.java
+++ b/src/java/org/apache/cassandra/gms/Gossiper.java
@@ -647,9 +647,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
public boolean usesHostId(InetAddress endpoint)
{
- if (MessagingService.instance().knowsVersion(endpoint) && MessagingService.instance().getVersion(endpoint) >= MessagingService.VERSION_12)
+ if (MessagingService.instance().knowsVersion(endpoint))
return true;
- else if (getEndpointStateForEndpoint(endpoint).getApplicationState(ApplicationState.NET_VERSION) != null && Integer.parseInt(getEndpointStateForEndpoint(endpoint).getApplicationState(ApplicationState.NET_VERSION).value) >= MessagingService.VERSION_12)
+ else if (getEndpointStateForEndpoint(endpoint).getApplicationState(ApplicationState.NET_VERSION) != null)
return true;
return false;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/gms/VersionedValue.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/gms/VersionedValue.java b/src/java/org/apache/cassandra/gms/VersionedValue.java
index c6e726d..b4cfa41 100644
--- a/src/java/org/apache/cassandra/gms/VersionedValue.java
+++ b/src/java/org/apache/cassandra/gms/VersionedValue.java
@@ -251,35 +251,7 @@ public class VersionedValue implements Comparable<VersionedValue>
private String outValue(VersionedValue value, int version)
{
- String outValue = value.value;
-
- if (version < MessagingService.VERSION_12)
- {
- String[] pieces = value.value.split(DELIMITER_STR, -1);
- String type = pieces[0];
-
- if ((type.equals(STATUS_NORMAL)) || type.equals(STATUS_BOOTSTRAPPING))
- {
- assert pieces.length >= 2;
- outValue = versionString(pieces[0], pieces[1]);
- }
-
- if (type.equals(STATUS_LEFT))
- {
- assert pieces.length >= 3;
-
- // three component 'left' was adopted starting from Cassandra 1.0
- // previous versions have '<type>:<token>' format
- outValue = (version < MessagingService.VERSION_10)
- ? versionString(pieces[0], pieces[2])
- : versionString(pieces[0], pieces[2], pieces[1]);
- }
-
- if ((type.equals(REMOVAL_COORDINATOR)) || (type.equals(REMOVING_TOKEN)) || (type.equals(REMOVED_TOKEN)))
- throw new RuntimeException(String.format("Unable to serialize %s(%s...) for nodes older than 1.2",
- VersionedValue.class.getName(), type));
- }
- return outValue;
+ return value.value;
}
public VersionedValue deserialize(DataInput in, int version) throws IOException
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/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 3e21025..51256ff 100644
--- a/src/java/org/apache/cassandra/io/sstable/Descriptor.java
+++ b/src/java/org/apache/cassandra/io/sstable/Descriptor.java
@@ -22,7 +22,6 @@ import java.util.StringTokenizer;
import com.google.common.base.Objects;
-import org.apache.cassandra.utils.FilterFactory;
import org.apache.cassandra.utils.Pair;
import static org.apache.cassandra.io.sstable.Component.separator;
@@ -41,31 +40,12 @@ public class Descriptor
// or have their size changed.
//
// Minor versions were introduced with version "hb" for Cassandra 1.0.3; prior to that,
- // we always incremented the major version. In particular, versions g and h are
- // forwards-compatible with version f, so if the above convention had been followed,
- // we would have labeled them fb and fc.
+ // we always incremented the major version.
public static class Version
{
// This needs to be at the begining for initialization sake
public static final String current_version = "ja";
- public static final Version LEGACY = new Version("a"); // "pre-history"
- // b (0.7.0): added version to sstable filenames
- // c (0.7.0): bloom filter component computes hashes over raw key bytes instead of strings
- // d (0.7.0): row size in data component becomes a long instead of int
- // e (0.7.0): stores undecorated keys in data and index components
- // f (0.7.0): switched bloom filter implementations in data component
- // g (0.8): tracks flushed-at context in metadata component
- // h (1.0): tracks max client timestamp in metadata component
- // hb (1.0.3): records compression ration in metadata component
- // hc (1.0.4): records partitioner in metadata component
- // hd (1.0.10): includes row tombstones in maxtimestamp
- // he (1.1.3): includes ancestors generation in metadata component
- // hf (1.1.6): marker that replay position corresponds to 1.1.5+ millis-based id (see CASSANDRA-4782)
- // ia (1.2.0): column indexes are promoted to the index file
- // records estimated histogram of deletion times in tombstones
- // bloom filter (keys and columns) upgraded to Murmur3
- // ib (1.2.1): tracks min client timestamp in metadata component
// ic (1.2.5): omits per-row bloom filter of column names
// ja (2.0.0): super columns are serialized as composites (note that there is no real format change,
// this is mostly a marker to know if we should expect super columns or not. We do need
@@ -78,51 +58,18 @@ public class Descriptor
private final String version;
- public final boolean hasStringsInBloomFilter;
- public final boolean hasIntRowSize;
- public final boolean hasEncodedKeys;
public final boolean isLatestVersion;
- public final boolean metadataIncludesReplayPosition;
- public final boolean metadataIncludesModernReplayPosition;
- public final boolean tracksMaxTimestamp;
- public final boolean tracksMinTimestamp;
- public final boolean hasCompressionRatio;
- public final boolean hasPartitioner;
- public final boolean tracksTombstones;
- public final boolean hasPromotedIndexes;
- public final FilterFactory.Type filterType;
- public final boolean hasAncestors;
public final boolean hasSuperColumns;
public final boolean tracksMaxLocalDeletionTime;
public final boolean hasBloomFilterFPChance;
- public final boolean hasRowLevelBF;
public Version(String version)
{
this.version = version;
- hasStringsInBloomFilter = version.compareTo("c") < 0;
- hasIntRowSize = version.compareTo("d") < 0;
- hasEncodedKeys = version.compareTo("e") < 0;
- metadataIncludesReplayPosition = version.compareTo("g") >= 0;
- hasCompressionRatio = version.compareTo("hb") >= 0;
- hasPartitioner = version.compareTo("hc") >= 0;
- tracksMaxTimestamp = version.compareTo("hd") >= 0;
- tracksMinTimestamp = version.compareTo("ib") >= 0;
tracksMaxLocalDeletionTime = version.compareTo("ja") >= 0;
- hasAncestors = version.compareTo("he") >= 0;
- metadataIncludesModernReplayPosition = version.compareTo("hf") >= 0;
- tracksTombstones = version.compareTo("ia") >= 0;
- hasPromotedIndexes = version.compareTo("ia") >= 0;
isLatestVersion = version.compareTo(current_version) == 0;
- if (version.compareTo("f") < 0)
- filterType = FilterFactory.Type.SHA;
- else if (version.compareTo("ia") < 0)
- filterType = FilterFactory.Type.MURMUR2;
- else
- filterType = FilterFactory.Type.MURMUR3;
hasSuperColumns = version.compareTo("ja") < 0;
hasBloomFilterFPChance = version.compareTo("ja") >= 0;
- hasRowLevelBF = version.compareTo("ic") < 0;
}
/**
@@ -137,25 +84,12 @@ public class Descriptor
public boolean isCompatible()
{
- return version.charAt(0) <= CURRENT.version.charAt(0);
+ return version.compareTo("ic") >= 0 && version.charAt(0) <= CURRENT.version.charAt(0);
}
public boolean isStreamCompatible()
{
- // we could add compatibility for earlier versions with the new single-pass streaming
- // (see SSTableWriter.appendFromStream) but versions earlier than 0.7.1 don't have the
- // MessagingService version awareness anyway so there's no point.
- return isCompatible() && version.charAt(0) >= 'i';
- }
-
- /**
- * Versions [h..hc] contained a timestamp value that was computed incorrectly, ignoring row tombstones.
- * containsTimestamp returns true if there is a timestamp value in the metadata file; to know if it
- * actually contains a *correct* timestamp, see tracksMaxTimestamp.
- */
- public boolean containsTimestamp()
- {
- return version.compareTo("h") >= 0;
+ return isCompatible() && version.charAt(0) >= 'j';
}
@Override
@@ -233,8 +167,7 @@ public class Descriptor
buff.append(cfname).append(separator);
if (temporary)
buff.append(SSTable.TEMPFILE_MARKER).append(separator);
- if (!Version.LEGACY.equals(version))
- buff.append(version).append(separator);
+ buff.append(version).append(separator);
buff.append(generation);
return buff.toString();
}
@@ -286,13 +219,11 @@ public class Descriptor
nexttok = st.nextToken();
}
- // optional version string
- Version version = Version.LEGACY;
- if (Version.validate(nexttok))
- {
- version = new Version(nexttok);
- nexttok = st.nextToken();
- }
+ if (!Version.validate(nexttok))
+ throw new UnsupportedOperationException("SSTable " + name + " is too old to open. Upgrade to 1.2.5 first, and run upgradesstables");
+ Version version = new Version(nexttok);
+
+ nexttok = st.nextToken();
int generation = Integer.parseInt(nexttok);
// component suffix
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/io/sstable/KeyIterator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/KeyIterator.java b/src/java/org/apache/cassandra/io/sstable/KeyIterator.java
index 9fe1309..0af4537 100644
--- a/src/java/org/apache/cassandra/io/sstable/KeyIterator.java
+++ b/src/java/org/apache/cassandra/io/sstable/KeyIterator.java
@@ -47,8 +47,8 @@ public class KeyIterator extends AbstractIterator<DecoratedKey> implements Close
{
if (in.isEOF())
return endOfData();
- DecoratedKey key = SSTableReader.decodeKey(StorageService.getPartitioner(), desc, ByteBufferUtil.readWithShortLength(in));
- RowIndexEntry.serializer.skip(in, desc.version); // skip remainder of the entry
+ DecoratedKey key = StorageService.getPartitioner().decorateKey(ByteBufferUtil.readWithShortLength(in));
+ RowIndexEntry.serializer.skip(in); // skip remainder of the entry
return key;
}
catch (IOException e)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/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 f5786cf..c225687 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTable.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTable.java
@@ -244,7 +244,7 @@ public abstract class SSTable
while (ifile.getFilePointer() < BYTES_CAP && keys < SAMPLES_CAP)
{
ByteBufferUtil.skipShortLength(ifile);
- RowIndexEntry.serializer.skip(ifile, descriptor.version);
+ RowIndexEntry.serializer.skip(ifile);
keys++;
}
assert keys > 0 && ifile.getFilePointer() > 0 && ifile.length() > 0 : "Unexpected empty index file: " + ifile;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
index 7c3f146..8170b29 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
@@ -122,38 +122,8 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
if (dataStart + dataSize > file.length())
throw new IOException(String.format("dataSize of %s starting at %s would be larger than file %s length %s",
dataSize, dataStart, file.getPath(), file.length()));
- if (checkData && !dataVersion.hasPromotedIndexes)
- {
- try
- {
- IndexHelper.skipBloomFilter(file);
- }
- catch (Exception e)
- {
- if (e instanceof EOFException)
- throw (EOFException) e;
-
- logger.debug("Invalid bloom filter in {}; will rebuild it", sstable);
- }
- try
- {
- // skipping the old row-level BF should have left the file position ready to deserialize index
- IndexHelper.deserializeIndex(file);
- }
- catch (Exception e)
- {
- logger.debug("Invalid row summary in {}; will rebuild it", sstable);
- }
- file.seek(this.dataStart);
- inputWithTracker.reset(0);
- }
}
- if (sstable != null && !dataVersion.hasPromotedIndexes)
- {
- IndexHelper.skipBloomFilter(inputWithTracker);
- IndexHelper.skipIndex(inputWithTracker);
- }
columnFamily = EmptyColumns.factory.create(metadata);
columnFamily.delete(DeletionInfo.serializer().deserializeFromSSTable(inputWithTracker, dataVersion));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/io/sstable/SSTableMetadata.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableMetadata.java b/src/java/org/apache/cassandra/io/sstable/SSTableMetadata.java
index aa75d75..ba6bfb5 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableMetadata.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableMetadata.java
@@ -356,29 +356,19 @@ public class SSTableMetadata
{
EstimatedHistogram.serializer.serialize(sstableStats.estimatedRowSize, out);
EstimatedHistogram.serializer.serialize(sstableStats.estimatedColumnCount, out);
- if (legacyDesc.version.metadataIncludesReplayPosition)
- ReplayPosition.serializer.serialize(sstableStats.replayPosition, out);
- if (legacyDesc.version.tracksMinTimestamp)
- out.writeLong(sstableStats.minTimestamp);
- if (legacyDesc.version.tracksMaxTimestamp)
- out.writeLong(sstableStats.maxTimestamp);
+ ReplayPosition.serializer.serialize(sstableStats.replayPosition, out);
+ out.writeLong(sstableStats.minTimestamp);
+ out.writeLong(sstableStats.maxTimestamp);
if (legacyDesc.version.tracksMaxLocalDeletionTime)
out.writeInt(sstableStats.maxLocalDeletionTime);
if (legacyDesc.version.hasBloomFilterFPChance)
out.writeDouble(sstableStats.bloomFilterFPChance);
- if (legacyDesc.version.hasCompressionRatio)
- out.writeDouble(sstableStats.compressionRatio);
- if (legacyDesc.version.hasPartitioner)
- out.writeUTF(sstableStats.partitioner);
- if (legacyDesc.version.hasAncestors)
- {
- out.writeInt(sstableStats.ancestors.size());
- for (Integer g : sstableStats.ancestors)
- out.writeInt(g);
- }
- if (legacyDesc.version.tracksTombstones)
- StreamingHistogram.serializer.serialize(sstableStats.estimatedTombstoneDropTime, out);
-
+ out.writeDouble(sstableStats.compressionRatio);
+ out.writeUTF(sstableStats.partitioner);
+ out.writeInt(sstableStats.ancestors.size());
+ for (Integer g : sstableStats.ancestors)
+ out.writeInt(g);
+ StreamingHistogram.serializer.serialize(sstableStats.estimatedTombstoneDropTime, out);
out.writeInt(sstableStats.sstableLevel);
}
@@ -416,30 +406,18 @@ public class SSTableMetadata
{
EstimatedHistogram rowSizes = EstimatedHistogram.serializer.deserialize(in);
EstimatedHistogram columnCounts = EstimatedHistogram.serializer.deserialize(in);
- ReplayPosition replayPosition = desc.version.metadataIncludesReplayPosition
- ? ReplayPosition.serializer.deserialize(in)
- : ReplayPosition.NONE;
- if (!desc.version.metadataIncludesModernReplayPosition)
- {
- // replay position may be "from the future" thanks to older versions generating them with nanotime.
- // make sure we don't omit replaying something that we should. see CASSANDRA-4782
- replayPosition = ReplayPosition.NONE;
- }
- long minTimestamp = desc.version.tracksMinTimestamp ? in.readLong() : Long.MIN_VALUE;
- long maxTimestamp = desc.version.containsTimestamp() ? in.readLong() : Long.MAX_VALUE;
- if (!desc.version.tracksMaxTimestamp) // see javadoc to Descriptor.containsTimestamp
- maxTimestamp = Long.MAX_VALUE;
+ ReplayPosition replayPosition = ReplayPosition.serializer.deserialize(in);
+ long minTimestamp = in.readLong();
+ long maxTimestamp = in.readLong();
int maxLocalDeletionTime = desc.version.tracksMaxLocalDeletionTime ? in.readInt() : Integer.MAX_VALUE;
double bloomFilterFPChance = desc.version.hasBloomFilterFPChance ? in.readDouble() : NO_BLOOM_FLITER_FP_CHANCE;
- double compressionRatio = desc.version.hasCompressionRatio ? in.readDouble() : NO_COMPRESSION_RATIO;
- String partitioner = desc.version.hasPartitioner ? in.readUTF() : null;
- int nbAncestors = desc.version.hasAncestors ? in.readInt() : 0;
+ double compressionRatio = in.readDouble();
+ String partitioner = in.readUTF();
+ int nbAncestors = in.readInt();
Set<Integer> ancestors = new HashSet<Integer>(nbAncestors);
for (int i = 0; i < nbAncestors; i++)
ancestors.add(in.readInt());
- StreamingHistogram tombstoneHistogram = desc.version.tracksTombstones
- ? StreamingHistogram.serializer.deserialize(in)
- : defaultTombstoneDropTimeHistogram();
+ StreamingHistogram tombstoneHistogram = StreamingHistogram.serializer.deserialize(in);
int sstableLevel = 0;
if (loadSSTableLevel && in.available() > 0)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
index f92e9ec..4e89b37 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
@@ -325,11 +325,6 @@ public class SSTableReader extends SSTable
// bf is enabled, but filter component is missing.
load(true);
}
- else if (descriptor.version.hasStringsInBloomFilter)
- {
- // versions before 'c' encoded keys as utf-16 before hashing to the filter.
- load(true);
- }
else if (descriptor.version.hasBloomFilterFPChance && sstableMetadata.bloomFilterFPChance != metadata.getBloomFilterFpChance())
{
// bf fp chance in sstable metadata and it has changed since compaction.
@@ -349,7 +344,7 @@ public class SSTableReader extends SSTable
try
{
stream = new DataInputStream(new BufferedInputStream(new FileInputStream(descriptor.filenameFor(Component.FILTER))));
- bf = FilterFactory.deserialize(stream, descriptor.version.filterType, true);
+ bf = FilterFactory.deserialize(stream, true);
}
finally
{
@@ -393,7 +388,7 @@ public class SSTableReader extends SSTable
{
ByteBuffer key = ByteBufferUtil.readWithShortLength(primaryIndex);
RowIndexEntry indexEntry = RowIndexEntry.serializer.deserialize(primaryIndex, descriptor.version);
- DecoratedKey decoratedKey = decodeKey(partitioner, descriptor, key);
+ DecoratedKey decoratedKey = partitioner.decorateKey(key);
if (first == null)
first = decoratedKey;
last = decoratedKey;
@@ -444,8 +439,8 @@ public class SSTableReader extends SSTable
FileUtils.deleteWithConfirm(summariesFile);
return false;
}
- reader.first = decodeKey(reader.partitioner, reader.descriptor, ByteBufferUtil.readWithLength(iStream));
- reader.last = decodeKey(reader.partitioner, reader.descriptor, ByteBufferUtil.readWithLength(iStream));
+ reader.first = reader.partitioner.decorateKey(ByteBufferUtil.readWithLength(iStream));
+ reader.last = reader.partitioner.decorateKey(ByteBufferUtil.readWithLength(iStream));
ibuilder.deserializeBounds(iStream);
dbuilder.deserializeBounds(iStream);
}
@@ -537,7 +532,7 @@ public class SSTableReader extends SSTable
*/
public void forceFilterFailures()
{
- bf = LegacyBloomFilter.alwaysMatchingBloomFilter();
+ bf = new AlwaysPresentFilter();
}
public IFilter getBloomFilter()
@@ -547,7 +542,7 @@ public class SSTableReader extends SSTable
public long getBloomFilterSerializedSize()
{
- return FilterFactory.serializedSize(bf, descriptor.version.filterType);
+ return FilterFactory.serializedSize(bf);
}
/**
@@ -847,7 +842,7 @@ public class SSTableReader extends SSTable
}
else
{
- DecoratedKey indexDecoratedKey = decodeKey(partitioner, descriptor, indexKey);
+ DecoratedKey indexDecoratedKey = partitioner.decorateKey(indexKey);
int comparison = indexDecoratedKey.compareTo(key);
int v = op.apply(comparison);
opSatisfied = (v == 0);
@@ -872,7 +867,7 @@ public class SSTableReader extends SSTable
{
// expensive sanity check! see CASSANDRA-4687
FileDataInput fdi = dfile.getSegment(indexEntry.position);
- DecoratedKey keyInDisk = SSTableReader.decodeKey(partitioner, descriptor, ByteBufferUtil.readWithShortLength(fdi));
+ DecoratedKey keyInDisk = partitioner.decorateKey(ByteBufferUtil.readWithShortLength(fdi));
if (!keyInDisk.equals(key))
throw new AssertionError(String.format("%s != %s in %s", keyInDisk, key, fdi.getPath()));
fdi.close();
@@ -887,7 +882,7 @@ public class SSTableReader extends SSTable
return indexEntry;
}
- RowIndexEntry.serializer.skip(in, descriptor.version);
+ RowIndexEntry.serializer.skip(in);
}
}
catch (IOException e)
@@ -1045,13 +1040,6 @@ public class SSTableReader extends SSTable
return maxDataAge > age;
}
- public static long readRowSize(DataInput in, Descriptor d) throws IOException
- {
- if (d.version.hasIntRowSize)
- return in.readInt();
- return in.readLong();
- }
-
public void createLinks(String snapshotDirectoryPath)
{
for (Component component : components)
@@ -1063,21 +1051,6 @@ public class SSTableReader extends SSTable
}
/**
- * Conditionally use the deprecated 'IPartitioner.convertFromDiskFormat' method.
- */
- public static DecoratedKey decodeKey(IPartitioner p, Descriptor d, ByteBuffer bytes)
- {
- if (d.version.hasEncodedKeys)
- return p.convertFromDiskFormat(bytes);
- return p.decorateKey(bytes);
- }
-
- public DecoratedKey decodeKey(ByteBuffer bytes)
- {
- return decodeKey(partitioner, descriptor, bytes);
- }
-
- /**
* TODO: Move someplace reusable
*/
public abstract static class Operator
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
index 949acda..3089769 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
@@ -85,7 +85,7 @@ public class SSTableScanner implements ICompactionScanner
while (!ifile.isEOF())
{
long startPosition = ifile.getFilePointer();
- DecoratedKey indexDecoratedKey = sstable.decodeKey(ByteBufferUtil.readWithShortLength(ifile));
+ DecoratedKey indexDecoratedKey = sstable.partitioner.decorateKey(ByteBufferUtil.readWithShortLength(ifile));
int comparison = indexDecoratedKey.compareTo(seekKey);
if (comparison >= 0)
{
@@ -98,7 +98,7 @@ public class SSTableScanner implements ICompactionScanner
}
else
{
- RowIndexEntry.serializer.skip(ifile, sstable.descriptor.version);
+ RowIndexEntry.serializer.skip(ifile);
}
}
exhausted = true;
@@ -169,8 +169,8 @@ public class SSTableScanner implements ICompactionScanner
assert !dfile.isEOF();
// Read data header
- DecoratedKey key = sstable.decodeKey(ByteBufferUtil.readWithShortLength(dfile));
- long dataSize = SSTableReader.readRowSize(dfile, sstable.descriptor);
+ DecoratedKey key = sstable.partitioner.decorateKey(ByteBufferUtil.readWithShortLength(dfile));
+ long dataSize = dfile.readLong();
long dataStart = dfile.getFilePointer();
finishedAt = dataStart + dataSize;
@@ -217,7 +217,7 @@ public class SSTableScanner implements ICompactionScanner
if (row == null)
{
- currentKey = sstable.decodeKey(ByteBufferUtil.readWithShortLength(ifile));
+ currentKey = sstable.partitioner.decorateKey(ByteBufferUtil.readWithShortLength(ifile));
currentEntry = RowIndexEntry.serializer.deserialize(ifile, sstable.descriptor.version);
}
else
@@ -233,7 +233,7 @@ public class SSTableScanner implements ICompactionScanner
}
else
{
- nextKey = sstable.decodeKey(ByteBufferUtil.readWithShortLength(ifile));
+ nextKey = sstable.partitioner.decorateKey(ByteBufferUtil.readWithShortLength(ifile));
nextEntry = RowIndexEntry.serializer.deserialize(ifile, sstable.descriptor.version);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
index fb77c94..0cbafda 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
@@ -452,7 +452,7 @@ public class SSTableWriter extends SSTable
// bloom filter
FileOutputStream fos = new FileOutputStream(path);
DataOutputStream stream = new DataOutputStream(fos);
- FilterFactory.serialize(bf, stream, descriptor.version.filterType);
+ FilterFactory.serialize(bf, stream);
stream.flush();
fos.getFD().sync();
stream.close();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/net/IncomingTcpConnection.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/IncomingTcpConnection.java b/src/java/org/apache/cassandra/net/IncomingTcpConnection.java
index c5e9a52..809d8c9 100644
--- a/src/java/org/apache/cassandra/net/IncomingTcpConnection.java
+++ b/src/java/org/apache/cassandra/net/IncomingTcpConnection.java
@@ -138,34 +138,7 @@ public class IncomingTcpConnection extends Thread
private void handleLegacyVersion(int version) throws IOException
{
- DataInputStream in = new DataInputStream(new BufferedInputStream(socket.getInputStream(), 4096));
-
- from = receiveMessage(in, version); // why? see => CASSANDRA-4099
- logger.debug("Version for {} is {}", from, version);
- if (version > MessagingService.current_version)
- {
- // save the endpoint so gossip will reconnect to it
- Gossiper.instance.addSavedEndpoint(from);
- logger.info("Received messages from newer protocol version. Ignoring");
- return;
- }
- int lastVersion = MessagingService.instance().setVersion(from, version);
- logger.debug("set version for {} to {}", from, version);
- if (lastVersion < version)
- {
- logger.debug("breaking outbound connections to force version upgrade");
- MessagingService.instance().getConnectionPool(from).resetToNewerVersion(version);
- }
-
- while (true)
- {
- MessagingService.validateMagic(in.readInt());
- int header = in.readInt(); // legacy protocol re-sends header for each message
- assert !(MessagingService.getBits(header, 3, 1) == 1) : "Non-stream connection cannot change to stream";
- version = MessagingService.getBits(header, 15, 8);
- logger.trace("Version is now {}", version);
- receiveMessage(in, version);
- }
+ throw new UnsupportedOperationException("Unable to read obsolete message version " + version + "; the earliest version supported is 1.2.0");
}
private void handleStream(DataInputStream input, int version) throws IOException
@@ -187,9 +160,6 @@ public class IncomingTcpConnection extends Thread
private InetAddress receiveMessage(DataInputStream input, int version) throws IOException
{
- if (version < MessagingService.VERSION_12)
- input.readInt(); // size of entire message. in 1.0+ this is just a placeholder
-
int id;
if (version < MessagingService.VERSION_20)
id = Integer.valueOf(input.readUTF());
@@ -197,13 +167,10 @@ public class IncomingTcpConnection extends Thread
id = input.readInt();
long timestamp = System.currentTimeMillis();
- if (version >= MessagingService.VERSION_12)
- {
- // make sure to readInt, even if cross_node_to is not enabled
- int partial = input.readInt();
- if (DatabaseDescriptor.hasCrossNodeTimeout())
- timestamp = (timestamp & 0xFFFFFFFF00000000L) | (((partial & 0xFFFFFFFFL) << 2) >> 2);
- }
+ // make sure to readInt, even if cross_node_to is not enabled
+ int partial = input.readInt();
+ if (DatabaseDescriptor.hasCrossNodeTimeout())
+ timestamp = (timestamp & 0xFFFFFFFF00000000L) | (((partial & 0xFFFFFFFFL) << 2) >> 2);
MessageIn message = MessageIn.read(input, version, id);
if (message == null)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/net/MessagingService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java
index 32b0705..40cf665 100644
--- a/src/java/org/apache/cassandra/net/MessagingService.java
+++ b/src/java/org/apache/cassandra/net/MessagingService.java
@@ -73,9 +73,6 @@ public final class MessagingService implements MessagingServiceMBean
public static final String MBEAN_NAME = "org.apache.cassandra.net:type=MessagingService";
// 8 bits version, so don't waste versions
- public static final int VERSION_10 = 3;
- public static final int VERSION_11 = 4;
- public static final int VERSION_117 = 5;
public static final int VERSION_12 = 6;
public static final int VERSION_20 = 7;
public static final int current_version = VERSION_20;
@@ -209,7 +206,6 @@ public final class MessagingService implements MessagingServiceMBean
put(Verb.GOSSIP_DIGEST_SYN, GossipDigestSyn.serializer);
put(Verb.DEFINITIONS_UPDATE, MigrationManager.MigrationsSerializer.instance);
put(Verb.TRUNCATE, Truncation.serializer);
- put(Verb.INDEX_SCAN, IndexScanCommand.serializer);
put(Verb.REPLICATION_FINISHED, null);
put(Verb.COUNTER_MUTATION, CounterMutation.serializer);
put(Verb.SNAPSHOT, SnapshotCommand.serializer);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/net/OutboundTcpConnection.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/OutboundTcpConnection.java b/src/java/org/apache/cassandra/net/OutboundTcpConnection.java
index c55d64f..de53580 100644
--- a/src/java/org/apache/cassandra/net/OutboundTcpConnection.java
+++ b/src/java/org/apache/cassandra/net/OutboundTcpConnection.java
@@ -222,24 +222,15 @@ public class OutboundTcpConnection extends Thread
private void writeInternal(MessageOut message, int id, long timestamp) throws IOException
{
out.writeInt(MessagingService.PROTOCOL_MAGIC);
- if (targetVersion < MessagingService.VERSION_12)
- {
- writeHeader(out, targetVersion, false);
- // 0.8 included a total message size int. 1.0 doesn't need it but expects it to be there.
- out.writeInt(-1);
- }
if (targetVersion < MessagingService.VERSION_20)
out.writeUTF(String.valueOf(id));
else
out.writeInt(id);
- if (targetVersion >= MessagingService.VERSION_12)
- {
- // int cast cuts off the high-order half of the timestamp, which we can assume remains
- // the same between now and when the recipient reconstructs it.
- out.writeInt((int) timestamp);
- }
+ // int cast cuts off the high-order half of the timestamp, which we can assume remains
+ // the same between now and when the recipient reconstructs it.
+ out.writeInt((int) timestamp);
message.serialize(out, targetVersion);
}
@@ -311,38 +302,35 @@ public class OutboundTcpConnection extends Thread
}
out = new DataOutputStream(new BufferedOutputStream(socket.getOutputStream(), 4096));
- if (targetVersion >= MessagingService.VERSION_12)
- {
- out.writeInt(MessagingService.PROTOCOL_MAGIC);
- writeHeader(out, targetVersion, shouldCompressConnection());
- out.flush();
+ out.writeInt(MessagingService.PROTOCOL_MAGIC);
+ writeHeader(out, targetVersion, shouldCompressConnection());
+ out.flush();
- DataInputStream in = new DataInputStream(socket.getInputStream());
- int maxTargetVersion = in.readInt();
- if (targetVersion > maxTargetVersion)
- {
- logger.debug("Target max version is {}; will reconnect with that version", maxTargetVersion);
- MessagingService.instance().setVersion(poolReference.endPoint(), maxTargetVersion);
- disconnect();
- return false;
- }
+ DataInputStream in = new DataInputStream(socket.getInputStream());
+ int maxTargetVersion = in.readInt();
+ if (targetVersion > maxTargetVersion)
+ {
+ logger.debug("Target max version is {}; will reconnect with that version", maxTargetVersion);
+ MessagingService.instance().setVersion(poolReference.endPoint(), maxTargetVersion);
+ disconnect();
+ return false;
+ }
- if (targetVersion < maxTargetVersion && targetVersion < MessagingService.current_version)
- {
- logger.trace("Detected higher max version {} (using {}); will reconnect when queued messages are done",
- maxTargetVersion, targetVersion);
- MessagingService.instance().setVersion(poolReference.endPoint(), Math.min(MessagingService.current_version, maxTargetVersion));
- softCloseSocket();
- }
+ if (targetVersion < maxTargetVersion && targetVersion < MessagingService.current_version)
+ {
+ logger.trace("Detected higher max version {} (using {}); will reconnect when queued messages are done",
+ maxTargetVersion, targetVersion);
+ MessagingService.instance().setVersion(poolReference.endPoint(), Math.min(MessagingService.current_version, maxTargetVersion));
+ softCloseSocket();
+ }
- out.writeInt(MessagingService.current_version);
- CompactEndpointSerializationHelper.serialize(FBUtilities.getBroadcastAddress(), out);
- if (shouldCompressConnection())
- {
- out.flush();
- logger.trace("Upgrading OutputStream to be compressed");
- out = new DataOutputStream(new SnappyOutputStream(new BufferedOutputStream(socket.getOutputStream())));
- }
+ out.writeInt(MessagingService.current_version);
+ CompactEndpointSerializationHelper.serialize(FBUtilities.getBroadcastAddress(), out);
+ if (shouldCompressConnection())
+ {
+ out.flush();
+ logger.trace("Upgrading OutputStream to be compressed");
+ out = new DataOutputStream(new SnappyOutputStream(new BufferedOutputStream(socket.getOutputStream())));
}
return true;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/service/ActiveRepairService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/ActiveRepairService.java b/src/java/org/apache/cassandra/service/ActiveRepairService.java
index f7a5fa4..b692ab0 100644
--- a/src/java/org/apache/cassandra/service/ActiveRepairService.java
+++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java
@@ -658,12 +658,6 @@ public class ActiveRepairService
logger.error(String.format("[repair #%s] ", getName()) + message);
throw new IOException(message);
}
-
- if (MessagingService.instance().getVersion(endpoint) < MessagingService.VERSION_11 && isSequential)
- {
- logger.info(String.format("[repair #%s] Cannot repair using snapshots as node %s is pre-1.1", getName(), endpoint));
- return;
- }
}
ActiveRepairService.instance.sessions.put(getName(), this);
@@ -1003,9 +997,7 @@ public class ActiveRepairService
};
StreamingRepairTask task = StreamingRepairTask.create(r1.endpoint, r2.endpoint, tablename, cfname, differences, callback);
- // Pre 1.0, nodes don't know how to handle forwarded streaming task so don't bother
- if (task.isLocalTask() || MessagingService.instance().getVersion(task.dst) >= MessagingService.VERSION_10)
- task.run();
+ task.run();
}
public String toString()
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/service/CacheService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/CacheService.java b/src/java/org/apache/cassandra/service/CacheService.java
index c927bfb..8396d61 100644
--- a/src/java/org/apache/cassandra/service/CacheService.java
+++ b/src/java/org/apache/cassandra/service/CacheService.java
@@ -351,9 +351,7 @@ public class CacheService implements CacheServiceMBean
ByteBufferUtil.writeWithLength(key.key, out);
Descriptor desc = key.desc;
out.writeInt(desc.generation);
- out.writeBoolean(desc.version.hasPromotedIndexes);
- if (!desc.version.hasPromotedIndexes)
- return;
+ out.writeBoolean(true);
RowIndexEntry.serializer.serialize(entry, out);
}
@@ -368,10 +366,8 @@ public class CacheService implements CacheServiceMBean
return null;
}
RowIndexEntry entry;
- if (input.readBoolean())
- entry = RowIndexEntry.serializer.deserialize(input, reader.descriptor.version);
- else
- entry = reader.getPosition(reader.partitioner.decorateKey(key), Operator.EQ);
+ input.readBoolean(); // backwards compatibility for "promoted indexes" boolean
+ entry = RowIndexEntry.serializer.deserialize(input, reader.descriptor.version);
return Futures.immediateFuture(Pair.create(new KeyCacheKey(reader.descriptor, key), entry));
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/service/CassandraDaemon.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java
index 1d6cb0c..3d5240f 100644
--- a/src/java/org/apache/cassandra/service/CassandraDaemon.java
+++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java
@@ -207,10 +207,6 @@ public class CassandraDaemon
: String.format("Directory %s is not accessible.", dataDir);
}
- // Migrate sstables from pre-#2749 to the correct location
- if (Directories.sstablesNeedsMigration())
- Directories.migrateSSTables();
-
if (CacheService.instance == null) // should never happen
throw new RuntimeException("Failed to initialize Cache Service.");
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/service/IndexScanVerbHandler.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/IndexScanVerbHandler.java b/src/java/org/apache/cassandra/service/IndexScanVerbHandler.java
deleted file mode 100644
index 2e3a921..0000000
--- a/src/java/org/apache/cassandra/service/IndexScanVerbHandler.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.cassandra.service;
-
-import java.util.List;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.cassandra.db.*;
-import org.apache.cassandra.net.IVerbHandler;
-import org.apache.cassandra.net.MessageIn;
-import org.apache.cassandra.net.MessagingService;
-import org.apache.cassandra.thrift.ThriftValidation;
-import org.apache.cassandra.tracing.Tracing;
-
-@Deprecated // 1.1 implements index scan with RangeSliceVerb instead
-public class IndexScanVerbHandler implements IVerbHandler<IndexScanCommand>
-{
- private static final Logger logger = LoggerFactory.getLogger(IndexScanVerbHandler.class);
-
- public void doVerb(MessageIn<IndexScanCommand> message, int id)
- {
- try
- {
- IndexScanCommand command = message.payload;
- ColumnFamilyStore cfs = Table.open(command.keyspace).getColumnFamilyStore(command.column_family);
- List<Row> rows = cfs.search(command.index_clause.expressions,
- command.range,
- command.index_clause.count,
- ThriftValidation.asIFilter(command.predicate, cfs.metadata, null));
- RangeSliceReply reply = new RangeSliceReply(rows);
- Tracing.trace("Enqueuing response to {}", message.from);
- MessagingService.instance().sendReply(reply.createMessage(), id, message.from);
- }
- catch (Exception ex)
- {
- throw new RuntimeException(ex);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/service/MigrationManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/MigrationManager.java b/src/java/org/apache/cassandra/service/MigrationManager.java
index d02226a..4b5352c 100644
--- a/src/java/org/apache/cassandra/service/MigrationManager.java
+++ b/src/java/org/apache/cassandra/service/MigrationManager.java
@@ -112,10 +112,6 @@ public class MigrationManager implements IEndpointStateChangeSubscriber
*/
private static void maybeScheduleSchemaPull(final UUID theirVersion, final InetAddress endpoint)
{
- // Can't request migrations from nodes with versions younger than 1.1.7
- if (MessagingService.instance().getVersion(endpoint) < MessagingService.VERSION_117)
- return;
-
if (Gossiper.instance.isFatClient(endpoint))
return;
@@ -305,11 +301,7 @@ public class MigrationManager implements IEndpointStateChangeSubscriber
for (InetAddress endpoint : Gossiper.instance.getLiveMembers())
{
if (endpoint.equals(FBUtilities.getBroadcastAddress()))
- continue; // we've delt with localhost already
-
- // don't send migrations to the nodes with the versions older than < 1.2
- if (MessagingService.instance().getVersion(endpoint) < MessagingService.VERSION_12)
- continue;
+ continue; // we've dealt with localhost already
pushSchemaMutation(endpoint, schema);
}
@@ -360,37 +352,16 @@ public class MigrationManager implements IEndpointStateChangeSubscriber
// and due to broken timestamps in versions prior to 1.1.7
for (InetAddress node : liveEndpoints)
{
- if (MessagingService.instance().getVersion(node) >= MessagingService.VERSION_117)
- {
- if (logger.isDebugEnabled())
- logger.debug("Requesting schema from " + node);
+ if (logger.isDebugEnabled())
+ logger.debug("Requesting schema from " + node);
- FBUtilities.waitOnFuture(StageManager.getStage(Stage.MIGRATION).submit(new MigrationTask(node)));
- break;
- }
+ FBUtilities.waitOnFuture(StageManager.getStage(Stage.MIGRATION).submit(new MigrationTask(node)));
+ break;
}
logger.info("Local schema reset is complete.");
}
- /**
- * Used only in case node has old style migration schema (newly updated)
- * @return the UUID identifying version of the last applied migration
- */
- @Deprecated
- public static UUID getLastMigrationId()
- {
- DecoratedKey dkey = StorageService.getPartitioner().decorateKey(LAST_MIGRATION_KEY);
- Table defs = Table.open(Table.SYSTEM_KS);
- ColumnFamilyStore cfStore = defs.getColumnFamilyStore(DefsTable.OLD_SCHEMA_CF);
- QueryFilter filter = QueryFilter.getNamesFilter(dkey, DefsTable.OLD_SCHEMA_CF, LAST_MIGRATION_KEY);
- ColumnFamily cf = cfStore.getColumnFamily(filter);
- if (cf == null || Iterables.isEmpty(cf.getColumnNames()))
- return null;
- else
- return UUIDGen.getUUID(cf.getColumn(LAST_MIGRATION_KEY).value());
- }
-
public static class MigrationsSerializer implements IVersionedSerializer<Collection<RowMutation>>
{
public static MigrationsSerializer instance = new MigrationsSerializer();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index caffaaf..5673063 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -736,11 +736,6 @@ public class StorageProxy implements StorageProxyMBean
public static void writeHintForMutation(RowMutation mutation, InetAddress target) throws IOException
{
UUID hostId = StorageService.instance.getTokenMetadata().getHostId(target);
- if ((hostId == null) && (MessagingService.instance().getVersion(target) < MessagingService.VERSION_12))
- {
- logger.warn("Unable to store hint for host with missing ID, {} (old node?)", target.toString());
- return;
- }
assert hostId != null : "Missing host ID for " + target.getHostAddress();
RowMutation hintedMutation = HintedHandOffManager.hintFor(mutation, hostId);
hintedMutation.apply();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/service/StorageService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 8e14d79..d720df2 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -66,7 +66,6 @@ import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.DynamicEndpointSnitch;
import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.locator.TokenMetadata;
-import org.apache.cassandra.metrics.StorageMetrics;
import org.apache.cassandra.net.AsyncOneResponse;
import org.apache.cassandra.net.MessageOut;
import org.apache.cassandra.net.MessagingService;
@@ -138,8 +137,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
public static final StorageService instance = new StorageService();
- private static final StorageMetrics metrics = new StorageMetrics();
-
public static IPartitioner getPartitioner()
{
return DatabaseDescriptor.getPartitioner();
@@ -238,7 +235,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
MessagingService.instance().registerVerbHandlers(MessagingService.Verb.READ_REPAIR, new ReadRepairVerbHandler());
MessagingService.instance().registerVerbHandlers(MessagingService.Verb.READ, new ReadVerbHandler());
MessagingService.instance().registerVerbHandlers(MessagingService.Verb.RANGE_SLICE, new RangeSliceVerbHandler());
- MessagingService.instance().registerVerbHandlers(MessagingService.Verb.INDEX_SCAN, new IndexScanVerbHandler());
MessagingService.instance().registerVerbHandlers(MessagingService.Verb.COUNTER_MUTATION, new CounterMutationVerbHandler());
MessagingService.instance().registerVerbHandlers(MessagingService.Verb.TRUNCATE, new TruncateVerbHandler());
MessagingService.instance().registerVerbHandlers(MessagingService.Verb.PAXOS_PREPARE, new PrepareVerbHandler());
@@ -306,7 +302,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
if (!initialized)
{
logger.warn("Starting gossip by operator request");
- Gossiper.instance.start((int)(System.currentTimeMillis() / 1000));
+ Gossiper.instance.start((int) (System.currentTimeMillis() / 1000));
initialized = true;
}
}
@@ -605,10 +601,10 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
Set<InetAddress> current = new HashSet<InetAddress>();
Collection<Token> tokens;
logger.debug("Bootstrap variables: {} {} {} {}",
- new Object[]{ DatabaseDescriptor.isAutoBootstrap(),
- SystemTable.bootstrapInProgress(),
- SystemTable.bootstrapComplete(),
- DatabaseDescriptor.getSeeds().contains(FBUtilities.getBroadcastAddress())});
+ DatabaseDescriptor.isAutoBootstrap(),
+ SystemTable.bootstrapInProgress(),
+ SystemTable.bootstrapComplete(),
+ DatabaseDescriptor.getSeeds().contains(FBUtilities.getBroadcastAddress()));
if (DatabaseDescriptor.isAutoBootstrap()
&& !SystemTable.bootstrapComplete()
&& !DatabaseDescriptor.getSeeds().contains(FBUtilities.getBroadcastAddress()))
@@ -1051,7 +1047,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
*/
public List<String> describeRingJMX(String keyspace) throws IOException
{
- List<TokenRange> tokenRanges = null;
+ List<TokenRange> tokenRanges;
try
{
tokenRanges = describeRing(keyspace);
@@ -1400,7 +1396,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
else if (tokenMetadata.isRelocating(token))
{
logger.info("Token {} is relocating to {}, ignoring update from {}",
- new Object[]{token, tokenMetadata.getRelocatingRanges().get(token), endpoint});
+ token, tokenMetadata.getRelocatingRanges().get(token), endpoint);
}
else if (Gossiper.instance.compareEndpointStartup(endpoint, currentOwner) > 0)
{
@@ -1507,7 +1503,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
if (logger.isDebugEnabled())
logger.debug("Node " + endpoint + " state left, tokens " + tokens);
- excise(tokens, endpoint, extractExpireTime(pieces, version));
+ excise(tokens, endpoint, extractExpireTime(pieces));
}
/**
@@ -1579,7 +1575,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
if (VersionedValue.REMOVED_TOKEN.equals(state))
{
- excise(removeTokens, endpoint, extractExpireTime(pieces, MessagingService.instance().getVersion(endpoint)));
+ excise(removeTokens, endpoint, extractExpireTime(pieces));
}
else if (VersionedValue.REMOVING_TOKEN.equals(state))
{
@@ -1599,7 +1595,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
}
else // now that the gossiper has told us about this nonexistent member, notify the gossiper to remove it
{
- addExpireTimeIfFound(endpoint, extractExpireTime(pieces, MessagingService.instance().getVersion(endpoint)));
+ addExpireTimeIfFound(endpoint, extractExpireTime(pieces));
removeEndpoint(endpoint);
}
}
@@ -1641,21 +1637,12 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
}
}
- protected long extractExpireTime(String[] pieces, int version)
+ protected long extractExpireTime(String[] pieces)
{
- if (version < MessagingService.VERSION_12)
- {
- if (pieces.length >= 3)
- return Long.parseLong(pieces[2]);
- else
- return 0L;
- } else
- {
- if (VersionedValue.STATUS_LEFT.equals(pieces[0]))
- return Long.parseLong(pieces[1]);
- else
- return Long.parseLong(pieces[2]);
- }
+ if (VersionedValue.STATUS_LEFT.equals(pieces[0]))
+ return Long.parseLong(pieces[1]);
+ else
+ return Long.parseLong(pieces[2]);
}
/**
@@ -2371,7 +2358,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
Token parsedEndToken = getPartitioner().getTokenFactory().fromString(endToken);
logger.info("starting user-requested repair of range ({}, {}] for keyspace {} and column families {}",
- new Object[] {parsedBeginToken, parsedEndToken, tableName, columnFamilies});
+ parsedBeginToken, parsedEndToken, tableName, columnFamilies);
return forceRepairAsync(tableName, isSequential, isLocal, Collections.singleton(new Range<Token>(parsedBeginToken, parsedEndToken)), columnFamilies);
}
@@ -2411,7 +2398,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
private FutureTask<Object> createRepairTask(final int cmd, final String keyspace, final Collection<Range<Token>> ranges, final boolean isSequential, final boolean isLocal, final String... columnFamilies)
{
- FutureTask<Object> task = new FutureTask<Object>(new WrappedRunnable()
+ return new FutureTask<Object>(new WrappedRunnable()
{
protected void runMayThrow() throws Exception
{
@@ -2472,7 +2459,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
sendNotification("repair", String.format("Repair command #%d finished", cmd), new int[]{cmd, ActiveRepairService.Status.FINISHED.ordinal()});
}
}, null);
- return task;
}
public ActiveRepairService.RepairFuture forceTableRepair(final Range<Token> range, final String tableName, boolean isSequential, boolean isLocal, final String... columnFamilies) throws IOException
@@ -2499,24 +2485,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
/* End of MBean interface methods */
/**
- * This method returns the predecessor of the endpoint ep on the identifier
- * space.
- */
- InetAddress getPredecessor(Token token)
- {
- return tokenMetadata.getEndpoint(tokenMetadata.getPredecessor(token));
- }
-
- /*
- * This method returns the successor of the endpoint ep on the identifier
- * space.
- */
- public InetAddress getSuccessor(Token token)
- {
- return tokenMetadata.getEndpoint(tokenMetadata.getSuccessor(token));
- }
-
- /**
* Get the "primary ranges" for the specified keyspace and endpoint.
* "Primary ranges" are the ranges that the node is responsible for storing replica primarily.
* The node that stores replica primarily is defined as the first node returned
@@ -3547,7 +3515,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
IEndpointSnitch oldSnitch = DatabaseDescriptor.getEndpointSnitch();
// new snitch registers mbean during construction
- IEndpointSnitch newSnitch = null;
+ IEndpointSnitch newSnitch;
try
{
newSnitch = FBUtilities.construct(epSnitchClassName, "snitch");
@@ -3661,7 +3629,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
logger.warn("Streaming to " + targetAddr + " failed");
onSuccess(); // calling onSuccess for latch countdown
}
- };
+ }
/**
* Used to request ranges from endpoints in the ring (will block until all data is fetched and ready)
@@ -3881,7 +3849,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
rangeXferExecutor.tearDown();
}
- @Override
public void disableAutoCompaction(String ks, String... columnFamilies) throws IOException
{
for (ColumnFamilyStore cfs : getValidColumnFamilies(true, true, ks, columnFamilies))
@@ -3890,7 +3857,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
}
}
- @Override
public void enableAutoCompaction(String ks, String... columnFamilies) throws IOException
{
for (ColumnFamilyStore cfs : getValidColumnFamilies(true, true, ks, columnFamilies))
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/service/StorageServiceMBean.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
index 84d0346..9548fcc 100644
--- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java
+++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
@@ -474,4 +474,5 @@ public interface StorageServiceMBean extends NotificationEmitter
void disableAutoCompaction(String ks, String ... columnFamilies) throws IOException;
void enableAutoCompaction(String ks, String ... columnFamilies) throws IOException;
+ public void deliverHints(String host) throws UnknownHostException;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java b/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java
index c552105..2eca71e 100644
--- a/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java
+++ b/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java
@@ -151,8 +151,8 @@ public class IncomingStreamReader
while (bytesRead < length)
{
in.reset(0);
- key = SSTableReader.decodeKey(StorageService.getPartitioner(), localFile.desc, ByteBufferUtil.readWithShortLength(in));
- long dataSize = SSTableReader.readRowSize(in, localFile.desc);
+ key = StorageService.getPartitioner().decorateKey(ByteBufferUtil.readWithShortLength(in));
+ long dataSize = in.readLong();
if (cfs.containsCachedRow(key) && remoteFile.type == OperationType.AES && dataSize <= DatabaseDescriptor.getInMemoryCompactionLimit())
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/streaming/PendingFile.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/PendingFile.java b/src/java/org/apache/cassandra/streaming/PendingFile.java
index 7474bd6..129b3f7 100644
--- a/src/java/org/apache/cassandra/streaming/PendingFile.java
+++ b/src/java/org/apache/cassandra/streaming/PendingFile.java
@@ -138,8 +138,7 @@ public class PendingFile
}
out.writeUTF(sc.type.name());
out.writeLong(sc.estimatedKeys);
- if (version > MessagingService.VERSION_11)
- CompressionInfo.serializer.serialize(sc.compressionInfo, out, version);
+ CompressionInfo.serializer.serialize(sc.compressionInfo, out, version);
}
public PendingFile deserialize(DataInput in, int version) throws IOException
@@ -159,8 +158,7 @@ public class PendingFile
type = OperationType.valueOf(in.readUTF());
long estimatedKeys = in.readLong();
CompressionInfo info = null;
- if (version > MessagingService.VERSION_11)
- info = CompressionInfo.serializer.deserialize(in, version);
+ info = CompressionInfo.serializer.deserialize(in, version);
return new PendingFile(null, desc, component, sections, type, estimatedKeys, info);
}
@@ -176,8 +174,7 @@ public class PendingFile
size += TypeSizes.NATIVE.sizeof(section.left) + TypeSizes.NATIVE.sizeof(section.right);
size += TypeSizes.NATIVE.sizeof(pf.type.name());
size += TypeSizes.NATIVE.sizeof(pf.estimatedKeys);
- if (version > MessagingService.VERSION_11)
- size += CompressionInfo.serializer.serializedSize(pf.compressionInfo, version);
+ size += CompressionInfo.serializer.serializedSize(pf.compressionInfo, version);
return size;
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7f2c3a8e/src/java/org/apache/cassandra/tools/StandaloneScrubber.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tools/StandaloneScrubber.java b/src/java/org/apache/cassandra/tools/StandaloneScrubber.java
index c9d39bb..9db565d 100644
--- a/src/java/org/apache/cassandra/tools/StandaloneScrubber.java
+++ b/src/java/org/apache/cassandra/tools/StandaloneScrubber.java
@@ -56,10 +56,6 @@ public class StandaloneScrubber
Options options = Options.parseArgs(args);
try
{
- // Migrate sstables from pre-#2749 to the correct location
- if (Directories.sstablesNeedsMigration())
- Directories.migrateSSTables();
-
// load keyspace descriptions.
DatabaseDescriptor.loadSchemas();