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 2011/10/13 07:14:08 UTC
svn commit: r1182654 [2/3] - in /cassandra/trunk: ./ contrib/
contrib/pig/src/java/org/apache/cassandra/hadoop/pig/
interface/thrift/gen-java/org/apache/cassandra/thrift/
src/java/org/apache/cassandra/cache/ src/java/org/apache/cassandra/cli/
src/java/...
Modified: cassandra/trunk/src/java/org/apache/cassandra/gms/EndpointState.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/gms/EndpointState.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/gms/EndpointState.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/gms/EndpointState.java Thu Oct 13 05:14:05 2011
@@ -18,15 +18,13 @@
package org.apache.cassandra.gms;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import java.io.*;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
/**
@@ -39,7 +37,7 @@ public class EndpointState
{
protected static Logger logger = LoggerFactory.getLogger(EndpointState.class);
- private final static ICompactSerializer<EndpointState> serializer = new EndpointStateSerializer();
+ private final static IVersionedSerializer<EndpointState> serializer = new EndpointStateSerializer();
private volatile HeartBeatState hbState;
final Map<ApplicationState, VersionedValue> applicationState = new NonBlockingHashMap<ApplicationState, VersionedValue>();
@@ -54,7 +52,7 @@ public class EndpointState
// fat client and will be removed automatically from gossip.
private volatile boolean hasToken;
- public static ICompactSerializer<EndpointState> serializer()
+ public static IVersionedSerializer<EndpointState> serializer()
{
return serializer;
}
@@ -134,11 +132,11 @@ public class EndpointState
}
}
-class EndpointStateSerializer implements ICompactSerializer<EndpointState>
+class EndpointStateSerializer implements IVersionedSerializer<EndpointState>
{
private static Logger logger = LoggerFactory.getLogger(EndpointStateSerializer.class);
- public void serialize(EndpointState epState, DataOutputStream dos, int version) throws IOException
+ public void serialize(EndpointState epState, DataOutput dos, int version) throws IOException
{
/* serialize the HeartBeatState */
HeartBeatState hbState = epState.getHeartBeatState();
@@ -150,15 +148,12 @@ class EndpointStateSerializer implements
for (Map.Entry<ApplicationState, VersionedValue> entry : epState.applicationState.entrySet())
{
VersionedValue value = entry.getValue();
- if (value != null)
- {
- dos.writeInt(entry.getKey().ordinal());
- VersionedValue.serializer.serialize(value, dos, version);
- }
+ dos.writeInt(entry.getKey().ordinal());
+ VersionedValue.serializer.serialize(value, dos, version);
}
}
- public EndpointState deserialize(DataInputStream dis, int version) throws IOException
+ public EndpointState deserialize(DataInput dis, int version) throws IOException
{
HeartBeatState hbState = HeartBeatState.serializer().deserialize(dis, version);
EndpointState epState = new EndpointState(hbState);
@@ -166,15 +161,15 @@ class EndpointStateSerializer implements
int appStateSize = dis.readInt();
for ( int i = 0; i < appStateSize; ++i )
{
- if ( dis.available() == 0 )
- {
- break;
- }
-
int key = dis.readInt();
VersionedValue value = VersionedValue.serializer.deserialize(dis, version);
epState.addApplicationState(Gossiper.STATES[key], value);
}
return epState;
}
+
+ public long serializedSize(EndpointState endpointState, int version)
+ {
+ throw new UnsupportedOperationException();
+ }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/gms/GossipDigest.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/gms/GossipDigest.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/gms/GossipDigest.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/gms/GossipDigest.java Thu Oct 13 05:14:05 2011
@@ -18,12 +18,10 @@
package org.apache.cassandra.gms;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import java.io.*;
import java.net.InetAddress;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.net.CompactEndpointSerializationHelper;
/**
@@ -33,7 +31,7 @@ import org.apache.cassandra.net.CompactE
public class GossipDigest implements Comparable<GossipDigest>
{
- private static ICompactSerializer<GossipDigest> serializer;
+ private static IVersionedSerializer<GossipDigest> serializer;
static
{
serializer = new GossipDigestSerializer();
@@ -43,7 +41,7 @@ public class GossipDigest implements Com
int generation;
int maxVersion;
- public static ICompactSerializer<GossipDigest> serializer()
+ public static IVersionedSerializer<GossipDigest> serializer()
{
return serializer;
}
@@ -89,20 +87,25 @@ public class GossipDigest implements Com
}
}
-class GossipDigestSerializer implements ICompactSerializer<GossipDigest>
+class GossipDigestSerializer implements IVersionedSerializer<GossipDigest>
{
- public void serialize(GossipDigest gDigest, DataOutputStream dos, int version) throws IOException
+ public void serialize(GossipDigest gDigest, DataOutput dos, int version) throws IOException
{
CompactEndpointSerializationHelper.serialize(gDigest.endpoint, dos);
dos.writeInt(gDigest.generation);
dos.writeInt(gDigest.maxVersion);
}
- public GossipDigest deserialize(DataInputStream dis, int version) throws IOException
+ public GossipDigest deserialize(DataInput dis, int version) throws IOException
{
InetAddress endpoint = CompactEndpointSerializationHelper.deserialize(dis);
int generation = dis.readInt();
int maxVersion = dis.readInt();
return new GossipDigest(endpoint, generation, maxVersion);
}
+
+ public long serializedSize(GossipDigest gossipDigest, int version)
+ {
+ throw new UnsupportedOperationException();
+ }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/gms/GossipDigestAck2Message.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/gms/GossipDigestAck2Message.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/gms/GossipDigestAck2Message.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/gms/GossipDigestAck2Message.java Thu Oct 13 05:14:05 2011
@@ -18,14 +18,12 @@
package org.apache.cassandra.gms;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import java.io.*;
import java.net.InetAddress;
import java.util.HashMap;
import java.util.Map;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
/**
@@ -35,7 +33,7 @@ import org.apache.cassandra.io.ICompactS
class GossipDigestAck2Message
{
- private static ICompactSerializer<GossipDigestAck2Message> serializer_;
+ private static IVersionedSerializer<GossipDigestAck2Message> serializer_;
static
{
serializer_ = new GossipDigestAck2MessageSerializer();
@@ -43,7 +41,7 @@ class GossipDigestAck2Message
Map<InetAddress, EndpointState> epStateMap_ = new HashMap<InetAddress, EndpointState>();
- public static ICompactSerializer<GossipDigestAck2Message> serializer()
+ public static IVersionedSerializer<GossipDigestAck2Message> serializer()
{
return serializer_;
}
@@ -59,18 +57,23 @@ class GossipDigestAck2Message
}
}
-class GossipDigestAck2MessageSerializer implements ICompactSerializer<GossipDigestAck2Message>
+class GossipDigestAck2MessageSerializer implements IVersionedSerializer<GossipDigestAck2Message>
{
- public void serialize(GossipDigestAck2Message gDigestAck2Message, DataOutputStream dos, int version) throws IOException
+ public void serialize(GossipDigestAck2Message gDigestAck2Message, DataOutput dos, int version) throws IOException
{
/* Use the EndpointState */
EndpointStatesSerializationHelper.serialize(gDigestAck2Message.epStateMap_, dos, version);
}
- public GossipDigestAck2Message deserialize(DataInputStream dis, int version) throws IOException
+ public GossipDigestAck2Message deserialize(DataInput dis, int version) throws IOException
{
Map<InetAddress, EndpointState> epStateMap = EndpointStatesSerializationHelper.deserialize(dis, version);
return new GossipDigestAck2Message(epStateMap);
}
+
+ public long serializedSize(GossipDigestAck2Message gossipDigestAck2Message, int version)
+ {
+ throw new UnsupportedOperationException();
+ }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/gms/GossipDigestAckMessage.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/gms/GossipDigestAckMessage.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/gms/GossipDigestAckMessage.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/gms/GossipDigestAckMessage.java Thu Oct 13 05:14:05 2011
@@ -18,16 +18,14 @@
package org.apache.cassandra.gms;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import java.io.*;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
@@ -38,7 +36,7 @@ import org.apache.cassandra.io.ICompactS
class GossipDigestAckMessage
{
- private static ICompactSerializer<GossipDigestAckMessage> serializer_;
+ private static IVersionedSerializer<GossipDigestAckMessage> serializer_;
static
{
serializer_ = new GossipDigestAckMessageSerializer();
@@ -47,7 +45,7 @@ class GossipDigestAckMessage
List<GossipDigest> gDigestList_ = new ArrayList<GossipDigest>();
Map<InetAddress, EndpointState> epStateMap_ = new HashMap<InetAddress, EndpointState>();
- static ICompactSerializer<GossipDigestAckMessage> serializer()
+ static IVersionedSerializer<GossipDigestAckMessage> serializer()
{
return serializer_;
}
@@ -69,20 +67,25 @@ class GossipDigestAckMessage
}
}
-class GossipDigestAckMessageSerializer implements ICompactSerializer<GossipDigestAckMessage>
+class GossipDigestAckMessageSerializer implements IVersionedSerializer<GossipDigestAckMessage>
{
- public void serialize(GossipDigestAckMessage gDigestAckMessage, DataOutputStream dos, int version) throws IOException
+ public void serialize(GossipDigestAckMessage gDigestAckMessage, DataOutput dos, int version) throws IOException
{
GossipDigestSerializationHelper.serialize(gDigestAckMessage.gDigestList_, dos, version);
dos.writeBoolean(true); // 0.6 compatibility
EndpointStatesSerializationHelper.serialize(gDigestAckMessage.epStateMap_, dos, version);
}
- public GossipDigestAckMessage deserialize(DataInputStream dis, int version) throws IOException
+ public GossipDigestAckMessage deserialize(DataInput dis, int version) throws IOException
{
List<GossipDigest> gDigestList = GossipDigestSerializationHelper.deserialize(dis, version);
dis.readBoolean(); // 0.6 compatibility
Map<InetAddress, EndpointState> epStateMap = EndpointStatesSerializationHelper.deserialize(dis, version);
return new GossipDigestAckMessage(gDigestList, epStateMap);
}
+
+ public long serializedSize(GossipDigestAckMessage gossipDigestAckMessage, int version)
+ {
+ throw new UnsupportedOperationException();
+ }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/gms/GossipDigestSynMessage.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/gms/GossipDigestSynMessage.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/gms/GossipDigestSynMessage.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/gms/GossipDigestSynMessage.java Thu Oct 13 05:14:05 2011
@@ -18,9 +18,7 @@
package org.apache.cassandra.gms;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import java.io.*;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.HashMap;
@@ -31,7 +29,7 @@ import java.util.Map.Entry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.net.CompactEndpointSerializationHelper;
@@ -42,7 +40,7 @@ import org.apache.cassandra.net.CompactE
class GossipDigestSynMessage
{
- private static ICompactSerializer<GossipDigestSynMessage> serializer_;
+ private static IVersionedSerializer<GossipDigestSynMessage> serializer_;
static
{
serializer_ = new GossipDigestSynMessageSerializer();
@@ -51,7 +49,7 @@ class GossipDigestSynMessage
String clusterId_;
List<GossipDigest> gDigests_ = new ArrayList<GossipDigest>();
- public static ICompactSerializer<GossipDigestSynMessage> serializer()
+ public static IVersionedSerializer<GossipDigestSynMessage> serializer()
{
return serializer_;
}
@@ -72,7 +70,7 @@ class GossipDigestSerializationHelper
{
private static Logger logger_ = LoggerFactory.getLogger(GossipDigestSerializationHelper.class);
- static void serialize(List<GossipDigest> gDigestList, DataOutputStream dos, int version) throws IOException
+ static void serialize(List<GossipDigest> gDigestList, DataOutput dos, int version) throws IOException
{
dos.writeInt(gDigestList.size());
for ( GossipDigest gDigest : gDigestList )
@@ -81,15 +79,14 @@ class GossipDigestSerializationHelper
}
}
- static List<GossipDigest> deserialize(DataInputStream dis, int version) throws IOException
+ static List<GossipDigest> deserialize(DataInput dis, int version) throws IOException
{
int size = dis.readInt();
List<GossipDigest> gDigests = new ArrayList<GossipDigest>(size);
for ( int i = 0; i < size; ++i )
{
- assert dis.available() > 0;
- gDigests.add(GossipDigest.serializer().deserialize(dis, version));
+ gDigests.add(GossipDigest.serializer().deserialize(dis, version));
}
return gDigests;
}
@@ -99,7 +96,7 @@ class EndpointStatesSerializationHelper
{
private static final Logger logger_ = LoggerFactory.getLogger(EndpointStatesSerializationHelper.class);
- static void serialize(Map<InetAddress, EndpointState> epStateMap, DataOutputStream dos, int version) throws IOException
+ static void serialize(Map<InetAddress, EndpointState> epStateMap, DataOutput dos, int version) throws IOException
{
dos.writeInt(epStateMap.size());
for (Entry<InetAddress, EndpointState> entry : epStateMap.entrySet())
@@ -110,14 +107,13 @@ class EndpointStatesSerializationHelper
}
}
- static Map<InetAddress, EndpointState> deserialize(DataInputStream dis, int version) throws IOException
+ static Map<InetAddress, EndpointState> deserialize(DataInput dis, int version) throws IOException
{
int size = dis.readInt();
Map<InetAddress, EndpointState> epStateMap = new HashMap<InetAddress, EndpointState>(size);
for ( int i = 0; i < size; ++i )
{
- assert dis.available() > 0;
InetAddress ep = CompactEndpointSerializationHelper.deserialize(dis);
EndpointState epState = EndpointState.serializer().deserialize(dis, version);
epStateMap.put(ep, epState);
@@ -126,20 +122,24 @@ class EndpointStatesSerializationHelper
}
}
-class GossipDigestSynMessageSerializer implements ICompactSerializer<GossipDigestSynMessage>
+class GossipDigestSynMessageSerializer implements IVersionedSerializer<GossipDigestSynMessage>
{
- public void serialize(GossipDigestSynMessage gDigestSynMessage, DataOutputStream dos, int version) throws IOException
+ public void serialize(GossipDigestSynMessage gDigestSynMessage, DataOutput dos, int version) throws IOException
{
dos.writeUTF(gDigestSynMessage.clusterId_);
GossipDigestSerializationHelper.serialize(gDigestSynMessage.gDigests_, dos, version);
}
- public GossipDigestSynMessage deserialize(DataInputStream dis, int version) throws IOException
+ public GossipDigestSynMessage deserialize(DataInput dis, int version) throws IOException
{
String clusterId = dis.readUTF();
List<GossipDigest> gDigests = GossipDigestSerializationHelper.deserialize(dis, version);
return new GossipDigestSynMessage(clusterId, gDigests);
}
+ public long serializedSize(GossipDigestSynMessage gossipDigestSynMessage, int version)
+ {
+ throw new UnsupportedOperationException();
+ }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/gms/Gossiper.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/gms/Gossiper.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/gms/Gossiper.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/gms/Gossiper.java Thu Oct 13 05:14:05 2011
@@ -75,7 +75,7 @@ public class Gossiper implements IFailur
private static Logger logger = LoggerFactory.getLogger(Gossiper.class);
public static final Gossiper instance = new Gossiper();
- private long aVeryLongTime;
+ public static final long aVeryLongTime = 259200 * 1000; // 3 days
private long FatClientTimeout;
private Random random = new Random();
private Comparator<InetAddress> inetcomparator = new Comparator<InetAddress>()
@@ -107,6 +107,8 @@ public class Gossiper implements IFailur
* gossip gets propagated to all nodes */
private Map<InetAddress, Long> justRemovedEndpoints = new ConcurrentHashMap<InetAddress, Long>();
+ private Map<InetAddress, Long> expireTimeEndpointMap = new ConcurrentHashMap<InetAddress, Long>();
+
// protocol versions of the other nodes in the cluster
private final ConcurrentMap<InetAddress, Integer> versions = new NonBlockingHashMap<InetAddress, Integer>();
@@ -174,8 +176,6 @@ public class Gossiper implements IFailur
private Gossiper()
{
- // 3 days
- aVeryLongTime = 259200 * 1000;
// half of QUARATINE_DELAY, to ensure justRemovedEndpoints has enough leeway to prevent re-gossip
FatClientTimeout = (long)(QUARANTINE_DELAY / 2);
/* register with the Failure Detector for receiving Failure detector events */
@@ -296,6 +296,7 @@ public class Gossiper implements IFailur
{
unreachableEndpoints.remove(endpoint);
endpointStateMap.remove(endpoint);
+ expireTimeEndpointMap.remove(endpoint);
justRemovedEndpoints.put(endpoint, System.currentTimeMillis());
if (logger.isDebugEnabled())
logger.debug("evicting " + endpoint + " from gossip");
@@ -417,7 +418,7 @@ public class Gossiper implements IFailur
EndpointState epState = endpointStateMap.get(endpoint);
epState.updateTimestamp(); // make sure we don't evict it too soon
epState.getHeartBeatState().forceNewerGenerationUnsafe();
- epState.addApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.removedNonlocal(token));
+ epState.addApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.removedNonlocal(token,computeExpireTime()));
logger.info("Completing removal of " + endpoint);
endpointStateMap.put(endpoint, epState);
// ensure at least one gossip round occurs before returning
@@ -572,8 +573,14 @@ public class Gossiper implements IFailur
evictFromMembership(endpoint); // can get rid of the state immediately
}
- if ( !epState.isAlive() && (duration > aVeryLongTime) && (!StorageService.instance.getTokenMetadata().isMember(endpoint)))
+ long expireTime = getExpireTimeForEndpoint(endpoint);
+ if (!epState.isAlive() && (now > expireTime)
+ && (!StorageService.instance.getTokenMetadata().isMember(endpoint)))
{
+ if (logger.isDebugEnabled())
+ {
+ logger.debug("time is expiring for endpoint : " + endpoint + " (" + expireTime + ")");
+ }
evictFromMembership(endpoint);
}
}
@@ -592,6 +599,17 @@ public class Gossiper implements IFailur
}
}
}
+
+ protected long getExpireTimeForEndpoint(InetAddress endpoint)
+ {
+ /* default expireTime is aVeryLongTime */
+ long expireTime = computeExpireTime();
+ if (expireTimeEndpointMap.containsKey(endpoint))
+ {
+ expireTime = expireTimeEndpointMap.get(endpoint);
+ }
+ return expireTime;
+ }
public EndpointState getEndpointStateForEndpoint(InetAddress ep)
{
@@ -719,6 +737,8 @@ public class Gossiper implements IFailur
localState.updateTimestamp(); // prevents doStatusCheck from racing us and evicting if it was down > aVeryLongTime
liveEndpoints.add(addr);
unreachableEndpoints.remove(addr);
+ expireTimeEndpointMap.remove(addr);
+ logger.debug("removing expire time for endpoint : " + addr);
logger.info("InetAddress {} is now UP", addr);
for (IEndpointStateChangeSubscriber subscriber : subscribers)
subscriber.onAlive(addr, localState);
@@ -1066,5 +1086,18 @@ public class Gossiper implements IFailur
{
return getCurrentGenerationNumber(InetAddress.getByName(address));
}
+
+ public void addExpireTimeForEndpoint(InetAddress endpoint, long expireTime)
+ {
+ if (logger.isDebugEnabled())
+ {
+ logger.debug("adding expire time for endpoint : " + endpoint + " (" + expireTime + ")");
+ }
+ expireTimeEndpointMap.put(endpoint, expireTime);
+ }
+
+ public static long computeExpireTime() {
+ return System.currentTimeMillis() + Gossiper.aVeryLongTime;
+ }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/gms/HeartBeatState.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/gms/HeartBeatState.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/gms/HeartBeatState.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/gms/HeartBeatState.java Thu Oct 13 05:14:05 2011
@@ -18,11 +18,9 @@
package org.apache.cassandra.gms;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import java.io.*;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
/**
@@ -31,7 +29,7 @@ import org.apache.cassandra.io.ICompactS
class HeartBeatState
{
- private static ICompactSerializer<HeartBeatState> serializer;
+ private static IVersionedSerializer<HeartBeatState> serializer;
static
{
@@ -52,7 +50,7 @@ class HeartBeatState
version = ver;
}
- public static ICompactSerializer<HeartBeatState> serializer()
+ public static IVersionedSerializer<HeartBeatState> serializer()
{
return serializer;
}
@@ -78,16 +76,21 @@ class HeartBeatState
}
}
-class HeartBeatStateSerializer implements ICompactSerializer<HeartBeatState>
+class HeartBeatStateSerializer implements IVersionedSerializer<HeartBeatState>
{
- public void serialize(HeartBeatState hbState, DataOutputStream dos, int version) throws IOException
+ public void serialize(HeartBeatState hbState, DataOutput dos, int version) throws IOException
{
dos.writeInt(hbState.getGeneration());
dos.writeInt(hbState.getHeartBeatVersion());
}
- public HeartBeatState deserialize(DataInputStream dis, int version) throws IOException
+ public HeartBeatState deserialize(DataInput dis, int version) throws IOException
{
return new HeartBeatState(dis.readInt(), dis.readInt());
}
+
+ public long serializedSize(HeartBeatState heartBeatState, int version)
+ {
+ throw new UnsupportedOperationException();
+ }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/gms/VersionedValue.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/gms/VersionedValue.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/gms/VersionedValue.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/gms/VersionedValue.java Thu Oct 13 05:14:05 2011
@@ -18,15 +18,13 @@
package org.apache.cassandra.gms;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import java.io.*;
import java.net.InetAddress;
import java.util.UUID;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.utils.FBUtilities;
@@ -44,7 +42,7 @@ import org.apache.cassandra.utils.FBUtil
public class VersionedValue implements Comparable<VersionedValue>
{
- public static final ICompactSerializer<VersionedValue> serializer = new VersionedValueSerializer();
+ public static final IVersionedSerializer<VersionedValue> serializer = new VersionedValueSerializer();
// this must be a char that cannot be present in any token
public final static char DELIMITER = ',';
@@ -70,14 +68,14 @@ public class VersionedValue implements C
private VersionedValue(String value, int version)
{
+ assert value != null;
this.value = value;
this.version = version;
}
private VersionedValue(String value)
{
- this.value = value;
- version = VersionGenerator.getNextVersion();
+ this(value, VersionGenerator.getNextVersion());
}
public int compareTo(VersionedValue value)
@@ -125,9 +123,10 @@ public class VersionedValue implements C
return new VersionedValue(VersionedValue.STATUS_LEAVING + VersionedValue.DELIMITER + partitioner.getTokenFactory().toString(token));
}
- public VersionedValue left(Token token)
+ public VersionedValue left(Token token, long expireTime)
{
- return new VersionedValue(VersionedValue.STATUS_LEFT + VersionedValue.DELIMITER + partitioner.getTokenFactory().toString(token));
+ return new VersionedValue(VersionedValue.STATUS_LEFT + VersionedValue.DELIMITER
+ + partitioner.getTokenFactory().toString(token) + VersionedValue.DELIMITER + expireTime);
}
public VersionedValue moving(Token token)
@@ -140,9 +139,10 @@ public class VersionedValue implements C
return new VersionedValue(VersionedValue.REMOVING_TOKEN + VersionedValue.DELIMITER + partitioner.getTokenFactory().toString(token));
}
- public VersionedValue removedNonlocal(Token token)
+ public VersionedValue removedNonlocal(Token token, long expireTime)
{
- return new VersionedValue(VersionedValue.REMOVED_TOKEN + VersionedValue.DELIMITER + partitioner.getTokenFactory().toString(token));
+ return new VersionedValue(VersionedValue.REMOVED_TOKEN + VersionedValue.DELIMITER
+ + partitioner.getTokenFactory().toString(token) + VersionedValue.DELIMITER + expireTime);
}
public VersionedValue removalCoordinator(Token token)
@@ -181,20 +181,25 @@ public class VersionedValue implements C
}
}
- private static class VersionedValueSerializer implements ICompactSerializer<VersionedValue>
+ private static class VersionedValueSerializer implements IVersionedSerializer<VersionedValue>
{
- public void serialize(VersionedValue value, DataOutputStream dos, int version) throws IOException
+ public void serialize(VersionedValue value, DataOutput dos, int version) throws IOException
{
dos.writeUTF(value.value);
dos.writeInt(value.version);
}
- public VersionedValue deserialize(DataInputStream dis, int version) throws IOException
+ public VersionedValue deserialize(DataInput dis, int version) throws IOException
{
String value = dis.readUTF();
int valVersion = dis.readInt();
return new VersionedValue(value, valVersion);
}
+
+ public long serializedSize(VersionedValue value, int version)
+ {
+ throw new UnsupportedOperationException();
+ }
}
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/io/IColumnSerializer.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/io/IColumnSerializer.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/io/IColumnSerializer.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/io/IColumnSerializer.java Thu Oct 13 05:14:05 2011
@@ -25,7 +25,7 @@ import java.io.IOException;
import org.apache.cassandra.db.IColumn;
-public interface IColumnSerializer extends ICompactSerializer2<IColumn>
+public interface IColumnSerializer extends ISerializer<IColumn>
{
public IColumn deserialize(DataInput in, boolean fromRemote, int expireBefore) throws IOException;
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/io/SerDeUtils.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/io/SerDeUtils.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/io/SerDeUtils.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/io/SerDeUtils.java Thu Oct 13 05:14:05 2011
@@ -38,7 +38,7 @@ import org.apache.cassandra.io.util.Outp
import org.apache.cassandra.utils.ByteBufferUtil;
/**
- * Static serialization/deserialization utility functions, intended to eventually replace ICompactSerializers.
+ * Static serialization/deserialization utility functions, intended to eventually replace IVersionedSerializers.
*/
public final class SerDeUtils
{
Modified: cassandra/trunk/src/java/org/apache/cassandra/io/sstable/IndexHelper.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/io/sstable/IndexHelper.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/io/sstable/IndexHelper.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/io/sstable/IndexHelper.java Thu Oct 13 05:14:05 2011
@@ -131,8 +131,8 @@ public class IndexHelper
DataInputStream stream = new DataInputStream(ByteBufferUtil.inputStream(bytes));
return useOldBuffer
- ? LegacyBloomFilter.serializer().deserialize(stream, 0) // version means nothing there.
- : BloomFilter.serializer().deserialize(stream);
+ ? LegacyBloomFilter.serializer().deserialize(stream)
+ : BloomFilter.serializer().deserialize(stream);
}
/**
Modified: cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableMetadata.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableMetadata.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableMetadata.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableMetadata.java Thu Oct 13 05:14:05 2011
@@ -31,7 +31,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.commitlog.ReplayPosition;
-import org.apache.cassandra.io.ICompactSerializer2;
+import org.apache.cassandra.io.ISerializer;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.EstimatedHistogram;
@@ -168,7 +168,7 @@ public class SSTableMetadata
}
}
- public static class SSTableMetadataSerializer implements ICompactSerializer2<SSTableMetadata>
+ public static class SSTableMetadataSerializer implements ISerializer<SSTableMetadata>
{
private static final Logger logger = LoggerFactory.getLogger(SSTableMetadataSerializer.class);
@@ -220,5 +220,10 @@ public class SSTableMetadata
long maxTimestamp = dis.readLong();
return new SSTableMetadata(rowSizes, columnCounts, replayPosition, maxTimestamp);
}
+
+ public long serializedSize(SSTableMetadata object)
+ {
+ throw new UnsupportedOperationException();
+ }
}
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableReader.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableReader.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableReader.java Thu Oct 13 05:14:05 2011
@@ -292,7 +292,7 @@ public class SSTableReader extends SSTab
stream = new DataInputStream(new BufferedInputStream(new FileInputStream(descriptor.filenameFor(Component.FILTER))));
if (descriptor.usesOldBloomFilter)
{
- bf = LegacyBloomFilter.serializer().deserialize(stream, 0); // version means nothing.
+ bf = LegacyBloomFilter.serializer().deserialize(stream);
}
else
{
Modified: cassandra/trunk/src/java/org/apache/cassandra/net/CompactEndpointSerializationHelper.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/net/CompactEndpointSerializationHelper.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/net/CompactEndpointSerializationHelper.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/net/CompactEndpointSerializationHelper.java Thu Oct 13 05:14:05 2011
@@ -18,23 +18,21 @@
package org.apache.cassandra.net;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import java.io.*;
import java.net.Inet4Address;
import java.net.Inet6Address;
import java.net.InetAddress;
public class CompactEndpointSerializationHelper
{
- public static void serialize(InetAddress endpoint, DataOutputStream dos) throws IOException
+ public static void serialize(InetAddress endpoint, DataOutput dos) throws IOException
{
byte[] buf = endpoint.getAddress();
dos.writeByte(buf.length);
dos.write(buf);
}
- public static InetAddress deserialize(DataInputStream dis) throws IOException
+ public static InetAddress deserialize(DataInput dis) throws IOException
{
byte[] bytes = new byte[dis.readByte()];
dis.readFully(bytes, 0, bytes.length);
Modified: cassandra/trunk/src/java/org/apache/cassandra/net/Header.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/net/Header.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/net/Header.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/net/Header.java Thu Oct 13 05:14:05 2011
@@ -18,28 +18,25 @@
package org.apache.cassandra.net;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import java.io.*;
import java.net.InetAddress;
import java.util.Hashtable;
import java.util.Map;
-import java.util.Set;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
public class Header
{
- private static ICompactSerializer<Header> serializer_;
+ private static IVersionedSerializer<Header> serializer_;
static
{
serializer_ = new HeaderSerializer();
}
- static ICompactSerializer<Header> serializer()
+ static IVersionedSerializer<Header> serializer()
{
return serializer_;
}
@@ -105,9 +102,9 @@ public class Header
}
}
-class HeaderSerializer implements ICompactSerializer<Header>
+class HeaderSerializer implements IVersionedSerializer<Header>
{
- public void serialize(Header t, DataOutputStream dos, int version) throws IOException
+ public void serialize(Header t, DataOutput dos, int version) throws IOException
{
CompactEndpointSerializationHelper.serialize(t.getFrom(), dos);
dos.writeInt(t.getVerb().ordinal());
@@ -121,7 +118,7 @@ class HeaderSerializer implements ICompa
}
}
- public Header deserialize(DataInputStream dis, int version) throws IOException
+ public Header deserialize(DataInput dis, int version) throws IOException
{
InetAddress from = CompactEndpointSerializationHelper.deserialize(dis);
int verbOrdinal = dis.readInt();
@@ -137,6 +134,11 @@ class HeaderSerializer implements ICompa
}
return new Header(from, StorageService.VERBS[verbOrdinal], details);
}
+
+ public long serializedSize(Header header, int version)
+ {
+ throw new UnsupportedOperationException();
+ }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/net/Message.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/net/Message.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/net/Message.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/net/Message.java Thu Oct 13 05:14:05 2011
@@ -18,13 +18,9 @@
package org.apache.cassandra.net;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
import java.net.InetAddress;
import org.apache.cassandra.concurrent.Stage;
-import org.apache.cassandra.io.ICompactSerializer;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
Modified: cassandra/trunk/src/java/org/apache/cassandra/service/AntiEntropyService.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/AntiEntropyService.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/AntiEntropyService.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/AntiEntropyService.java Thu Oct 13 05:14:05 2011
@@ -42,7 +42,7 @@ import org.apache.cassandra.dht.Abstract
import org.apache.cassandra.dht.RandomPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.gms.*;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.FastByteArrayInputStream;
import org.apache.cassandra.io.util.FastByteArrayOutputStream;
import org.apache.cassandra.net.CompactEndpointSerializationHelper;
@@ -393,7 +393,7 @@ public class AntiEntropyService
* Handler for requests from remote nodes to generate a valid tree.
* The payload is a CFPair representing the columnfamily to validate.
*/
- public static class TreeRequestVerbHandler implements IVerbHandler, ICompactSerializer<TreeRequest>
+ public static class TreeRequestVerbHandler implements IVerbHandler
{
public static final TreeRequestVerbHandler SERIALIZER = new TreeRequestVerbHandler();
static Message makeVerb(TreeRequest request, int version)
@@ -411,7 +411,7 @@ public class AntiEntropyService
}
}
- public void serialize(TreeRequest request, DataOutputStream dos, int version) throws IOException
+ public void serialize(TreeRequest request, DataOutput dos, int version) throws IOException
{
dos.writeUTF(request.sessionid);
CompactEndpointSerializationHelper.serialize(request.endpoint, dos);
@@ -421,7 +421,7 @@ public class AntiEntropyService
AbstractBounds.serializer().serialize(request.range, dos);
}
- public TreeRequest deserialize(DataInputStream dis, int version) throws IOException
+ public TreeRequest deserialize(DataInput dis, int version) throws IOException
{
String sessId = dis.readUTF();
InetAddress endpoint = CompactEndpointSerializationHelper.deserialize(dis);
@@ -465,7 +465,7 @@ public class AntiEntropyService
* Handler for responses from remote nodes which contain a valid tree.
* The payload is a completed Validator object from the remote endpoint.
*/
- public static class TreeResponseVerbHandler implements IVerbHandler, ICompactSerializer<Validator>
+ public static class TreeResponseVerbHandler implements IVerbHandler
{
public static final TreeResponseVerbHandler SERIALIZER = new TreeResponseVerbHandler();
static Message makeVerb(InetAddress local, Validator validator)
Modified: cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java Thu Oct 13 05:14:05 2011
@@ -68,6 +68,7 @@ import org.apache.cassandra.utils.Pair;
public class StorageProxy implements StorageProxyMBean
{
private static final Logger logger = LoggerFactory.getLogger(StorageProxy.class);
+ private static final boolean OPTIMIZE_LOCAL_REQUESTS = true; // set to false to test messagingservice path on single node
// mbean stuff
private static final LatencyTracker readStats = new LatencyTracker();
@@ -288,7 +289,7 @@ public class StorageProxy implements Sto
{
String dc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(destination);
- if (destination.equals(FBUtilities.getBroadcastAddress()))
+ if (destination.equals(FBUtilities.getBroadcastAddress()) && OPTIMIZE_LOCAL_REQUESTS)
{
insertLocal(rm, responseHandler);
}
@@ -634,7 +635,7 @@ public class StorageProxy implements Sto
// The data-request message is sent to dataPoint, the node that will actually get the data for us
InetAddress dataPoint = handler.endpoints.get(0);
- if (dataPoint.equals(FBUtilities.getBroadcastAddress()))
+ if (dataPoint.equals(FBUtilities.getBroadcastAddress()) && OPTIMIZE_LOCAL_REQUESTS)
{
logger.debug("reading data locally");
StageManager.getStage(Stage.READ).execute(new LocalReadRunnable(command, handler));
Modified: cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java Thu Oct 13 05:14:05 2011
@@ -969,7 +969,7 @@ public class StorageService implements I
if (logger_.isDebugEnabled())
logger_.debug("Node " + endpoint + " state left, token " + token);
- excise(token, endpoint);
+ excise(token, endpoint, extractExpireTime(pieces));
}
/**
@@ -1021,7 +1021,7 @@ public class StorageService implements I
if (VersionedValue.REMOVED_TOKEN.equals(state))
{
- excise(removeToken, endpoint);
+ excise(removeToken, endpoint, extractExpireTime(pieces));
}
else if (VersionedValue.REMOVING_TOKEN.equals(state))
{
@@ -1054,6 +1054,30 @@ public class StorageService implements I
SystemTable.removeToken(token);
}
}
+
+ private void excise(Token token, InetAddress endpoint, long expireTime)
+ {
+ addExpireTimeIfFound(endpoint, expireTime);
+ excise(token, endpoint);
+ }
+
+ protected void addExpireTimeIfFound(InetAddress endpoint, long expireTime)
+ {
+ if (expireTime != 0L)
+ {
+ Gossiper.instance.addExpireTimeForEndpoint(endpoint, expireTime);
+ }
+ }
+
+ protected long extractExpireTime(String[] pieces)
+ {
+ long expireTime = 0L;
+ if (pieces.length >= 3)
+ {
+ expireTime = Long.parseLong(pieces[2]);
+ }
+ return expireTime;
+ }
/**
* Calculate pending ranges according to bootsrapping and leaving nodes. Reasoning is:
@@ -1978,7 +2002,7 @@ public class StorageService implements I
tokenMetadata_.removeEndpoint(FBUtilities.getBroadcastAddress());
calculatePendingRanges();
- Gossiper.instance.addLocalApplicationState(ApplicationState.STATUS, valueFactory.left(getLocalToken()));
+ Gossiper.instance.addLocalApplicationState(ApplicationState.STATUS, valueFactory.left(getLocalToken(),Gossiper.computeExpireTime()));
logger_.info("Announcing that I have left the ring for " + RING_DELAY + "ms");
try
{
Modified: cassandra/trunk/src/java/org/apache/cassandra/streaming/PendingFile.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/streaming/PendingFile.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/streaming/PendingFile.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/streaming/PendingFile.java Thu Oct 13 05:14:05 2011
@@ -21,13 +21,11 @@ package org.apache.cassandra.streaming;
*/
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import java.io.*;
import java.util.ArrayList;
import java.util.List;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.net.MessagingService;
@@ -108,9 +106,9 @@ public class PendingFile
return getFilename() + " sections=" + sections.size() + " progress=" + progress + "/" + size + " - " + progress*100/size + "%";
}
- public static class PendingFileSerializer implements ICompactSerializer<PendingFile>
+ public static class PendingFileSerializer implements IVersionedSerializer<PendingFile>
{
- public void serialize(PendingFile sc, DataOutputStream dos, int version) throws IOException
+ public void serialize(PendingFile sc, DataOutput dos, int version) throws IOException
{
if (sc == null)
{
@@ -131,7 +129,7 @@ public class PendingFile
dos.writeLong(sc.estimatedKeys);
}
- public PendingFile deserialize(DataInputStream dis, int version) throws IOException
+ public PendingFile deserialize(DataInput dis, int version) throws IOException
{
String filename = dis.readUTF();
if (filename.isEmpty())
@@ -152,5 +150,10 @@ public class PendingFile
estimatedKeys = dis.readLong();
return new PendingFile(null, desc, component, sections, type, estimatedKeys);
}
+
+ public long serializedSize(PendingFile pendingFile, int version)
+ {
+ throw new UnsupportedOperationException();
+ }
}
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/streaming/StreamHeader.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/streaming/StreamHeader.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/streaming/StreamHeader.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/streaming/StreamHeader.java Thu Oct 13 05:14:05 2011
@@ -21,26 +21,24 @@ package org.apache.cassandra.streaming;
*
*/
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import java.io.*;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
public class StreamHeader
{
- private static ICompactSerializer<StreamHeader> serializer;
+ private static IVersionedSerializer<StreamHeader> serializer;
static
{
serializer = new StreamHeaderSerializer();
}
- public static ICompactSerializer<StreamHeader> serializer()
+ public static IVersionedSerializer<StreamHeader> serializer()
{
return serializer;
}
@@ -69,9 +67,9 @@ public class StreamHeader
this.pendingFiles = pendingFiles;
}
- private static class StreamHeaderSerializer implements ICompactSerializer<StreamHeader>
+ private static class StreamHeaderSerializer implements IVersionedSerializer<StreamHeader>
{
- public void serialize(StreamHeader sh, DataOutputStream dos, int version) throws IOException
+ public void serialize(StreamHeader sh, DataOutput dos, int version) throws IOException
{
dos.writeUTF(sh.table);
dos.writeLong(sh.sessionId);
@@ -83,7 +81,7 @@ public class StreamHeader
}
}
- public StreamHeader deserialize(DataInputStream dis, int version) throws IOException
+ public StreamHeader deserialize(DataInput dis, int version) throws IOException
{
String table = dis.readUTF();
long sessionId = dis.readLong();
@@ -98,5 +96,10 @@ public class StreamHeader
return new StreamHeader(table, sessionId, file, pendingFiles);
}
+
+ public long serializedSize(StreamHeader streamHeader, int version)
+ {
+ throw new UnsupportedOperationException();
+ }
}
}
\ No newline at end of file
Modified: cassandra/trunk/src/java/org/apache/cassandra/streaming/StreamReply.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/streaming/StreamReply.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/streaming/StreamReply.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/streaming/StreamReply.java Thu Oct 13 05:14:05 2011
@@ -21,11 +21,9 @@ package org.apache.cassandra.streaming;
*/
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import java.io.*;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.FastByteArrayOutputStream;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessageProducer;
@@ -41,7 +39,7 @@ class StreamReply implements MessageProd
SESSION_FINISHED,
}
- public static final ICompactSerializer<StreamReply> serializer = new FileStatusSerializer();
+ public static final IVersionedSerializer<StreamReply> serializer = new FileStatusSerializer();
public final long sessionId;
public final String file;
@@ -72,21 +70,26 @@ class StreamReply implements MessageProd
')';
}
- private static class FileStatusSerializer implements ICompactSerializer<StreamReply>
+ private static class FileStatusSerializer implements IVersionedSerializer<StreamReply>
{
- public void serialize(StreamReply reply, DataOutputStream dos, int version) throws IOException
+ public void serialize(StreamReply reply, DataOutput dos, int version) throws IOException
{
dos.writeLong(reply.sessionId);
dos.writeUTF(reply.file);
dos.writeInt(reply.action.ordinal());
}
- public StreamReply deserialize(DataInputStream dis, int version) throws IOException
+ public StreamReply deserialize(DataInput dis, int version) throws IOException
{
long sessionId = dis.readLong();
String targetFile = dis.readUTF();
Status action = Status.values()[dis.readInt()];
return new StreamReply(targetFile, sessionId, action);
}
+
+ public long serializedSize(StreamReply streamReply, int version)
+ {
+ throw new UnsupportedOperationException();
+ }
}
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/streaming/StreamRequestMessage.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/streaming/StreamRequestMessage.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/streaming/StreamRequestMessage.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/streaming/StreamRequestMessage.java Thu Oct 13 05:14:05 2011
@@ -27,12 +27,11 @@ import java.util.*;
import com.google.common.collect.Iterables;
-import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Range;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.FastByteArrayOutputStream;
import org.apache.cassandra.net.CompactEndpointSerializationHelper;
import org.apache.cassandra.net.Message;
@@ -50,13 +49,13 @@ import org.apache.cassandra.utils.FBUtil
*/
class StreamRequestMessage implements MessageProducer
{
- private static ICompactSerializer<StreamRequestMessage> serializer_;
+ private static IVersionedSerializer<StreamRequestMessage> serializer_;
static
{
serializer_ = new StreamRequestMessageSerializer();
}
- protected static ICompactSerializer<StreamRequestMessage> serializer()
+ protected static IVersionedSerializer<StreamRequestMessage> serializer()
{
return serializer_;
}
@@ -135,9 +134,9 @@ class StreamRequestMessage implements Me
return sb.toString();
}
- private static class StreamRequestMessageSerializer implements ICompactSerializer<StreamRequestMessage>
+ private static class StreamRequestMessageSerializer implements IVersionedSerializer<StreamRequestMessage>
{
- public void serialize(StreamRequestMessage srm, DataOutputStream dos, int version) throws IOException
+ public void serialize(StreamRequestMessage srm, DataOutput dos, int version) throws IOException
{
dos.writeLong(srm.sessionId);
CompactEndpointSerializationHelper.serialize(srm.target, dos);
@@ -168,7 +167,7 @@ class StreamRequestMessage implements Me
}
}
- public StreamRequestMessage deserialize(DataInputStream dis, int version) throws IOException
+ public StreamRequestMessage deserialize(DataInput dis, int version) throws IOException
{
long sessionId = dis.readLong();
InetAddress target = CompactEndpointSerializationHelper.deserialize(dis);
@@ -202,5 +201,10 @@ class StreamRequestMessage implements Me
return new StreamRequestMessage(target, ranges, table, stores, sessionId, type);
}
}
+
+ public long serializedSize(StreamRequestMessage streamRequestMessage, int version)
+ {
+ throw new UnsupportedOperationException();
+ }
}
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/streaming/StreamingRepairTask.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/streaming/StreamingRepairTask.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/streaming/StreamingRepairTask.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/streaming/StreamingRepairTask.java Thu Oct 13 05:14:05 2011
@@ -33,7 +33,7 @@ import org.apache.cassandra.db.Table;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.gms.Gossiper;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.net.*;
import org.apache.cassandra.service.StorageService;
@@ -267,9 +267,9 @@ public class StreamingRepairTask impleme
}
}
- private static class StreamingRepairTaskSerializer implements ICompactSerializer<StreamingRepairTask>
+ private static class StreamingRepairTaskSerializer implements IVersionedSerializer<StreamingRepairTask>
{
- public void serialize(StreamingRepairTask task, DataOutputStream dos, int version) throws IOException
+ public void serialize(StreamingRepairTask task, DataOutput dos, int version) throws IOException
{
UUIDGen.write(task.id, dos);
CompactEndpointSerializationHelper.serialize(task.owner, dos);
@@ -285,14 +285,14 @@ public class StreamingRepairTask impleme
// We don't serialize the callback on purpose
}
- public StreamingRepairTask deserialize(DataInputStream dis, int version) throws IOException
+ public StreamingRepairTask deserialize(DataInput dis, int version) throws IOException
{
UUID id = UUIDGen.read(dis);
InetAddress owner = CompactEndpointSerializationHelper.deserialize(dis);
InetAddress src = CompactEndpointSerializationHelper.deserialize(dis);
InetAddress dst = CompactEndpointSerializationHelper.deserialize(dis);
- String tableName = dis.readUTF(dis);
- String cfName = dis.readUTF(dis);
+ String tableName = dis.readUTF();
+ String cfName = dis.readUTF();
int rangesCount = dis.readInt();
List<Range> ranges = new ArrayList<Range>(rangesCount);
for (int i = 0; i < rangesCount; ++i)
@@ -301,5 +301,10 @@ public class StreamingRepairTask impleme
}
return new StreamingRepairTask(id, owner, src, dst, tableName, cfName, ranges, makeReplyingCallback(owner, id));
}
+
+ public long serializedSize(StreamingRepairTask task, int version)
+ {
+ throw new UnsupportedOperationException();
+ }
}
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/utils/BloomFilter.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/utils/BloomFilter.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/utils/BloomFilter.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/utils/BloomFilter.java Thu Oct 13 05:14:05 2011
@@ -20,7 +20,6 @@ package org.apache.cassandra.utils;
import java.nio.ByteBuffer;
-import org.apache.cassandra.io.ICompactSerializer2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -31,7 +30,7 @@ public class BloomFilter extends Filter
private static final Logger logger = LoggerFactory.getLogger(BloomFilter.class);
private static final int EXCESS = 20;
- static ICompactSerializer2<BloomFilter> serializer_ = new BloomFilterSerializer();
+ static BloomFilterSerializer serializer_ = new BloomFilterSerializer();
public OpenBitSet bitset;
@@ -46,7 +45,7 @@ public class BloomFilter extends Filter
return new BloomFilter(0, bucketsFor(0, 0));
}
- public static ICompactSerializer2<BloomFilter> serializer()
+ public static BloomFilterSerializer serializer()
{
return serializer_;
}
@@ -141,8 +140,8 @@ public class BloomFilter extends Filter
bitset.clear(0, bitset.size());
}
- public int serializedSize()
+ public long serializedSize()
{
- return BloomFilterSerializer.serializedSize(this);
+ return serializer_.serializedSize(this);
}
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/utils/BloomFilterSerializer.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/utils/BloomFilterSerializer.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/utils/BloomFilterSerializer.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/utils/BloomFilterSerializer.java Thu Oct 13 05:14:05 2011
@@ -26,10 +26,10 @@ import java.io.DataOutput;
import java.io.IOException;
import org.apache.cassandra.db.DBConstants;
-import org.apache.cassandra.io.ICompactSerializer2;
+import org.apache.cassandra.io.ISerializer;
import org.apache.cassandra.utils.obs.OpenBitSet;
-class BloomFilterSerializer implements ICompactSerializer2<BloomFilter>
+public class BloomFilterSerializer implements ISerializer<BloomFilter>
{
public void serialize(BloomFilter bf, DataOutput dos) throws IOException
{
@@ -62,7 +62,7 @@ class BloomFilterSerializer implements I
*
* @return serialized size of the given bloom filter
*/
- public static int serializedSize(BloomFilter bf)
+ public long serializedSize(BloomFilter bf)
{
return DBConstants.intSize // hash count
+ DBConstants.intSize // length
Modified: cassandra/trunk/src/java/org/apache/cassandra/utils/EstimatedHistogram.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/utils/EstimatedHistogram.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/utils/EstimatedHistogram.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/utils/EstimatedHistogram.java Thu Oct 13 05:14:05 2011
@@ -24,7 +24,7 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.concurrent.atomic.AtomicLongArray;
-import org.apache.cassandra.io.ICompactSerializer2;
+import org.apache.cassandra.io.ISerializer;
public class EstimatedHistogram
{
@@ -202,7 +202,7 @@ public class EstimatedHistogram
Arrays.equals(getBuckets(false), that.getBuckets(false));
}
- public static class EstimatedHistogramSerializer implements ICompactSerializer2<EstimatedHistogram>
+ public static class EstimatedHistogramSerializer implements ISerializer<EstimatedHistogram>
{
public void serialize(EstimatedHistogram eh, DataOutput dos) throws IOException
{
@@ -228,5 +228,10 @@ public class EstimatedHistogram
}
return new EstimatedHistogram(offsets, buckets);
}
+
+ public long serializedSize(EstimatedHistogram object)
+ {
+ throw new UnsupportedOperationException();
+ }
}
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/utils/FBUtilities.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/utils/FBUtilities.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/utils/FBUtilities.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/utils/FBUtilities.java Thu Oct 13 05:14:05 2011
@@ -49,6 +49,8 @@ import org.apache.cassandra.db.Decorated
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.locator.PropertyFileSnitch;
import org.apache.cassandra.net.IAsyncResult;
import org.apache.thrift.TBase;
@@ -670,4 +672,15 @@ public class FBUtilities
public void close() {}
}
+
+ public static <T> byte[] serialize(T object, IVersionedSerializer<T> serializer, int version) throws IOException
+ {
+ int size = (int) serializer.serializedSize(object, version);
+ DataOutputBuffer buffer = new DataOutputBuffer(size);
+ serializer.serialize(object, buffer, version);
+ assert buffer.getLength() == size && buffer.getData().length == size
+ : String.format("Final buffer length %s to accomodate data size of %s (predicted %s)",
+ buffer.getData().length, buffer.getLength(), size);
+ return buffer.getData();
+ }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/utils/LegacyBloomFilter.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/utils/LegacyBloomFilter.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/utils/LegacyBloomFilter.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/utils/LegacyBloomFilter.java Thu Oct 13 05:14:05 2011
@@ -21,7 +21,7 @@ package org.apache.cassandra.utils;
import java.nio.ByteBuffer;
import java.util.BitSet;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,9 +29,9 @@ public class LegacyBloomFilter extends F
{
private static final int EXCESS = 20;
private static final Logger logger = LoggerFactory.getLogger(LegacyBloomFilter.class);
- static ICompactSerializer<LegacyBloomFilter> serializer_ = new LegacyBloomFilterSerializer();
+ static LegacyBloomFilterSerializer serializer_ = new LegacyBloomFilterSerializer();
- public static ICompactSerializer<LegacyBloomFilter> serializer()
+ public static LegacyBloomFilterSerializer serializer()
{
return serializer_;
}
@@ -121,11 +121,6 @@ public class LegacyBloomFilter extends F
return filter_.toString();
}
- ICompactSerializer<LegacyBloomFilter> tserializer()
- {
- return serializer_;
- }
-
int emptyBuckets()
{
int n = 0;
Modified: cassandra/trunk/src/java/org/apache/cassandra/utils/LegacyBloomFilterSerializer.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/utils/LegacyBloomFilterSerializer.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/utils/LegacyBloomFilterSerializer.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/utils/LegacyBloomFilterSerializer.java Thu Oct 13 05:14:05 2011
@@ -21,16 +21,14 @@ package org.apache.cassandra.utils;
*/
+import java.io.*;
import java.util.BitSet;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.io.ObjectInputStream;
-import org.apache.cassandra.io.ICompactSerializer;
-class LegacyBloomFilterSerializer implements ICompactSerializer<LegacyBloomFilter>
+import org.apache.cassandra.io.IVersionedSerializer;
+
+public class LegacyBloomFilterSerializer
{
- public void serialize(LegacyBloomFilter bf, DataOutputStream dos, int version)
+ public void serialize(LegacyBloomFilter bf, DataOutput dos)
throws IOException
{
throw new UnsupportedOperationException("Shouldn't be serializing legacy bloom filters");
@@ -40,7 +38,7 @@ class LegacyBloomFilterSerializer implem
// oos.flush();
}
- public LegacyBloomFilter deserialize(DataInputStream dis, int version) throws IOException
+ public LegacyBloomFilter deserialize(DataInputStream dis) throws IOException
{
int hashes = dis.readInt();
ObjectInputStream ois = new ObjectInputStream(dis);
@@ -53,4 +51,9 @@ class LegacyBloomFilterSerializer implem
throw new RuntimeException(e);
}
}
+
+ public long serializedSize(LegacyBloomFilter legacyBloomFilter)
+ {
+ throw new UnsupportedOperationException();
+ }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/utils/MerkleTree.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/utils/MerkleTree.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/utils/MerkleTree.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/utils/MerkleTree.java Thu Oct 13 05:14:05 2011
@@ -18,12 +18,7 @@
*/
package org.apache.cassandra.utils;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
-import java.io.Serializable;
+import java.io.*;
import java.util.*;
import com.google.common.collect.AbstractIterator;
@@ -32,7 +27,7 @@ import com.google.common.collect.Peeking
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.net.MessagingService;
/**
@@ -84,7 +79,7 @@ public class MerkleTree implements Seria
private long size;
private Hashable root;
- public static class MerkleTreeSerializer implements ICompactSerializer<MerkleTree>
+ public static class MerkleTreeSerializer
{
public void serialize(MerkleTree mt, DataOutputStream dos, int version) throws IOException
{
@@ -632,7 +627,7 @@ public class MerkleTree implements Seria
private Hashable lchild;
private Hashable rchild;
- private static ICompactSerializer<Inner> serializer = new InnerSerializer();
+ private static InnerSerializer serializer = new InnerSerializer();
/**
* Constructs an Inner with the given token and children, and a null hash.
@@ -701,9 +696,9 @@ public class MerkleTree implements Seria
return buff.toString();
}
- private static class InnerSerializer implements ICompactSerializer<Inner>
+ private static class InnerSerializer
{
- public void serialize(Inner inner, DataOutputStream dos, int version) throws IOException
+ public void serialize(Inner inner, DataOutput dos, int version) throws IOException
{
if (inner.hash == null)
dos.writeInt(-1);
@@ -717,7 +712,7 @@ public class MerkleTree implements Seria
Hashable.serializer.serialize(inner.rchild, dos, version);
}
- public Inner deserialize(DataInputStream dis, int version) throws IOException
+ public Inner deserialize(DataInput dis, int version) throws IOException
{
int hashLen = dis.readInt();
byte[] hash = hashLen >= 0 ? new byte[hashLen] : null;
@@ -744,7 +739,7 @@ public class MerkleTree implements Seria
{
public static final long serialVersionUID = 1L;
static final byte IDENT = 1;
- private static ICompactSerializer<Leaf> serializer = new LeafSerializer();
+ private static LeafSerializer serializer = new LeafSerializer();
/**
* Constructs a null hash.
@@ -775,9 +770,9 @@ public class MerkleTree implements Seria
return "#<Leaf " + Hashable.toString(hash()) + ">";
}
- private static class LeafSerializer implements ICompactSerializer<Leaf>
+ private static class LeafSerializer
{
- public void serialize(Leaf leaf, DataOutputStream dos, int version) throws IOException
+ public void serialize(Leaf leaf, DataOutput dos) throws IOException
{
if (leaf.hash == null)
dos.writeInt(-1);
@@ -788,7 +783,7 @@ public class MerkleTree implements Seria
}
}
- public Leaf deserialize(DataInputStream dis, int version) throws IOException
+ public Leaf deserialize(DataInput dis) throws IOException
{
int hashLen = dis.readInt();
byte[] hash = hashLen < 0 ? null : new byte[hashLen];
@@ -827,7 +822,7 @@ public class MerkleTree implements Seria
static abstract class Hashable implements Serializable
{
private static final long serialVersionUID = 1L;
- private static ICompactSerializer<Hashable> serializer = new HashableSerializer();
+ private static IVersionedSerializer<Hashable> serializer = new HashableSerializer();
protected byte[] hash;
@@ -886,9 +881,9 @@ public class MerkleTree implements Seria
return "[" + Hex.bytesToHex(hash) + "]";
}
- private static class HashableSerializer implements ICompactSerializer<Hashable>
+ private static class HashableSerializer implements IVersionedSerializer<Hashable>
{
- public void serialize(Hashable h, DataOutputStream dos, int version) throws IOException
+ public void serialize(Hashable h, DataOutput dos, int version) throws IOException
{
if (h instanceof Inner)
{
@@ -898,22 +893,27 @@ public class MerkleTree implements Seria
else if (h instanceof Leaf)
{
dos.writeByte(Leaf.IDENT);
- Leaf.serializer.serialize((Leaf)h, dos, version);
+ Leaf.serializer.serialize((Leaf)h, dos);
}
else
throw new IOException("Unexpected Hashable: " + h.getClass().getCanonicalName());
}
- public Hashable deserialize(DataInputStream dis, int version) throws IOException
+ public Hashable deserialize(DataInput dis, int version) throws IOException
{
byte ident = dis.readByte();
if (Inner.IDENT == ident)
return Inner.serializer.deserialize(dis, version);
else if (Leaf.IDENT == ident)
- return Leaf.serializer.deserialize(dis, version);
+ return Leaf.serializer.deserialize(dis);
else
throw new IOException("Unexpected Hashable: " + ident);
}
+
+ public long serializedSize(Hashable hashable, int version)
+ {
+ throw new UnsupportedOperationException();
+ }
}
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/utils/UUIDGen.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/utils/UUIDGen.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/utils/UUIDGen.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/utils/UUIDGen.java Thu Oct 13 05:14:05 2011
@@ -20,9 +20,7 @@ package org.apache.cassandra.utils;
*
*/
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import java.io.*;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
@@ -71,13 +69,13 @@ public class UUIDGen
}
/** reads a uuid from an input stream. */
- public static UUID read(DataInputStream dis) throws IOException
+ public static UUID read(DataInput dis) throws IOException
{
return new UUID(dis.readLong(), dis.readLong());
}
/** writes a uuid to an output stream. */
- public static void write(UUID uuid, DataOutputStream dos) throws IOException
+ public static void write(UUID uuid, DataOutput dos) throws IOException
{
dos.writeLong(uuid.getMostSignificantBits());
dos.writeLong(uuid.getLeastSignificantBits());
@@ -142,7 +140,7 @@ public class UUIDGen
*
* @param uuid a type-1 (time-based) UUID
* @return the number of milliseconds since the unix epoch
- * @throws InvalidArgumentException if the UUID is not version 1
+ * @throws IllegalArgumentException if the UUID is not version 1
*/
public static long getAdjustedTimestamp(UUID uuid)
{
Modified: cassandra/trunk/test/unit/org/apache/cassandra/gms/GossipDigestTest.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/unit/org/apache/cassandra/gms/GossipDigestTest.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/test/unit/org/apache/cassandra/gms/GossipDigestTest.java (original)
+++ cassandra/trunk/test/unit/org/apache/cassandra/gms/GossipDigestTest.java Thu Oct 13 05:14:05 2011
@@ -35,7 +35,6 @@ import org.junit.Test;
public class GossipDigestTest
{
-
@Test
public void test() throws IOException
{
@@ -56,5 +55,4 @@ public class GossipDigestTest
GossipDigest actual = GossipDigest.serializer().deserialize(new DataInputStream(input), MessagingService.version_);
assertEquals(0, expected.compareTo(actual));
}
-
}
Modified: cassandra/trunk/test/unit/org/apache/cassandra/io/CompactSerializerTest.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/unit/org/apache/cassandra/io/CompactSerializerTest.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/test/unit/org/apache/cassandra/io/CompactSerializerTest.java (original)
+++ cassandra/trunk/test/unit/org/apache/cassandra/io/CompactSerializerTest.java Thu Oct 13 05:14:05 2011
@@ -61,18 +61,12 @@ public class CompactSerializerTest exten
expectedClassNames.add("VersionedValueSerializer");
expectedClassNames.add("HeaderSerializer");
expectedClassNames.add("MessageSerializer");
- expectedClassNames.add("TreeRequestVerbHandler");
- expectedClassNames.add("TreeResponseVerbHandler");
expectedClassNames.add("PendingFileSerializer");
expectedClassNames.add("StreamHeaderSerializer");
expectedClassNames.add("FileStatusSerializer");
expectedClassNames.add("StreamRequestMessageSerializer");
- expectedClassNames.add("LegacyBloomFilterSerializer");
expectedClassNames.add("CounterMutationSerializer");
expectedClassNames.add("HashableSerializer");
- expectedClassNames.add("InnerSerializer");
- expectedClassNames.add("LeafSerializer");
- expectedClassNames.add("MerkleTreeSerializer");
expectedClassNames.add("StreamingRepairTaskSerializer");
discoveredClassNames = new ArrayList<String>();
@@ -107,7 +101,7 @@ public class CompactSerializerTest exten
if(t instanceof ParameterizedType)
{
ParameterizedType pt = (ParameterizedType)t;
- if (((Class)pt.getRawType()).getSimpleName().equals("ICompactSerializer"))
+ if (((Class)pt.getRawType()).getSimpleName().equals("IVersionedSerializer"))
{
discoveredClassNames.add(simpleName);
break classTraversal;
Modified: cassandra/trunk/test/unit/org/apache/cassandra/net/MessageSerializer.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/unit/org/apache/cassandra/net/MessageSerializer.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/test/unit/org/apache/cassandra/net/MessageSerializer.java (original)
+++ cassandra/trunk/test/unit/org/apache/cassandra/net/MessageSerializer.java Thu Oct 13 05:14:05 2011
@@ -21,15 +21,13 @@ package org.apache.cassandra.net;
*/
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import java.io.*;
-import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.IVersionedSerializer;
-public class MessageSerializer implements ICompactSerializer<Message>
+public class MessageSerializer implements IVersionedSerializer<Message>
{
- public void serialize(Message t, DataOutputStream dos, int version) throws IOException
+ public void serialize(Message t, DataOutput dos, int version) throws IOException
{
assert t.getVersion() == version : "internode protocol version mismatch"; // indicates programmer error.
Header.serializer().serialize( t.header_, dos, version);
@@ -38,7 +36,7 @@ public class MessageSerializer implement
dos.write(bytes);
}
- public Message deserialize(DataInputStream dis, int version) throws IOException
+ public Message deserialize(DataInput dis, int version) throws IOException
{
Header header = Header.serializer().deserialize(dis, version);
int size = dis.readInt();
@@ -46,4 +44,9 @@ public class MessageSerializer implement
dis.readFully(bytes);
return new Message(header, bytes, version);
}
+
+ public long serializedSize(Message message, int version)
+ {
+ throw new UnsupportedOperationException();
+ }
}
Modified: cassandra/trunk/test/unit/org/apache/cassandra/service/LeaveAndBootstrapTest.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/unit/org/apache/cassandra/service/LeaveAndBootstrapTest.java?rev=1182654&r1=1182653&r2=1182654&view=diff
==============================================================================
--- cassandra/trunk/test/unit/org/apache/cassandra/service/LeaveAndBootstrapTest.java (original)
+++ cassandra/trunk/test/unit/org/apache/cassandra/service/LeaveAndBootstrapTest.java Thu Oct 13 05:14:05 2011
@@ -35,6 +35,7 @@ import org.apache.cassandra.Util;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.dht.*;
import org.apache.cassandra.gms.ApplicationState;
+import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.VersionedValue;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.SimpleSnitch;
@@ -299,8 +300,10 @@ public class LeaveAndBootstrapTest exten
// Now finish node 6 and node 9 leaving, as well as boot1 (after this node 8 is still
// leaving and boot2 in progress
- ss.onChange(hosts.get(LEAVING[0]), ApplicationState.STATUS, valueFactory.left(endpointTokens.get(LEAVING[0])));
- ss.onChange(hosts.get(LEAVING[2]), ApplicationState.STATUS, valueFactory.left(endpointTokens.get(LEAVING[2])));
+ ss.onChange(hosts.get(LEAVING[0]), ApplicationState.STATUS,
+ valueFactory.left(endpointTokens.get(LEAVING[0]), Gossiper.computeExpireTime()));
+ ss.onChange(hosts.get(LEAVING[2]), ApplicationState.STATUS,
+ valueFactory.left(endpointTokens.get(LEAVING[2]), Gossiper.computeExpireTime()));
ss.onChange(boot1, ApplicationState.STATUS, valueFactory.normal(keyTokens.get(5)));
// adjust precalcuated results. this changes what the epected endpoints are.
@@ -506,7 +509,8 @@ public class LeaveAndBootstrapTest exten
// node 3 goes through leave and left and then jumps to normal at its new token
ss.onChange(hosts.get(2), ApplicationState.STATUS, valueFactory.leaving(keyTokens.get(2)));
- ss.onChange(hosts.get(2), ApplicationState.STATUS, valueFactory.left(keyTokens.get(2)));
+ ss.onChange(hosts.get(2), ApplicationState.STATUS,
+ valueFactory.left(keyTokens.get(2), Gossiper.computeExpireTime()));
ss.onChange(hosts.get(2), ApplicationState.STATUS, valueFactory.normal(keyTokens.get(4)));
assertTrue(tmd.getBootstrapTokens().isEmpty());
@@ -556,7 +560,8 @@ public class LeaveAndBootstrapTest exten
assertTrue(tmd.getBootstrapTokens().isEmpty());
// go to state left
- ss.onChange(hosts.get(2), ApplicationState.STATUS, valueFactory.left(keyTokens.get(1)));
+ ss.onChange(hosts.get(2), ApplicationState.STATUS,
+ valueFactory.left(keyTokens.get(1), Gossiper.computeExpireTime()));
assertFalse(tmd.isMember(hosts.get(2)));
assertFalse(tmd.isLeaving(hosts.get(2)));
@@ -583,7 +588,8 @@ public class LeaveAndBootstrapTest exten
Util.createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, 7);
// node hosts.get(2) goes jumps to left
- ss.onChange(hosts.get(2), ApplicationState.STATUS, valueFactory.left(endpointTokens.get(2)));
+ ss.onChange(hosts.get(2), ApplicationState.STATUS,
+ valueFactory.left(endpointTokens.get(2), Gossiper.computeExpireTime()));
assertFalse(tmd.isMember(hosts.get(2)));
@@ -595,7 +601,8 @@ public class LeaveAndBootstrapTest exten
assertTrue(tmd.getBootstrapTokens().get(keyTokens.get(1)).equals(hosts.get(3)));
// and then directly to 'left'
- ss.onChange(hosts.get(2), ApplicationState.STATUS, valueFactory.left(keyTokens.get(1)));
+ ss.onChange(hosts.get(2), ApplicationState.STATUS,
+ valueFactory.left(keyTokens.get(1), Gossiper.computeExpireTime()));
assertTrue(tmd.getBootstrapTokens().size() == 0);
assertFalse(tmd.isMember(hosts.get(2)));