You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by yu...@apache.org on 2013/07/18 16:28:36 UTC
git commit: fix bulkloader patch by yukim reviewed by slebresne for
CASSANDRA-5542
Updated Branches:
refs/heads/trunk 4f0d7b456 -> d881024ee
fix bulkloader patch by yukim reviewed by slebresne for CASSANDRA-5542
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/d881024e
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/d881024e
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/d881024e
Branch: refs/heads/trunk
Commit: d881024eeff2406809732c7a24016e02ab129d34
Parents: 4f0d7b4
Author: Yuki Morishita <yu...@apache.org>
Authored: Thu Jul 18 09:28:26 2013 -0500
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Jul 18 09:28:26 2013 -0500
----------------------------------------------------------------------
.../org/apache/cassandra/config/CFMetaData.java | 22 +++++-
.../apache/cassandra/cql3/UntypedResultSet.java | 5 ++
.../cassandra/hadoop/BulkRecordWriter.java | 13 ++--
.../cassandra/io/sstable/SSTableLoader.java | 21 ++----
.../net/IncomingStreamingConnection.java | 28 +------
.../net/OutboundTcpConnectionPool.java | 23 +++---
.../cassandra/service/StorageService.java | 4 +-
.../cassandra/streaming/ConnectionHandler.java | 77 ++++++++++++--------
.../cassandra/streaming/StreamManager.java | 39 +++++++---
.../cassandra/streaming/StreamResultFuture.java | 44 +++++------
.../cassandra/streaming/StreamSession.java | 46 +++---------
.../streaming/messages/StreamInitMessage.java | 17 ++---
.../org/apache/cassandra/tools/BulkLoader.java | 66 +++++++++++------
13 files changed, 216 insertions(+), 189 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d881024e/src/java/org/apache/cassandra/config/CFMetaData.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java
index 3611986..573819a 100644
--- a/src/java/org/apache/cassandra/config/CFMetaData.java
+++ b/src/java/org/apache/cassandra/config/CFMetaData.java
@@ -39,8 +39,8 @@ import org.apache.commons.lang.builder.ToStringBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.cassandra.cql3.ColumnNameBuilder;
import org.apache.cassandra.cql3.CFDefinition;
+import org.apache.cassandra.cql3.ColumnNameBuilder;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.cql3.statements.CreateTableStatement;
@@ -58,6 +58,7 @@ import org.apache.cassandra.io.compress.CompressionParameters;
import org.apache.cassandra.io.compress.LZ4Compressor;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.serializers.MarshalException;
+import org.apache.cassandra.thrift.CqlRow;
import org.apache.cassandra.thrift.IndexType;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.ByteBufferUtil;
@@ -937,6 +938,25 @@ public final class CFMetaData
}
}
+ /**
+ * Create CFMetaData from thrift {@link CqlRow} that contains columns from schema_columnfamilies.
+ *
+ * @param row CqlRow containing columns from schema_columnfamilies.
+ * @return CFMetaData derived from CqlRow
+ */
+ public static CFMetaData fromThriftCqlRow(CqlRow row)
+ {
+ Map<String, ByteBuffer> columns = new HashMap<>();
+ try
+ {
+ for (org.apache.cassandra.thrift.Column column : row.getColumns())
+ columns.put(ByteBufferUtil.string(column.bufferForName()), column.value);
+ }
+ catch (CharacterCodingException ignore) {}
+ UntypedResultSet.Row cql3row = new UntypedResultSet.Row(columns);
+ return fromSchemaNoColumns(cql3row);
+ }
+
public void reload()
{
Row cfDefRow = SystemKeyspace.readSchemaRow(ksName, cfName);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d881024e/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
index 25294d6..3ef5c95 100644
--- a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
+++ b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
@@ -79,6 +79,11 @@ public class UntypedResultSet implements Iterable<UntypedResultSet.Row>
{
Map<String, ByteBuffer> data = new HashMap<String, ByteBuffer>();
+ public Row(Map<String, ByteBuffer> data)
+ {
+ this.data.putAll(data);
+ }
+
public Row(List<ColumnSpecification> names, List<ByteBuffer> columns)
{
for (int i = 0; i < names.size(); i++)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d881024e/src/java/org/apache/cassandra/hadoop/BulkRecordWriter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/BulkRecordWriter.java b/src/java/org/apache/cassandra/hadoop/BulkRecordWriter.java
index 704d19f..b6a7e75 100644
--- a/src/java/org/apache/cassandra/hadoop/BulkRecordWriter.java
+++ b/src/java/org/apache/cassandra/hadoop/BulkRecordWriter.java
@@ -32,6 +32,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.auth.IAuthenticator;
+import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.marshal.AbstractType;
@@ -253,7 +254,7 @@ implements org.apache.hadoop.mapred.RecordWriter<ByteBuffer,List<Mutation>>
static class ExternalClient extends SSTableLoader.Client
{
- private final Map<String, Set<String>> knownCfs = new HashMap<String, Set<String>>();
+ private final Map<String, Map<String, CFMetaData>> knownCfs = new HashMap<>();
private final String hostlist;
private final int rpcPort;
private final String username;
@@ -319,9 +320,9 @@ implements org.apache.hadoop.mapred.RecordWriter<ByteBuffer,List<Mutation>>
for (KsDef ksDef : ksDefs)
{
- Set<String> cfs = new HashSet<String>(ksDef.cf_defs.size());
+ Map<String, CFMetaData> cfs = new HashMap<>(ksDef.cf_defs.size());
for (CfDef cfDef : ksDef.cf_defs)
- cfs.add(cfDef.name);
+ cfs.put(cfDef.name, CFMetaData.fromThrift(cfDef));
knownCfs.put(ksDef.name, cfs);
}
break;
@@ -334,10 +335,10 @@ implements org.apache.hadoop.mapred.RecordWriter<ByteBuffer,List<Mutation>>
}
}
- public boolean validateColumnFamily(String keyspace, String cfName)
+ public CFMetaData getCFMetaData(String keyspace, String cfName)
{
- Set<String> cfs = knownCfs.get(keyspace);
- return cfs != null && cfs.contains(cfName);
+ Map<String, CFMetaData> cfs = knownCfs.get(keyspace);
+ return cfs != null ? cfs.get(cfName) : null;
}
private static Cassandra.Client createThriftClient(String host, int port) throws TTransportException
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d881024e/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
index 4d54cd9..d945a48 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
@@ -23,6 +23,7 @@ import java.io.IOException;
import java.net.InetAddress;
import java.util.*;
+import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.IPartitioner;
@@ -80,7 +81,8 @@ public class SSTableLoader implements StreamEventHandler
return false;
}
- if (!client.validateColumnFamily(keyspace, desc.cfname))
+ CFMetaData metadata = client.getCFMetaData(keyspace, desc.cfname);
+ if (metadata == null)
{
outputHandler.output(String.format("Skipping file %s: column family %s.%s doesn't exist", name, keyspace, desc.cfname));
return false;
@@ -96,7 +98,7 @@ public class SSTableLoader implements StreamEventHandler
try
{
- sstables.add(SSTableReader.open(desc, components, null, client.getPartitioner()));
+ sstables.add(SSTableReader.open(desc, components, metadata, client.getPartitioner()));
}
catch (IOException e)
{
@@ -147,18 +149,7 @@ public class SSTableLoader implements StreamEventHandler
public void handleStreamEvent(StreamEvent event)
{
- if (event.eventType == StreamEvent.Type.FILE_PROGRESS)
- {
- ProgressInfo progress = ((StreamEvent.ProgressEvent) event).progress;
- StringBuilder sb = new StringBuilder("\r");
- sb.append(progress.fileName);
- sb.append(": ");
- sb.append(progress.currentBytes).append("/").append(progress.totalBytes);
- System.out.print(sb.toString());
- if (progress.currentBytes == progress.totalBytes)
- System.out.println();
- }
- else if (event.eventType == StreamEvent.Type.STREAM_COMPLETE)
+ if (event.eventType == StreamEvent.Type.STREAM_COMPLETE)
{
StreamEvent.SessionCompleteEvent se = (StreamEvent.SessionCompleteEvent) event;
if (!se.success)
@@ -204,7 +195,7 @@ public class SSTableLoader implements StreamEventHandler
* Validate that {@code keyspace} is an existing keyspace and {@code
* cfName} one of its existing column family.
*/
- public abstract boolean validateColumnFamily(String keyspace, String cfName);
+ public abstract CFMetaData getCFMetaData(String keyspace, String cfName);
public Map<InetAddress, Collection<Range<Token>>> getEndpointToRangesMap()
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d881024e/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
index 24b2bab..20392f2 100644
--- a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
+++ b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
@@ -25,9 +25,7 @@ import java.net.Socket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.cassandra.streaming.StreamManager;
import org.apache.cassandra.streaming.StreamResultFuture;
-import org.apache.cassandra.streaming.StreamSession;
import org.apache.cassandra.streaming.messages.StreamInitMessage;
import org.apache.cassandra.streaming.messages.StreamMessage;
@@ -43,7 +41,7 @@ public class IncomingStreamingConnection extends Thread
public IncomingStreamingConnection(int version, Socket socket)
{
- super("stream-init " + socket.getRemoteSocketAddress());
+ super("STREAM-INIT-" + socket.getRemoteSocketAddress());
this.version = version;
this.socket = socket;
}
@@ -60,29 +58,11 @@ public class IncomingStreamingConnection extends Thread
DataInput input = new DataInputStream(socket.getInputStream());
StreamInitMessage init = StreamInitMessage.serializer.deserialize(input, version);
- // We will use the current socket to incoming stream. So if the other side is the
- // stream initiator, we must first create an outgoing stream, after which real streaming
- // will start. If we were the initiator however, this socket will just be our incoming
- // stream, everything is setup and we can initiate real streaming by sending the prepare message.
+ // The initiator makes two connections, one for incoming and one for outgoing.
+ // The receiving side distinguish two connections by looking at StreamInitMessage#isForOutgoing.
// Note: we cannot use the same socket for incoming and outgoing streams because we want to
// parallelize said streams and the socket is blocking, so we might deadlock.
- if (init.sentByInitiator)
- {
- StreamResultFuture.initReceivingSide(init.planId, init.description, init.from, socket, version);
- }
- else
- {
- StreamResultFuture stream = StreamManager.instance.getStream(init.planId);
- if (stream == null)
- {
- // This should not happen. All we can do is close the socket to inform the other side, but that's a bug.
- logger.error("Got StreamInit message for a stream we are supposed to be the initiator of, but stream not found.");
- socket.close();
- return;
- }
- // We're fully setup for this session, start the actual streaming
- stream.startStreaming(init.from, socket, version);
- }
+ StreamResultFuture.initReceivingSide(init.planId, init.description, init.from, socket, init.isForOutgoing, version);
}
catch (IOException e)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d881024e/src/java/org/apache/cassandra/net/OutboundTcpConnectionPool.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/OutboundTcpConnectionPool.java b/src/java/org/apache/cassandra/net/OutboundTcpConnectionPool.java
index 4efb507..81168c6 100644
--- a/src/java/org/apache/cassandra/net/OutboundTcpConnectionPool.java
+++ b/src/java/org/apache/cassandra/net/OutboundTcpConnectionPool.java
@@ -34,7 +34,6 @@ import org.apache.cassandra.utils.FBUtilities;
public class OutboundTcpConnectionPool
{
- private final IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
// pointer for the real Address.
private final InetAddress id;
public final OutboundTcpConnection cmdCon;
@@ -117,17 +116,22 @@ public class OutboundTcpConnectionPool
public Socket newSocket() throws IOException
{
+ return newSocket(endPoint());
+ }
+
+ public static Socket newSocket(InetAddress endpoint) throws IOException
+ {
// zero means 'bind on any available port.'
- if (isEncryptedChannel())
+ if (isEncryptedChannel(endpoint))
{
if (Config.getOutboundBindAny())
- return SSLFactory.getSocket(DatabaseDescriptor.getServerEncryptionOptions(), endPoint(), DatabaseDescriptor.getSSLStoragePort());
+ return SSLFactory.getSocket(DatabaseDescriptor.getServerEncryptionOptions(), endpoint, DatabaseDescriptor.getSSLStoragePort());
else
- return SSLFactory.getSocket(DatabaseDescriptor.getServerEncryptionOptions(), endPoint(), DatabaseDescriptor.getSSLStoragePort(), FBUtilities.getLocalAddress(), 0);
+ return SSLFactory.getSocket(DatabaseDescriptor.getServerEncryptionOptions(), endpoint, DatabaseDescriptor.getSSLStoragePort(), FBUtilities.getLocalAddress(), 0);
}
else
{
- Socket socket = SocketChannel.open(new InetSocketAddress(endPoint(), DatabaseDescriptor.getStoragePort())).socket();
+ Socket socket = SocketChannel.open(new InetSocketAddress(endpoint, DatabaseDescriptor.getStoragePort())).socket();
if (Config.getOutboundBindAny() && !socket.isBound())
socket.bind(new InetSocketAddress(FBUtilities.getLocalAddress(), 0));
return socket;
@@ -141,8 +145,9 @@ public class OutboundTcpConnectionPool
return resetedEndpoint == null ? id : resetedEndpoint;
}
- boolean isEncryptedChannel()
+ public static boolean isEncryptedChannel(InetAddress address)
{
+ IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
switch (DatabaseDescriptor.getServerEncryptionOptions().internode_encryption)
{
case none:
@@ -150,13 +155,13 @@ public class OutboundTcpConnectionPool
case all:
break;
case dc:
- if (snitch.getDatacenter(id).equals(snitch.getDatacenter(FBUtilities.getBroadcastAddress())))
+ if (snitch.getDatacenter(address).equals(snitch.getDatacenter(FBUtilities.getBroadcastAddress())))
return false;
break;
case rack:
// for rack then check if the DC's are the same.
- if (snitch.getRack(id).equals(snitch.getRack(FBUtilities.getBroadcastAddress()))
- && snitch.getDatacenter(id).equals(snitch.getDatacenter(FBUtilities.getBroadcastAddress())))
+ if (snitch.getRack(address).equals(snitch.getRack(FBUtilities.getBroadcastAddress()))
+ && snitch.getDatacenter(address).equals(snitch.getDatacenter(FBUtilities.getBroadcastAddress())))
return false;
break;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d881024e/src/java/org/apache/cassandra/service/StorageService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index a31c10a..6e12e9d 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -3545,9 +3545,9 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
}
@Override
- public boolean validateColumnFamily(String keyspace, String cfName)
+ public CFMetaData getCFMetaData(String keyspace, String cfName)
{
- return Schema.instance.getCFMetaData(keyspace, cfName) != null;
+ return Schema.instance.getCFMetaData(keyspace, cfName);
}
};
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d881024e/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
index 6be1f42..c4f2181 100644
--- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
+++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
@@ -23,17 +23,13 @@ import java.net.Socket;
import java.net.SocketException;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
-import java.nio.channels.SocketChannel;
import java.nio.channels.WritableByteChannel;
import java.util.Collection;
import java.util.Comparator;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
-import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
@@ -41,7 +37,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.OutboundTcpConnectionPool;
import org.apache.cassandra.streaming.messages.StreamInitMessage;
import org.apache.cassandra.streaming.messages.StreamMessage;
import org.apache.cassandra.utils.FBUtilities;
@@ -70,38 +66,47 @@ public class ConnectionHandler
this.session = session;
}
- public ConnectionHandler initiate() throws IOException
- {
- // Connect to other side and use that as the outgoing socket. Once the receiving
- // peer send back his init message, we'll have our incoming handling.
- outgoing = new OutgoingMessageHandler(session, connect(session.peer), StreamMessage.CURRENT_VERSION);
-
- logger.debug("Sending stream init... for {}", session.planId());
- outgoing.sendInitMessage(true);
- outgoing.start();
-
- return this;
- }
-
- public ConnectionHandler initiateOnReceivingSide(Socket incomingSocket, int version) throws IOException
+ /**
+ * Set up incoming message handler and initiate streaming.
+ *
+ * This method is called once on initiator.
+ *
+ * @throws IOException
+ */
+ public void initiate() throws IOException
{
- // Create and start the incoming handler
- incoming = new IncomingMessageHandler(session, incomingSocket, version);
+ logger.debug("[Stream #{}] Sending stream init for incoming stream", session.planId());
+ Socket incomingSocket = connect(session.peer);
+ incoming = new IncomingMessageHandler(session, incomingSocket, StreamMessage.CURRENT_VERSION);
+ incoming.sendInitMessage(true);
incoming.start();
- // Connect back to the other side, and use that new socket for the outgoing handler
- outgoing = new OutgoingMessageHandler(session, connect(session.peer), version);
-
- logger.debug("Sending stream init back to initiator...");
+ logger.debug("[Stream #{}] Sending stream init for outgoing stream", session.planId());
+ Socket outgoingSocket = connect(session.peer);
+ outgoing = new OutgoingMessageHandler(session, outgoingSocket, StreamMessage.CURRENT_VERSION);
outgoing.sendInitMessage(false);
outgoing.start();
- return this;
}
- public void attachIncomingSocket(Socket incomingSocket, int version) throws IOException
+ /**
+ * Set up outgoing message handler on receiving side.
+ *
+ * @param socket socket to use for {@link OutgoingMessageHandler}.
+ * @param version Streaming message version
+ * @throws IOException
+ */
+ public void initiateOnReceivingSide(Socket socket, boolean isForOutgoing, int version) throws IOException
{
- incoming = new IncomingMessageHandler(session, incomingSocket, version);
- incoming.start();
+ if (isForOutgoing)
+ {
+ outgoing = new OutgoingMessageHandler(session, socket, version);
+ outgoing.start();
+ }
+ else
+ {
+ incoming = new IncomingMessageHandler(session, socket, version);
+ incoming.start();
+ }
}
/**
@@ -121,7 +126,7 @@ public class ConnectionHandler
try
{
logger.info("Connecting to {} for streaming", peer);
- Socket socket = MessagingService.instance().getConnectionPool(peer).newSocket();
+ Socket socket = OutboundTcpConnectionPool.newSocket(peer);
socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
return socket;
}
@@ -173,6 +178,14 @@ public class ConnectionHandler
outgoing.enqueue(message);
}
+ /**
+ * @return true if outgoing connection is opened and ready to send messages
+ */
+ public boolean isOutgoingConnected()
+ {
+ return outgoing != null && !outgoing.isClosed();
+ }
+
abstract static class MessageHandler implements Runnable
{
protected final StreamSession session;
@@ -209,9 +222,9 @@ public class ConnectionHandler
: in;
}
- public void sendInitMessage(boolean sentByInitiator) throws IOException
+ public void sendInitMessage(boolean isForOutgoing) throws IOException
{
- StreamInitMessage message = new StreamInitMessage(FBUtilities.getBroadcastAddress(), session.planId(), session.description(), sentByInitiator);
+ StreamInitMessage message = new StreamInitMessage(FBUtilities.getBroadcastAddress(), session.planId(), session.description(), isForOutgoing);
getWriteChannel().write(message.createMessage(false, protocolVersion));
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d881024e/src/java/org/apache/cassandra/streaming/StreamManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamManager.java b/src/java/org/apache/cassandra/streaming/StreamManager.java
index 50f8978..5fc1c75 100644
--- a/src/java/org/apache/cassandra/streaming/StreamManager.java
+++ b/src/java/org/apache/cassandra/streaming/StreamManager.java
@@ -17,19 +17,15 @@
*/
package org.apache.cassandra.streaming;
-import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import com.google.common.base.Function;
import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.RateLimiter;
import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.RateLimiter;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
import org.apache.cassandra.config.DatabaseDescriptor;
@@ -64,12 +60,17 @@ public class StreamManager implements StreamManagerMBean
return limiter;
}
- /** Currently running stream plans. Removed after completion/failure. */
- private final Map<UUID, StreamResultFuture> currentStreams = new NonBlockingHashMap<>();
+ /*
+ * Currently running streams. Removed after completion/failure.
+ * We manage them in two different maps to distinguish plan from initiated ones to
+ * receiving ones withing the same JVM.
+ */
+ private final Map<UUID, StreamResultFuture> initiatedStreams = new NonBlockingHashMap<>();
+ private final Map<UUID, StreamResultFuture> receivingStreams = new NonBlockingHashMap<>();
public Set<StreamState> getCurrentStreams()
{
- return Sets.newHashSet(Iterables.transform(currentStreams.values(), new Function<StreamResultFuture, StreamState>()
+ return Sets.newHashSet(Iterables.transform(Iterables.concat(initiatedStreams.values(), receivingStreams.values()), new Function<StreamResultFuture, StreamState>()
{
public StreamState apply(StreamResultFuture input)
{
@@ -85,15 +86,29 @@ public class StreamManager implements StreamManagerMBean
{
public void run()
{
- currentStreams.remove(result.planId);
+ initiatedStreams.remove(result.planId);
+ }
+ }, MoreExecutors.sameThreadExecutor());
+
+ initiatedStreams.put(result.planId, result);
+ }
+
+ public void registerReceiving(final StreamResultFuture result)
+ {
+ // Make sure we remove the stream on completion (whether successful or not)
+ result.addListener(new Runnable()
+ {
+ public void run()
+ {
+ receivingStreams.remove(result.planId);
}
}, MoreExecutors.sameThreadExecutor());
- currentStreams.put(result.planId, result);
+ receivingStreams.put(result.planId, result);
}
- public StreamResultFuture getStream(UUID planId)
+ public StreamResultFuture getReceivingStream(UUID planId)
{
- return currentStreams.get(planId);
+ return receivingStreams.get(planId);
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d881024e/src/java/org/apache/cassandra/streaming/StreamResultFuture.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamResultFuture.java b/src/java/org/apache/cassandra/streaming/StreamResultFuture.java
index cc2432a..d9189c5 100644
--- a/src/java/org/apache/cassandra/streaming/StreamResultFuture.java
+++ b/src/java/org/apache/cassandra/streaming/StreamResultFuture.java
@@ -21,22 +21,14 @@ import java.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import java.util.*;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
import com.google.common.collect.ImmutableSet;
import com.google.common.util.concurrent.AbstractFuture;
import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListeningExecutorService;
-import com.google.common.util.concurrent.MoreExecutors;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.cassandra.utils.FBUtilities;
-
-
/**
* A future on the result ({@link StreamState}) of a streaming plan.
*
@@ -68,7 +60,6 @@ public final class StreamResultFuture extends AbstractFuture<StreamState>
*
* @param planId Stream plan ID
* @param description Stream description
- * @param numberOfSessions number of sessions to wait for complete
*/
private StreamResultFuture(UUID planId, String description, Collection<StreamSession> sessions)
{
@@ -76,7 +67,7 @@ public final class StreamResultFuture extends AbstractFuture<StreamState>
this.description = description;
this.ongoingSessions = new HashMap<>(sessions.size());
for (StreamSession session : sessions)
- this.ongoingSessions.put(session.peer, session);;
+ this.ongoingSessions.put(session.peer, session);
// if there is no session to listen to, we immediately set result for returning
if (sessions.isEmpty())
@@ -97,16 +88,29 @@ public final class StreamResultFuture extends AbstractFuture<StreamState>
return future;
}
- public static StreamResultFuture initReceivingSide(UUID planId, String description, InetAddress from, Socket socket, int version)
+ public static synchronized StreamResultFuture initReceivingSide(UUID planId,
+ String description,
+ InetAddress from,
+ Socket socket,
+ boolean isForOutgoing,
+ int version) throws IOException
{
- final StreamSession session = new StreamSession(from);
-
- // The main reason we create a StreamResultFuture on the receiving side is for JMX exposure.
- StreamResultFuture future = createAndRegister(planId, description, Collections.singleton(session));
+ StreamResultFuture future = StreamManager.instance.getReceivingStream(planId);
+ if (future == null)
+ {
+ final StreamSession session = new StreamSession(from);
- session.init(future);
- session.start(socket, version);
+ // The main reason we create a StreamResultFuture on the receiving side is for JMX exposure.
+ future = new StreamResultFuture(planId, description, Collections.singleton(session));
+ StreamManager.instance.registerReceiving(future);
+ session.init(future);
+ session.handler.initiateOnReceivingSide(socket, isForOutgoing, version);
+ }
+ else
+ {
+ future.attachSocket(from, socket, isForOutgoing, version);
+ }
return future;
}
@@ -117,14 +121,12 @@ public final class StreamResultFuture extends AbstractFuture<StreamState>
return future;
}
- public void startStreaming(InetAddress from, Socket socket, int version) throws IOException
+ public void attachSocket(InetAddress from, Socket socket, boolean isForOutgoing, int version) throws IOException
{
StreamSession session = ongoingSessions.get(from);
if (session == null)
throw new RuntimeException(String.format("Got connection from %s for stream session %s but no such session locally", from, planId));
-
- session.handler.attachIncomingSocket(socket, version);
- session.onInitializationComplete();
+ session.handler.initiateOnReceivingSide(socket, isForOutgoing, version);
}
public void addEventListener(StreamEventHandler listener)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d881024e/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamSession.java b/src/java/org/apache/cassandra/streaming/StreamSession.java
index 6a771a1..d766ff8 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -30,8 +30,8 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
-import org.apache.cassandra.db.RowPosition;
import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.RowPosition;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
@@ -43,7 +43,6 @@ import org.apache.cassandra.metrics.StreamingMetrics;
import org.apache.cassandra.streaming.messages.*;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
-import org.apache.cassandra.utils.UUIDGen;
/**
* Handles the streaming a one or more section of one of more sstables to and from a specific
@@ -56,17 +55,15 @@ import org.apache.cassandra.utils.UUIDGen;
*
* (a) A node (the initiator in the following) create a new StreamSession, initialize it (init())
* and then start it (start()). Start will create a {@link ConnectionHandler} that will create
- * a connection to the remote node (the follower in the following) with whom to stream and send
- * a StreamInit message. This first connection will be the outgoing connection for the
- * initiator.
+ * two connections to the remote node (the follower in the following) with whom to stream and send
+ * a StreamInit message. The first connection will be the incoming connection for the
+ * initiator, and the second connection will be the outgoing.
* (b) Upon reception of that StreamInit message, the follower creates its own StreamSession,
- * initialize it and start it using start(Socket, int). This creates the follower
- * ConnectionHandler, which will use the just opened connection as incoming connection. It
- * will then connect back to the initiator and send its own StreamInit message on that new
- * connection. This new connection will be the outgoing connection for the follower.
- * (c) On receiving the follower StreamInit message, the initiator will record that new connection
- * as it's own incoming connection and call the Session onInitializationComplete() method to start
- * the streaming prepare phase (StreamResultFuture.startStreaming()).
+ * initialize it if it still does not exist, and attach connecting socket to its ConnectionHandler
+ * according to StreamInit message's isForOutgoing flag.
+ * (d) When the both incoming and outgoing connections are established, StreamSession calls
+ * StreamSession#onInitializationComplete method to start the streaming prepare phase
+ * (StreamResultFuture.startStreaming()).
*
* 2. Streaming preparation phase
*
@@ -175,7 +172,6 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
* perform pre-streaming initialization.
*
* @param streamResult result to report to
- * @return this object for chaining
*/
public void init(StreamResultFuture streamResult)
{
@@ -184,7 +180,6 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
// register to gossiper/FD to fail on node failure
Gossiper.instance.register(this);
FailureDetector.instance.registerFailureDetectionEventListener(this);
-
}
public void start()
@@ -203,6 +198,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
try
{
handler.initiate();
+ onInitializationComplete();
}
catch (IOException e)
{
@@ -212,25 +208,6 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
});
}
- public void start(final Socket socket, final int version)
- {
- streamExecutor.execute(new Runnable()
- {
- public void run()
- {
- try
- {
- handler.initiateOnReceivingSide(socket, version);
- }
- catch (IOException e)
- {
- onError(e);
- }
- }
- });
- }
-
-
/**
* Request data fetch task to this session.
*
@@ -410,7 +387,8 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
{
logger.error("Streaming error occurred", e);
// send session failure message
- handler.sendMessage(new SessionFailedMessage());
+ if (handler.isOutgoingConnected())
+ handler.sendMessage(new SessionFailedMessage());
// fail session
closeSession(State.FAILED);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d881024e/src/java/org/apache/cassandra/streaming/messages/StreamInitMessage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/messages/StreamInitMessage.java b/src/java/org/apache/cassandra/streaming/messages/StreamInitMessage.java
index 685ad41..025daab 100644
--- a/src/java/org/apache/cassandra/streaming/messages/StreamInitMessage.java
+++ b/src/java/org/apache/cassandra/streaming/messages/StreamInitMessage.java
@@ -43,15 +43,15 @@ public class StreamInitMessage
public final UUID planId;
public final String description;
- // Whether the sender of this message is the stream initiator
- public final boolean sentByInitiator;
+ // true if this init message is to connect for outgoing message on receiving side
+ public final boolean isForOutgoing;
- public StreamInitMessage(InetAddress from, UUID planId, String description, boolean sentByInitiator)
+ public StreamInitMessage(InetAddress from, UUID planId, String description, boolean isForOutgoing)
{
this.from = from;
this.planId = planId;
this.description = description;
- this.sentByInitiator = sentByInitiator;
+ this.isForOutgoing = isForOutgoing;
}
/**
@@ -72,11 +72,6 @@ public class StreamInitMessage
// Setting up the version bit
header |= (version << 8);
- /* Adding the StreamHeader which contains the session Id along
- * with the pendingfile info for the stream.
- * | Session Id | Pending File Size | Pending File | Bool more files |
- * | No. of Pending files | Pending Files ... |
- */
byte[] bytes;
try
{
@@ -106,7 +101,7 @@ public class StreamInitMessage
CompactEndpointSerializationHelper.serialize(message.from, out);
UUIDSerializer.serializer.serialize(message.planId, out, MessagingService.current_version);
out.writeUTF(message.description);
- out.writeBoolean(message.sentByInitiator);
+ out.writeBoolean(message.isForOutgoing);
}
public StreamInitMessage deserialize(DataInput in, int version) throws IOException
@@ -123,7 +118,7 @@ public class StreamInitMessage
long size = CompactEndpointSerializationHelper.serializedSize(message.from);
size += UUIDSerializer.serializer.serializedSize(message.planId, MessagingService.current_version);
size += TypeSizes.NATIVE.sizeof(message.description);
- size += TypeSizes.NATIVE.sizeof(message.sentByInitiator);
+ size += TypeSizes.NATIVE.sizeof(message.isForOutgoing);
return size;
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/d881024e/src/java/org/apache/cassandra/tools/BulkLoader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tools/BulkLoader.java b/src/java/org/apache/cassandra/tools/BulkLoader.java
index 45d298a..0f175f0 100644
--- a/src/java/org/apache/cassandra/tools/BulkLoader.java
+++ b/src/java/org/apache/cassandra/tools/BulkLoader.java
@@ -31,6 +31,7 @@ import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.cassandra.auth.IAuthenticator;
+import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.SystemKeyspace;
@@ -63,7 +64,7 @@ public class BulkLoader
SSTableLoader loader = new SSTableLoader(options.directory, new ExternalClient(options.hosts, options.rpcPort, options.user, options.passwd), handler);
DatabaseDescriptor.setStreamThroughputOutboundMegabitsPerSec(options.throttle);
StreamResultFuture future = loader.stream(options.ignores);
- future.addEventListener(new ProgressIndicator(handler));
+ future.addEventListener(new ProgressIndicator());
try
{
future.get();
@@ -83,16 +84,15 @@ public class BulkLoader
// Return true when everything is at 100%
static class ProgressIndicator implements StreamEventHandler
{
- private final Set<ProgressInfo> progresses = new HashSet<>();
- private final OutputHandler handler;
+ private final Map<InetAddress, SessionInfo> sessionsByHost = new HashMap<>();
+ private final Map<InetAddress, Set<ProgressInfo>> progressByHost = new HashMap<>();
private long start;
private long lastProgress;
private long lastTime;
- public ProgressIndicator(OutputHandler handler)
+ public ProgressIndicator()
{
- this.handler = handler;
start = lastTime = System.nanoTime();
}
@@ -101,11 +101,22 @@ public class BulkLoader
public void handleStreamEvent(StreamEvent event)
{
- if (event.eventType == StreamEvent.Type.FILE_PROGRESS)
+ if (event.eventType == StreamEvent.Type.STREAM_PREPARED)
+ {
+ SessionInfo session = ((StreamEvent.SessionPreparedEvent) event).session;
+ sessionsByHost.put(session.peer, session);
+ }
+ else if (event.eventType == StreamEvent.Type.FILE_PROGRESS)
{
ProgressInfo progressInfo = ((StreamEvent.ProgressEvent) event).progress;
// update progress
+ Set<ProgressInfo> progresses = progressByHost.get(progressInfo.peer);
+ if (progresses == null)
+ {
+ progresses = new HashSet<>();
+ progressByHost.put(progressInfo.peer, progresses);
+ }
if (progresses.contains(progressInfo))
progresses.remove(progressInfo);
progresses.add(progressInfo);
@@ -115,16 +126,24 @@ public class BulkLoader
long totalProgress = 0;
long totalSize = 0;
- long completed = 0;
- for (ProgressInfo entry : progresses)
+ for (Map.Entry<InetAddress, Set<ProgressInfo>> entry : progressByHost.entrySet())
{
- if (entry.currentBytes == entry.totalBytes)
- completed++;
- totalProgress += entry.currentBytes;
- totalSize += entry.totalBytes;
- sb.append("[").append(entry.peer);
- sb.append(" ").append(completed).append("/").append(progresses.size());
- sb.append(" (").append(entry.totalBytes == 0 ? 100L : entry.currentBytes * 100L / entry.totalBytes).append(")] ");
+ SessionInfo session = sessionsByHost.get(entry.getKey());
+
+ long size = session.getTotalSizeToSend();
+ long current = 0;
+ int completed = 0;
+ for (ProgressInfo progress : entry.getValue())
+ {
+ if (progress.currentBytes == progress.totalBytes)
+ completed++;
+ current += progress.currentBytes;
+ }
+ totalProgress += current;
+ totalSize += size;
+ sb.append("[").append(entry.getKey());
+ sb.append(" ").append(completed).append("/").append(session.getTotalFilesToSend());
+ sb.append(" (").append(size == 0 ? 100L : current * 100L / size).append("%)] ");
}
long time = System.nanoTime();
long deltaTime = TimeUnit.NANOSECONDS.toMillis(time - lastTime);
@@ -132,11 +151,11 @@ public class BulkLoader
long deltaProgress = totalProgress - lastProgress;
lastProgress = totalProgress;
- sb.append("[total: ").append(totalSize == 0 ? 100L : totalProgress * 100L / totalSize).append(" - ");
+ sb.append("[total: ").append(totalSize == 0 ? 100L : totalProgress * 100L / totalSize).append("% - ");
sb.append(mbPerSec(deltaProgress, deltaTime)).append("MB/s");
sb.append(" (avg: ").append(mbPerSec(totalProgress, TimeUnit.NANOSECONDS.toMillis(time - start))).append("MB/s)]");
- handler.output(sb.toString());
+ System.out.print(sb.toString());
}
}
@@ -149,7 +168,7 @@ public class BulkLoader
static class ExternalClient extends SSTableLoader.Client
{
- private final Set<String> knownCfs = new HashSet<>();
+ private final Map<String, CFMetaData> knownCfs = new HashMap<>();
private final Set<InetAddress> hosts;
private final int rpcPort;
private final String user;
@@ -187,13 +206,16 @@ public class BulkLoader
}
}
- String query = String.format("SELECT columnfamily_name FROM %s.%s WHERE keyspace_name = '%s'",
+ String query = String.format("SELECT * FROM %s.%s WHERE keyspace_name = '%s'",
Keyspace.SYSTEM_KS,
SystemKeyspace.SCHEMA_COLUMNFAMILIES_CF,
keyspace);
CqlResult result = client.execute_cql3_query(ByteBufferUtil.bytes(query), Compression.NONE, ConsistencyLevel.ONE);
for (CqlRow row : result.rows)
- knownCfs.add(new String(row.getColumns().get(0).getValue(), "UTF8"));
+ {
+ CFMetaData metadata = CFMetaData.fromThriftCqlRow(row);
+ knownCfs.put(metadata.cfName, metadata);
+ }
break;
}
catch (Exception e)
@@ -204,9 +226,9 @@ public class BulkLoader
}
}
- public boolean validateColumnFamily(String keyspace, String cfName)
+ public CFMetaData getCFMetaData(String keyspace, String cfName)
{
- return knownCfs.contains(cfName);
+ return knownCfs.get(cfName);
}
private static Cassandra.Client createThriftClient(String host, int port, String user, String passwd) throws Exception