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 2014/08/15 20:26:58 UTC
[1/6] git commit: Fix sstableloader unable to connect encrypted node
Repository: cassandra
Updated Branches:
refs/heads/cassandra-2.0 ad6ba3d24 -> 563cea14b
refs/heads/cassandra-2.1 54fbb0abb -> de5bb5854
refs/heads/trunk fe8829fa6 -> ff88f2376
Fix sstableloader unable to connect encrypted node
patch by yukim; reviewed by krummas for CASSANDRA-7585
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/563cea14
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/563cea14
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/563cea14
Branch: refs/heads/cassandra-2.0
Commit: 563cea14b4bb87cd37ab10399904f08757c34d27
Parents: ad6ba3d
Author: Yuki Morishita <yu...@apache.org>
Authored: Fri Aug 15 12:31:59 2014 -0500
Committer: Yuki Morishita <yu...@apache.org>
Committed: Fri Aug 15 12:31:59 2014 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../config/YamlConfigurationLoader.java | 6 +-
.../cassandra/io/sstable/SSTableLoader.java | 22 ++-
.../cassandra/streaming/ConnectionHandler.java | 48 +-----
.../streaming/DefaultConnectionFactory.java | 74 +++++++++
.../streaming/StreamConnectionFactory.java | 30 ++++
.../apache/cassandra/streaming/StreamPlan.java | 16 +-
.../cassandra/streaming/StreamResultFuture.java | 2 +-
.../cassandra/streaming/StreamSession.java | 13 +-
.../tools/BulkLoadConnectionFactory.java | 68 +++++++++
.../org/apache/cassandra/tools/BulkLoader.java | 149 +++++++++++++------
.../streaming/StreamTransferTaskTest.java | 2 +-
12 files changed, 330 insertions(+), 101 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 4306de5..e335484 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -47,6 +47,7 @@
* Backport CASSANDRA-6747 (CASSANDRA-7560)
* Track max/min timestamps for range tombstones (CASSANDRA-7647)
* Fix NPE when listing saved caches dir (CASSANDRA-7632)
+ * Fix sstableloader unable to connect encrypted node (CASSANDRA-7585)
Merged from 1.2:
* Remove duplicates from StorageService.getJoiningNodes (CASSANDRA-7478)
* Clone token map outside of hot gossip loops (CASSANDRA-7758)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java b/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
index 6b5a152..b520d07 100644
--- a/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
+++ b/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
@@ -69,10 +69,14 @@ public class YamlConfigurationLoader implements ConfigurationLoader
public Config loadConfig() throws ConfigurationException
{
+ return loadConfig(getStorageConfigURL());
+ }
+
+ public Config loadConfig(URL url) throws ConfigurationException
+ {
InputStream input = null;
try
{
- URL url = getStorageConfigURL();
logger.info("Loading settings from {}", url);
try
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/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 4a1604d..85dc0e4 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
@@ -50,7 +50,7 @@ public class SSTableLoader implements StreamEventHandler
private final OutputHandler outputHandler;
private final Set<InetAddress> failedHosts = new HashSet<>();
- private final List<SSTableReader> sstables = new ArrayList<SSTableReader>();
+ private final List<SSTableReader> sstables = new ArrayList<>();
private final Multimap<InetAddress, StreamSession.SSTableStreamingSections> streamingDetails = HashMultimap.create();
static
@@ -94,7 +94,7 @@ public class SSTableLoader implements StreamEventHandler
return false;
}
- Set<Component> components = new HashSet<Component>();
+ Set<Component> components = new HashSet<>();
components.add(Component.DATA);
components.add(Component.PRIMARY_INDEX);
if (new File(desc.filenameFor(Component.SUMMARY)).exists())
@@ -149,7 +149,7 @@ public class SSTableLoader implements StreamEventHandler
client.init(keyspace);
outputHandler.output("Established connection to initial hosts");
- StreamPlan plan = new StreamPlan("Bulk Load");
+ StreamPlan plan = new StreamPlan("Bulk Load").connectionFactory(client.getConnectionFactory());
Map<InetAddress, Collection<Range<Token>>> endpointToRanges = client.getEndpointToRangesMap();
openSSTables(endpointToRanges);
@@ -220,7 +220,7 @@ public class SSTableLoader implements StreamEventHandler
public static abstract class Client
{
- private final Map<InetAddress, Collection<Range<Token>>> endpointToRanges = new HashMap<InetAddress, Collection<Range<Token>>>();
+ private final Map<InetAddress, Collection<Range<Token>>> endpointToRanges = new HashMap<>();
private IPartitioner partitioner;
/**
@@ -240,6 +240,17 @@ public class SSTableLoader implements StreamEventHandler
public void stop() {}
/**
+ * Provides connection factory.
+ * By default, it uses DefaultConnectionFactory.
+ *
+ * @return StreamConnectionFactory to use
+ */
+ public StreamConnectionFactory getConnectionFactory()
+ {
+ return new DefaultConnectionFactory();
+ }
+
+ /**
* Validate that {@code keyspace} is an existing keyspace and {@code
* cfName} one of its existing column family.
*/
@@ -258,6 +269,7 @@ public class SSTableLoader implements StreamEventHandler
protected void setPartitioner(IPartitioner partitioner)
{
this.partitioner = partitioner;
+ // the following is still necessary since Range/Token reference partitioner through StorageService.getPartitioner
DatabaseDescriptor.setPartitioner(partitioner);
}
@@ -271,7 +283,7 @@ public class SSTableLoader implements StreamEventHandler
Collection<Range<Token>> ranges = endpointToRanges.get(endpoint);
if (ranges == null)
{
- ranges = new HashSet<Range<Token>>();
+ ranges = new HashSet<>();
endpointToRanges.put(endpoint, ranges);
}
ranges.add(range);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/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 b06a818..8fba41b 100644
--- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
+++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
@@ -18,7 +18,6 @@
package org.apache.cassandra.streaming;
import java.io.IOException;
-import java.net.InetAddress;
import java.net.Socket;
import java.net.SocketException;
import java.nio.ByteBuffer;
@@ -37,8 +36,6 @@ import com.google.common.util.concurrent.SettableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.cassandra.config.DatabaseDescriptor;
-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;
@@ -55,8 +52,6 @@ public class ConnectionHandler
{
private static final Logger logger = LoggerFactory.getLogger(ConnectionHandler.class);
- private static final int MAX_CONNECT_ATTEMPTS = 3;
-
private final StreamSession session;
private IncomingMessageHandler incoming;
@@ -79,12 +74,12 @@ public class ConnectionHandler
public void initiate() throws IOException
{
logger.debug("[Stream #{}] Sending stream init for incoming stream", session.planId());
- Socket incomingSocket = connect(session.peer);
+ Socket incomingSocket = session.createConnection();
incoming.start(incomingSocket, StreamMessage.CURRENT_VERSION);
incoming.sendInitMessage(incomingSocket, true);
logger.debug("[Stream #{}] Sending stream init for outgoing stream", session.planId());
- Socket outgoingSocket = connect(session.peer);
+ Socket outgoingSocket = session.createConnection();
outgoing.start(outgoingSocket, StreamMessage.CURRENT_VERSION);
outgoing.sendInitMessage(outgoingSocket, false);
}
@@ -104,45 +99,6 @@ public class ConnectionHandler
incoming.start(socket, version);
}
- /**
- * Connect to peer and start exchanging message.
- * When connect attempt fails, this retries for maximum of MAX_CONNECT_ATTEMPTS times.
- *
- * @param peer the peer to connect to.
- * @return the created socket.
- *
- * @throws IOException when connection failed.
- */
- private static Socket connect(InetAddress peer) throws IOException
- {
- int attempts = 0;
- while (true)
- {
- try
- {
- Socket socket = OutboundTcpConnectionPool.newSocket(peer);
- socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
- return socket;
- }
- catch (IOException e)
- {
- if (++attempts >= MAX_CONNECT_ATTEMPTS)
- throw e;
-
- long waitms = DatabaseDescriptor.getRpcTimeout() * (long)Math.pow(2, attempts);
- logger.warn("Failed attempt " + attempts + " to connect to " + peer + ". Retrying in " + waitms + " ms. (" + e + ")");
- try
- {
- Thread.sleep(waitms);
- }
- catch (InterruptedException wtf)
- {
- throw new IOException("interrupted", wtf);
- }
- }
- }
- }
-
public ListenableFuture<?> close()
{
logger.debug("[Stream #{}] Closing stream connection handler on {}", session.planId(), session.peer);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java b/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
new file mode 100644
index 0000000..53af4c8
--- /dev/null
+++ b/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.streaming;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.Socket;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.net.OutboundTcpConnectionPool;
+
+public class DefaultConnectionFactory implements StreamConnectionFactory
+{
+ private static final Logger logger = LoggerFactory.getLogger(DefaultConnectionFactory.class);
+
+ private static final int MAX_CONNECT_ATTEMPTS = 3;
+
+ /**
+ * Connect to peer and start exchanging message.
+ * When connect attempt fails, this retries for maximum of MAX_CONNECT_ATTEMPTS times.
+ *
+ * @param peer the peer to connect to.
+ * @return the created socket.
+ *
+ * @throws IOException when connection failed.
+ */
+ public Socket createConnection(InetAddress peer) throws IOException
+ {
+ int attempts = 0;
+ while (true)
+ {
+ try
+ {
+ Socket socket = OutboundTcpConnectionPool.newSocket(peer);
+ socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
+ return socket;
+ }
+ catch (IOException e)
+ {
+ if (++attempts >= MAX_CONNECT_ATTEMPTS)
+ throw e;
+
+ long waitms = DatabaseDescriptor.getRpcTimeout() * (long)Math.pow(2, attempts);
+ logger.warn("Failed attempt " + attempts + " to connect to " + peer + ". Retrying in " + waitms + " ms. (" + e + ")");
+ try
+ {
+ Thread.sleep(waitms);
+ }
+ catch (InterruptedException wtf)
+ {
+ throw new IOException("interrupted", wtf);
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java b/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java
new file mode 100644
index 0000000..dd99611
--- /dev/null
+++ b/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.streaming;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.Socket;
+
+/**
+ * Interface that creates connection used by streaming.
+ */
+public interface StreamConnectionFactory
+{
+ Socket createConnection(InetAddress peer) throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/src/java/org/apache/cassandra/streaming/StreamPlan.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamPlan.java b/src/java/org/apache/cassandra/streaming/StreamPlan.java
index b57e097..e582c79 100644
--- a/src/java/org/apache/cassandra/streaming/StreamPlan.java
+++ b/src/java/org/apache/cassandra/streaming/StreamPlan.java
@@ -38,6 +38,8 @@ public class StreamPlan
// sessions per InetAddress of the other end.
private final Map<InetAddress, StreamSession> sessions = new HashMap<>();
+ private StreamConnectionFactory connectionFactory = new DefaultConnectionFactory();
+
private boolean flushBeforeTransfer = true;
/**
@@ -132,6 +134,18 @@ public class StreamPlan
}
/**
+ * Set custom StreamConnectionFactory to be used for establishing connection
+ *
+ * @param factory StreamConnectionFactory to use
+ * @return self
+ */
+ public StreamPlan connectionFactory(StreamConnectionFactory factory)
+ {
+ this.connectionFactory = factory;
+ return this;
+ }
+
+ /**
* @return true if this plan has no plan to execute
*/
public boolean isEmpty()
@@ -167,7 +181,7 @@ public class StreamPlan
StreamSession session = sessions.get(peer);
if (session == null)
{
- session = new StreamSession(peer);
+ session = new StreamSession(peer, connectionFactory);
sessions.put(peer, session);
}
return session;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/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 dcffaff..add14f7 100644
--- a/src/java/org/apache/cassandra/streaming/StreamResultFuture.java
+++ b/src/java/org/apache/cassandra/streaming/StreamResultFuture.java
@@ -106,7 +106,7 @@ public final class StreamResultFuture extends AbstractFuture<StreamState>
StreamResultFuture future = StreamManager.instance.getReceivingStream(planId);
if (future == null)
{
- final StreamSession session = new StreamSession(from);
+ final StreamSession session = new StreamSession(from, null);
// The main reason we create a StreamResultFuture on the receiving side is for JMX exposure.
future = new StreamResultFuture(planId, description, Collections.singleton(session));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/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 55e30f0..4fcbe36 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -19,6 +19,7 @@ package org.apache.cassandra.streaming;
import java.io.IOException;
import java.net.InetAddress;
+import java.net.Socket;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -128,6 +129,8 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
// data receivers, filled after receiving prepare message
private final Map<UUID, StreamReceiveTask> receivers = new ConcurrentHashMap<>();
private final StreamingMetrics metrics;
+ /* can be null when session is created in remote */
+ private final StreamConnectionFactory factory;
public final ConnectionHandler handler;
@@ -152,10 +155,12 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
* Create new streaming session with the peer.
*
* @param peer Address of streaming peer
+ * @param factory is used for establishing connection
*/
- public StreamSession(InetAddress peer)
+ public StreamSession(InetAddress peer, StreamConnectionFactory factory)
{
this.peer = peer;
+ this.factory = factory;
this.handler = new ConnectionHandler(this);
this.metrics = StreamingMetrics.get(peer);
}
@@ -211,6 +216,12 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
});
}
+ public Socket createConnection() throws IOException
+ {
+ assert factory != null;
+ return factory.createConnection(peer);
+ }
+
/**
* Request data fetch task to this session.
*
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java b/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java
new file mode 100644
index 0000000..399344e
--- /dev/null
+++ b/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.tools;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.nio.channels.SocketChannel;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.cassandra.config.EncryptionOptions;
+import org.apache.cassandra.security.SSLFactory;
+import org.apache.cassandra.streaming.StreamConnectionFactory;
+import org.apache.cassandra.utils.FBUtilities;
+
+public class BulkLoadConnectionFactory implements StreamConnectionFactory
+{
+ private final boolean outboundBindAny;
+ private final int storagePort;
+ private final int secureStoragePort;
+ private final EncryptionOptions.ServerEncryptionOptions encryptionOptions;
+
+ public BulkLoadConnectionFactory(int storagePort, int secureStoragePort, EncryptionOptions.ServerEncryptionOptions encryptionOptions, boolean outboundBindAny)
+ {
+ this.storagePort = storagePort;
+ this.secureStoragePort = secureStoragePort;
+ this.encryptionOptions = encryptionOptions;
+ this.outboundBindAny = outboundBindAny;
+ }
+
+ public Socket createConnection(InetAddress peer) throws IOException
+ {
+ // Connect to secure port for all peers if ServerEncryptionOptions is configured other than 'none'
+ // When 'all', 'dc' and 'rack', server nodes always have SSL port open, and since thin client like sstableloader
+ // does not know which node is in which dc/rack, connecting to SSL port is always the option.
+ if (encryptionOptions != null && encryptionOptions.internode_encryption != EncryptionOptions.ServerEncryptionOptions.InternodeEncryption.none)
+ {
+ if (outboundBindAny)
+ return SSLFactory.getSocket(encryptionOptions, peer, secureStoragePort);
+ else
+ return SSLFactory.getSocket(encryptionOptions, peer, secureStoragePort, FBUtilities.getLocalAddress(), 0);
+ }
+ else
+ {
+ Socket socket = SocketChannel.open(new InetSocketAddress(peer, storagePort)).socket();
+ if (outboundBindAny && !socket.isBound())
+ socket.bind(new InetSocketAddress(FBUtilities.getLocalAddress(), 0));
+ return socket;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/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 37ec635..4077722 100644
--- a/src/java/org/apache/cassandra/tools/BulkLoader.java
+++ b/src/java/org/apache/cassandra/tools/BulkLoader.java
@@ -18,29 +18,27 @@
package org.apache.cassandra.tools;
import java.io.File;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
+import java.net.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import com.google.common.base.Joiner;
import com.google.common.collect.Sets;
-import org.apache.cassandra.config.EncryptionOptions;
+
+import org.apache.cassandra.config.*;
+
import org.apache.commons.cli.*;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
-import org.apache.thrift.transport.TFramedTransport;
-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;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.sstable.SSTableLoader;
import org.apache.cassandra.streaming.*;
import org.apache.cassandra.thrift.*;
@@ -60,7 +58,10 @@ public class BulkLoader
private static final String USER_OPTION = "username";
private static final String PASSWD_OPTION = "password";
private static final String THROTTLE_MBITS = "throttle";
+
private static final String TRANSPORT_FACTORY = "transport-factory";
+
+ /* client encryption options */
private static final String SSL_TRUSTSTORE = "truststore";
private static final String SSL_TRUSTSTORE_PW = "truststore-password";
private static final String SSL_KEYSTORE = "keystore";
@@ -69,12 +70,20 @@ public class BulkLoader
private static final String SSL_ALGORITHM = "ssl-alg";
private static final String SSL_STORE_TYPE = "store-type";
private static final String SSL_CIPHER_SUITES = "ssl-ciphers";
+ private static final String CONFIG_PATH = "conf-path";
public static void main(String args[])
{
LoaderOptions options = LoaderOptions.parseArgs(args);
OutputHandler handler = new OutputHandler.SystemOutput(options.verbose, options.debug);
- SSTableLoader loader = new SSTableLoader(options.directory, new ExternalClient(options.hosts, options.rpcPort, options.user, options.passwd, options.transportFactory), handler);
+ SSTableLoader loader = new SSTableLoader(options.directory, new ExternalClient(options.hosts,
+ options.rpcPort,
+ options.user,
+ options.passwd,
+ options.transportFactory,
+ options.storagePort,
+ options.sslStoragePort,
+ options.serverEncOptions), handler);
DatabaseDescriptor.setStreamThroughputOutboundMegabitsPerSec(options.throttle);
StreamResultFuture future = null;
try
@@ -207,8 +216,18 @@ public class BulkLoader
private final String user;
private final String passwd;
private final ITransportFactory transportFactory;
-
- public ExternalClient(Set<InetAddress> hosts, int port, String user, String passwd, ITransportFactory transportFactory)
+ private final int storagePort;
+ private final int sslStoragePort;
+ private final EncryptionOptions.ServerEncryptionOptions serverEncOptions;
+
+ public ExternalClient(Set<InetAddress> hosts,
+ int port,
+ String user,
+ String passwd,
+ ITransportFactory transportFactory,
+ int storagePort,
+ int sslStoragePort,
+ EncryptionOptions.ServerEncryptionOptions serverEncryptionOptions)
{
super();
this.hosts = hosts;
@@ -216,8 +235,12 @@ public class BulkLoader
this.user = user;
this.passwd = passwd;
this.transportFactory = transportFactory;
+ this.storagePort = storagePort;
+ this.sslStoragePort = sslStoragePort;
+ this.serverEncOptions = serverEncryptionOptions;
}
+ @Override
public void init(String keyspace)
{
Iterator<InetAddress> hostiter = hosts.iterator();
@@ -234,7 +257,7 @@ public class BulkLoader
for (TokenRange tr : client.describe_ring(keyspace))
{
- Range<Token> range = new Range<>(tkFactory.fromString(tr.start_token), tkFactory.fromString(tr.end_token));
+ Range<Token> range = new Range<>(tkFactory.fromString(tr.start_token), tkFactory.fromString(tr.end_token), getPartitioner());
for (String ep : tr.endpoints)
{
addRangeForEndpoint(range, InetAddress.getByName(ep));
@@ -261,6 +284,13 @@ public class BulkLoader
}
}
+ @Override
+ public StreamConnectionFactory getConnectionFactory()
+ {
+ return new BulkLoadConnectionFactory(storagePort, sslStoragePort, serverEncOptions, false);
+ }
+
+ @Override
public CFMetaData getCFMetaData(String keyspace, String cfName)
{
return knownCfs.get(cfName);
@@ -273,7 +303,7 @@ public class BulkLoader
Cassandra.Client client = new Cassandra.Client(protocol);
if (user != null && passwd != null)
{
- Map<String, String> credentials = new HashMap<String, String>();
+ Map<String, String> credentials = new HashMap<>();
credentials.put(IAuthenticator.USERNAME_KEY, user);
credentials.put(IAuthenticator.PASSWORD_KEY, passwd);
AuthenticationRequest authenticationRequest = new AuthenticationRequest(credentials);
@@ -294,11 +324,14 @@ public class BulkLoader
public String user;
public String passwd;
public int throttle = 0;
+ public int storagePort;
+ public int sslStoragePort;
public ITransportFactory transportFactory = new TFramedTransportFactory();
public EncryptionOptions encOptions = new EncryptionOptions.ClientEncryptionOptions();
+ public EncryptionOptions.ServerEncryptionOptions serverEncOptions = new EncryptionOptions.ServerEncryptionOptions();
- public final Set<InetAddress> hosts = new HashSet<InetAddress>();
- public final Set<InetAddress> ignores = new HashSet<InetAddress>();
+ public final Set<InetAddress> hosts = new HashSet<>();
+ public final Set<InetAddress> ignores = new HashSet<>();
LoaderOptions(File directory)
{
@@ -349,9 +382,6 @@ public class BulkLoader
opts.verbose = cmd.hasOption(VERBOSE_OPTION);
opts.noProgress = cmd.hasOption(NOPROGRESS_OPTION);
- if (cmd.hasOption(THROTTLE_MBITS))
- opts.throttle = Integer.parseInt(cmd.getOptionValue(THROTTLE_MBITS));
-
if (cmd.hasOption(RPC_PORT_OPTION))
opts.rpcPort = Integer.parseInt(cmd.getOptionValue(RPC_PORT_OPTION));
@@ -400,44 +430,71 @@ public class BulkLoader
}
}
- if(cmd.hasOption(SSL_TRUSTSTORE))
+ // try to load config file first, so that values can be rewritten with other option values.
+ // otherwise use default config.
+ Config config;
+ if (cmd.hasOption(CONFIG_PATH))
+ {
+ File configFile = new File(cmd.getOptionValue(CONFIG_PATH));
+ if (!configFile.exists())
+ {
+ errorMsg("Config file not found", options);
+ }
+ config = new YamlConfigurationLoader().loadConfig(configFile.toURI().toURL());
+ }
+ else
+ {
+ config = new Config();
+ }
+ opts.storagePort = config.storage_port;
+ opts.sslStoragePort = config.ssl_storage_port;
+ opts.throttle = config.stream_throughput_outbound_megabits_per_sec;
+ opts.encOptions = config.client_encryption_options;
+ opts.serverEncOptions = config.server_encryption_options;
+
+ if (cmd.hasOption(THROTTLE_MBITS))
+ {
+ opts.throttle = Integer.parseInt(cmd.getOptionValue(THROTTLE_MBITS));
+ }
+
+ if (cmd.hasOption(SSL_TRUSTSTORE))
{
opts.encOptions.truststore = cmd.getOptionValue(SSL_TRUSTSTORE);
}
- if(cmd.hasOption(SSL_TRUSTSTORE_PW))
+ if (cmd.hasOption(SSL_TRUSTSTORE_PW))
{
opts.encOptions.truststore_password = cmd.getOptionValue(SSL_TRUSTSTORE_PW);
}
- if(cmd.hasOption(SSL_KEYSTORE))
+ if (cmd.hasOption(SSL_KEYSTORE))
{
opts.encOptions.keystore = cmd.getOptionValue(SSL_KEYSTORE);
// if a keystore was provided, lets assume we'll need to use it
opts.encOptions.require_client_auth = true;
}
- if(cmd.hasOption(SSL_KEYSTORE_PW))
+ if (cmd.hasOption(SSL_KEYSTORE_PW))
{
opts.encOptions.keystore_password = cmd.getOptionValue(SSL_KEYSTORE_PW);
}
- if(cmd.hasOption(SSL_PROTOCOL))
+ if (cmd.hasOption(SSL_PROTOCOL))
{
opts.encOptions.protocol = cmd.getOptionValue(SSL_PROTOCOL);
}
- if(cmd.hasOption(SSL_ALGORITHM))
+ if (cmd.hasOption(SSL_ALGORITHM))
{
opts.encOptions.algorithm = cmd.getOptionValue(SSL_ALGORITHM);
}
- if(cmd.hasOption(SSL_STORE_TYPE))
+ if (cmd.hasOption(SSL_STORE_TYPE))
{
opts.encOptions.store_type = cmd.getOptionValue(SSL_STORE_TYPE);
}
- if(cmd.hasOption(SSL_CIPHER_SUITES))
+ if (cmd.hasOption(SSL_CIPHER_SUITES))
{
opts.encOptions.cipher_suites = cmd.getOptionValue(SSL_CIPHER_SUITES).split(",");
}
@@ -451,7 +508,7 @@ public class BulkLoader
return opts;
}
- catch (ParseException e)
+ catch (ParseException | ConfigurationException | MalformedURLException e)
{
errorMsg(e.getMessage(), options);
return null;
@@ -508,6 +565,7 @@ public class BulkLoader
printUsage(options);
System.exit(1);
}
+
private static CmdLineOptions getCmdLineOptions()
{
CmdLineOptions options = new CmdLineOptions();
@@ -516,37 +574,38 @@ public class BulkLoader
options.addOption("h", HELP_OPTION, "display this help message");
options.addOption(null, NOPROGRESS_OPTION, "don't display progress");
options.addOption("i", IGNORE_NODES_OPTION, "NODES", "don't stream to this (comma separated) list of nodes");
- options.addOption("d", INITIAL_HOST_ADDRESS_OPTION, "initial hosts", "try to connect to these hosts (comma separated) initially for ring information");
+ options.addOption("d", INITIAL_HOST_ADDRESS_OPTION, "initial hosts", "Required. try to connect to these hosts (comma separated) initially for ring information");
options.addOption("p", RPC_PORT_OPTION, "rpc port", "port used for rpc (default 9160)");
options.addOption("t", THROTTLE_MBITS, "throttle", "throttle speed in Mbits (default unlimited)");
options.addOption("u", USER_OPTION, "username", "username for cassandra authentication");
options.addOption("pw", PASSWD_OPTION, "password", "password for cassandra authentication");
options.addOption("tf", TRANSPORT_FACTORY, "transport factory", "Fully-qualified ITransportFactory class name for creating a connection to cassandra");
// ssl connection-related options
- options.addOption("ts", SSL_TRUSTSTORE, "TRUSTSTORE", "SSL: full path to truststore");
- options.addOption("tspw", SSL_TRUSTSTORE_PW, "TRUSTSTORE-PASSWORD", "SSL: password of the truststore");
- options.addOption("ks", SSL_KEYSTORE, "KEYSTORE", "SSL: full path to keystore");
- options.addOption("kspw", SSL_KEYSTORE_PW, "KEYSTORE-PASSWORD", "SSL: password of the keystore");
- options.addOption("prtcl", SSL_PROTOCOL, "PROTOCOL", "SSL: connections protocol to use (default: TLS)");
- options.addOption("alg", SSL_ALGORITHM, "ALGORITHM", "SSL: algorithm (default: SunX509)");
- options.addOption("st", SSL_STORE_TYPE, "STORE-TYPE", "SSL: type of store");
- options.addOption("ciphers", SSL_CIPHER_SUITES, "CIPHER-SUITES", "SSL: comma-separated list of encryption suites to use");
+ options.addOption("ts", SSL_TRUSTSTORE, "TRUSTSTORE", "Client SSL: full path to truststore");
+ options.addOption("tspw", SSL_TRUSTSTORE_PW, "TRUSTSTORE-PASSWORD", "Client SSL: password of the truststore");
+ options.addOption("ks", SSL_KEYSTORE, "KEYSTORE", "Client SSL: full path to keystore");
+ options.addOption("kspw", SSL_KEYSTORE_PW, "KEYSTORE-PASSWORD", "Client SSL: password of the keystore");
+ options.addOption("prtcl", SSL_PROTOCOL, "PROTOCOL", "Client SSL: connections protocol to use (default: TLS)");
+ options.addOption("alg", SSL_ALGORITHM, "ALGORITHM", "Client SSL: algorithm (default: SunX509)");
+ options.addOption("st", SSL_STORE_TYPE, "STORE-TYPE", "Client SSL: type of store");
+ options.addOption("ciphers", SSL_CIPHER_SUITES, "CIPHER-SUITES", "Client SSL: comma-separated list of encryption suites to use");
+ options.addOption("f", CONFIG_PATH, "path to config file", "cassandra.yaml file path for streaming throughput and client/server SSL.");
return options;
}
public static void printUsage(Options options)
{
String usage = String.format("%s [options] <dir_path>", TOOL_NAME);
- StringBuilder header = new StringBuilder();
- header.append("--\n");
- header.append("Bulk load the sstables found in the directory <dir_path> to the configured cluster." );
- header.append("The parent directory of <dir_path> is used as the keyspace name. ");
- header.append("So for instance, to load an sstable named Standard1-g-1-Data.db into keyspace Keyspace1, ");
- header.append("you will need to have the files Standard1-g-1-Data.db and Standard1-g-1-Index.db in a ");
- header.append("directory Keyspace1/Standard1/ in the directory and call: sstableloader Keyspace1/Standard1");
- header.append("\n--\n");
- header.append("Options are:");
- new HelpFormatter().printHelp(usage, header.toString(), options, "");
+ String header = System.lineSeparator() +
+ "Bulk load the sstables found in the directory <dir_path> to the configured cluster." +
+ "The parent directories of <dir_path> are used as the target keyspace/table name. " +
+ "So for instance, to load an sstable named Standard1-g-1-Data.db into Keyspace1/Standard1, " +
+ "you will need to have the files Standard1-g-1-Data.db and Standard1-g-1-Index.db into a directory /path/to/Keyspace1/Standard1/.";
+ String footer = System.lineSeparator() +
+ "You can provide cassandra.yaml file with -f command line option to set up streaming throughput, client and server encryption options. " +
+ "Only stream_throughput_outbound_megabits_per_sec, server_encryption_options and client_encryption_options are read from yaml. " +
+ "You can override options read from cassandra.yaml with corresponding command line options.";
+ new HelpFormatter().printHelp(usage, header, options, footer);
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java b/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
index 9b02817..ce0f9d0 100644
--- a/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
+++ b/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
@@ -43,7 +43,7 @@ public class StreamTransferTaskTest extends SchemaLoader
String ks = "Keyspace1";
String cf = "Standard1";
- StreamSession session = new StreamSession(FBUtilities.getBroadcastAddress());
+ StreamSession session = new StreamSession(FBUtilities.getBroadcastAddress(), null);
ColumnFamilyStore cfs = Keyspace.open(ks).getColumnFamilyStore(cf);
// create two sstables
[2/6] git commit: Fix sstableloader unable to connect encrypted node
Posted by yu...@apache.org.
Fix sstableloader unable to connect encrypted node
patch by yukim; reviewed by krummas for CASSANDRA-7585
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/563cea14
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/563cea14
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/563cea14
Branch: refs/heads/cassandra-2.1
Commit: 563cea14b4bb87cd37ab10399904f08757c34d27
Parents: ad6ba3d
Author: Yuki Morishita <yu...@apache.org>
Authored: Fri Aug 15 12:31:59 2014 -0500
Committer: Yuki Morishita <yu...@apache.org>
Committed: Fri Aug 15 12:31:59 2014 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../config/YamlConfigurationLoader.java | 6 +-
.../cassandra/io/sstable/SSTableLoader.java | 22 ++-
.../cassandra/streaming/ConnectionHandler.java | 48 +-----
.../streaming/DefaultConnectionFactory.java | 74 +++++++++
.../streaming/StreamConnectionFactory.java | 30 ++++
.../apache/cassandra/streaming/StreamPlan.java | 16 +-
.../cassandra/streaming/StreamResultFuture.java | 2 +-
.../cassandra/streaming/StreamSession.java | 13 +-
.../tools/BulkLoadConnectionFactory.java | 68 +++++++++
.../org/apache/cassandra/tools/BulkLoader.java | 149 +++++++++++++------
.../streaming/StreamTransferTaskTest.java | 2 +-
12 files changed, 330 insertions(+), 101 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 4306de5..e335484 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -47,6 +47,7 @@
* Backport CASSANDRA-6747 (CASSANDRA-7560)
* Track max/min timestamps for range tombstones (CASSANDRA-7647)
* Fix NPE when listing saved caches dir (CASSANDRA-7632)
+ * Fix sstableloader unable to connect encrypted node (CASSANDRA-7585)
Merged from 1.2:
* Remove duplicates from StorageService.getJoiningNodes (CASSANDRA-7478)
* Clone token map outside of hot gossip loops (CASSANDRA-7758)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java b/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
index 6b5a152..b520d07 100644
--- a/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
+++ b/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
@@ -69,10 +69,14 @@ public class YamlConfigurationLoader implements ConfigurationLoader
public Config loadConfig() throws ConfigurationException
{
+ return loadConfig(getStorageConfigURL());
+ }
+
+ public Config loadConfig(URL url) throws ConfigurationException
+ {
InputStream input = null;
try
{
- URL url = getStorageConfigURL();
logger.info("Loading settings from {}", url);
try
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/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 4a1604d..85dc0e4 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
@@ -50,7 +50,7 @@ public class SSTableLoader implements StreamEventHandler
private final OutputHandler outputHandler;
private final Set<InetAddress> failedHosts = new HashSet<>();
- private final List<SSTableReader> sstables = new ArrayList<SSTableReader>();
+ private final List<SSTableReader> sstables = new ArrayList<>();
private final Multimap<InetAddress, StreamSession.SSTableStreamingSections> streamingDetails = HashMultimap.create();
static
@@ -94,7 +94,7 @@ public class SSTableLoader implements StreamEventHandler
return false;
}
- Set<Component> components = new HashSet<Component>();
+ Set<Component> components = new HashSet<>();
components.add(Component.DATA);
components.add(Component.PRIMARY_INDEX);
if (new File(desc.filenameFor(Component.SUMMARY)).exists())
@@ -149,7 +149,7 @@ public class SSTableLoader implements StreamEventHandler
client.init(keyspace);
outputHandler.output("Established connection to initial hosts");
- StreamPlan plan = new StreamPlan("Bulk Load");
+ StreamPlan plan = new StreamPlan("Bulk Load").connectionFactory(client.getConnectionFactory());
Map<InetAddress, Collection<Range<Token>>> endpointToRanges = client.getEndpointToRangesMap();
openSSTables(endpointToRanges);
@@ -220,7 +220,7 @@ public class SSTableLoader implements StreamEventHandler
public static abstract class Client
{
- private final Map<InetAddress, Collection<Range<Token>>> endpointToRanges = new HashMap<InetAddress, Collection<Range<Token>>>();
+ private final Map<InetAddress, Collection<Range<Token>>> endpointToRanges = new HashMap<>();
private IPartitioner partitioner;
/**
@@ -240,6 +240,17 @@ public class SSTableLoader implements StreamEventHandler
public void stop() {}
/**
+ * Provides connection factory.
+ * By default, it uses DefaultConnectionFactory.
+ *
+ * @return StreamConnectionFactory to use
+ */
+ public StreamConnectionFactory getConnectionFactory()
+ {
+ return new DefaultConnectionFactory();
+ }
+
+ /**
* Validate that {@code keyspace} is an existing keyspace and {@code
* cfName} one of its existing column family.
*/
@@ -258,6 +269,7 @@ public class SSTableLoader implements StreamEventHandler
protected void setPartitioner(IPartitioner partitioner)
{
this.partitioner = partitioner;
+ // the following is still necessary since Range/Token reference partitioner through StorageService.getPartitioner
DatabaseDescriptor.setPartitioner(partitioner);
}
@@ -271,7 +283,7 @@ public class SSTableLoader implements StreamEventHandler
Collection<Range<Token>> ranges = endpointToRanges.get(endpoint);
if (ranges == null)
{
- ranges = new HashSet<Range<Token>>();
+ ranges = new HashSet<>();
endpointToRanges.put(endpoint, ranges);
}
ranges.add(range);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/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 b06a818..8fba41b 100644
--- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
+++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
@@ -18,7 +18,6 @@
package org.apache.cassandra.streaming;
import java.io.IOException;
-import java.net.InetAddress;
import java.net.Socket;
import java.net.SocketException;
import java.nio.ByteBuffer;
@@ -37,8 +36,6 @@ import com.google.common.util.concurrent.SettableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.cassandra.config.DatabaseDescriptor;
-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;
@@ -55,8 +52,6 @@ public class ConnectionHandler
{
private static final Logger logger = LoggerFactory.getLogger(ConnectionHandler.class);
- private static final int MAX_CONNECT_ATTEMPTS = 3;
-
private final StreamSession session;
private IncomingMessageHandler incoming;
@@ -79,12 +74,12 @@ public class ConnectionHandler
public void initiate() throws IOException
{
logger.debug("[Stream #{}] Sending stream init for incoming stream", session.planId());
- Socket incomingSocket = connect(session.peer);
+ Socket incomingSocket = session.createConnection();
incoming.start(incomingSocket, StreamMessage.CURRENT_VERSION);
incoming.sendInitMessage(incomingSocket, true);
logger.debug("[Stream #{}] Sending stream init for outgoing stream", session.planId());
- Socket outgoingSocket = connect(session.peer);
+ Socket outgoingSocket = session.createConnection();
outgoing.start(outgoingSocket, StreamMessage.CURRENT_VERSION);
outgoing.sendInitMessage(outgoingSocket, false);
}
@@ -104,45 +99,6 @@ public class ConnectionHandler
incoming.start(socket, version);
}
- /**
- * Connect to peer and start exchanging message.
- * When connect attempt fails, this retries for maximum of MAX_CONNECT_ATTEMPTS times.
- *
- * @param peer the peer to connect to.
- * @return the created socket.
- *
- * @throws IOException when connection failed.
- */
- private static Socket connect(InetAddress peer) throws IOException
- {
- int attempts = 0;
- while (true)
- {
- try
- {
- Socket socket = OutboundTcpConnectionPool.newSocket(peer);
- socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
- return socket;
- }
- catch (IOException e)
- {
- if (++attempts >= MAX_CONNECT_ATTEMPTS)
- throw e;
-
- long waitms = DatabaseDescriptor.getRpcTimeout() * (long)Math.pow(2, attempts);
- logger.warn("Failed attempt " + attempts + " to connect to " + peer + ". Retrying in " + waitms + " ms. (" + e + ")");
- try
- {
- Thread.sleep(waitms);
- }
- catch (InterruptedException wtf)
- {
- throw new IOException("interrupted", wtf);
- }
- }
- }
- }
-
public ListenableFuture<?> close()
{
logger.debug("[Stream #{}] Closing stream connection handler on {}", session.planId(), session.peer);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java b/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
new file mode 100644
index 0000000..53af4c8
--- /dev/null
+++ b/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.streaming;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.Socket;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.net.OutboundTcpConnectionPool;
+
+public class DefaultConnectionFactory implements StreamConnectionFactory
+{
+ private static final Logger logger = LoggerFactory.getLogger(DefaultConnectionFactory.class);
+
+ private static final int MAX_CONNECT_ATTEMPTS = 3;
+
+ /**
+ * Connect to peer and start exchanging message.
+ * When connect attempt fails, this retries for maximum of MAX_CONNECT_ATTEMPTS times.
+ *
+ * @param peer the peer to connect to.
+ * @return the created socket.
+ *
+ * @throws IOException when connection failed.
+ */
+ public Socket createConnection(InetAddress peer) throws IOException
+ {
+ int attempts = 0;
+ while (true)
+ {
+ try
+ {
+ Socket socket = OutboundTcpConnectionPool.newSocket(peer);
+ socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
+ return socket;
+ }
+ catch (IOException e)
+ {
+ if (++attempts >= MAX_CONNECT_ATTEMPTS)
+ throw e;
+
+ long waitms = DatabaseDescriptor.getRpcTimeout() * (long)Math.pow(2, attempts);
+ logger.warn("Failed attempt " + attempts + " to connect to " + peer + ". Retrying in " + waitms + " ms. (" + e + ")");
+ try
+ {
+ Thread.sleep(waitms);
+ }
+ catch (InterruptedException wtf)
+ {
+ throw new IOException("interrupted", wtf);
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java b/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java
new file mode 100644
index 0000000..dd99611
--- /dev/null
+++ b/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.streaming;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.Socket;
+
+/**
+ * Interface that creates connection used by streaming.
+ */
+public interface StreamConnectionFactory
+{
+ Socket createConnection(InetAddress peer) throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/src/java/org/apache/cassandra/streaming/StreamPlan.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamPlan.java b/src/java/org/apache/cassandra/streaming/StreamPlan.java
index b57e097..e582c79 100644
--- a/src/java/org/apache/cassandra/streaming/StreamPlan.java
+++ b/src/java/org/apache/cassandra/streaming/StreamPlan.java
@@ -38,6 +38,8 @@ public class StreamPlan
// sessions per InetAddress of the other end.
private final Map<InetAddress, StreamSession> sessions = new HashMap<>();
+ private StreamConnectionFactory connectionFactory = new DefaultConnectionFactory();
+
private boolean flushBeforeTransfer = true;
/**
@@ -132,6 +134,18 @@ public class StreamPlan
}
/**
+ * Set custom StreamConnectionFactory to be used for establishing connection
+ *
+ * @param factory StreamConnectionFactory to use
+ * @return self
+ */
+ public StreamPlan connectionFactory(StreamConnectionFactory factory)
+ {
+ this.connectionFactory = factory;
+ return this;
+ }
+
+ /**
* @return true if this plan has no plan to execute
*/
public boolean isEmpty()
@@ -167,7 +181,7 @@ public class StreamPlan
StreamSession session = sessions.get(peer);
if (session == null)
{
- session = new StreamSession(peer);
+ session = new StreamSession(peer, connectionFactory);
sessions.put(peer, session);
}
return session;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/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 dcffaff..add14f7 100644
--- a/src/java/org/apache/cassandra/streaming/StreamResultFuture.java
+++ b/src/java/org/apache/cassandra/streaming/StreamResultFuture.java
@@ -106,7 +106,7 @@ public final class StreamResultFuture extends AbstractFuture<StreamState>
StreamResultFuture future = StreamManager.instance.getReceivingStream(planId);
if (future == null)
{
- final StreamSession session = new StreamSession(from);
+ final StreamSession session = new StreamSession(from, null);
// The main reason we create a StreamResultFuture on the receiving side is for JMX exposure.
future = new StreamResultFuture(planId, description, Collections.singleton(session));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/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 55e30f0..4fcbe36 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -19,6 +19,7 @@ package org.apache.cassandra.streaming;
import java.io.IOException;
import java.net.InetAddress;
+import java.net.Socket;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -128,6 +129,8 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
// data receivers, filled after receiving prepare message
private final Map<UUID, StreamReceiveTask> receivers = new ConcurrentHashMap<>();
private final StreamingMetrics metrics;
+ /* can be null when session is created in remote */
+ private final StreamConnectionFactory factory;
public final ConnectionHandler handler;
@@ -152,10 +155,12 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
* Create new streaming session with the peer.
*
* @param peer Address of streaming peer
+ * @param factory is used for establishing connection
*/
- public StreamSession(InetAddress peer)
+ public StreamSession(InetAddress peer, StreamConnectionFactory factory)
{
this.peer = peer;
+ this.factory = factory;
this.handler = new ConnectionHandler(this);
this.metrics = StreamingMetrics.get(peer);
}
@@ -211,6 +216,12 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
});
}
+ public Socket createConnection() throws IOException
+ {
+ assert factory != null;
+ return factory.createConnection(peer);
+ }
+
/**
* Request data fetch task to this session.
*
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java b/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java
new file mode 100644
index 0000000..399344e
--- /dev/null
+++ b/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.tools;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.nio.channels.SocketChannel;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.cassandra.config.EncryptionOptions;
+import org.apache.cassandra.security.SSLFactory;
+import org.apache.cassandra.streaming.StreamConnectionFactory;
+import org.apache.cassandra.utils.FBUtilities;
+
+public class BulkLoadConnectionFactory implements StreamConnectionFactory
+{
+ private final boolean outboundBindAny;
+ private final int storagePort;
+ private final int secureStoragePort;
+ private final EncryptionOptions.ServerEncryptionOptions encryptionOptions;
+
+ public BulkLoadConnectionFactory(int storagePort, int secureStoragePort, EncryptionOptions.ServerEncryptionOptions encryptionOptions, boolean outboundBindAny)
+ {
+ this.storagePort = storagePort;
+ this.secureStoragePort = secureStoragePort;
+ this.encryptionOptions = encryptionOptions;
+ this.outboundBindAny = outboundBindAny;
+ }
+
+ public Socket createConnection(InetAddress peer) throws IOException
+ {
+ // Connect to secure port for all peers if ServerEncryptionOptions is configured other than 'none'
+ // When 'all', 'dc' and 'rack', server nodes always have SSL port open, and since thin client like sstableloader
+ // does not know which node is in which dc/rack, connecting to SSL port is always the option.
+ if (encryptionOptions != null && encryptionOptions.internode_encryption != EncryptionOptions.ServerEncryptionOptions.InternodeEncryption.none)
+ {
+ if (outboundBindAny)
+ return SSLFactory.getSocket(encryptionOptions, peer, secureStoragePort);
+ else
+ return SSLFactory.getSocket(encryptionOptions, peer, secureStoragePort, FBUtilities.getLocalAddress(), 0);
+ }
+ else
+ {
+ Socket socket = SocketChannel.open(new InetSocketAddress(peer, storagePort)).socket();
+ if (outboundBindAny && !socket.isBound())
+ socket.bind(new InetSocketAddress(FBUtilities.getLocalAddress(), 0));
+ return socket;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/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 37ec635..4077722 100644
--- a/src/java/org/apache/cassandra/tools/BulkLoader.java
+++ b/src/java/org/apache/cassandra/tools/BulkLoader.java
@@ -18,29 +18,27 @@
package org.apache.cassandra.tools;
import java.io.File;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
+import java.net.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import com.google.common.base.Joiner;
import com.google.common.collect.Sets;
-import org.apache.cassandra.config.EncryptionOptions;
+
+import org.apache.cassandra.config.*;
+
import org.apache.commons.cli.*;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
-import org.apache.thrift.transport.TFramedTransport;
-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;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.sstable.SSTableLoader;
import org.apache.cassandra.streaming.*;
import org.apache.cassandra.thrift.*;
@@ -60,7 +58,10 @@ public class BulkLoader
private static final String USER_OPTION = "username";
private static final String PASSWD_OPTION = "password";
private static final String THROTTLE_MBITS = "throttle";
+
private static final String TRANSPORT_FACTORY = "transport-factory";
+
+ /* client encryption options */
private static final String SSL_TRUSTSTORE = "truststore";
private static final String SSL_TRUSTSTORE_PW = "truststore-password";
private static final String SSL_KEYSTORE = "keystore";
@@ -69,12 +70,20 @@ public class BulkLoader
private static final String SSL_ALGORITHM = "ssl-alg";
private static final String SSL_STORE_TYPE = "store-type";
private static final String SSL_CIPHER_SUITES = "ssl-ciphers";
+ private static final String CONFIG_PATH = "conf-path";
public static void main(String args[])
{
LoaderOptions options = LoaderOptions.parseArgs(args);
OutputHandler handler = new OutputHandler.SystemOutput(options.verbose, options.debug);
- SSTableLoader loader = new SSTableLoader(options.directory, new ExternalClient(options.hosts, options.rpcPort, options.user, options.passwd, options.transportFactory), handler);
+ SSTableLoader loader = new SSTableLoader(options.directory, new ExternalClient(options.hosts,
+ options.rpcPort,
+ options.user,
+ options.passwd,
+ options.transportFactory,
+ options.storagePort,
+ options.sslStoragePort,
+ options.serverEncOptions), handler);
DatabaseDescriptor.setStreamThroughputOutboundMegabitsPerSec(options.throttle);
StreamResultFuture future = null;
try
@@ -207,8 +216,18 @@ public class BulkLoader
private final String user;
private final String passwd;
private final ITransportFactory transportFactory;
-
- public ExternalClient(Set<InetAddress> hosts, int port, String user, String passwd, ITransportFactory transportFactory)
+ private final int storagePort;
+ private final int sslStoragePort;
+ private final EncryptionOptions.ServerEncryptionOptions serverEncOptions;
+
+ public ExternalClient(Set<InetAddress> hosts,
+ int port,
+ String user,
+ String passwd,
+ ITransportFactory transportFactory,
+ int storagePort,
+ int sslStoragePort,
+ EncryptionOptions.ServerEncryptionOptions serverEncryptionOptions)
{
super();
this.hosts = hosts;
@@ -216,8 +235,12 @@ public class BulkLoader
this.user = user;
this.passwd = passwd;
this.transportFactory = transportFactory;
+ this.storagePort = storagePort;
+ this.sslStoragePort = sslStoragePort;
+ this.serverEncOptions = serverEncryptionOptions;
}
+ @Override
public void init(String keyspace)
{
Iterator<InetAddress> hostiter = hosts.iterator();
@@ -234,7 +257,7 @@ public class BulkLoader
for (TokenRange tr : client.describe_ring(keyspace))
{
- Range<Token> range = new Range<>(tkFactory.fromString(tr.start_token), tkFactory.fromString(tr.end_token));
+ Range<Token> range = new Range<>(tkFactory.fromString(tr.start_token), tkFactory.fromString(tr.end_token), getPartitioner());
for (String ep : tr.endpoints)
{
addRangeForEndpoint(range, InetAddress.getByName(ep));
@@ -261,6 +284,13 @@ public class BulkLoader
}
}
+ @Override
+ public StreamConnectionFactory getConnectionFactory()
+ {
+ return new BulkLoadConnectionFactory(storagePort, sslStoragePort, serverEncOptions, false);
+ }
+
+ @Override
public CFMetaData getCFMetaData(String keyspace, String cfName)
{
return knownCfs.get(cfName);
@@ -273,7 +303,7 @@ public class BulkLoader
Cassandra.Client client = new Cassandra.Client(protocol);
if (user != null && passwd != null)
{
- Map<String, String> credentials = new HashMap<String, String>();
+ Map<String, String> credentials = new HashMap<>();
credentials.put(IAuthenticator.USERNAME_KEY, user);
credentials.put(IAuthenticator.PASSWORD_KEY, passwd);
AuthenticationRequest authenticationRequest = new AuthenticationRequest(credentials);
@@ -294,11 +324,14 @@ public class BulkLoader
public String user;
public String passwd;
public int throttle = 0;
+ public int storagePort;
+ public int sslStoragePort;
public ITransportFactory transportFactory = new TFramedTransportFactory();
public EncryptionOptions encOptions = new EncryptionOptions.ClientEncryptionOptions();
+ public EncryptionOptions.ServerEncryptionOptions serverEncOptions = new EncryptionOptions.ServerEncryptionOptions();
- public final Set<InetAddress> hosts = new HashSet<InetAddress>();
- public final Set<InetAddress> ignores = new HashSet<InetAddress>();
+ public final Set<InetAddress> hosts = new HashSet<>();
+ public final Set<InetAddress> ignores = new HashSet<>();
LoaderOptions(File directory)
{
@@ -349,9 +382,6 @@ public class BulkLoader
opts.verbose = cmd.hasOption(VERBOSE_OPTION);
opts.noProgress = cmd.hasOption(NOPROGRESS_OPTION);
- if (cmd.hasOption(THROTTLE_MBITS))
- opts.throttle = Integer.parseInt(cmd.getOptionValue(THROTTLE_MBITS));
-
if (cmd.hasOption(RPC_PORT_OPTION))
opts.rpcPort = Integer.parseInt(cmd.getOptionValue(RPC_PORT_OPTION));
@@ -400,44 +430,71 @@ public class BulkLoader
}
}
- if(cmd.hasOption(SSL_TRUSTSTORE))
+ // try to load config file first, so that values can be rewritten with other option values.
+ // otherwise use default config.
+ Config config;
+ if (cmd.hasOption(CONFIG_PATH))
+ {
+ File configFile = new File(cmd.getOptionValue(CONFIG_PATH));
+ if (!configFile.exists())
+ {
+ errorMsg("Config file not found", options);
+ }
+ config = new YamlConfigurationLoader().loadConfig(configFile.toURI().toURL());
+ }
+ else
+ {
+ config = new Config();
+ }
+ opts.storagePort = config.storage_port;
+ opts.sslStoragePort = config.ssl_storage_port;
+ opts.throttle = config.stream_throughput_outbound_megabits_per_sec;
+ opts.encOptions = config.client_encryption_options;
+ opts.serverEncOptions = config.server_encryption_options;
+
+ if (cmd.hasOption(THROTTLE_MBITS))
+ {
+ opts.throttle = Integer.parseInt(cmd.getOptionValue(THROTTLE_MBITS));
+ }
+
+ if (cmd.hasOption(SSL_TRUSTSTORE))
{
opts.encOptions.truststore = cmd.getOptionValue(SSL_TRUSTSTORE);
}
- if(cmd.hasOption(SSL_TRUSTSTORE_PW))
+ if (cmd.hasOption(SSL_TRUSTSTORE_PW))
{
opts.encOptions.truststore_password = cmd.getOptionValue(SSL_TRUSTSTORE_PW);
}
- if(cmd.hasOption(SSL_KEYSTORE))
+ if (cmd.hasOption(SSL_KEYSTORE))
{
opts.encOptions.keystore = cmd.getOptionValue(SSL_KEYSTORE);
// if a keystore was provided, lets assume we'll need to use it
opts.encOptions.require_client_auth = true;
}
- if(cmd.hasOption(SSL_KEYSTORE_PW))
+ if (cmd.hasOption(SSL_KEYSTORE_PW))
{
opts.encOptions.keystore_password = cmd.getOptionValue(SSL_KEYSTORE_PW);
}
- if(cmd.hasOption(SSL_PROTOCOL))
+ if (cmd.hasOption(SSL_PROTOCOL))
{
opts.encOptions.protocol = cmd.getOptionValue(SSL_PROTOCOL);
}
- if(cmd.hasOption(SSL_ALGORITHM))
+ if (cmd.hasOption(SSL_ALGORITHM))
{
opts.encOptions.algorithm = cmd.getOptionValue(SSL_ALGORITHM);
}
- if(cmd.hasOption(SSL_STORE_TYPE))
+ if (cmd.hasOption(SSL_STORE_TYPE))
{
opts.encOptions.store_type = cmd.getOptionValue(SSL_STORE_TYPE);
}
- if(cmd.hasOption(SSL_CIPHER_SUITES))
+ if (cmd.hasOption(SSL_CIPHER_SUITES))
{
opts.encOptions.cipher_suites = cmd.getOptionValue(SSL_CIPHER_SUITES).split(",");
}
@@ -451,7 +508,7 @@ public class BulkLoader
return opts;
}
- catch (ParseException e)
+ catch (ParseException | ConfigurationException | MalformedURLException e)
{
errorMsg(e.getMessage(), options);
return null;
@@ -508,6 +565,7 @@ public class BulkLoader
printUsage(options);
System.exit(1);
}
+
private static CmdLineOptions getCmdLineOptions()
{
CmdLineOptions options = new CmdLineOptions();
@@ -516,37 +574,38 @@ public class BulkLoader
options.addOption("h", HELP_OPTION, "display this help message");
options.addOption(null, NOPROGRESS_OPTION, "don't display progress");
options.addOption("i", IGNORE_NODES_OPTION, "NODES", "don't stream to this (comma separated) list of nodes");
- options.addOption("d", INITIAL_HOST_ADDRESS_OPTION, "initial hosts", "try to connect to these hosts (comma separated) initially for ring information");
+ options.addOption("d", INITIAL_HOST_ADDRESS_OPTION, "initial hosts", "Required. try to connect to these hosts (comma separated) initially for ring information");
options.addOption("p", RPC_PORT_OPTION, "rpc port", "port used for rpc (default 9160)");
options.addOption("t", THROTTLE_MBITS, "throttle", "throttle speed in Mbits (default unlimited)");
options.addOption("u", USER_OPTION, "username", "username for cassandra authentication");
options.addOption("pw", PASSWD_OPTION, "password", "password for cassandra authentication");
options.addOption("tf", TRANSPORT_FACTORY, "transport factory", "Fully-qualified ITransportFactory class name for creating a connection to cassandra");
// ssl connection-related options
- options.addOption("ts", SSL_TRUSTSTORE, "TRUSTSTORE", "SSL: full path to truststore");
- options.addOption("tspw", SSL_TRUSTSTORE_PW, "TRUSTSTORE-PASSWORD", "SSL: password of the truststore");
- options.addOption("ks", SSL_KEYSTORE, "KEYSTORE", "SSL: full path to keystore");
- options.addOption("kspw", SSL_KEYSTORE_PW, "KEYSTORE-PASSWORD", "SSL: password of the keystore");
- options.addOption("prtcl", SSL_PROTOCOL, "PROTOCOL", "SSL: connections protocol to use (default: TLS)");
- options.addOption("alg", SSL_ALGORITHM, "ALGORITHM", "SSL: algorithm (default: SunX509)");
- options.addOption("st", SSL_STORE_TYPE, "STORE-TYPE", "SSL: type of store");
- options.addOption("ciphers", SSL_CIPHER_SUITES, "CIPHER-SUITES", "SSL: comma-separated list of encryption suites to use");
+ options.addOption("ts", SSL_TRUSTSTORE, "TRUSTSTORE", "Client SSL: full path to truststore");
+ options.addOption("tspw", SSL_TRUSTSTORE_PW, "TRUSTSTORE-PASSWORD", "Client SSL: password of the truststore");
+ options.addOption("ks", SSL_KEYSTORE, "KEYSTORE", "Client SSL: full path to keystore");
+ options.addOption("kspw", SSL_KEYSTORE_PW, "KEYSTORE-PASSWORD", "Client SSL: password of the keystore");
+ options.addOption("prtcl", SSL_PROTOCOL, "PROTOCOL", "Client SSL: connections protocol to use (default: TLS)");
+ options.addOption("alg", SSL_ALGORITHM, "ALGORITHM", "Client SSL: algorithm (default: SunX509)");
+ options.addOption("st", SSL_STORE_TYPE, "STORE-TYPE", "Client SSL: type of store");
+ options.addOption("ciphers", SSL_CIPHER_SUITES, "CIPHER-SUITES", "Client SSL: comma-separated list of encryption suites to use");
+ options.addOption("f", CONFIG_PATH, "path to config file", "cassandra.yaml file path for streaming throughput and client/server SSL.");
return options;
}
public static void printUsage(Options options)
{
String usage = String.format("%s [options] <dir_path>", TOOL_NAME);
- StringBuilder header = new StringBuilder();
- header.append("--\n");
- header.append("Bulk load the sstables found in the directory <dir_path> to the configured cluster." );
- header.append("The parent directory of <dir_path> is used as the keyspace name. ");
- header.append("So for instance, to load an sstable named Standard1-g-1-Data.db into keyspace Keyspace1, ");
- header.append("you will need to have the files Standard1-g-1-Data.db and Standard1-g-1-Index.db in a ");
- header.append("directory Keyspace1/Standard1/ in the directory and call: sstableloader Keyspace1/Standard1");
- header.append("\n--\n");
- header.append("Options are:");
- new HelpFormatter().printHelp(usage, header.toString(), options, "");
+ String header = System.lineSeparator() +
+ "Bulk load the sstables found in the directory <dir_path> to the configured cluster." +
+ "The parent directories of <dir_path> are used as the target keyspace/table name. " +
+ "So for instance, to load an sstable named Standard1-g-1-Data.db into Keyspace1/Standard1, " +
+ "you will need to have the files Standard1-g-1-Data.db and Standard1-g-1-Index.db into a directory /path/to/Keyspace1/Standard1/.";
+ String footer = System.lineSeparator() +
+ "You can provide cassandra.yaml file with -f command line option to set up streaming throughput, client and server encryption options. " +
+ "Only stream_throughput_outbound_megabits_per_sec, server_encryption_options and client_encryption_options are read from yaml. " +
+ "You can override options read from cassandra.yaml with corresponding command line options.";
+ new HelpFormatter().printHelp(usage, header, options, footer);
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java b/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
index 9b02817..ce0f9d0 100644
--- a/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
+++ b/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
@@ -43,7 +43,7 @@ public class StreamTransferTaskTest extends SchemaLoader
String ks = "Keyspace1";
String cf = "Standard1";
- StreamSession session = new StreamSession(FBUtilities.getBroadcastAddress());
+ StreamSession session = new StreamSession(FBUtilities.getBroadcastAddress(), null);
ColumnFamilyStore cfs = Keyspace.open(ks).getColumnFamilyStore(cf);
// create two sstables
[6/6] git commit: Merge branch 'cassandra-2.1' into trunk
Posted by yu...@apache.org.
Merge branch 'cassandra-2.1' into trunk
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/ff88f237
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/ff88f237
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/ff88f237
Branch: refs/heads/trunk
Commit: ff88f237662e0ad3389373bc4466bf80c10420d1
Parents: fe8829f de5bb58
Author: Yuki Morishita <yu...@apache.org>
Authored: Fri Aug 15 13:26:39 2014 -0500
Committer: Yuki Morishita <yu...@apache.org>
Committed: Fri Aug 15 13:26:39 2014 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../config/YamlConfigurationLoader.java | 7 +-
.../cassandra/io/sstable/SSTableLoader.java | 22 ++-
.../cassandra/streaming/ConnectionHandler.java | 49 +------
.../streaming/DefaultConnectionFactory.java | 75 ++++++++++
.../streaming/StreamConnectionFactory.java | 30 ++++
.../cassandra/streaming/StreamCoordinator.java | 13 +-
.../apache/cassandra/streaming/StreamPlan.java | 17 ++-
.../cassandra/streaming/StreamResultFuture.java | 2 +-
.../cassandra/streaming/StreamSession.java | 13 +-
.../tools/BulkLoadConnectionFactory.java | 68 +++++++++
.../org/apache/cassandra/tools/BulkLoader.java | 139 +++++++++++++------
.../streaming/StreamTransferTaskTest.java | 2 +-
13 files changed, 337 insertions(+), 101 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ff88f237/CHANGES.txt
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ff88f237/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ff88f237/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ff88f237/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
index f9ae82e,4043ac8..5d246f4
--- a/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
+++ b/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
@@@ -57,10 -40,10 +57,10 @@@ public class StreamTransferTaskTes
@Test
public void testScheduleTimeout() throws Exception
{
- String ks = "Keyspace1";
+ String ks = KEYSPACE1;
String cf = "Standard1";
- StreamSession session = new StreamSession(FBUtilities.getBroadcastAddress(), 0);
+ StreamSession session = new StreamSession(FBUtilities.getBroadcastAddress(), null, 0);
ColumnFamilyStore cfs = Keyspace.open(ks).getColumnFamilyStore(cf);
// create two sstables
[3/6] git commit: Fix sstableloader unable to connect encrypted node
Posted by yu...@apache.org.
Fix sstableloader unable to connect encrypted node
patch by yukim; reviewed by krummas for CASSANDRA-7585
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/563cea14
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/563cea14
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/563cea14
Branch: refs/heads/trunk
Commit: 563cea14b4bb87cd37ab10399904f08757c34d27
Parents: ad6ba3d
Author: Yuki Morishita <yu...@apache.org>
Authored: Fri Aug 15 12:31:59 2014 -0500
Committer: Yuki Morishita <yu...@apache.org>
Committed: Fri Aug 15 12:31:59 2014 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../config/YamlConfigurationLoader.java | 6 +-
.../cassandra/io/sstable/SSTableLoader.java | 22 ++-
.../cassandra/streaming/ConnectionHandler.java | 48 +-----
.../streaming/DefaultConnectionFactory.java | 74 +++++++++
.../streaming/StreamConnectionFactory.java | 30 ++++
.../apache/cassandra/streaming/StreamPlan.java | 16 +-
.../cassandra/streaming/StreamResultFuture.java | 2 +-
.../cassandra/streaming/StreamSession.java | 13 +-
.../tools/BulkLoadConnectionFactory.java | 68 +++++++++
.../org/apache/cassandra/tools/BulkLoader.java | 149 +++++++++++++------
.../streaming/StreamTransferTaskTest.java | 2 +-
12 files changed, 330 insertions(+), 101 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 4306de5..e335484 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -47,6 +47,7 @@
* Backport CASSANDRA-6747 (CASSANDRA-7560)
* Track max/min timestamps for range tombstones (CASSANDRA-7647)
* Fix NPE when listing saved caches dir (CASSANDRA-7632)
+ * Fix sstableloader unable to connect encrypted node (CASSANDRA-7585)
Merged from 1.2:
* Remove duplicates from StorageService.getJoiningNodes (CASSANDRA-7478)
* Clone token map outside of hot gossip loops (CASSANDRA-7758)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java b/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
index 6b5a152..b520d07 100644
--- a/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
+++ b/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
@@ -69,10 +69,14 @@ public class YamlConfigurationLoader implements ConfigurationLoader
public Config loadConfig() throws ConfigurationException
{
+ return loadConfig(getStorageConfigURL());
+ }
+
+ public Config loadConfig(URL url) throws ConfigurationException
+ {
InputStream input = null;
try
{
- URL url = getStorageConfigURL();
logger.info("Loading settings from {}", url);
try
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/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 4a1604d..85dc0e4 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
@@ -50,7 +50,7 @@ public class SSTableLoader implements StreamEventHandler
private final OutputHandler outputHandler;
private final Set<InetAddress> failedHosts = new HashSet<>();
- private final List<SSTableReader> sstables = new ArrayList<SSTableReader>();
+ private final List<SSTableReader> sstables = new ArrayList<>();
private final Multimap<InetAddress, StreamSession.SSTableStreamingSections> streamingDetails = HashMultimap.create();
static
@@ -94,7 +94,7 @@ public class SSTableLoader implements StreamEventHandler
return false;
}
- Set<Component> components = new HashSet<Component>();
+ Set<Component> components = new HashSet<>();
components.add(Component.DATA);
components.add(Component.PRIMARY_INDEX);
if (new File(desc.filenameFor(Component.SUMMARY)).exists())
@@ -149,7 +149,7 @@ public class SSTableLoader implements StreamEventHandler
client.init(keyspace);
outputHandler.output("Established connection to initial hosts");
- StreamPlan plan = new StreamPlan("Bulk Load");
+ StreamPlan plan = new StreamPlan("Bulk Load").connectionFactory(client.getConnectionFactory());
Map<InetAddress, Collection<Range<Token>>> endpointToRanges = client.getEndpointToRangesMap();
openSSTables(endpointToRanges);
@@ -220,7 +220,7 @@ public class SSTableLoader implements StreamEventHandler
public static abstract class Client
{
- private final Map<InetAddress, Collection<Range<Token>>> endpointToRanges = new HashMap<InetAddress, Collection<Range<Token>>>();
+ private final Map<InetAddress, Collection<Range<Token>>> endpointToRanges = new HashMap<>();
private IPartitioner partitioner;
/**
@@ -240,6 +240,17 @@ public class SSTableLoader implements StreamEventHandler
public void stop() {}
/**
+ * Provides connection factory.
+ * By default, it uses DefaultConnectionFactory.
+ *
+ * @return StreamConnectionFactory to use
+ */
+ public StreamConnectionFactory getConnectionFactory()
+ {
+ return new DefaultConnectionFactory();
+ }
+
+ /**
* Validate that {@code keyspace} is an existing keyspace and {@code
* cfName} one of its existing column family.
*/
@@ -258,6 +269,7 @@ public class SSTableLoader implements StreamEventHandler
protected void setPartitioner(IPartitioner partitioner)
{
this.partitioner = partitioner;
+ // the following is still necessary since Range/Token reference partitioner through StorageService.getPartitioner
DatabaseDescriptor.setPartitioner(partitioner);
}
@@ -271,7 +283,7 @@ public class SSTableLoader implements StreamEventHandler
Collection<Range<Token>> ranges = endpointToRanges.get(endpoint);
if (ranges == null)
{
- ranges = new HashSet<Range<Token>>();
+ ranges = new HashSet<>();
endpointToRanges.put(endpoint, ranges);
}
ranges.add(range);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/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 b06a818..8fba41b 100644
--- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
+++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
@@ -18,7 +18,6 @@
package org.apache.cassandra.streaming;
import java.io.IOException;
-import java.net.InetAddress;
import java.net.Socket;
import java.net.SocketException;
import java.nio.ByteBuffer;
@@ -37,8 +36,6 @@ import com.google.common.util.concurrent.SettableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.cassandra.config.DatabaseDescriptor;
-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;
@@ -55,8 +52,6 @@ public class ConnectionHandler
{
private static final Logger logger = LoggerFactory.getLogger(ConnectionHandler.class);
- private static final int MAX_CONNECT_ATTEMPTS = 3;
-
private final StreamSession session;
private IncomingMessageHandler incoming;
@@ -79,12 +74,12 @@ public class ConnectionHandler
public void initiate() throws IOException
{
logger.debug("[Stream #{}] Sending stream init for incoming stream", session.planId());
- Socket incomingSocket = connect(session.peer);
+ Socket incomingSocket = session.createConnection();
incoming.start(incomingSocket, StreamMessage.CURRENT_VERSION);
incoming.sendInitMessage(incomingSocket, true);
logger.debug("[Stream #{}] Sending stream init for outgoing stream", session.planId());
- Socket outgoingSocket = connect(session.peer);
+ Socket outgoingSocket = session.createConnection();
outgoing.start(outgoingSocket, StreamMessage.CURRENT_VERSION);
outgoing.sendInitMessage(outgoingSocket, false);
}
@@ -104,45 +99,6 @@ public class ConnectionHandler
incoming.start(socket, version);
}
- /**
- * Connect to peer and start exchanging message.
- * When connect attempt fails, this retries for maximum of MAX_CONNECT_ATTEMPTS times.
- *
- * @param peer the peer to connect to.
- * @return the created socket.
- *
- * @throws IOException when connection failed.
- */
- private static Socket connect(InetAddress peer) throws IOException
- {
- int attempts = 0;
- while (true)
- {
- try
- {
- Socket socket = OutboundTcpConnectionPool.newSocket(peer);
- socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
- return socket;
- }
- catch (IOException e)
- {
- if (++attempts >= MAX_CONNECT_ATTEMPTS)
- throw e;
-
- long waitms = DatabaseDescriptor.getRpcTimeout() * (long)Math.pow(2, attempts);
- logger.warn("Failed attempt " + attempts + " to connect to " + peer + ". Retrying in " + waitms + " ms. (" + e + ")");
- try
- {
- Thread.sleep(waitms);
- }
- catch (InterruptedException wtf)
- {
- throw new IOException("interrupted", wtf);
- }
- }
- }
- }
-
public ListenableFuture<?> close()
{
logger.debug("[Stream #{}] Closing stream connection handler on {}", session.planId(), session.peer);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java b/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
new file mode 100644
index 0000000..53af4c8
--- /dev/null
+++ b/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.streaming;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.Socket;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.net.OutboundTcpConnectionPool;
+
+public class DefaultConnectionFactory implements StreamConnectionFactory
+{
+ private static final Logger logger = LoggerFactory.getLogger(DefaultConnectionFactory.class);
+
+ private static final int MAX_CONNECT_ATTEMPTS = 3;
+
+ /**
+ * Connect to peer and start exchanging message.
+ * When connect attempt fails, this retries for maximum of MAX_CONNECT_ATTEMPTS times.
+ *
+ * @param peer the peer to connect to.
+ * @return the created socket.
+ *
+ * @throws IOException when connection failed.
+ */
+ public Socket createConnection(InetAddress peer) throws IOException
+ {
+ int attempts = 0;
+ while (true)
+ {
+ try
+ {
+ Socket socket = OutboundTcpConnectionPool.newSocket(peer);
+ socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
+ return socket;
+ }
+ catch (IOException e)
+ {
+ if (++attempts >= MAX_CONNECT_ATTEMPTS)
+ throw e;
+
+ long waitms = DatabaseDescriptor.getRpcTimeout() * (long)Math.pow(2, attempts);
+ logger.warn("Failed attempt " + attempts + " to connect to " + peer + ". Retrying in " + waitms + " ms. (" + e + ")");
+ try
+ {
+ Thread.sleep(waitms);
+ }
+ catch (InterruptedException wtf)
+ {
+ throw new IOException("interrupted", wtf);
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java b/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java
new file mode 100644
index 0000000..dd99611
--- /dev/null
+++ b/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.streaming;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.Socket;
+
+/**
+ * Interface that creates connection used by streaming.
+ */
+public interface StreamConnectionFactory
+{
+ Socket createConnection(InetAddress peer) throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/src/java/org/apache/cassandra/streaming/StreamPlan.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamPlan.java b/src/java/org/apache/cassandra/streaming/StreamPlan.java
index b57e097..e582c79 100644
--- a/src/java/org/apache/cassandra/streaming/StreamPlan.java
+++ b/src/java/org/apache/cassandra/streaming/StreamPlan.java
@@ -38,6 +38,8 @@ public class StreamPlan
// sessions per InetAddress of the other end.
private final Map<InetAddress, StreamSession> sessions = new HashMap<>();
+ private StreamConnectionFactory connectionFactory = new DefaultConnectionFactory();
+
private boolean flushBeforeTransfer = true;
/**
@@ -132,6 +134,18 @@ public class StreamPlan
}
/**
+ * Set custom StreamConnectionFactory to be used for establishing connection
+ *
+ * @param factory StreamConnectionFactory to use
+ * @return self
+ */
+ public StreamPlan connectionFactory(StreamConnectionFactory factory)
+ {
+ this.connectionFactory = factory;
+ return this;
+ }
+
+ /**
* @return true if this plan has no plan to execute
*/
public boolean isEmpty()
@@ -167,7 +181,7 @@ public class StreamPlan
StreamSession session = sessions.get(peer);
if (session == null)
{
- session = new StreamSession(peer);
+ session = new StreamSession(peer, connectionFactory);
sessions.put(peer, session);
}
return session;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/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 dcffaff..add14f7 100644
--- a/src/java/org/apache/cassandra/streaming/StreamResultFuture.java
+++ b/src/java/org/apache/cassandra/streaming/StreamResultFuture.java
@@ -106,7 +106,7 @@ public final class StreamResultFuture extends AbstractFuture<StreamState>
StreamResultFuture future = StreamManager.instance.getReceivingStream(planId);
if (future == null)
{
- final StreamSession session = new StreamSession(from);
+ final StreamSession session = new StreamSession(from, null);
// The main reason we create a StreamResultFuture on the receiving side is for JMX exposure.
future = new StreamResultFuture(planId, description, Collections.singleton(session));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/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 55e30f0..4fcbe36 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -19,6 +19,7 @@ package org.apache.cassandra.streaming;
import java.io.IOException;
import java.net.InetAddress;
+import java.net.Socket;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -128,6 +129,8 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
// data receivers, filled after receiving prepare message
private final Map<UUID, StreamReceiveTask> receivers = new ConcurrentHashMap<>();
private final StreamingMetrics metrics;
+ /* can be null when session is created in remote */
+ private final StreamConnectionFactory factory;
public final ConnectionHandler handler;
@@ -152,10 +155,12 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
* Create new streaming session with the peer.
*
* @param peer Address of streaming peer
+ * @param factory is used for establishing connection
*/
- public StreamSession(InetAddress peer)
+ public StreamSession(InetAddress peer, StreamConnectionFactory factory)
{
this.peer = peer;
+ this.factory = factory;
this.handler = new ConnectionHandler(this);
this.metrics = StreamingMetrics.get(peer);
}
@@ -211,6 +216,12 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
});
}
+ public Socket createConnection() throws IOException
+ {
+ assert factory != null;
+ return factory.createConnection(peer);
+ }
+
/**
* Request data fetch task to this session.
*
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java b/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java
new file mode 100644
index 0000000..399344e
--- /dev/null
+++ b/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.tools;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.nio.channels.SocketChannel;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.cassandra.config.EncryptionOptions;
+import org.apache.cassandra.security.SSLFactory;
+import org.apache.cassandra.streaming.StreamConnectionFactory;
+import org.apache.cassandra.utils.FBUtilities;
+
+public class BulkLoadConnectionFactory implements StreamConnectionFactory
+{
+ private final boolean outboundBindAny;
+ private final int storagePort;
+ private final int secureStoragePort;
+ private final EncryptionOptions.ServerEncryptionOptions encryptionOptions;
+
+ public BulkLoadConnectionFactory(int storagePort, int secureStoragePort, EncryptionOptions.ServerEncryptionOptions encryptionOptions, boolean outboundBindAny)
+ {
+ this.storagePort = storagePort;
+ this.secureStoragePort = secureStoragePort;
+ this.encryptionOptions = encryptionOptions;
+ this.outboundBindAny = outboundBindAny;
+ }
+
+ public Socket createConnection(InetAddress peer) throws IOException
+ {
+ // Connect to secure port for all peers if ServerEncryptionOptions is configured other than 'none'
+ // When 'all', 'dc' and 'rack', server nodes always have SSL port open, and since thin client like sstableloader
+ // does not know which node is in which dc/rack, connecting to SSL port is always the option.
+ if (encryptionOptions != null && encryptionOptions.internode_encryption != EncryptionOptions.ServerEncryptionOptions.InternodeEncryption.none)
+ {
+ if (outboundBindAny)
+ return SSLFactory.getSocket(encryptionOptions, peer, secureStoragePort);
+ else
+ return SSLFactory.getSocket(encryptionOptions, peer, secureStoragePort, FBUtilities.getLocalAddress(), 0);
+ }
+ else
+ {
+ Socket socket = SocketChannel.open(new InetSocketAddress(peer, storagePort)).socket();
+ if (outboundBindAny && !socket.isBound())
+ socket.bind(new InetSocketAddress(FBUtilities.getLocalAddress(), 0));
+ return socket;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/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 37ec635..4077722 100644
--- a/src/java/org/apache/cassandra/tools/BulkLoader.java
+++ b/src/java/org/apache/cassandra/tools/BulkLoader.java
@@ -18,29 +18,27 @@
package org.apache.cassandra.tools;
import java.io.File;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
+import java.net.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import com.google.common.base.Joiner;
import com.google.common.collect.Sets;
-import org.apache.cassandra.config.EncryptionOptions;
+
+import org.apache.cassandra.config.*;
+
import org.apache.commons.cli.*;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
-import org.apache.thrift.transport.TFramedTransport;
-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;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.sstable.SSTableLoader;
import org.apache.cassandra.streaming.*;
import org.apache.cassandra.thrift.*;
@@ -60,7 +58,10 @@ public class BulkLoader
private static final String USER_OPTION = "username";
private static final String PASSWD_OPTION = "password";
private static final String THROTTLE_MBITS = "throttle";
+
private static final String TRANSPORT_FACTORY = "transport-factory";
+
+ /* client encryption options */
private static final String SSL_TRUSTSTORE = "truststore";
private static final String SSL_TRUSTSTORE_PW = "truststore-password";
private static final String SSL_KEYSTORE = "keystore";
@@ -69,12 +70,20 @@ public class BulkLoader
private static final String SSL_ALGORITHM = "ssl-alg";
private static final String SSL_STORE_TYPE = "store-type";
private static final String SSL_CIPHER_SUITES = "ssl-ciphers";
+ private static final String CONFIG_PATH = "conf-path";
public static void main(String args[])
{
LoaderOptions options = LoaderOptions.parseArgs(args);
OutputHandler handler = new OutputHandler.SystemOutput(options.verbose, options.debug);
- SSTableLoader loader = new SSTableLoader(options.directory, new ExternalClient(options.hosts, options.rpcPort, options.user, options.passwd, options.transportFactory), handler);
+ SSTableLoader loader = new SSTableLoader(options.directory, new ExternalClient(options.hosts,
+ options.rpcPort,
+ options.user,
+ options.passwd,
+ options.transportFactory,
+ options.storagePort,
+ options.sslStoragePort,
+ options.serverEncOptions), handler);
DatabaseDescriptor.setStreamThroughputOutboundMegabitsPerSec(options.throttle);
StreamResultFuture future = null;
try
@@ -207,8 +216,18 @@ public class BulkLoader
private final String user;
private final String passwd;
private final ITransportFactory transportFactory;
-
- public ExternalClient(Set<InetAddress> hosts, int port, String user, String passwd, ITransportFactory transportFactory)
+ private final int storagePort;
+ private final int sslStoragePort;
+ private final EncryptionOptions.ServerEncryptionOptions serverEncOptions;
+
+ public ExternalClient(Set<InetAddress> hosts,
+ int port,
+ String user,
+ String passwd,
+ ITransportFactory transportFactory,
+ int storagePort,
+ int sslStoragePort,
+ EncryptionOptions.ServerEncryptionOptions serverEncryptionOptions)
{
super();
this.hosts = hosts;
@@ -216,8 +235,12 @@ public class BulkLoader
this.user = user;
this.passwd = passwd;
this.transportFactory = transportFactory;
+ this.storagePort = storagePort;
+ this.sslStoragePort = sslStoragePort;
+ this.serverEncOptions = serverEncryptionOptions;
}
+ @Override
public void init(String keyspace)
{
Iterator<InetAddress> hostiter = hosts.iterator();
@@ -234,7 +257,7 @@ public class BulkLoader
for (TokenRange tr : client.describe_ring(keyspace))
{
- Range<Token> range = new Range<>(tkFactory.fromString(tr.start_token), tkFactory.fromString(tr.end_token));
+ Range<Token> range = new Range<>(tkFactory.fromString(tr.start_token), tkFactory.fromString(tr.end_token), getPartitioner());
for (String ep : tr.endpoints)
{
addRangeForEndpoint(range, InetAddress.getByName(ep));
@@ -261,6 +284,13 @@ public class BulkLoader
}
}
+ @Override
+ public StreamConnectionFactory getConnectionFactory()
+ {
+ return new BulkLoadConnectionFactory(storagePort, sslStoragePort, serverEncOptions, false);
+ }
+
+ @Override
public CFMetaData getCFMetaData(String keyspace, String cfName)
{
return knownCfs.get(cfName);
@@ -273,7 +303,7 @@ public class BulkLoader
Cassandra.Client client = new Cassandra.Client(protocol);
if (user != null && passwd != null)
{
- Map<String, String> credentials = new HashMap<String, String>();
+ Map<String, String> credentials = new HashMap<>();
credentials.put(IAuthenticator.USERNAME_KEY, user);
credentials.put(IAuthenticator.PASSWORD_KEY, passwd);
AuthenticationRequest authenticationRequest = new AuthenticationRequest(credentials);
@@ -294,11 +324,14 @@ public class BulkLoader
public String user;
public String passwd;
public int throttle = 0;
+ public int storagePort;
+ public int sslStoragePort;
public ITransportFactory transportFactory = new TFramedTransportFactory();
public EncryptionOptions encOptions = new EncryptionOptions.ClientEncryptionOptions();
+ public EncryptionOptions.ServerEncryptionOptions serverEncOptions = new EncryptionOptions.ServerEncryptionOptions();
- public final Set<InetAddress> hosts = new HashSet<InetAddress>();
- public final Set<InetAddress> ignores = new HashSet<InetAddress>();
+ public final Set<InetAddress> hosts = new HashSet<>();
+ public final Set<InetAddress> ignores = new HashSet<>();
LoaderOptions(File directory)
{
@@ -349,9 +382,6 @@ public class BulkLoader
opts.verbose = cmd.hasOption(VERBOSE_OPTION);
opts.noProgress = cmd.hasOption(NOPROGRESS_OPTION);
- if (cmd.hasOption(THROTTLE_MBITS))
- opts.throttle = Integer.parseInt(cmd.getOptionValue(THROTTLE_MBITS));
-
if (cmd.hasOption(RPC_PORT_OPTION))
opts.rpcPort = Integer.parseInt(cmd.getOptionValue(RPC_PORT_OPTION));
@@ -400,44 +430,71 @@ public class BulkLoader
}
}
- if(cmd.hasOption(SSL_TRUSTSTORE))
+ // try to load config file first, so that values can be rewritten with other option values.
+ // otherwise use default config.
+ Config config;
+ if (cmd.hasOption(CONFIG_PATH))
+ {
+ File configFile = new File(cmd.getOptionValue(CONFIG_PATH));
+ if (!configFile.exists())
+ {
+ errorMsg("Config file not found", options);
+ }
+ config = new YamlConfigurationLoader().loadConfig(configFile.toURI().toURL());
+ }
+ else
+ {
+ config = new Config();
+ }
+ opts.storagePort = config.storage_port;
+ opts.sslStoragePort = config.ssl_storage_port;
+ opts.throttle = config.stream_throughput_outbound_megabits_per_sec;
+ opts.encOptions = config.client_encryption_options;
+ opts.serverEncOptions = config.server_encryption_options;
+
+ if (cmd.hasOption(THROTTLE_MBITS))
+ {
+ opts.throttle = Integer.parseInt(cmd.getOptionValue(THROTTLE_MBITS));
+ }
+
+ if (cmd.hasOption(SSL_TRUSTSTORE))
{
opts.encOptions.truststore = cmd.getOptionValue(SSL_TRUSTSTORE);
}
- if(cmd.hasOption(SSL_TRUSTSTORE_PW))
+ if (cmd.hasOption(SSL_TRUSTSTORE_PW))
{
opts.encOptions.truststore_password = cmd.getOptionValue(SSL_TRUSTSTORE_PW);
}
- if(cmd.hasOption(SSL_KEYSTORE))
+ if (cmd.hasOption(SSL_KEYSTORE))
{
opts.encOptions.keystore = cmd.getOptionValue(SSL_KEYSTORE);
// if a keystore was provided, lets assume we'll need to use it
opts.encOptions.require_client_auth = true;
}
- if(cmd.hasOption(SSL_KEYSTORE_PW))
+ if (cmd.hasOption(SSL_KEYSTORE_PW))
{
opts.encOptions.keystore_password = cmd.getOptionValue(SSL_KEYSTORE_PW);
}
- if(cmd.hasOption(SSL_PROTOCOL))
+ if (cmd.hasOption(SSL_PROTOCOL))
{
opts.encOptions.protocol = cmd.getOptionValue(SSL_PROTOCOL);
}
- if(cmd.hasOption(SSL_ALGORITHM))
+ if (cmd.hasOption(SSL_ALGORITHM))
{
opts.encOptions.algorithm = cmd.getOptionValue(SSL_ALGORITHM);
}
- if(cmd.hasOption(SSL_STORE_TYPE))
+ if (cmd.hasOption(SSL_STORE_TYPE))
{
opts.encOptions.store_type = cmd.getOptionValue(SSL_STORE_TYPE);
}
- if(cmd.hasOption(SSL_CIPHER_SUITES))
+ if (cmd.hasOption(SSL_CIPHER_SUITES))
{
opts.encOptions.cipher_suites = cmd.getOptionValue(SSL_CIPHER_SUITES).split(",");
}
@@ -451,7 +508,7 @@ public class BulkLoader
return opts;
}
- catch (ParseException e)
+ catch (ParseException | ConfigurationException | MalformedURLException e)
{
errorMsg(e.getMessage(), options);
return null;
@@ -508,6 +565,7 @@ public class BulkLoader
printUsage(options);
System.exit(1);
}
+
private static CmdLineOptions getCmdLineOptions()
{
CmdLineOptions options = new CmdLineOptions();
@@ -516,37 +574,38 @@ public class BulkLoader
options.addOption("h", HELP_OPTION, "display this help message");
options.addOption(null, NOPROGRESS_OPTION, "don't display progress");
options.addOption("i", IGNORE_NODES_OPTION, "NODES", "don't stream to this (comma separated) list of nodes");
- options.addOption("d", INITIAL_HOST_ADDRESS_OPTION, "initial hosts", "try to connect to these hosts (comma separated) initially for ring information");
+ options.addOption("d", INITIAL_HOST_ADDRESS_OPTION, "initial hosts", "Required. try to connect to these hosts (comma separated) initially for ring information");
options.addOption("p", RPC_PORT_OPTION, "rpc port", "port used for rpc (default 9160)");
options.addOption("t", THROTTLE_MBITS, "throttle", "throttle speed in Mbits (default unlimited)");
options.addOption("u", USER_OPTION, "username", "username for cassandra authentication");
options.addOption("pw", PASSWD_OPTION, "password", "password for cassandra authentication");
options.addOption("tf", TRANSPORT_FACTORY, "transport factory", "Fully-qualified ITransportFactory class name for creating a connection to cassandra");
// ssl connection-related options
- options.addOption("ts", SSL_TRUSTSTORE, "TRUSTSTORE", "SSL: full path to truststore");
- options.addOption("tspw", SSL_TRUSTSTORE_PW, "TRUSTSTORE-PASSWORD", "SSL: password of the truststore");
- options.addOption("ks", SSL_KEYSTORE, "KEYSTORE", "SSL: full path to keystore");
- options.addOption("kspw", SSL_KEYSTORE_PW, "KEYSTORE-PASSWORD", "SSL: password of the keystore");
- options.addOption("prtcl", SSL_PROTOCOL, "PROTOCOL", "SSL: connections protocol to use (default: TLS)");
- options.addOption("alg", SSL_ALGORITHM, "ALGORITHM", "SSL: algorithm (default: SunX509)");
- options.addOption("st", SSL_STORE_TYPE, "STORE-TYPE", "SSL: type of store");
- options.addOption("ciphers", SSL_CIPHER_SUITES, "CIPHER-SUITES", "SSL: comma-separated list of encryption suites to use");
+ options.addOption("ts", SSL_TRUSTSTORE, "TRUSTSTORE", "Client SSL: full path to truststore");
+ options.addOption("tspw", SSL_TRUSTSTORE_PW, "TRUSTSTORE-PASSWORD", "Client SSL: password of the truststore");
+ options.addOption("ks", SSL_KEYSTORE, "KEYSTORE", "Client SSL: full path to keystore");
+ options.addOption("kspw", SSL_KEYSTORE_PW, "KEYSTORE-PASSWORD", "Client SSL: password of the keystore");
+ options.addOption("prtcl", SSL_PROTOCOL, "PROTOCOL", "Client SSL: connections protocol to use (default: TLS)");
+ options.addOption("alg", SSL_ALGORITHM, "ALGORITHM", "Client SSL: algorithm (default: SunX509)");
+ options.addOption("st", SSL_STORE_TYPE, "STORE-TYPE", "Client SSL: type of store");
+ options.addOption("ciphers", SSL_CIPHER_SUITES, "CIPHER-SUITES", "Client SSL: comma-separated list of encryption suites to use");
+ options.addOption("f", CONFIG_PATH, "path to config file", "cassandra.yaml file path for streaming throughput and client/server SSL.");
return options;
}
public static void printUsage(Options options)
{
String usage = String.format("%s [options] <dir_path>", TOOL_NAME);
- StringBuilder header = new StringBuilder();
- header.append("--\n");
- header.append("Bulk load the sstables found in the directory <dir_path> to the configured cluster." );
- header.append("The parent directory of <dir_path> is used as the keyspace name. ");
- header.append("So for instance, to load an sstable named Standard1-g-1-Data.db into keyspace Keyspace1, ");
- header.append("you will need to have the files Standard1-g-1-Data.db and Standard1-g-1-Index.db in a ");
- header.append("directory Keyspace1/Standard1/ in the directory and call: sstableloader Keyspace1/Standard1");
- header.append("\n--\n");
- header.append("Options are:");
- new HelpFormatter().printHelp(usage, header.toString(), options, "");
+ String header = System.lineSeparator() +
+ "Bulk load the sstables found in the directory <dir_path> to the configured cluster." +
+ "The parent directories of <dir_path> are used as the target keyspace/table name. " +
+ "So for instance, to load an sstable named Standard1-g-1-Data.db into Keyspace1/Standard1, " +
+ "you will need to have the files Standard1-g-1-Data.db and Standard1-g-1-Index.db into a directory /path/to/Keyspace1/Standard1/.";
+ String footer = System.lineSeparator() +
+ "You can provide cassandra.yaml file with -f command line option to set up streaming throughput, client and server encryption options. " +
+ "Only stream_throughput_outbound_megabits_per_sec, server_encryption_options and client_encryption_options are read from yaml. " +
+ "You can override options read from cassandra.yaml with corresponding command line options.";
+ new HelpFormatter().printHelp(usage, header, options, footer);
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/563cea14/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java b/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
index 9b02817..ce0f9d0 100644
--- a/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
+++ b/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
@@ -43,7 +43,7 @@ public class StreamTransferTaskTest extends SchemaLoader
String ks = "Keyspace1";
String cf = "Standard1";
- StreamSession session = new StreamSession(FBUtilities.getBroadcastAddress());
+ StreamSession session = new StreamSession(FBUtilities.getBroadcastAddress(), null);
ColumnFamilyStore cfs = Keyspace.open(ks).getColumnFamilyStore(cf);
// create two sstables
[4/6] git commit: Merge branch 'cassandra-2.0' into cassandra-2.1
Posted by yu...@apache.org.
Merge branch 'cassandra-2.0' into cassandra-2.1
Conflicts:
src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
src/java/org/apache/cassandra/streaming/ConnectionHandler.java
src/java/org/apache/cassandra/streaming/StreamPlan.java
src/java/org/apache/cassandra/streaming/StreamResultFuture.java
src/java/org/apache/cassandra/streaming/StreamSession.java
src/java/org/apache/cassandra/tools/BulkLoader.java
test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/de5bb585
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/de5bb585
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/de5bb585
Branch: refs/heads/trunk
Commit: de5bb5854de5c02e13c3311e0e4f70dc44f5bb83
Parents: 54fbb0a 563cea1
Author: Yuki Morishita <yu...@apache.org>
Authored: Fri Aug 15 13:24:31 2014 -0500
Committer: Yuki Morishita <yu...@apache.org>
Committed: Fri Aug 15 13:24:31 2014 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../config/YamlConfigurationLoader.java | 7 +-
.../cassandra/io/sstable/SSTableLoader.java | 22 ++-
.../cassandra/streaming/ConnectionHandler.java | 49 +------
.../streaming/DefaultConnectionFactory.java | 75 ++++++++++
.../streaming/StreamConnectionFactory.java | 30 ++++
.../cassandra/streaming/StreamCoordinator.java | 13 +-
.../apache/cassandra/streaming/StreamPlan.java | 17 ++-
.../cassandra/streaming/StreamResultFuture.java | 2 +-
.../cassandra/streaming/StreamSession.java | 13 +-
.../tools/BulkLoadConnectionFactory.java | 68 +++++++++
.../org/apache/cassandra/tools/BulkLoader.java | 139 +++++++++++++------
.../streaming/StreamTransferTaskTest.java | 2 +-
13 files changed, 337 insertions(+), 101 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 191b187,e335484..ced91fd
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -171,7 -47,9 +171,8 @@@ Merged from 2.0
* Backport CASSANDRA-6747 (CASSANDRA-7560)
* Track max/min timestamps for range tombstones (CASSANDRA-7647)
* Fix NPE when listing saved caches dir (CASSANDRA-7632)
+ * Fix sstableloader unable to connect encrypted node (CASSANDRA-7585)
Merged from 1.2:
- * Remove duplicates from StorageService.getJoiningNodes (CASSANDRA-7478)
* Clone token map outside of hot gossip loops (CASSANDRA-7758)
* Add stop method to EmbeddedCassandraService (CASSANDRA-7595)
* Support connecting to ipv6 jmx with nodetool (CASSANDRA-7669)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
index 78621f2,b520d07..0b62ff4
--- a/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
+++ b/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
@@@ -81,14 -69,18 +81,19 @@@ public class YamlConfigurationLoader im
public Config loadConfig() throws ConfigurationException
{
+ return loadConfig(getStorageConfigURL());
+ }
+
+ public Config loadConfig(URL url) throws ConfigurationException
+ {
+ InputStream input = null;
try
{
- URL url = getStorageConfigURL();
logger.info("Loading settings from {}", url);
- try
+ byte[] configBytes;
+ try (InputStream is = url.openStream())
{
- input = url.openStream();
+ configBytes = ByteStreams.toByteArray(is);
}
catch (IOException e)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
index bbb1277,85dc0e4..3d7eea7
--- a/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
@@@ -157,7 -149,7 +157,7 @@@ public class SSTableLoader implements S
client.init(keyspace);
outputHandler.output("Established connection to initial hosts");
- StreamPlan plan = new StreamPlan("Bulk Load", 0, connectionsPerHost);
- StreamPlan plan = new StreamPlan("Bulk Load").connectionFactory(client.getConnectionFactory());
++ StreamPlan plan = new StreamPlan("Bulk Load", 0, connectionsPerHost).connectionFactory(client.getConnectionFactory());
Map<InetAddress, Collection<Range<Token>>> endpointToRanges = client.getEndpointToRangesMap();
openSSTables(endpointToRanges);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/ConnectionHandler.java
index 5716ae9,8fba41b..a20eb34
--- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
+++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
@@@ -37,9 -36,6 +36,7 @@@ import com.google.common.util.concurren
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
- import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.io.util.DataOutputStreamAndChannel;
- 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;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
index 0000000,53af4c8..f711490
mode 000000,100644..100644
--- a/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
+++ b/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
@@@ -1,0 -1,74 +1,75 @@@
+ /*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ package org.apache.cassandra.streaming;
+
+ import java.io.IOException;
+ import java.net.InetAddress;
+ import java.net.Socket;
+
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+
+ import org.apache.cassandra.config.DatabaseDescriptor;
+ import org.apache.cassandra.net.OutboundTcpConnectionPool;
+
+ public class DefaultConnectionFactory implements StreamConnectionFactory
+ {
+ private static final Logger logger = LoggerFactory.getLogger(DefaultConnectionFactory.class);
+
+ private static final int MAX_CONNECT_ATTEMPTS = 3;
+
+ /**
+ * Connect to peer and start exchanging message.
+ * When connect attempt fails, this retries for maximum of MAX_CONNECT_ATTEMPTS times.
+ *
+ * @param peer the peer to connect to.
+ * @return the created socket.
+ *
+ * @throws IOException when connection failed.
+ */
+ public Socket createConnection(InetAddress peer) throws IOException
+ {
+ int attempts = 0;
+ while (true)
+ {
+ try
+ {
+ Socket socket = OutboundTcpConnectionPool.newSocket(peer);
+ socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
++ socket.setKeepAlive(true);
+ return socket;
+ }
+ catch (IOException e)
+ {
+ if (++attempts >= MAX_CONNECT_ATTEMPTS)
+ throw e;
+
+ long waitms = DatabaseDescriptor.getRpcTimeout() * (long)Math.pow(2, attempts);
+ logger.warn("Failed attempt " + attempts + " to connect to " + peer + ". Retrying in " + waitms + " ms. (" + e + ")");
+ try
+ {
+ Thread.sleep(waitms);
+ }
+ catch (InterruptedException wtf)
+ {
+ throw new IOException("interrupted", wtf);
+ }
+ }
+ }
+ }
+ }
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/streaming/StreamCoordinator.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/StreamCoordinator.java
index 14dad29,0000000..c8c0f14
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/streaming/StreamCoordinator.java
+++ b/src/java/org/apache/cassandra/streaming/StreamCoordinator.java
@@@ -1,278 -1,0 +1,285 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.streaming;
+
+import java.net.InetAddress;
+import java.util.*;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
+import org.apache.cassandra.utils.FBUtilities;
+
+/**
+ * {@link StreamCoordinator} is a helper class that abstracts away maintaining multiple
+ * StreamSession and ProgressInfo instances per peer.
+ *
+ * This class coordinates multiple SessionStreams per peer in both the outgoing StreamPlan context and on the
+ * inbound StreamResultFuture context.
+ */
+public class StreamCoordinator
+{
+ private static final Logger logger = LoggerFactory.getLogger(StreamCoordinator.class);
+
+ // Executor strictly for establishing the initial connections. Once we're connected to the other end the rest of the
+ // streaming is handled directly by the ConnectionHandler's incoming and outgoing threads.
+ private static final DebuggableThreadPoolExecutor streamExecutor = DebuggableThreadPoolExecutor.createWithFixedPoolSize("StreamConnectionEstablisher",
+ FBUtilities.getAvailableProcessors());
+
+ private Map<InetAddress, HostStreamingData> peerSessions = new HashMap<>();
+ private final int connectionsPerHost;
++ private StreamConnectionFactory factory;
+
- public StreamCoordinator(int connectionsPerHost)
++ public StreamCoordinator(int connectionsPerHost, StreamConnectionFactory factory)
+ {
+ this.connectionsPerHost = connectionsPerHost;
++ this.factory = factory;
++ }
++
++ public void setConnectionFactory(StreamConnectionFactory factory)
++ {
++ this.factory = factory;
+ }
+
+ /**
+ * @return true if any stream session is active
+ */
+ public synchronized boolean hasActiveSessions()
+ {
+ for (HostStreamingData data : peerSessions.values())
+ {
+ if (data.hasActiveSessions())
+ return true;
+ }
+ return false;
+ }
+
+ public synchronized Collection<StreamSession> getAllStreamSessions()
+ {
+ Collection<StreamSession> results = new ArrayList<>();
+ for (HostStreamingData data : peerSessions.values())
+ {
+ results.addAll(data.getAllStreamSessions());
+ }
+ return results;
+ }
+
+ public boolean isReceiving()
+ {
+ return connectionsPerHost == 0;
+ }
+
+ public void connectAllStreamSessions()
+ {
+ for (HostStreamingData data : peerSessions.values())
+ data.connectAllStreamSessions();
+ }
+
+ public synchronized Set<InetAddress> getPeers()
+ {
+ return new HashSet<>(peerSessions.keySet());
+ }
+
+ public synchronized StreamSession getOrCreateNextSession(InetAddress peer)
+ {
+ return getOrCreateHostData(peer).getOrCreateNextSession(peer);
+ }
+
+ public synchronized StreamSession getOrCreateSessionById(InetAddress peer, int id)
+ {
+ return getOrCreateHostData(peer).getOrCreateSessionById(peer, id);
+ }
+
+ public synchronized void updateProgress(ProgressInfo info)
+ {
+ getHostData(info.peer).updateProgress(info);
+ }
+
+ public synchronized void addSessionInfo(SessionInfo session)
+ {
+ HostStreamingData data = getOrCreateHostData(session.peer);
+ data.addSessionInfo(session);
+ }
+
+ public synchronized Set<SessionInfo> getAllSessionInfo()
+ {
+ Set<SessionInfo> result = new HashSet<>();
+ for (HostStreamingData data : peerSessions.values())
+ {
+ result.addAll(data.getAllSessionInfo());
+ }
+ return result;
+ }
+
+ public synchronized void transferFiles(InetAddress to, Collection<StreamSession.SSTableStreamingSections> sstableDetails)
+ {
+ HostStreamingData sessionList = getOrCreateHostData(to);
+
+ if (connectionsPerHost > 1)
+ {
+ List<List<StreamSession.SSTableStreamingSections>> buckets = sliceSSTableDetails(sstableDetails);
+
+ for (List<StreamSession.SSTableStreamingSections> subList : buckets)
+ {
+ StreamSession session = sessionList.getOrCreateNextSession(to);
+ session.addTransferFiles(subList);
+ }
+ }
+ else
+ {
+ StreamSession session = sessionList.getOrCreateNextSession(to);
+ session.addTransferFiles(sstableDetails);
+ }
+ }
+
+ private List<List<StreamSession.SSTableStreamingSections>> sliceSSTableDetails(Collection<StreamSession.SSTableStreamingSections> sstableDetails)
+ {
+ // There's no point in divvying things up into more buckets than we have sstableDetails
+ int targetSlices = Math.min(sstableDetails.size(), connectionsPerHost);
+ int step = Math.round((float) sstableDetails.size() / (float) targetSlices);
+ int index = 0;
+
+ List<List<StreamSession.SSTableStreamingSections>> result = new ArrayList<>();
+ List<StreamSession.SSTableStreamingSections> slice = null;
+ for (StreamSession.SSTableStreamingSections streamSession : sstableDetails)
+ {
+ if (index % step == 0)
+ {
+ slice = new ArrayList<>();
+ result.add(slice);
+ }
+ slice.add(streamSession);
+ ++index;
+ }
+
+ return result;
+ }
+
+ private HostStreamingData getHostData(InetAddress peer)
+ {
+ HostStreamingData data = peerSessions.get(peer);
+ if (data == null)
+ throw new IllegalArgumentException("Unknown peer requested: " + peer.toString());
+ return data;
+ }
+
+ private HostStreamingData getOrCreateHostData(InetAddress peer)
+ {
+ HostStreamingData data = peerSessions.get(peer);
+ if (data == null)
+ {
+ data = new HostStreamingData();
+ peerSessions.put(peer, data);
+ }
+ return data;
+ }
+
+ private static class StreamSessionConnector implements Runnable
+ {
+ private final StreamSession session;
+ public StreamSessionConnector(StreamSession session)
+ {
+ this.session = session;
+ }
+
+ @Override
+ public void run()
+ {
+ session.start();
+ logger.info("[Stream #{}, ID#{}] Beginning stream session with {}", session.planId(), session.sessionIndex(), session.peer);
+ }
+ }
+
+ private class HostStreamingData
+ {
+ private Map<Integer, StreamSession> streamSessions = new HashMap<>();
+ private Map<Integer, SessionInfo> sessionInfos = new HashMap<>();
+
+ private int lastReturned = -1;
+
+ public boolean hasActiveSessions()
+ {
+ for (StreamSession session : streamSessions.values())
+ {
+ StreamSession.State state = session.state();
+ if (state != StreamSession.State.COMPLETE && state != StreamSession.State.FAILED)
+ return true;
+ }
+ return false;
+ }
+
+ public StreamSession getOrCreateNextSession(InetAddress peer)
+ {
+ // create
+ if (streamSessions.size() < connectionsPerHost)
+ {
- StreamSession session = new StreamSession(peer, streamSessions.size());
++ StreamSession session = new StreamSession(peer, factory, streamSessions.size());
+ streamSessions.put(++lastReturned, session);
+ return session;
+ }
+ // get
+ else
+ {
+ if (lastReturned >= streamSessions.size() - 1)
+ lastReturned = 0;
+
+ return streamSessions.get(lastReturned++);
+ }
+ }
+
+ public void connectAllStreamSessions()
+ {
+ for (StreamSession session : streamSessions.values())
+ {
+ streamExecutor.execute(new StreamSessionConnector(session));
+ }
+ }
+
+ public Collection<StreamSession> getAllStreamSessions()
+ {
+ return Collections.unmodifiableCollection(streamSessions.values());
+ }
+
+ public StreamSession getOrCreateSessionById(InetAddress peer, int id)
+ {
+ StreamSession session = streamSessions.get(id);
+ if (session == null)
+ {
- session = new StreamSession(peer, id);
++ session = new StreamSession(peer, factory, id);
+ streamSessions.put(id, session);
+ }
+ return session;
+ }
+
+ public void updateProgress(ProgressInfo info)
+ {
+ sessionInfos.get(info.sessionIndex).updateProgress(info);
+ }
+
+ public void addSessionInfo(SessionInfo info)
+ {
+ sessionInfos.put(info.sessionIndex, info);
+ }
+
+ public Collection<SessionInfo> getAllSessionInfo()
+ {
+ return sessionInfos.values();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/streaming/StreamPlan.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/StreamPlan.java
index e775c90,e582c79..c423a33
--- a/src/java/org/apache/cassandra/streaming/StreamPlan.java
+++ b/src/java/org/apache/cassandra/streaming/StreamPlan.java
@@@ -35,9 -34,12 +35,11 @@@ public class StreamPla
private final UUID planId = UUIDGen.getTimeUUID();
private final String description;
private final List<StreamEventHandler> handlers = new ArrayList<>();
-
- // sessions per InetAddress of the other end.
- private final Map<InetAddress, StreamSession> sessions = new HashMap<>();
+ private final long repairedAt;
+ private final StreamCoordinator coordinator;
+ private StreamConnectionFactory connectionFactory = new DefaultConnectionFactory();
+
private boolean flushBeforeTransfer = true;
/**
@@@ -47,14 -49,7 +49,14 @@@
*/
public StreamPlan(String description)
{
+ this(description, ActiveRepairService.UNREPAIRED_SSTABLE, 1);
+ }
+
+ public StreamPlan(String description, long repairedAt, int connectionsPerHost)
+ {
this.description = description;
+ this.repairedAt = repairedAt;
- this.coordinator = new StreamCoordinator(connectionsPerHost);
++ this.coordinator = new StreamCoordinator(connectionsPerHost, connectionFactory);
}
/**
@@@ -139,6 -134,18 +141,18 @@@
}
/**
+ * Set custom StreamConnectionFactory to be used for establishing connection
+ *
+ * @param factory StreamConnectionFactory to use
+ * @return self
+ */
+ public StreamPlan connectionFactory(StreamConnectionFactory factory)
+ {
- this.connectionFactory = factory;
++ this.coordinator.setConnectionFactory(factory);
+ return this;
+ }
+
+ /**
* @return true if this plan has no plan to execute
*/
public boolean isEmpty()
@@@ -168,4 -175,16 +182,5 @@@
this.flushBeforeTransfer = flushBeforeTransfer;
return this;
}
+
- private StreamSession getOrCreateSession(InetAddress peer)
- {
- StreamSession session = sessions.get(peer);
- if (session == null)
- {
- session = new StreamSession(peer, connectionFactory);
- sessions.put(peer, session);
- }
- return session;
- }
-
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/streaming/StreamResultFuture.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/StreamResultFuture.java
index 6a143d1,add14f7..f28a937
--- a/src/java/org/apache/cassandra/streaming/StreamResultFuture.java
+++ b/src/java/org/apache/cassandra/streaming/StreamResultFuture.java
@@@ -69,14 -75,9 +69,14 @@@ public final class StreamResultFuture e
set(getCurrentState());
}
- static StreamResultFuture init(UUID planId, String description, Collection<StreamSession> sessions, Collection<StreamEventHandler> listeners)
+ private StreamResultFuture(UUID planId, String description)
+ {
- this(planId, description, new StreamCoordinator(0));
++ this(planId, description, new StreamCoordinator(0, new DefaultConnectionFactory()));
+ }
+
+ static StreamResultFuture init(UUID planId, String description, Collection<StreamEventHandler> listeners, StreamCoordinator coordinator)
{
- StreamResultFuture future = createAndRegister(planId, description, sessions);
+ StreamResultFuture future = createAndRegister(planId, description, coordinator);
if (listeners != null)
{
for (StreamEventHandler listener : listeners)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/StreamSession.java
index 1afc07e,4fcbe36..2efa00d
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@@ -148,11 -155,12 +151,13 @@@ public class StreamSession implements I
* Create new streaming session with the peer.
*
* @param peer Address of streaming peer
+ * @param factory is used for establishing connection
*/
- public StreamSession(InetAddress peer, int index)
- public StreamSession(InetAddress peer, StreamConnectionFactory factory)
++ public StreamSession(InetAddress peer, StreamConnectionFactory factory, int index)
{
this.peer = peer;
+ this.index = index;
+ this.factory = factory;
this.handler = new ConnectionHandler(this);
this.metrics = StreamingMetrics.get(peer);
}
@@@ -192,18 -199,29 +197,24 @@@
return;
}
- streamExecutor.execute(new Runnable()
+ try
{
- public void run()
- {
- try
- {
- handler.initiate();
- onInitializationComplete();
- }
- catch (IOException e)
- {
- onError(e);
- }
- }
- });
+ logger.info("[Stream #{}, ID#{}] Beginning stream session with {}", planId(), sessionIndex(), peer);
+ handler.initiate();
+ onInitializationComplete();
+ }
+ catch (Exception e)
+ {
+ onError(e);
+ }
}
+ public Socket createConnection() throws IOException
+ {
+ assert factory != null;
+ return factory.createConnection(peer);
+ }
+
/**
* Request data fetch task to this session.
*
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/tools/BulkLoader.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/tools/BulkLoader.java
index 74b803a,4077722..50e340b
--- a/src/java/org/apache/cassandra/tools/BulkLoader.java
+++ b/src/java/org/apache/cassandra/tools/BulkLoader.java
@@@ -18,27 -18,27 +18,27 @@@
package org.apache.cassandra.tools;
import java.io.File;
- import java.net.InetAddress;
- import java.net.UnknownHostException;
+ import java.net.*;
import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.TimeUnit;
import com.google.common.base.Joiner;
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
+ import com.google.common.collect.Sets;
+
-import org.apache.cassandra.config.*;
-
import org.apache.commons.cli.*;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
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.config.EncryptionOptions;
++import org.apache.cassandra.config.*;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
+ import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.sstable.SSTableLoader;
import org.apache.cassandra.streaming.*;
import org.apache.cassandra.thrift.*;
@@@ -66,26 -70,22 +69,30 @@@ public class BulkLoade
private static final String SSL_ALGORITHM = "ssl-alg";
private static final String SSL_STORE_TYPE = "store-type";
private static final String SSL_CIPHER_SUITES = "ssl-ciphers";
+ private static final String CONNECTIONS_PER_HOST = "connections-per-host";
+ private static final String CONFIG_PATH = "conf-path";
public static void main(String args[])
{
LoaderOptions options = LoaderOptions.parseArgs(args);
- OutputHandler handler = new OutputHandler.SystemOutput(options.verbose, true);
+ OutputHandler handler = new OutputHandler.SystemOutput(options.verbose, options.debug);
- SSTableLoader loader = new SSTableLoader(options.directory, new ExternalClient(options.hosts,
- options.rpcPort,
- options.user,
- options.passwd,
- options.transportFactory,
- options.storagePort,
- options.sslStoragePort,
- options.serverEncOptions), handler);
+ SSTableLoader loader = new SSTableLoader(
+ options.directory,
+ new ExternalClient(
+ options.hosts,
+ options.rpcPort,
+ options.user,
+ options.passwd,
- options.transportFactory),
++ options.transportFactory,
++ options.storagePort,
++ options.sslStoragePort,
++ options.serverEncOptions),
+ handler,
+ options.connectionsPerHost);
DatabaseDescriptor.setStreamThroughputOutboundMegabitsPerSec(options.throttle);
StreamResultFuture future = null;
+
+ ProgressIndicator indicator = new ProgressIndicator();
try
{
if (options.noProgress)
@@@ -349,9 -324,11 +377,12 @@@
public String user;
public String passwd;
public int throttle = 0;
+ public int storagePort;
+ public int sslStoragePort;
public ITransportFactory transportFactory = new TFramedTransportFactory();
public EncryptionOptions encOptions = new EncryptionOptions.ClientEncryptionOptions();
+ public int connectionsPerHost = 1;
+ public EncryptionOptions.ServerEncryptionOptions serverEncOptions = new EncryptionOptions.ServerEncryptionOptions();
public final Set<InetAddress> hosts = new HashSet<>();
public final Set<InetAddress> ignores = new HashSet<>();
@@@ -455,10 -430,34 +483,37 @@@
}
}
+ if (cmd.hasOption(CONNECTIONS_PER_HOST))
+ opts.connectionsPerHost = Integer.parseInt(cmd.getOptionValue(CONNECTIONS_PER_HOST));
+
- if(cmd.hasOption(SSL_TRUSTSTORE))
+ // try to load config file first, so that values can be rewritten with other option values.
+ // otherwise use default config.
+ Config config;
+ if (cmd.hasOption(CONFIG_PATH))
+ {
+ File configFile = new File(cmd.getOptionValue(CONFIG_PATH));
+ if (!configFile.exists())
+ {
+ errorMsg("Config file not found", options);
+ }
+ config = new YamlConfigurationLoader().loadConfig(configFile.toURI().toURL());
+ }
+ else
+ {
+ config = new Config();
+ }
+ opts.storagePort = config.storage_port;
+ opts.sslStoragePort = config.ssl_storage_port;
+ opts.throttle = config.stream_throughput_outbound_megabits_per_sec;
+ opts.encOptions = config.client_encryption_options;
+ opts.serverEncOptions = config.server_encryption_options;
+
+ if (cmd.hasOption(THROTTLE_MBITS))
+ {
+ opts.throttle = Integer.parseInt(cmd.getOptionValue(THROTTLE_MBITS));
+ }
+
+ if (cmd.hasOption(SSL_TRUSTSTORE))
{
opts.encOptions.truststore = cmd.getOptionValue(SSL_TRUSTSTORE);
}
@@@ -579,16 -580,16 +635,17 @@@
options.addOption("u", USER_OPTION, "username", "username for cassandra authentication");
options.addOption("pw", PASSWD_OPTION, "password", "password for cassandra authentication");
options.addOption("tf", TRANSPORT_FACTORY, "transport factory", "Fully-qualified ITransportFactory class name for creating a connection to cassandra");
+ options.addOption("cph", CONNECTIONS_PER_HOST, "connectionsPerHost", "number of concurrent connections-per-host.");
// ssl connection-related options
- options.addOption("ts", SSL_TRUSTSTORE, "TRUSTSTORE", "SSL: full path to truststore");
- options.addOption("tspw", SSL_TRUSTSTORE_PW, "TRUSTSTORE-PASSWORD", "SSL: password of the truststore");
- options.addOption("ks", SSL_KEYSTORE, "KEYSTORE", "SSL: full path to keystore");
- options.addOption("kspw", SSL_KEYSTORE_PW, "KEYSTORE-PASSWORD", "SSL: password of the keystore");
- options.addOption("prtcl", SSL_PROTOCOL, "PROTOCOL", "SSL: connections protocol to use (default: TLS)");
- options.addOption("alg", SSL_ALGORITHM, "ALGORITHM", "SSL: algorithm (default: SunX509)");
- options.addOption("st", SSL_STORE_TYPE, "STORE-TYPE", "SSL: type of store");
- options.addOption("ciphers", SSL_CIPHER_SUITES, "CIPHER-SUITES", "SSL: comma-separated list of encryption suites to use");
+ options.addOption("ts", SSL_TRUSTSTORE, "TRUSTSTORE", "Client SSL: full path to truststore");
+ options.addOption("tspw", SSL_TRUSTSTORE_PW, "TRUSTSTORE-PASSWORD", "Client SSL: password of the truststore");
+ options.addOption("ks", SSL_KEYSTORE, "KEYSTORE", "Client SSL: full path to keystore");
+ options.addOption("kspw", SSL_KEYSTORE_PW, "KEYSTORE-PASSWORD", "Client SSL: password of the keystore");
+ options.addOption("prtcl", SSL_PROTOCOL, "PROTOCOL", "Client SSL: connections protocol to use (default: TLS)");
+ options.addOption("alg", SSL_ALGORITHM, "ALGORITHM", "Client SSL: algorithm (default: SunX509)");
+ options.addOption("st", SSL_STORE_TYPE, "STORE-TYPE", "Client SSL: type of store");
+ options.addOption("ciphers", SSL_CIPHER_SUITES, "CIPHER-SUITES", "Client SSL: comma-separated list of encryption suites to use");
+ options.addOption("f", CONFIG_PATH, "path to config file", "cassandra.yaml file path for streaming throughput and client/server SSL.");
return options;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
index c6c04b0,ce0f9d0..4043ac8
--- a/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
+++ b/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
@@@ -43,7 -43,7 +43,7 @@@ public class StreamTransferTaskTest ext
String ks = "Keyspace1";
String cf = "Standard1";
- StreamSession session = new StreamSession(FBUtilities.getBroadcastAddress(), 0);
- StreamSession session = new StreamSession(FBUtilities.getBroadcastAddress(), null);
++ StreamSession session = new StreamSession(FBUtilities.getBroadcastAddress(), null, 0);
ColumnFamilyStore cfs = Keyspace.open(ks).getColumnFamilyStore(cf);
// create two sstables
[5/6] git commit: Merge branch 'cassandra-2.0' into cassandra-2.1
Posted by yu...@apache.org.
Merge branch 'cassandra-2.0' into cassandra-2.1
Conflicts:
src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
src/java/org/apache/cassandra/streaming/ConnectionHandler.java
src/java/org/apache/cassandra/streaming/StreamPlan.java
src/java/org/apache/cassandra/streaming/StreamResultFuture.java
src/java/org/apache/cassandra/streaming/StreamSession.java
src/java/org/apache/cassandra/tools/BulkLoader.java
test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/de5bb585
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/de5bb585
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/de5bb585
Branch: refs/heads/cassandra-2.1
Commit: de5bb5854de5c02e13c3311e0e4f70dc44f5bb83
Parents: 54fbb0a 563cea1
Author: Yuki Morishita <yu...@apache.org>
Authored: Fri Aug 15 13:24:31 2014 -0500
Committer: Yuki Morishita <yu...@apache.org>
Committed: Fri Aug 15 13:24:31 2014 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../config/YamlConfigurationLoader.java | 7 +-
.../cassandra/io/sstable/SSTableLoader.java | 22 ++-
.../cassandra/streaming/ConnectionHandler.java | 49 +------
.../streaming/DefaultConnectionFactory.java | 75 ++++++++++
.../streaming/StreamConnectionFactory.java | 30 ++++
.../cassandra/streaming/StreamCoordinator.java | 13 +-
.../apache/cassandra/streaming/StreamPlan.java | 17 ++-
.../cassandra/streaming/StreamResultFuture.java | 2 +-
.../cassandra/streaming/StreamSession.java | 13 +-
.../tools/BulkLoadConnectionFactory.java | 68 +++++++++
.../org/apache/cassandra/tools/BulkLoader.java | 139 +++++++++++++------
.../streaming/StreamTransferTaskTest.java | 2 +-
13 files changed, 337 insertions(+), 101 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 191b187,e335484..ced91fd
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -171,7 -47,9 +171,8 @@@ Merged from 2.0
* Backport CASSANDRA-6747 (CASSANDRA-7560)
* Track max/min timestamps for range tombstones (CASSANDRA-7647)
* Fix NPE when listing saved caches dir (CASSANDRA-7632)
+ * Fix sstableloader unable to connect encrypted node (CASSANDRA-7585)
Merged from 1.2:
- * Remove duplicates from StorageService.getJoiningNodes (CASSANDRA-7478)
* Clone token map outside of hot gossip loops (CASSANDRA-7758)
* Add stop method to EmbeddedCassandraService (CASSANDRA-7595)
* Support connecting to ipv6 jmx with nodetool (CASSANDRA-7669)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
index 78621f2,b520d07..0b62ff4
--- a/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
+++ b/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
@@@ -81,14 -69,18 +81,19 @@@ public class YamlConfigurationLoader im
public Config loadConfig() throws ConfigurationException
{
+ return loadConfig(getStorageConfigURL());
+ }
+
+ public Config loadConfig(URL url) throws ConfigurationException
+ {
+ InputStream input = null;
try
{
- URL url = getStorageConfigURL();
logger.info("Loading settings from {}", url);
- try
+ byte[] configBytes;
+ try (InputStream is = url.openStream())
{
- input = url.openStream();
+ configBytes = ByteStreams.toByteArray(is);
}
catch (IOException e)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
index bbb1277,85dc0e4..3d7eea7
--- a/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
@@@ -157,7 -149,7 +157,7 @@@ public class SSTableLoader implements S
client.init(keyspace);
outputHandler.output("Established connection to initial hosts");
- StreamPlan plan = new StreamPlan("Bulk Load", 0, connectionsPerHost);
- StreamPlan plan = new StreamPlan("Bulk Load").connectionFactory(client.getConnectionFactory());
++ StreamPlan plan = new StreamPlan("Bulk Load", 0, connectionsPerHost).connectionFactory(client.getConnectionFactory());
Map<InetAddress, Collection<Range<Token>>> endpointToRanges = client.getEndpointToRangesMap();
openSSTables(endpointToRanges);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/ConnectionHandler.java
index 5716ae9,8fba41b..a20eb34
--- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
+++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
@@@ -37,9 -36,6 +36,7 @@@ import com.google.common.util.concurren
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
- import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.io.util.DataOutputStreamAndChannel;
- 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;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
index 0000000,53af4c8..f711490
mode 000000,100644..100644
--- a/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
+++ b/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
@@@ -1,0 -1,74 +1,75 @@@
+ /*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ package org.apache.cassandra.streaming;
+
+ import java.io.IOException;
+ import java.net.InetAddress;
+ import java.net.Socket;
+
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+
+ import org.apache.cassandra.config.DatabaseDescriptor;
+ import org.apache.cassandra.net.OutboundTcpConnectionPool;
+
+ public class DefaultConnectionFactory implements StreamConnectionFactory
+ {
+ private static final Logger logger = LoggerFactory.getLogger(DefaultConnectionFactory.class);
+
+ private static final int MAX_CONNECT_ATTEMPTS = 3;
+
+ /**
+ * Connect to peer and start exchanging message.
+ * When connect attempt fails, this retries for maximum of MAX_CONNECT_ATTEMPTS times.
+ *
+ * @param peer the peer to connect to.
+ * @return the created socket.
+ *
+ * @throws IOException when connection failed.
+ */
+ public Socket createConnection(InetAddress peer) throws IOException
+ {
+ int attempts = 0;
+ while (true)
+ {
+ try
+ {
+ Socket socket = OutboundTcpConnectionPool.newSocket(peer);
+ socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
++ socket.setKeepAlive(true);
+ return socket;
+ }
+ catch (IOException e)
+ {
+ if (++attempts >= MAX_CONNECT_ATTEMPTS)
+ throw e;
+
+ long waitms = DatabaseDescriptor.getRpcTimeout() * (long)Math.pow(2, attempts);
+ logger.warn("Failed attempt " + attempts + " to connect to " + peer + ". Retrying in " + waitms + " ms. (" + e + ")");
+ try
+ {
+ Thread.sleep(waitms);
+ }
+ catch (InterruptedException wtf)
+ {
+ throw new IOException("interrupted", wtf);
+ }
+ }
+ }
+ }
+ }
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/streaming/StreamCoordinator.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/StreamCoordinator.java
index 14dad29,0000000..c8c0f14
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/streaming/StreamCoordinator.java
+++ b/src/java/org/apache/cassandra/streaming/StreamCoordinator.java
@@@ -1,278 -1,0 +1,285 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.streaming;
+
+import java.net.InetAddress;
+import java.util.*;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
+import org.apache.cassandra.utils.FBUtilities;
+
+/**
+ * {@link StreamCoordinator} is a helper class that abstracts away maintaining multiple
+ * StreamSession and ProgressInfo instances per peer.
+ *
+ * This class coordinates multiple SessionStreams per peer in both the outgoing StreamPlan context and on the
+ * inbound StreamResultFuture context.
+ */
+public class StreamCoordinator
+{
+ private static final Logger logger = LoggerFactory.getLogger(StreamCoordinator.class);
+
+ // Executor strictly for establishing the initial connections. Once we're connected to the other end the rest of the
+ // streaming is handled directly by the ConnectionHandler's incoming and outgoing threads.
+ private static final DebuggableThreadPoolExecutor streamExecutor = DebuggableThreadPoolExecutor.createWithFixedPoolSize("StreamConnectionEstablisher",
+ FBUtilities.getAvailableProcessors());
+
+ private Map<InetAddress, HostStreamingData> peerSessions = new HashMap<>();
+ private final int connectionsPerHost;
++ private StreamConnectionFactory factory;
+
- public StreamCoordinator(int connectionsPerHost)
++ public StreamCoordinator(int connectionsPerHost, StreamConnectionFactory factory)
+ {
+ this.connectionsPerHost = connectionsPerHost;
++ this.factory = factory;
++ }
++
++ public void setConnectionFactory(StreamConnectionFactory factory)
++ {
++ this.factory = factory;
+ }
+
+ /**
+ * @return true if any stream session is active
+ */
+ public synchronized boolean hasActiveSessions()
+ {
+ for (HostStreamingData data : peerSessions.values())
+ {
+ if (data.hasActiveSessions())
+ return true;
+ }
+ return false;
+ }
+
+ public synchronized Collection<StreamSession> getAllStreamSessions()
+ {
+ Collection<StreamSession> results = new ArrayList<>();
+ for (HostStreamingData data : peerSessions.values())
+ {
+ results.addAll(data.getAllStreamSessions());
+ }
+ return results;
+ }
+
+ public boolean isReceiving()
+ {
+ return connectionsPerHost == 0;
+ }
+
+ public void connectAllStreamSessions()
+ {
+ for (HostStreamingData data : peerSessions.values())
+ data.connectAllStreamSessions();
+ }
+
+ public synchronized Set<InetAddress> getPeers()
+ {
+ return new HashSet<>(peerSessions.keySet());
+ }
+
+ public synchronized StreamSession getOrCreateNextSession(InetAddress peer)
+ {
+ return getOrCreateHostData(peer).getOrCreateNextSession(peer);
+ }
+
+ public synchronized StreamSession getOrCreateSessionById(InetAddress peer, int id)
+ {
+ return getOrCreateHostData(peer).getOrCreateSessionById(peer, id);
+ }
+
+ public synchronized void updateProgress(ProgressInfo info)
+ {
+ getHostData(info.peer).updateProgress(info);
+ }
+
+ public synchronized void addSessionInfo(SessionInfo session)
+ {
+ HostStreamingData data = getOrCreateHostData(session.peer);
+ data.addSessionInfo(session);
+ }
+
+ public synchronized Set<SessionInfo> getAllSessionInfo()
+ {
+ Set<SessionInfo> result = new HashSet<>();
+ for (HostStreamingData data : peerSessions.values())
+ {
+ result.addAll(data.getAllSessionInfo());
+ }
+ return result;
+ }
+
+ public synchronized void transferFiles(InetAddress to, Collection<StreamSession.SSTableStreamingSections> sstableDetails)
+ {
+ HostStreamingData sessionList = getOrCreateHostData(to);
+
+ if (connectionsPerHost > 1)
+ {
+ List<List<StreamSession.SSTableStreamingSections>> buckets = sliceSSTableDetails(sstableDetails);
+
+ for (List<StreamSession.SSTableStreamingSections> subList : buckets)
+ {
+ StreamSession session = sessionList.getOrCreateNextSession(to);
+ session.addTransferFiles(subList);
+ }
+ }
+ else
+ {
+ StreamSession session = sessionList.getOrCreateNextSession(to);
+ session.addTransferFiles(sstableDetails);
+ }
+ }
+
+ private List<List<StreamSession.SSTableStreamingSections>> sliceSSTableDetails(Collection<StreamSession.SSTableStreamingSections> sstableDetails)
+ {
+ // There's no point in divvying things up into more buckets than we have sstableDetails
+ int targetSlices = Math.min(sstableDetails.size(), connectionsPerHost);
+ int step = Math.round((float) sstableDetails.size() / (float) targetSlices);
+ int index = 0;
+
+ List<List<StreamSession.SSTableStreamingSections>> result = new ArrayList<>();
+ List<StreamSession.SSTableStreamingSections> slice = null;
+ for (StreamSession.SSTableStreamingSections streamSession : sstableDetails)
+ {
+ if (index % step == 0)
+ {
+ slice = new ArrayList<>();
+ result.add(slice);
+ }
+ slice.add(streamSession);
+ ++index;
+ }
+
+ return result;
+ }
+
+ private HostStreamingData getHostData(InetAddress peer)
+ {
+ HostStreamingData data = peerSessions.get(peer);
+ if (data == null)
+ throw new IllegalArgumentException("Unknown peer requested: " + peer.toString());
+ return data;
+ }
+
+ private HostStreamingData getOrCreateHostData(InetAddress peer)
+ {
+ HostStreamingData data = peerSessions.get(peer);
+ if (data == null)
+ {
+ data = new HostStreamingData();
+ peerSessions.put(peer, data);
+ }
+ return data;
+ }
+
+ private static class StreamSessionConnector implements Runnable
+ {
+ private final StreamSession session;
+ public StreamSessionConnector(StreamSession session)
+ {
+ this.session = session;
+ }
+
+ @Override
+ public void run()
+ {
+ session.start();
+ logger.info("[Stream #{}, ID#{}] Beginning stream session with {}", session.planId(), session.sessionIndex(), session.peer);
+ }
+ }
+
+ private class HostStreamingData
+ {
+ private Map<Integer, StreamSession> streamSessions = new HashMap<>();
+ private Map<Integer, SessionInfo> sessionInfos = new HashMap<>();
+
+ private int lastReturned = -1;
+
+ public boolean hasActiveSessions()
+ {
+ for (StreamSession session : streamSessions.values())
+ {
+ StreamSession.State state = session.state();
+ if (state != StreamSession.State.COMPLETE && state != StreamSession.State.FAILED)
+ return true;
+ }
+ return false;
+ }
+
+ public StreamSession getOrCreateNextSession(InetAddress peer)
+ {
+ // create
+ if (streamSessions.size() < connectionsPerHost)
+ {
- StreamSession session = new StreamSession(peer, streamSessions.size());
++ StreamSession session = new StreamSession(peer, factory, streamSessions.size());
+ streamSessions.put(++lastReturned, session);
+ return session;
+ }
+ // get
+ else
+ {
+ if (lastReturned >= streamSessions.size() - 1)
+ lastReturned = 0;
+
+ return streamSessions.get(lastReturned++);
+ }
+ }
+
+ public void connectAllStreamSessions()
+ {
+ for (StreamSession session : streamSessions.values())
+ {
+ streamExecutor.execute(new StreamSessionConnector(session));
+ }
+ }
+
+ public Collection<StreamSession> getAllStreamSessions()
+ {
+ return Collections.unmodifiableCollection(streamSessions.values());
+ }
+
+ public StreamSession getOrCreateSessionById(InetAddress peer, int id)
+ {
+ StreamSession session = streamSessions.get(id);
+ if (session == null)
+ {
- session = new StreamSession(peer, id);
++ session = new StreamSession(peer, factory, id);
+ streamSessions.put(id, session);
+ }
+ return session;
+ }
+
+ public void updateProgress(ProgressInfo info)
+ {
+ sessionInfos.get(info.sessionIndex).updateProgress(info);
+ }
+
+ public void addSessionInfo(SessionInfo info)
+ {
+ sessionInfos.put(info.sessionIndex, info);
+ }
+
+ public Collection<SessionInfo> getAllSessionInfo()
+ {
+ return sessionInfos.values();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/streaming/StreamPlan.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/StreamPlan.java
index e775c90,e582c79..c423a33
--- a/src/java/org/apache/cassandra/streaming/StreamPlan.java
+++ b/src/java/org/apache/cassandra/streaming/StreamPlan.java
@@@ -35,9 -34,12 +35,11 @@@ public class StreamPla
private final UUID planId = UUIDGen.getTimeUUID();
private final String description;
private final List<StreamEventHandler> handlers = new ArrayList<>();
-
- // sessions per InetAddress of the other end.
- private final Map<InetAddress, StreamSession> sessions = new HashMap<>();
+ private final long repairedAt;
+ private final StreamCoordinator coordinator;
+ private StreamConnectionFactory connectionFactory = new DefaultConnectionFactory();
+
private boolean flushBeforeTransfer = true;
/**
@@@ -47,14 -49,7 +49,14 @@@
*/
public StreamPlan(String description)
{
+ this(description, ActiveRepairService.UNREPAIRED_SSTABLE, 1);
+ }
+
+ public StreamPlan(String description, long repairedAt, int connectionsPerHost)
+ {
this.description = description;
+ this.repairedAt = repairedAt;
- this.coordinator = new StreamCoordinator(connectionsPerHost);
++ this.coordinator = new StreamCoordinator(connectionsPerHost, connectionFactory);
}
/**
@@@ -139,6 -134,18 +141,18 @@@
}
/**
+ * Set custom StreamConnectionFactory to be used for establishing connection
+ *
+ * @param factory StreamConnectionFactory to use
+ * @return self
+ */
+ public StreamPlan connectionFactory(StreamConnectionFactory factory)
+ {
- this.connectionFactory = factory;
++ this.coordinator.setConnectionFactory(factory);
+ return this;
+ }
+
+ /**
* @return true if this plan has no plan to execute
*/
public boolean isEmpty()
@@@ -168,4 -175,16 +182,5 @@@
this.flushBeforeTransfer = flushBeforeTransfer;
return this;
}
+
- private StreamSession getOrCreateSession(InetAddress peer)
- {
- StreamSession session = sessions.get(peer);
- if (session == null)
- {
- session = new StreamSession(peer, connectionFactory);
- sessions.put(peer, session);
- }
- return session;
- }
-
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/streaming/StreamResultFuture.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/StreamResultFuture.java
index 6a143d1,add14f7..f28a937
--- a/src/java/org/apache/cassandra/streaming/StreamResultFuture.java
+++ b/src/java/org/apache/cassandra/streaming/StreamResultFuture.java
@@@ -69,14 -75,9 +69,14 @@@ public final class StreamResultFuture e
set(getCurrentState());
}
- static StreamResultFuture init(UUID planId, String description, Collection<StreamSession> sessions, Collection<StreamEventHandler> listeners)
+ private StreamResultFuture(UUID planId, String description)
+ {
- this(planId, description, new StreamCoordinator(0));
++ this(planId, description, new StreamCoordinator(0, new DefaultConnectionFactory()));
+ }
+
+ static StreamResultFuture init(UUID planId, String description, Collection<StreamEventHandler> listeners, StreamCoordinator coordinator)
{
- StreamResultFuture future = createAndRegister(planId, description, sessions);
+ StreamResultFuture future = createAndRegister(planId, description, coordinator);
if (listeners != null)
{
for (StreamEventHandler listener : listeners)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/StreamSession.java
index 1afc07e,4fcbe36..2efa00d
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@@ -148,11 -155,12 +151,13 @@@ public class StreamSession implements I
* Create new streaming session with the peer.
*
* @param peer Address of streaming peer
+ * @param factory is used for establishing connection
*/
- public StreamSession(InetAddress peer, int index)
- public StreamSession(InetAddress peer, StreamConnectionFactory factory)
++ public StreamSession(InetAddress peer, StreamConnectionFactory factory, int index)
{
this.peer = peer;
+ this.index = index;
+ this.factory = factory;
this.handler = new ConnectionHandler(this);
this.metrics = StreamingMetrics.get(peer);
}
@@@ -192,18 -199,29 +197,24 @@@
return;
}
- streamExecutor.execute(new Runnable()
+ try
{
- public void run()
- {
- try
- {
- handler.initiate();
- onInitializationComplete();
- }
- catch (IOException e)
- {
- onError(e);
- }
- }
- });
+ logger.info("[Stream #{}, ID#{}] Beginning stream session with {}", planId(), sessionIndex(), peer);
+ handler.initiate();
+ onInitializationComplete();
+ }
+ catch (Exception e)
+ {
+ onError(e);
+ }
}
+ public Socket createConnection() throws IOException
+ {
+ assert factory != null;
+ return factory.createConnection(peer);
+ }
+
/**
* Request data fetch task to this session.
*
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/src/java/org/apache/cassandra/tools/BulkLoader.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/tools/BulkLoader.java
index 74b803a,4077722..50e340b
--- a/src/java/org/apache/cassandra/tools/BulkLoader.java
+++ b/src/java/org/apache/cassandra/tools/BulkLoader.java
@@@ -18,27 -18,27 +18,27 @@@
package org.apache.cassandra.tools;
import java.io.File;
- import java.net.InetAddress;
- import java.net.UnknownHostException;
+ import java.net.*;
import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.TimeUnit;
import com.google.common.base.Joiner;
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
+ import com.google.common.collect.Sets;
+
-import org.apache.cassandra.config.*;
-
import org.apache.commons.cli.*;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
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.config.EncryptionOptions;
++import org.apache.cassandra.config.*;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
+ import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.sstable.SSTableLoader;
import org.apache.cassandra.streaming.*;
import org.apache.cassandra.thrift.*;
@@@ -66,26 -70,22 +69,30 @@@ public class BulkLoade
private static final String SSL_ALGORITHM = "ssl-alg";
private static final String SSL_STORE_TYPE = "store-type";
private static final String SSL_CIPHER_SUITES = "ssl-ciphers";
+ private static final String CONNECTIONS_PER_HOST = "connections-per-host";
+ private static final String CONFIG_PATH = "conf-path";
public static void main(String args[])
{
LoaderOptions options = LoaderOptions.parseArgs(args);
- OutputHandler handler = new OutputHandler.SystemOutput(options.verbose, true);
+ OutputHandler handler = new OutputHandler.SystemOutput(options.verbose, options.debug);
- SSTableLoader loader = new SSTableLoader(options.directory, new ExternalClient(options.hosts,
- options.rpcPort,
- options.user,
- options.passwd,
- options.transportFactory,
- options.storagePort,
- options.sslStoragePort,
- options.serverEncOptions), handler);
+ SSTableLoader loader = new SSTableLoader(
+ options.directory,
+ new ExternalClient(
+ options.hosts,
+ options.rpcPort,
+ options.user,
+ options.passwd,
- options.transportFactory),
++ options.transportFactory,
++ options.storagePort,
++ options.sslStoragePort,
++ options.serverEncOptions),
+ handler,
+ options.connectionsPerHost);
DatabaseDescriptor.setStreamThroughputOutboundMegabitsPerSec(options.throttle);
StreamResultFuture future = null;
+
+ ProgressIndicator indicator = new ProgressIndicator();
try
{
if (options.noProgress)
@@@ -349,9 -324,11 +377,12 @@@
public String user;
public String passwd;
public int throttle = 0;
+ public int storagePort;
+ public int sslStoragePort;
public ITransportFactory transportFactory = new TFramedTransportFactory();
public EncryptionOptions encOptions = new EncryptionOptions.ClientEncryptionOptions();
+ public int connectionsPerHost = 1;
+ public EncryptionOptions.ServerEncryptionOptions serverEncOptions = new EncryptionOptions.ServerEncryptionOptions();
public final Set<InetAddress> hosts = new HashSet<>();
public final Set<InetAddress> ignores = new HashSet<>();
@@@ -455,10 -430,34 +483,37 @@@
}
}
+ if (cmd.hasOption(CONNECTIONS_PER_HOST))
+ opts.connectionsPerHost = Integer.parseInt(cmd.getOptionValue(CONNECTIONS_PER_HOST));
+
- if(cmd.hasOption(SSL_TRUSTSTORE))
+ // try to load config file first, so that values can be rewritten with other option values.
+ // otherwise use default config.
+ Config config;
+ if (cmd.hasOption(CONFIG_PATH))
+ {
+ File configFile = new File(cmd.getOptionValue(CONFIG_PATH));
+ if (!configFile.exists())
+ {
+ errorMsg("Config file not found", options);
+ }
+ config = new YamlConfigurationLoader().loadConfig(configFile.toURI().toURL());
+ }
+ else
+ {
+ config = new Config();
+ }
+ opts.storagePort = config.storage_port;
+ opts.sslStoragePort = config.ssl_storage_port;
+ opts.throttle = config.stream_throughput_outbound_megabits_per_sec;
+ opts.encOptions = config.client_encryption_options;
+ opts.serverEncOptions = config.server_encryption_options;
+
+ if (cmd.hasOption(THROTTLE_MBITS))
+ {
+ opts.throttle = Integer.parseInt(cmd.getOptionValue(THROTTLE_MBITS));
+ }
+
+ if (cmd.hasOption(SSL_TRUSTSTORE))
{
opts.encOptions.truststore = cmd.getOptionValue(SSL_TRUSTSTORE);
}
@@@ -579,16 -580,16 +635,17 @@@
options.addOption("u", USER_OPTION, "username", "username for cassandra authentication");
options.addOption("pw", PASSWD_OPTION, "password", "password for cassandra authentication");
options.addOption("tf", TRANSPORT_FACTORY, "transport factory", "Fully-qualified ITransportFactory class name for creating a connection to cassandra");
+ options.addOption("cph", CONNECTIONS_PER_HOST, "connectionsPerHost", "number of concurrent connections-per-host.");
// ssl connection-related options
- options.addOption("ts", SSL_TRUSTSTORE, "TRUSTSTORE", "SSL: full path to truststore");
- options.addOption("tspw", SSL_TRUSTSTORE_PW, "TRUSTSTORE-PASSWORD", "SSL: password of the truststore");
- options.addOption("ks", SSL_KEYSTORE, "KEYSTORE", "SSL: full path to keystore");
- options.addOption("kspw", SSL_KEYSTORE_PW, "KEYSTORE-PASSWORD", "SSL: password of the keystore");
- options.addOption("prtcl", SSL_PROTOCOL, "PROTOCOL", "SSL: connections protocol to use (default: TLS)");
- options.addOption("alg", SSL_ALGORITHM, "ALGORITHM", "SSL: algorithm (default: SunX509)");
- options.addOption("st", SSL_STORE_TYPE, "STORE-TYPE", "SSL: type of store");
- options.addOption("ciphers", SSL_CIPHER_SUITES, "CIPHER-SUITES", "SSL: comma-separated list of encryption suites to use");
+ options.addOption("ts", SSL_TRUSTSTORE, "TRUSTSTORE", "Client SSL: full path to truststore");
+ options.addOption("tspw", SSL_TRUSTSTORE_PW, "TRUSTSTORE-PASSWORD", "Client SSL: password of the truststore");
+ options.addOption("ks", SSL_KEYSTORE, "KEYSTORE", "Client SSL: full path to keystore");
+ options.addOption("kspw", SSL_KEYSTORE_PW, "KEYSTORE-PASSWORD", "Client SSL: password of the keystore");
+ options.addOption("prtcl", SSL_PROTOCOL, "PROTOCOL", "Client SSL: connections protocol to use (default: TLS)");
+ options.addOption("alg", SSL_ALGORITHM, "ALGORITHM", "Client SSL: algorithm (default: SunX509)");
+ options.addOption("st", SSL_STORE_TYPE, "STORE-TYPE", "Client SSL: type of store");
+ options.addOption("ciphers", SSL_CIPHER_SUITES, "CIPHER-SUITES", "Client SSL: comma-separated list of encryption suites to use");
+ options.addOption("f", CONFIG_PATH, "path to config file", "cassandra.yaml file path for streaming throughput and client/server SSL.");
return options;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/de5bb585/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
index c6c04b0,ce0f9d0..4043ac8
--- a/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
+++ b/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
@@@ -43,7 -43,7 +43,7 @@@ public class StreamTransferTaskTest ext
String ks = "Keyspace1";
String cf = "Standard1";
- StreamSession session = new StreamSession(FBUtilities.getBroadcastAddress(), 0);
- StreamSession session = new StreamSession(FBUtilities.getBroadcastAddress(), null);
++ StreamSession session = new StreamSession(FBUtilities.getBroadcastAddress(), null, 0);
ColumnFamilyStore cfs = Keyspace.open(ks).getColumnFamilyStore(cf);
// create two sstables