natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

revert last 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1041489 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-12-02 17:42:15 +00:00
parent af5e0b7dd4
commit f0104cdce5
4 changed files with 604 additions and 176 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

551
contrib/maven/pom.xml Normal file
View File

@ -0,0 +1,551 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
this work for additional information regarding copyright ownership.
The ASF licenses this file to You under the Apache License, Version 2.0
(the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<parent>
<groupId>org.apache</groupId>
<artifactId>apache</artifactId>
<version>6</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>org.apache.cassandra</groupId>
<artifactId>cassandra</artifactId>
<version>0.6-SNAPSHOT</version>
<packaging>jar</packaging>
<name>Cassandra</name>
<inceptionYear>2009</inceptionYear>
<prerequisites>
<maven>2.0.9</maven>
</prerequisites>
<url>http://incubator.apache.org/cassandra</url>
<mailingLists>
<mailingList>
<name>cassandra-user</name>
<subscribe>cassandra-user-subscribe@incubator.apache.org</subscribe>
<unsubscribe>cassandra-user-unsubscribe@incubator.apache.org</unsubscribe>
<post>cassandra-user@incubator.apache.org</post>
<archive>http://mail-archives.apache.org/mod_mbox/incubator-cassandra-user/</archive>
</mailingList>
<mailingList>
<name>cassandra-dev</name>
<subscribe>cassandra-dev-subscribe@incubator.apache.org</subscribe>
<unsubscribe>cassandra-dev-unsubscribe@incubator.apache.org</unsubscribe>
<post>cassandra-dev@incubator.apache.org</post>
<archive>http://mail-archives.apache.org/mod_mbox/incubator-cassandra-dev/</archive>
</mailingList>
<mailingList>
<name>cassandra-commits</name>
<subscribe>cassandra-commits-subscribe@incubator.apache.org</subscribe>
<unsubscribe>cassandra-commits-unsubscribe@incubator.apache.org</unsubscribe>
<post>cassandra-commits@incubator.apache.org</post>
<archive>http://mail-archives.apache.org/mod_mbox/incubator-cassandra-commits/</archive>
</mailingList>
</mailingLists>
<!-- This section intentionally left blank (see: CASSANDRA-415) -->
<developers>
<developer>
</developer>
</developers>
<issueManagement>
<system>JIRA</system>
<url>https://issues.apache.org/jira/browse/CASSANDRA</url>
</issueManagement>
<!-- see: https://hudson.dev.java.net/maven-hudson-plugin (to kick off hudson jobs via mvn) -->
<ciManagement>
<system>hudson</system>
<url>http://hudson.zones.apache.org/hudson</url>
<notifiers>
<!-- Hopefully one day hudson will automatically load this data -->
<notifier>
<type>mail</type>
<sendOnError>true</sendOnError>
<sendOnFailure>true</sendOnFailure>
<sendOnSuccess>false</sendOnSuccess>
<sendOnWarning>false</sendOnWarning>
<configuration><address>cassandra-commits@incubator.apache.org</address></configuration>
</notifier>
</notifiers>
</ciManagement>
<distributionManagement>
<site>
<id>cassandra-website</id>
<url>scp://people.apache.org/x1/www/incubator.apache.org/cassandra/maven/${pom.version}</url>
</site>
</distributionManagement>
<scm>
<connection>scm:svn:http://svn.apache.org/repos/asf/incubator/cassandra/trunk</connection>
<developerConnection>scm:svn:https://svn.apache.org/repos/asf/incubator/cassandra/trunk</developerConnection>
<url>http://svn.apache.org/viewvc/incubator/cassandra/trunk/</url>
</scm>
<dependencies>
<dependency>
<groupId>commons-collections</groupId>
<artifactId>commons-collections</artifactId>
<version>3.2.1</version>
</dependency>
<dependency>
<groupId>commons-cli</groupId>
<artifactId>commons-cli</artifactId>
<version>1.1</version>
</dependency>
<dependency>
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
<version>2.4</version>
</dependency>
<dependency>
<groupId>jline</groupId>
<artifactId>jline</artifactId>
<version>0.9.94</version>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.15</version>
<!-- And the exclusion below illustrate why we should try to make as many of our jars optional -->
<exclusions>
<exclusion>
<groupId>javax.jms</groupId>
<artifactId>jms</artifactId>
</exclusion>
<exclusion>
<groupId>com.sun.jmx</groupId>
<artifactId>jmxri</artifactId>
</exclusion>
<exclusion>
<groupId>com.sun.jdmk</groupId>
<artifactId>jmxtools</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.5.8</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.5.8</version>
</dependency>
<dependency>
<groupId>org.antlr</groupId>
<artifactId>antlr-runtime</artifactId>
<version>3.1.3</version>
</dependency>
<dependency>
<groupId>com.google.collections</groupId>
<artifactId>google-collections</artifactId>
<version>1.0-rc1</version>
</dependency>
<!-- Dependencies which I could not find in the maven repos -->
<dependency>
<groupId>high-scale-lib</groupId>
<artifactId>high-scale-lib</artifactId>
<version>UNKNOWN</version>
<scope>system</scope>
<systemPath>${basedir}/lib/high-scale-lib.jar</systemPath>
</dependency>
<dependency>
<groupId>flexjson</groupId>
<artifactId>flexjson</artifactId>
<version>1.7</version>
<scope>system</scope>
<systemPath>${basedir}/lib/flexjson-1.7.jar</systemPath>
</dependency>
<dependency>
<groupId>libthrift</groupId>
<artifactId>libthrift</artifactId>
<version>UNKNOWN</version>
<scope>system</scope>
<systemPath>${basedir}/lib/libthrift-r820831.jar</systemPath>
</dependency>
<dependency>
<groupId>jsonsimple</groupId>
<artifactId>jsonsimple</artifactId>
<version>UNKNOWN</version>
<scope>system</scope>
<systemPath>${basedir}/lib/json_simple-1.1.jar</systemPath>
</dependency>
<dependency>
<groupId>com.reardencommerce</groupId>
<artifactId>clhm</artifactId>
<version>UNKNOWN</version>
<scope>system</scope>
<systemPath>${basedir}/lib/clhm-production.jar</systemPath>
</dependency>
<!-- Testing Dependencies -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.6</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<sourceDirectory>${basedir}/src/java</sourceDirectory>
<testSourceDirectory>${basedir}/test/unit</testSourceDirectory>
<outputDirectory>build/classes</outputDirectory>
<testResources>
<testResource>
<directory>${basedir}/test/conf</directory>
<includes>
<include>**/*</include>
</includes>
</testResource>
<testResource>
<directory>${basedir}/test/resources</directory>
<includes>
<include>*.json</include>
</includes>
</testResource>
</testResources>
<plugins>
<!-- builds the antlr sources -->
<plugin>
<groupId>org.antlr</groupId>
<artifactId>antlr3-maven-plugin</artifactId>
<version>3.1.3-1</version>
<executions>
<execution>
<phase>process-sources</phase>
<goals>
<goal>antlr</goal>
</goals>
<configuration>
<sourceDirectory>${basedir}/src/java</sourceDirectory>
</configuration>
</execution>
</executions>
</plugin>
<!-- Include the thrift generated sources in the build -->
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>build-helper-maven-plugin</artifactId>
<version>1.3</version>
<executions>
<execution>
<id>add-source</id>
<phase>generate-sources</phase>
<goals>
<goal>add-source</goal>
</goals>
<configuration>
<sources>
<source>${basedir}/interface/gen-java</source>
</sources>
</configuration>
</execution>
</executions>
</plugin>
<!-- Setup the build for JAVA 1.6 -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.6</source>
<target>1.6</target>
<optimize>true</optimize>
<debug>true</debug>
<showDeprecation>true</showDeprecation>
<showWarnings>true</showWarnings>
</configuration>
</plugin>
<!-- Configure the JUnit Env -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<systemProperties>
<property>
<name>storage-config</name>
<value>${basedir}/test/conf</value>
</property>
</systemProperties>
<forkMode>always</forkMode>
<excludes>
<exclude>**/TestRingCache.java</exclude>
</excludes>
</configuration>
</plugin>
<!-- Lets run with cobertura -->
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>cobertura-maven-plugin</artifactId>
<version>2.0</version>
<configuration>
</configuration>
</plugin>
<!-- These plugins help validate ASF legal requirements are met for the release -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-release-plugin</artifactId>
<version>2.0-beta-9</version>
<configuration>
<autoVersionSubmodules>true</autoVersionSubmodules>
<allowTimestampedSnapshots>false</allowTimestampedSnapshots>
<preparationGoals>clean install</preparationGoals>
<goals>deploy</goals>
<arguments>-Papache-release</arguments>
</configuration>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>ianal-maven-plugin</artifactId>
<version>1.0-alpha-1</version>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>rat-maven-plugin</artifactId>
<version>1.0-alpha-3</version>
<configuration>
<excludeSubProjects>false</excludeSubProjects>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-enforcer-plugin</artifactId>
<executions>
<execution>
<phase>validate</phase>
<goals>
<goal>enforce</goal>
</goals>
<configuration>
<rules>
<requireMavenVersion>
<version>[2.0.9,)</version>
</requireMavenVersion>
</rules>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>ianal-maven-plugin</artifactId>
<executions>
<execution>
<goals>
<goal>verify-legal-files</goal>
</goals>
<configuration>
<strict>true</strict>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
<reporting>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jxr-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-report-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-pmd-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>taglist-maven-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-javadoc-plugin</artifactId>
<configuration>
<links>
<link>http://java.sun.com/j2se/1.6.0/docs/api/</link>
<link>http://logging.apache.org/log4j/docs/api/</link>
</links>
<!--
<stylesheetfile>${basedir}/../etc/css/stylesheet.css</stylesheetfile>
-->
<linksource>true</linksource>
<maxmemory>900m</maxmemory>
<source>1.6</source>
</configuration>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>cobertura-maven-plugin</artifactId>
<version>2.2</version>
<configuration>
<formats>
<format>html</format>
<format>xml</format>
</formats>
</configuration>
</plugin>
</plugins>
</reporting>
<profiles>
<!-- run "mvn -P thrift" to generate the thrift classes -->
<profile>
<id>thrift</id>
<build>
<defaultGoal>process-sources</defaultGoal>
<plugins>
<plugin>
<artifactId>maven-antrun-plugin</artifactId>
<executions>
<execution>
<phase>process-sources</phase>
<configuration>
<tasks>
<ant antfile="${basedir}/build.xml" target="gen-thrift-java"/>
<ant antfile="${basedir}/build.xml" target="gen-thrift-py"/>
</tasks>
</configuration>
<goals>
<goal>run</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</profile>
<profile>
<id>apache-release</id>
<build>
<plugins>
<!-- We want to sign the artifact, the POM, and all attached artifacts -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-gpg-plugin</artifactId>
<configuration>
<passphrase>${gpg.passphrase}</passphrase>
</configuration>
<executions>
<execution>
<goals>
<goal>sign</goal>
</goals>
</execution>
</executions>
</plugin>
<!-- We want to deploy the artifact to a staging location for perusal -->
<plugin>
<inherited>true</inherited>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-deploy-plugin</artifactId>
<configuration>
<updateReleaseInfo>true</updateReleaseInfo>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-source-plugin</artifactId>
<executions>
<execution>
<id>attach-sources</id>
<goals>
<goal>jar</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-javadoc-plugin</artifactId>
<configuration>
<encoding>${project.build.sourceEncoding}</encoding>
</configuration>
<executions>
<execution>
<id>attach-javadocs</id>
<goals>
<goal>jar</goal>
</goals>
</execution>
</executions>
</plugin>
<!-- Deploy a source distro too -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<executions>
<execution>
<goals>
<goal>single</goal>
</goals>
<phase>package</phase>
<configuration>
<runOnlyAtExecutionRoot>true</runOnlyAtExecutionRoot>
<descriptorRefs>
<descriptorRef>
source-release
</descriptorRef>
</descriptorRefs>
</configuration>
</execution>
</executions>
<dependencies>
<dependency>
<groupId>org.apache.geronimo.genesis</groupId>
<artifactId>apache-source-release-assembly-descriptor</artifactId>
<version>2.0</version>
</dependency>
</dependencies>
</plugin>
</plugins>
</build>
</profile>
</profiles>
</project>

1
src/java/org/apache/cassandra/db/RowMutation.java
View File

@ -47,7 +47,6 @@ public class RowMutation
{
private static ICompactSerializer<RowMutation> serializer_;
public static final String HINT = "HINT";
public static final String FORWARD_HEADER = "FORWARD";
static
{

50
src/java/org/apache/cassandra/db/RowMutationVerbHandler.java
View File

@ -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;
}
}
}

178
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -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<RowMutation> mutations, ConsistencyLevel consistencyLevel) throws UnavailableException, TimeoutException
public static void mutate(List<RowMutation> mutations, ConsistencyLevel consistency_level) throws UnavailableException, TimeoutException
{
long startTime = System.nanoTime();
List<IWriteResponseHandler> responseHandlers = new ArrayList<IWriteResponseHandler>();
ArrayList<IWriteResponseHandler> responseHandlers = new ArrayList<IWriteResponseHandler>();
RowMutation mostRecentRowMutation = null;
StorageService ss = StorageService.instance;
String localDataCenter = getDataCenter(FBUtilities.getLocalAddress());
try
{
@ -115,67 +110,58 @@ public class StorageProxy implements StorageProxyMBean
Collection<InetAddress> writeEndpoints = ss.getTokenMetadata().getWriteEndpoints(StorageService.getPartitioner().getToken(rm.key()), table, naturalEndpoints);
Multimap<InetAddress, InetAddress> 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<String, Pair<Message, InetAddress>> dcMap = groupEndpointsByDataCenter(rm, hintedEndpoints, responseHandler);
// Traverse all dataCenters where messages will be sent to.
for (Map.Entry<String, Collection<Pair<Message, InetAddress>>> entry : dcMap.asMap().entrySet())
Message unhintedMessage = null;
for (Map.Entry<InetAddress, Collection<InetAddress>> 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<Pair<Message, InetAddress>> messagesForDataCenter = entry.getValue();
Iterator<Pair<Message, InetAddress>> iter = messagesForDataCenter.iterator();
assert iter.hasNext();
InetAddress destination = entry.getKey();
Collection<InetAddress> targets = entry.getValue();
// First endpoint in list is the destination for this group
Pair<Message, InetAddress> 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<String, Pair<Message, InetAddress>> groupEndpointsByDataCenter(RowMutation rm, Multimap<InetAddress, InetAddress> endpoints, final IWriteResponseHandler responseHandler) throws IOException
{
Set<Map.Entry<InetAddress, Collection<InetAddress>>> endpointSet = endpoints.asMap().entrySet();
Multimap<String, Pair<Message, InetAddress>> dcMap = HashMultimap.create(endpointSet.size(), 10);
Message unhintedMessage = null;
for (Map.Entry<InetAddress, Collection<InetAddress>> entry : endpointSet)
{
InetAddress destination = entry.getKey();
Collection<InetAddress> 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<Message,InetAddress>(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<Message,InetAddress>(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())