diff --git a/contrib/maven/pom.xml b/contrib/maven/pom.xml
new file mode 100644
index 0000000000..5d98eb76a8
--- /dev/null
+++ b/contrib/maven/pom.xml
@@ -0,0 +1,551 @@
+
+
+
+
+ org.apache
+ apache
+ 6
+
+ 4.0.0
+
+ org.apache.cassandra
+ cassandra
+ 0.6-SNAPSHOT
+ jar
+ Cassandra
+ 2009
+
+
+ 2.0.9
+
+
+ http://incubator.apache.org/cassandra
+
+
+
+ cassandra-user
+ cassandra-user-subscribe@incubator.apache.org
+ cassandra-user-unsubscribe@incubator.apache.org
+ cassandra-user@incubator.apache.org
+ http://mail-archives.apache.org/mod_mbox/incubator-cassandra-user/
+
+
+ cassandra-dev
+ cassandra-dev-subscribe@incubator.apache.org
+ cassandra-dev-unsubscribe@incubator.apache.org
+ cassandra-dev@incubator.apache.org
+ http://mail-archives.apache.org/mod_mbox/incubator-cassandra-dev/
+
+
+ cassandra-commits
+ cassandra-commits-subscribe@incubator.apache.org
+ cassandra-commits-unsubscribe@incubator.apache.org
+ cassandra-commits@incubator.apache.org
+ http://mail-archives.apache.org/mod_mbox/incubator-cassandra-commits/
+
+
+
+
+
+
+
+
+
+
+ JIRA
+ https://issues.apache.org/jira/browse/CASSANDRA
+
+
+
+
+ hudson
+ http://hudson.zones.apache.org/hudson
+
+
+
+ mail
+ true
+ true
+ false
+ false
+ cassandra-commits@incubator.apache.org
+
+
+
+
+
+
+ cassandra-website
+ scp://people.apache.org/x1/www/incubator.apache.org/cassandra/maven/${pom.version}
+
+
+
+
+ scm:svn:http://svn.apache.org/repos/asf/incubator/cassandra/trunk
+ scm:svn:https://svn.apache.org/repos/asf/incubator/cassandra/trunk
+ http://svn.apache.org/viewvc/incubator/cassandra/trunk/
+
+
+
+
+
+ commons-collections
+ commons-collections
+ 3.2.1
+
+
+ commons-cli
+ commons-cli
+ 1.1
+
+
+ commons-lang
+ commons-lang
+ 2.4
+
+
+ jline
+ jline
+ 0.9.94
+
+
+ log4j
+ log4j
+ 1.2.15
+
+
+
+ javax.jms
+ jms
+
+
+ com.sun.jmx
+ jmxri
+
+
+ com.sun.jdmk
+ jmxtools
+
+
+
+
+ org.slf4j
+ slf4j-api
+ 1.5.8
+
+
+ org.slf4j
+ slf4j-log4j12
+ 1.5.8
+
+
+ org.antlr
+ antlr-runtime
+ 3.1.3
+
+
+ com.google.collections
+ google-collections
+ 1.0-rc1
+
+
+
+
+
+ high-scale-lib
+ high-scale-lib
+ UNKNOWN
+ system
+ ${basedir}/lib/high-scale-lib.jar
+
+
+ flexjson
+ flexjson
+ 1.7
+ system
+ ${basedir}/lib/flexjson-1.7.jar
+
+
+ libthrift
+ libthrift
+ UNKNOWN
+ system
+ ${basedir}/lib/libthrift-r820831.jar
+
+
+ jsonsimple
+ jsonsimple
+ UNKNOWN
+ system
+ ${basedir}/lib/json_simple-1.1.jar
+
+
+ com.reardencommerce
+ clhm
+ UNKNOWN
+ system
+ ${basedir}/lib/clhm-production.jar
+
+
+
+
+ junit
+ junit
+ 4.6
+ test
+
+
+
+
+
+ ${basedir}/src/java
+ ${basedir}/test/unit
+ build/classes
+
+
+
+ ${basedir}/test/conf
+
+ **/*
+
+
+
+ ${basedir}/test/resources
+
+ *.json
+
+
+
+
+
+
+
+
+ org.antlr
+ antlr3-maven-plugin
+ 3.1.3-1
+
+
+ process-sources
+
+ antlr
+
+
+ ${basedir}/src/java
+
+
+
+
+
+
+
+ org.codehaus.mojo
+ build-helper-maven-plugin
+ 1.3
+
+
+ add-source
+ generate-sources
+
+ add-source
+
+
+
+ ${basedir}/interface/gen-java
+
+
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-compiler-plugin
+
+ 1.6
+ 1.6
+ true
+ true
+ true
+ true
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-surefire-plugin
+
+
+
+ storage-config
+ ${basedir}/test/conf
+
+
+ always
+
+ **/TestRingCache.java
+
+
+
+
+
+
+ org.codehaus.mojo
+ cobertura-maven-plugin
+ 2.0
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-release-plugin
+ 2.0-beta-9
+
+ true
+ false
+ clean install
+ deploy
+ -Papache-release
+
+
+
+ org.codehaus.mojo
+ ianal-maven-plugin
+ 1.0-alpha-1
+
+
+ org.codehaus.mojo
+ rat-maven-plugin
+ 1.0-alpha-3
+
+ false
+
+
+
+ org.apache.maven.plugins
+ maven-enforcer-plugin
+
+
+ validate
+
+ enforce
+
+
+
+
+ [2.0.9,)
+
+
+
+
+
+
+
+ org.codehaus.mojo
+ ianal-maven-plugin
+
+
+
+ verify-legal-files
+
+
+ true
+
+
+
+
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-jxr-plugin
+
+
+ org.apache.maven.plugins
+ maven-surefire-report-plugin
+
+
+ org.apache.maven.plugins
+ maven-pmd-plugin
+
+
+ org.codehaus.mojo
+ taglist-maven-plugin
+
+
+ org.apache.maven.plugins
+ maven-javadoc-plugin
+
+
+ http://java.sun.com/j2se/1.6.0/docs/api/
+ http://logging.apache.org/log4j/docs/api/
+
+
+ true
+ 900m
+ 1.6
+
+
+
+ org.codehaus.mojo
+ cobertura-maven-plugin
+ 2.2
+
+
+ html
+ xml
+
+
+
+
+
+
+
+
+
+
+ thrift
+
+ process-sources
+
+
+
+ maven-antrun-plugin
+
+
+ process-sources
+
+
+
+
+
+
+
+ run
+
+
+
+
+
+
+
+
+
+
+ apache-release
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-gpg-plugin
+
+ ${gpg.passphrase}
+
+
+
+
+ sign
+
+
+
+
+
+
+
+ true
+ org.apache.maven.plugins
+ maven-deploy-plugin
+
+ true
+
+
+
+ org.apache.maven.plugins
+ maven-source-plugin
+
+
+ attach-sources
+
+ jar
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-javadoc-plugin
+
+ ${project.build.sourceEncoding}
+
+
+
+ attach-javadocs
+
+ jar
+
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-assembly-plugin
+
+
+
+ single
+
+ package
+
+ true
+
+
+ source-release
+
+
+
+
+
+
+
+ org.apache.geronimo.genesis
+ apache-source-release-assembly-descriptor
+ 2.0
+
+
+
+
+
+
+
+
+
+
+
diff --git a/src/java/org/apache/cassandra/db/RowMutation.java b/src/java/org/apache/cassandra/db/RowMutation.java
index 60e0a0177b..5217d83ac0 100644
--- a/src/java/org/apache/cassandra/db/RowMutation.java
+++ b/src/java/org/apache/cassandra/db/RowMutation.java
@@ -47,7 +47,6 @@ public class RowMutation
{
private static ICompactSerializer serializer_;
public static final String HINT = "HINT";
- public static final String FORWARD_HEADER = "FORWARD";
static
{
diff --git a/src/java/org/apache/cassandra/db/RowMutationVerbHandler.java b/src/java/org/apache/cassandra/db/RowMutationVerbHandler.java
index 92b89b4b5d..bd6d8f29d2 100644
--- a/src/java/org/apache/cassandra/db/RowMutationVerbHandler.java
+++ b/src/java/org/apache/cassandra/db/RowMutationVerbHandler.java
@@ -18,23 +18,25 @@
package org.apache.cassandra.db;
-import java.io.ByteArrayInputStream;
-import java.io.DataInputStream;
-import java.io.IOException;
+import java.io.*;
+
import java.net.InetAddress;
-import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import com.google.common.base.Charsets;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
-import org.apache.cassandra.net.MessagingService;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.net.*;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
+import static com.google.common.base.Charsets.UTF_8;
+
public class RowMutationVerbHandler implements IVerbHandler
{
@@ -67,11 +69,6 @@ public class RowMutationVerbHandler implements IVerbHandler
hintedMutation.apply();
}
}
-
- // Check if there were any forwarding headers in this message
- byte[] forwardBytes = message.getHeader(RowMutation.FORWARD_HEADER);
- if (forwardBytes != null)
- forwardToLocalNodes(message, forwardBytes);
Table.open(rm.getTable()).apply(rm, bytes, true);
@@ -85,34 +82,5 @@ public class RowMutationVerbHandler implements IVerbHandler
{
logger_.error("Error in row mutation", e);
}
- }
-
- private void forwardToLocalNodes(Message message, byte[] forwardBytes) throws UnknownHostException
- {
- // remove fwds from message to avoid infinite loop
- message.setHeader(RowMutation.FORWARD_HEADER, null);
-
- int bytesPerInetAddress = FBUtilities.getLocalAddress().getAddress().length;
- assert forwardBytes.length >= bytesPerInetAddress;
- assert forwardBytes.length % bytesPerInetAddress == 0;
-
- int offset = 0;
- byte[] addressBytes = new byte[bytesPerInetAddress];
-
- // Send a message to each of the addresses on our Forward List
- while (offset < forwardBytes.length)
- {
- System.arraycopy(forwardBytes, offset, addressBytes, 0, bytesPerInetAddress);
- InetAddress address = InetAddress.getByAddress(addressBytes);
-
- if (logger_.isDebugEnabled())
- logger_.debug("Forwarding message to " + address);
-
- // Send the original message to the address specified by the FORWARD_HINT
- // Let the response go back to the coordinator
- MessagingService.instance.sendOneWay(message, message.getFrom());
-
- offset += bytesPerInetAddress;
- }
}
}
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index cc652ae3ee..9b96607d55 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -27,23 +27,20 @@ import java.util.concurrent.*;
import javax.management.MBeanServer;
import javax.management.ObjectName;
-import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
+import static com.google.common.base.Charsets.UTF_8;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.*;
-import org.apache.cassandra.db.filter.QueryFilter;
-import org.apache.cassandra.dht.AbstractBounds;
-import org.apache.cassandra.dht.Bounds;
-import org.apache.cassandra.dht.IPartitioner;
-import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.dht.*;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.TokenMetadata;
@@ -56,8 +53,7 @@ import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.LatencyTracker;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.WrappedRunnable;
-
-import static com.google.common.base.Charsets.UTF_8;
+import org.apache.cassandra.db.filter.QueryFilter;
public class StorageProxy implements StorageProxyMBean
{
@@ -94,14 +90,13 @@ public class StorageProxy implements StorageProxyMBean
* @param mutations the mutations to be applied across the replicas
* @param consistency_level the consistency level for the operation
*/
- public static void mutate(List mutations, ConsistencyLevel consistencyLevel) throws UnavailableException, TimeoutException
+ public static void mutate(List mutations, ConsistencyLevel consistency_level) throws UnavailableException, TimeoutException
{
long startTime = System.nanoTime();
- List responseHandlers = new ArrayList();
+ ArrayList responseHandlers = new ArrayList();
RowMutation mostRecentRowMutation = null;
StorageService ss = StorageService.instance;
- String localDataCenter = getDataCenter(FBUtilities.getLocalAddress());
try
{
@@ -115,67 +110,58 @@ public class StorageProxy implements StorageProxyMBean
Collection writeEndpoints = ss.getTokenMetadata().getWriteEndpoints(StorageService.getPartitioner().getToken(rm.key()), table, naturalEndpoints);
Multimap hintedEndpoints = rs.getHintedEndpoints(writeEndpoints);
- final IWriteResponseHandler responseHandler = rs.getWriteResponseHandler(writeEndpoints, hintedEndpoints, consistencyLevel);
-
- // exit early if we can't fulfuill the CL at this time
+ // send out the writes, as in mutate() above, but this time with a callback that tracks responses
+ final IWriteResponseHandler responseHandler = rs.getWriteResponseHandler(writeEndpoints, hintedEndpoints, consistency_level);
responseHandler.assureSufficientLiveNodes();
-
+
responseHandlers.add(responseHandler);
-
- // Creates a Multimap that holds onto all the messages and addresses meant for a specific datacenter.
- Multimap> dcMap = groupEndpointsByDataCenter(rm, hintedEndpoints, responseHandler);
-
- // Traverse all dataCenters where messages will be sent to.
- for (Map.Entry>> entry : dcMap.asMap().entrySet())
+ Message unhintedMessage = null;
+ for (Map.Entry> entry : hintedEndpoints.asMap().entrySet())
{
- String dataCenter = entry.getKey();
-
- // Grab a set of all the messages bound for this dataCenter and create an iterator over this set.
- Collection> messagesForDataCenter = entry.getValue();
- Iterator> iter = messagesForDataCenter.iterator();
- assert iter.hasNext();
+ InetAddress destination = entry.getKey();
+ Collection targets = entry.getValue();
- // First endpoint in list is the destination for this group
- Pair messageAndDestination = iter.next();
-
- Message primaryMessage = messageAndDestination.left;
- InetAddress target = messageAndDestination.right;
-
- // Add all the other destinations that are bound for the same dataCenter as a header in the primary message.
- while (iter.hasNext())
+ if (targets.size() == 1 && targets.iterator().next().equals(destination))
{
- messageAndDestination = iter.next();
- assert messageAndDestination.left == primaryMessage;
-
- if (dataCenter.equals(localDataCenter))
+ // unhinted writes
+ if (destination.equals(FBUtilities.getLocalAddress()))
{
- // direct write to local DC
- assert primaryMessage.getHeader(RowMutation.FORWARD_HEADER) == null;
- MessagingService.instance.sendOneWay(primaryMessage, target);
+ insertLocalMessage(rm, responseHandler);
}
else
{
- // group all nodes in this DC as forward headers on the primary message
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- DataOutputStream dos = new DataOutputStream(bos);
-
- // append to older addresses
- byte[] previousHints = primaryMessage.getHeader(RowMutation.FORWARD_HEADER);
- if (previousHints != null)
- dos.write(previousHints);
-
- dos.write(messageAndDestination.right.getAddress());
- primaryMessage.setHeader(RowMutation.FORWARD_HEADER, bos.toByteArray());
+ // belongs on a different server. send it there.
+ if (unhintedMessage == null)
+ {
+ unhintedMessage = rm.makeRowMutationMessage();
+ MessagingService.instance.addCallback(responseHandler, unhintedMessage.getMessageId());
+ }
+ if (logger.isDebugEnabled())
+ logger.debug("insert writing key " + FBUtilities.bytesToHex(rm.key()) + " to " + unhintedMessage.getMessageId() + "@" + destination);
+ MessagingService.instance.sendOneWay(unhintedMessage, destination);
}
- }
-
- MessagingService.instance.sendOneWay(primaryMessage, target);
+ }
+ else
+ {
+ // hinted
+ Message hintedMessage = rm.makeRowMutationMessage();
+ for (InetAddress target : targets)
+ {
+ if (!target.equals(destination))
+ {
+ addHintHeader(hintedMessage, target);
+ if (logger.isDebugEnabled())
+ logger.debug("insert writing key " + FBUtilities.bytesToHex(rm.key()) + " to " + hintedMessage.getMessageId() + "@" + destination + " for " + target);
+ }
+ }
+ responseHandler.addHintCallback(hintedMessage, destination);
+ MessagingService.instance.sendOneWay(hintedMessage, destination);
+ }
}
}
-
// wait for writes. throws timeoutexception if necessary
for (IWriteResponseHandler responseHandler : responseHandlers)
- {
+ {
responseHandler.get();
}
}
@@ -192,66 +178,6 @@ public class StorageProxy implements StorageProxyMBean
}
}
-
- private static Multimap> groupEndpointsByDataCenter(RowMutation rm, Multimap endpoints, final IWriteResponseHandler responseHandler) throws IOException
- {
-
- Set>> endpointSet = endpoints.asMap().entrySet();
- Multimap> dcMap = HashMultimap.create(endpointSet.size(), 10);
- Message unhintedMessage = null;
-
- for (Map.Entry> entry : endpointSet)
- {
- InetAddress destination = entry.getKey();
- Collection targets = entry.getValue();
-
- String dataCenter = getDataCenter(destination);
-
- if (targets.size() == 1 && targets.iterator().next().equals(destination))
- {
- // unhinted writes
- if (destination.equals(FBUtilities.getLocalAddress()))
- {
- insertLocalMessage(rm, responseHandler);
- }
- else
- {
- // belongs on a different server.
- if (unhintedMessage == null)
- {
- unhintedMessage = rm.makeRowMutationMessage();
- MessagingService.instance.addCallback(responseHandler, unhintedMessage.getMessageId());
- }
-
- if (logger.isDebugEnabled())
- logger.debug("insert writing key " + FBUtilities.bytesToHex(rm.key()) + " to " + unhintedMessage.getMessageId() + "@" + destination);
-
- dcMap.put(dataCenter, new Pair(unhintedMessage, destination));
- }
- }
- else
- {
- // hinted
- Message hintedMessage = rm.makeRowMutationMessage();
-
- for (InetAddress target : targets)
- {
- if (!target.equals(destination))
- {
- addHintHeader(hintedMessage, target);
- if (logger.isDebugEnabled())
- logger.debug("insert writing key " + FBUtilities.bytesToHex(rm.key()) + " to " + hintedMessage.getMessageId() + "@" + destination + " for " + target);
- }
- }
-
- responseHandler.addHintCallback(hintedMessage, destination);
- dcMap.put(dataCenter, new Pair(hintedMessage, destination));
- }
- }
-
- return dcMap;
- }
-
private static void addHintHeader(Message message, InetAddress target) throws IOException
{
@@ -266,22 +192,6 @@ public class StorageProxy implements StorageProxyMBean
message.setHeader(RowMutation.HINT, bos.toByteArray());
}
- private static String getDataCenter(InetAddress addr)
- {
- String dataCenter = null;
- try
- {
- dataCenter = DatabaseDescriptor.getEndpointSnitch().getDatacenter(addr);
- }
- catch (UnsupportedOperationException e)
- {
- // SimpleSnitch throws this
- dataCenter = "default";
- }
-
- return dataCenter;
- }
-
private static void insertLocalMessage(final RowMutation rm, final IWriteResponseHandler responseHandler)
{
if (logger.isDebugEnabled())