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 2016/03/10 20:04:33 UTC
[01/15] cassandra git commit: Fix streaming_socket_timeout_in_ms not
enforced
Repository: cassandra
Updated Branches:
refs/heads/cassandra-2.1 e94a2a034 -> 561000aa3
refs/heads/cassandra-2.2 7e220bc98 -> b9ff7fe13
refs/heads/cassandra-3.0 232e12b94 -> c9d2b7add
refs/heads/cassandra-3.5 f27ab2908 -> 10d61d46d
refs/heads/trunk bdd9e3b16 -> 9d527ed25
Fix streaming_socket_timeout_in_ms not enforced
Patch by Paulo Motta; reviewed by Yuki Morishita for CASSANDRA-11286
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/561000aa
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/561000aa
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/561000aa
Branch: refs/heads/cassandra-2.1
Commit: 561000aa3094699bab29766d9644ff50f6cb74f3
Parents: e94a2a0
Author: Paulo Motta <pa...@gmail.com>
Authored: Fri Feb 12 12:17:01 2016 -0300
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 10 12:54:24 2016 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
conf/cassandra.yaml | 7 +++----
.../net/IncomingStreamingConnection.java | 7 ++++++-
.../cassandra/streaming/ConnectionHandler.java | 21 +++++++++++++-------
.../cassandra/streaming/StreamSession.java | 2 ++
5 files changed, 26 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index e7c997a..4b505f8 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.1.14
+ * Fix streaming_socket_timeout_in_ms not enforced (CASSANDRA-11286)
* Avoid dropping message too quickly due to missing unit conversion (CASSANDRA-11302)
* COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053)
* InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/conf/cassandra.yaml
----------------------------------------------------------------------
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index 1fa04e6..0da4800 100644
--- a/conf/cassandra.yaml
+++ b/conf/cassandra.yaml
@@ -630,10 +630,9 @@ request_timeout_in_ms: 10000
# and the times are synchronized between the nodes.
cross_node_timeout: false
-# Enable socket timeout for streaming operation.
-# When a timeout occurs during streaming, streaming is retried from the start
-# of the current file. This _can_ involve re-streaming an important amount of
-# data, so you should avoid setting the value too low.
+# Set socket timeout for streaming operation.
+# The stream session is failed if no data is received by any of the
+# participants within that period.
# Default value is 3600000, which means streams timeout after an hour.
# streaming_socket_timeout_in_ms: 3600000
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
index 1f98bc4..5ced786 100644
--- a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
+++ b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
@@ -27,6 +27,7 @@ import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.streaming.StreamResultFuture;
import org.apache.cassandra.streaming.messages.StreamInitMessage;
import org.apache.cassandra.streaming.messages.StreamMessage;
@@ -62,6 +63,10 @@ public class IncomingStreamingConnection extends Thread implements Closeable
DataInput input = new DataInputStream(socket.getInputStream());
StreamInitMessage init = StreamInitMessage.serializer.deserialize(input, version);
+ //Set SO_TIMEOUT on follower side
+ if (!init.isForOutgoing)
+ socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
+
// The initiator makes two connections, one for incoming and one for outgoing.
// The receiving side distinguish two connections by looking at StreamInitMessage#isForOutgoing.
// Note: we cannot use the same socket for incoming and outgoing streams because we want to
@@ -74,7 +79,7 @@ public class IncomingStreamingConnection extends Thread implements Closeable
close();
}
}
-
+
@Override
public void close()
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/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 ac267f9..52268b2 100644
--- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
+++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
@@ -105,12 +105,22 @@ public class ConnectionHandler
{
logger.debug("[Stream #{}] Closing stream connection handler on {}", session.planId(), session.peer);
- ListenableFuture<?> inClosed = incoming == null ? Futures.immediateFuture(null) : incoming.close();
- ListenableFuture<?> outClosed = outgoing == null ? Futures.immediateFuture(null) : outgoing.close();
+ ListenableFuture<?> inClosed = closeIncoming();
+ ListenableFuture<?> outClosed = closeOutgoing();
return Futures.allAsList(inClosed, outClosed);
}
+ public ListenableFuture<?> closeOutgoing()
+ {
+ return outgoing == null ? Futures.immediateFuture(null) : outgoing.close();
+ }
+
+ public ListenableFuture<?> closeIncoming()
+ {
+ return incoming == null ? Futures.immediateFuture(null) : incoming.close();
+ }
+
/**
* Enqueue messages to be sent.
*
@@ -165,11 +175,8 @@ public class ConnectionHandler
protected static ReadableByteChannel getReadChannel(Socket socket) throws IOException
{
- ReadableByteChannel in = socket.getChannel();
- // socket channel is null when encrypted(SSL)
- return in == null
- ? Channels.newChannel(socket.getInputStream())
- : in;
+ //we do this instead of socket.getChannel() so socketSoTimeout is respected
+ return Channels.newChannel(socket.getInputStream());
}
public void sendInitMessage(Socket socket, boolean isForOutgoing) throws IOException
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/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 98a6f1f..642e837 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -609,6 +609,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
else
{
state(State.WAIT_COMPLETE);
+ handler.closeIncoming();
}
}
@@ -696,6 +697,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
handler.sendMessage(new CompleteMessage());
completeSent = true;
state(State.WAIT_COMPLETE);
+ handler.closeOutgoing();
}
}
return completed;
[15/15] cassandra git commit: Merge branch 'cassandra-3.5' into trunk
Posted by yu...@apache.org.
Merge branch 'cassandra-3.5' into trunk
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/9d527ed2
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/9d527ed2
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/9d527ed2
Branch: refs/heads/trunk
Commit: 9d527ed257ece570b1cd29e1ce77a74d10cfd605
Parents: bdd9e3b 10d61d4
Author: Yuki Morishita <yu...@apache.org>
Authored: Thu Mar 10 13:03:21 2016 -0600
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 10 13:03:21 2016 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
conf/cassandra.yaml | 7 +++----
.../net/IncomingStreamingConnection.java | 5 +++++
.../cassandra/streaming/ConnectionHandler.java | 21 +++++++++++++-------
.../cassandra/streaming/StreamSession.java | 2 ++
5 files changed, 25 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/9d527ed2/CHANGES.txt
----------------------------------------------------------------------
[11/15] cassandra git commit: Merge branch 'cassandra-2.2' into
cassandra-3.0
Posted by yu...@apache.org.
Merge branch 'cassandra-2.2' into cassandra-3.0
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/c9d2b7ad
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/c9d2b7ad
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/c9d2b7ad
Branch: refs/heads/cassandra-3.5
Commit: c9d2b7addcfefb4cfa2bfcb96215153030d9d624
Parents: 232e12b b9ff7fe
Author: Yuki Morishita <yu...@apache.org>
Authored: Thu Mar 10 13:01:30 2016 -0600
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 10 13:01:30 2016 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
conf/cassandra.yaml | 7 +++----
.../net/IncomingStreamingConnection.java | 5 +++++
.../cassandra/streaming/ConnectionHandler.java | 21 +++++++++++++-------
.../cassandra/streaming/StreamSession.java | 2 ++
5 files changed, 25 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c9d2b7ad/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 7b93083,045b867..4c00994
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -56,8 -26,8 +56,9 @@@ Merged from 2.2
* (cqlsh) Support utf-8/cp65001 encoding on Windows (CASSANDRA-11030)
* Fix paging on DISTINCT queries repeats result when first row in partition changes
(CASSANDRA-10010)
+ * cqlsh: change default encoding to UTF-8 (CASSANDRA-11124)
Merged from 2.1:
+ * Fix streaming_socket_timeout_in_ms not enforced (CASSANDRA-11286)
* Avoid dropping message too quickly due to missing unit conversion (CASSANDRA-11302)
* COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053)
* InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c9d2b7ad/conf/cassandra.yaml
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c9d2b7ad/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
index e58d227,842676d..19e0671
--- a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
+++ b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
@@@ -25,8 -27,7 +25,9 @@@ import java.util.Set
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataInputPlus.DataInputStreamPlus;
+ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.streaming.StreamResultFuture;
import org.apache.cassandra.streaming.messages.StreamInitMessage;
import org.apache.cassandra.streaming.messages.StreamMessage;
@@@ -60,9 -60,13 +61,13 @@@ public class IncomingStreamingConnectio
if (version != StreamMessage.CURRENT_VERSION)
throw new IOException(String.format("Received stream using protocol version %d (my version %d). Terminating connection", version, MessagingService.current_version));
- DataInput input = new DataInputStream(socket.getInputStream());
+ DataInputPlus input = new DataInputStreamPlus(socket.getInputStream());
StreamInitMessage init = StreamInitMessage.serializer.deserialize(input, version);
+ //Set SO_TIMEOUT on follower side
+ if (!init.isForOutgoing)
+ socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
+
// The initiator makes two connections, one for incoming and one for outgoing.
// The receiving side distinguish two connections by looking at StreamInitMessage#isForOutgoing.
// Note: we cannot use the same socket for incoming and outgoing streams because we want to
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c9d2b7ad/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
[04/15] cassandra git commit: Fix streaming_socket_timeout_in_ms not
enforced
Posted by yu...@apache.org.
Fix streaming_socket_timeout_in_ms not enforced
Patch by Paulo Motta; reviewed by Yuki Morishita for CASSANDRA-11286
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/561000aa
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/561000aa
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/561000aa
Branch: refs/heads/cassandra-3.0
Commit: 561000aa3094699bab29766d9644ff50f6cb74f3
Parents: e94a2a0
Author: Paulo Motta <pa...@gmail.com>
Authored: Fri Feb 12 12:17:01 2016 -0300
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 10 12:54:24 2016 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
conf/cassandra.yaml | 7 +++----
.../net/IncomingStreamingConnection.java | 7 ++++++-
.../cassandra/streaming/ConnectionHandler.java | 21 +++++++++++++-------
.../cassandra/streaming/StreamSession.java | 2 ++
5 files changed, 26 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index e7c997a..4b505f8 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.1.14
+ * Fix streaming_socket_timeout_in_ms not enforced (CASSANDRA-11286)
* Avoid dropping message too quickly due to missing unit conversion (CASSANDRA-11302)
* COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053)
* InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/conf/cassandra.yaml
----------------------------------------------------------------------
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index 1fa04e6..0da4800 100644
--- a/conf/cassandra.yaml
+++ b/conf/cassandra.yaml
@@ -630,10 +630,9 @@ request_timeout_in_ms: 10000
# and the times are synchronized between the nodes.
cross_node_timeout: false
-# Enable socket timeout for streaming operation.
-# When a timeout occurs during streaming, streaming is retried from the start
-# of the current file. This _can_ involve re-streaming an important amount of
-# data, so you should avoid setting the value too low.
+# Set socket timeout for streaming operation.
+# The stream session is failed if no data is received by any of the
+# participants within that period.
# Default value is 3600000, which means streams timeout after an hour.
# streaming_socket_timeout_in_ms: 3600000
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
index 1f98bc4..5ced786 100644
--- a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
+++ b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
@@ -27,6 +27,7 @@ import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.streaming.StreamResultFuture;
import org.apache.cassandra.streaming.messages.StreamInitMessage;
import org.apache.cassandra.streaming.messages.StreamMessage;
@@ -62,6 +63,10 @@ public class IncomingStreamingConnection extends Thread implements Closeable
DataInput input = new DataInputStream(socket.getInputStream());
StreamInitMessage init = StreamInitMessage.serializer.deserialize(input, version);
+ //Set SO_TIMEOUT on follower side
+ if (!init.isForOutgoing)
+ socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
+
// The initiator makes two connections, one for incoming and one for outgoing.
// The receiving side distinguish two connections by looking at StreamInitMessage#isForOutgoing.
// Note: we cannot use the same socket for incoming and outgoing streams because we want to
@@ -74,7 +79,7 @@ public class IncomingStreamingConnection extends Thread implements Closeable
close();
}
}
-
+
@Override
public void close()
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/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 ac267f9..52268b2 100644
--- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
+++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
@@ -105,12 +105,22 @@ public class ConnectionHandler
{
logger.debug("[Stream #{}] Closing stream connection handler on {}", session.planId(), session.peer);
- ListenableFuture<?> inClosed = incoming == null ? Futures.immediateFuture(null) : incoming.close();
- ListenableFuture<?> outClosed = outgoing == null ? Futures.immediateFuture(null) : outgoing.close();
+ ListenableFuture<?> inClosed = closeIncoming();
+ ListenableFuture<?> outClosed = closeOutgoing();
return Futures.allAsList(inClosed, outClosed);
}
+ public ListenableFuture<?> closeOutgoing()
+ {
+ return outgoing == null ? Futures.immediateFuture(null) : outgoing.close();
+ }
+
+ public ListenableFuture<?> closeIncoming()
+ {
+ return incoming == null ? Futures.immediateFuture(null) : incoming.close();
+ }
+
/**
* Enqueue messages to be sent.
*
@@ -165,11 +175,8 @@ public class ConnectionHandler
protected static ReadableByteChannel getReadChannel(Socket socket) throws IOException
{
- ReadableByteChannel in = socket.getChannel();
- // socket channel is null when encrypted(SSL)
- return in == null
- ? Channels.newChannel(socket.getInputStream())
- : in;
+ //we do this instead of socket.getChannel() so socketSoTimeout is respected
+ return Channels.newChannel(socket.getInputStream());
}
public void sendInitMessage(Socket socket, boolean isForOutgoing) throws IOException
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/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 98a6f1f..642e837 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -609,6 +609,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
else
{
state(State.WAIT_COMPLETE);
+ handler.closeIncoming();
}
}
@@ -696,6 +697,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
handler.sendMessage(new CompleteMessage());
completeSent = true;
state(State.WAIT_COMPLETE);
+ handler.closeOutgoing();
}
}
return completed;
[13/15] cassandra git commit: Merge branch 'cassandra-3.0' into
cassandra-3.5
Posted by yu...@apache.org.
Merge branch 'cassandra-3.0' into cassandra-3.5
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/10d61d46
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/10d61d46
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/10d61d46
Branch: refs/heads/cassandra-3.5
Commit: 10d61d46d36484ae0ae77108531528be5ba27619
Parents: f27ab29 c9d2b7a
Author: Yuki Morishita <yu...@apache.org>
Authored: Thu Mar 10 13:03:15 2016 -0600
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 10 13:03:15 2016 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
conf/cassandra.yaml | 7 +++----
.../net/IncomingStreamingConnection.java | 5 +++++
.../cassandra/streaming/ConnectionHandler.java | 21 +++++++++++++-------
.../cassandra/streaming/StreamSession.java | 2 ++
5 files changed, 25 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/10d61d46/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 68f5714,4c00994..292ebb5
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -88,9 -56,9 +88,10 @@@ Merged from 2.2
* (cqlsh) Support utf-8/cp65001 encoding on Windows (CASSANDRA-11030)
* Fix paging on DISTINCT queries repeats result when first row in partition changes
(CASSANDRA-10010)
+ * (cqlsh) Support timezone conversion using pytz (CASSANDRA-10397)
* cqlsh: change default encoding to UTF-8 (CASSANDRA-11124)
Merged from 2.1:
+ * Fix streaming_socket_timeout_in_ms not enforced (CASSANDRA-11286)
* Avoid dropping message too quickly due to missing unit conversion (CASSANDRA-11302)
* COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053)
* InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/10d61d46/conf/cassandra.yaml
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/10d61d46/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
[10/15] cassandra git commit: Merge branch 'cassandra-2.2' into
cassandra-3.0
Posted by yu...@apache.org.
Merge branch 'cassandra-2.2' into cassandra-3.0
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/c9d2b7ad
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/c9d2b7ad
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/c9d2b7ad
Branch: refs/heads/trunk
Commit: c9d2b7addcfefb4cfa2bfcb96215153030d9d624
Parents: 232e12b b9ff7fe
Author: Yuki Morishita <yu...@apache.org>
Authored: Thu Mar 10 13:01:30 2016 -0600
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 10 13:01:30 2016 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
conf/cassandra.yaml | 7 +++----
.../net/IncomingStreamingConnection.java | 5 +++++
.../cassandra/streaming/ConnectionHandler.java | 21 +++++++++++++-------
.../cassandra/streaming/StreamSession.java | 2 ++
5 files changed, 25 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c9d2b7ad/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 7b93083,045b867..4c00994
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -56,8 -26,8 +56,9 @@@ Merged from 2.2
* (cqlsh) Support utf-8/cp65001 encoding on Windows (CASSANDRA-11030)
* Fix paging on DISTINCT queries repeats result when first row in partition changes
(CASSANDRA-10010)
+ * cqlsh: change default encoding to UTF-8 (CASSANDRA-11124)
Merged from 2.1:
+ * Fix streaming_socket_timeout_in_ms not enforced (CASSANDRA-11286)
* Avoid dropping message too quickly due to missing unit conversion (CASSANDRA-11302)
* COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053)
* InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c9d2b7ad/conf/cassandra.yaml
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c9d2b7ad/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
index e58d227,842676d..19e0671
--- a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
+++ b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
@@@ -25,8 -27,7 +25,9 @@@ import java.util.Set
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataInputPlus.DataInputStreamPlus;
+ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.streaming.StreamResultFuture;
import org.apache.cassandra.streaming.messages.StreamInitMessage;
import org.apache.cassandra.streaming.messages.StreamMessage;
@@@ -60,9 -60,13 +61,13 @@@ public class IncomingStreamingConnectio
if (version != StreamMessage.CURRENT_VERSION)
throw new IOException(String.format("Received stream using protocol version %d (my version %d). Terminating connection", version, MessagingService.current_version));
- DataInput input = new DataInputStream(socket.getInputStream());
+ DataInputPlus input = new DataInputStreamPlus(socket.getInputStream());
StreamInitMessage init = StreamInitMessage.serializer.deserialize(input, version);
+ //Set SO_TIMEOUT on follower side
+ if (!init.isForOutgoing)
+ socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
+
// The initiator makes two connections, one for incoming and one for outgoing.
// The receiving side distinguish two connections by looking at StreamInitMessage#isForOutgoing.
// Note: we cannot use the same socket for incoming and outgoing streams because we want to
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c9d2b7ad/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
[06/15] cassandra git commit: Merge branch 'cassandra-2.1' into
cassandra-2.2
Posted by yu...@apache.org.
Merge branch 'cassandra-2.1' into cassandra-2.2
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/b9ff7fe1
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/b9ff7fe1
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/b9ff7fe1
Branch: refs/heads/trunk
Commit: b9ff7fe13740a21da25ed6414ea17fe5530652ea
Parents: 7e220bc 561000a
Author: Yuki Morishita <yu...@apache.org>
Authored: Thu Mar 10 13:00:09 2016 -0600
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 10 13:00:09 2016 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
conf/cassandra.yaml | 7 +++----
.../net/IncomingStreamingConnection.java | 7 ++++++-
.../cassandra/streaming/ConnectionHandler.java | 21 +++++++++++++-------
.../cassandra/streaming/StreamSession.java | 2 ++
5 files changed, 26 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index b9596d9,4b505f8..045b867
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,32 -1,5 +1,33 @@@
-2.1.14
+2.2.6
+ * Non-obsoleting compaction operations over compressed files can impose rate limit on normal reads (CASSANDRA-11301)
+ * Add missing newline at end of bin/cqlsh (CASSANDRA-11325)
+ * Fix AE in nodetool cfstats (backport CASSANDRA-10859) (CASSANDRA-11297)
+ * Unresolved hostname leads to replace being ignored (CASSANDRA-11210)
+ * Fix filtering on non-primary key columns for thrift static column families
+ (CASSANDRA-6377)
+ * Only log yaml config once, at startup (CASSANDRA-11217)
+ * Preserve order for preferred SSL cipher suites (CASSANDRA-11164)
+ * Reference leak with parallel repairs on the same table (CASSANDRA-11215)
+ * Range.compareTo() violates the contract of Comparable (CASSANDRA-11216)
+ * Avoid NPE when serializing ErrorMessage with null message (CASSANDRA-11167)
+ * Replacing an aggregate with a new version doesn't reset INITCOND (CASSANDRA-10840)
+ * (cqlsh) cqlsh cannot be called through symlink (CASSANDRA-11037)
+ * fix ohc and java-driver pom dependencies in build.xml (CASSANDRA-10793)
+ * Protect from keyspace dropped during repair (CASSANDRA-11065)
+ * Handle adding fields to a UDT in SELECT JSON and toJson() (CASSANDRA-11146)
+ * Better error message for cleanup (CASSANDRA-10991)
+ * cqlsh pg-style-strings broken if line ends with ';' (CASSANDRA-11123)
+ * Use cloned TokenMetadata in size estimates to avoid race against membership check
+ (CASSANDRA-10736)
+ * Always persist upsampled index summaries (CASSANDRA-10512)
+ * (cqlsh) Fix inconsistent auto-complete (CASSANDRA-10733)
+ * Make SELECT JSON and toJson() threadsafe (CASSANDRA-11048)
+ * Fix SELECT on tuple relations for mixed ASC/DESC clustering order (CASSANDRA-7281)
+ * (cqlsh) Support utf-8/cp65001 encoding on Windows (CASSANDRA-11030)
+ * Fix paging on DISTINCT queries repeats result when first row in partition changes
+ (CASSANDRA-10010)
+Merged from 2.1:
+ * Fix streaming_socket_timeout_in_ms not enforced (CASSANDRA-11286)
* Avoid dropping message too quickly due to missing unit conversion (CASSANDRA-11302)
* COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053)
* InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/conf/cassandra.yaml
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/ConnectionHandler.java
index 1ec7e1c,52268b2..d4662c7
--- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
+++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
@@@ -170,14 -175,10 +180,11 @@@ public class ConnectionHandle
protected static ReadableByteChannel getReadChannel(Socket socket) throws IOException
{
- ReadableByteChannel in = socket.getChannel();
- // socket channel is null when encrypted(SSL)
- return in == null
- ? Channels.newChannel(socket.getInputStream())
- : in;
+ //we do this instead of socket.getChannel() so socketSoTimeout is respected
+ return Channels.newChannel(socket.getInputStream());
}
+ @SuppressWarnings("resource")
public void sendInitMessage(Socket socket, boolean isForOutgoing) throws IOException
{
StreamInitMessage message = new StreamInitMessage(
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
[03/15] cassandra git commit: Fix streaming_socket_timeout_in_ms not
enforced
Posted by yu...@apache.org.
Fix streaming_socket_timeout_in_ms not enforced
Patch by Paulo Motta; reviewed by Yuki Morishita for CASSANDRA-11286
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/561000aa
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/561000aa
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/561000aa
Branch: refs/heads/trunk
Commit: 561000aa3094699bab29766d9644ff50f6cb74f3
Parents: e94a2a0
Author: Paulo Motta <pa...@gmail.com>
Authored: Fri Feb 12 12:17:01 2016 -0300
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 10 12:54:24 2016 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
conf/cassandra.yaml | 7 +++----
.../net/IncomingStreamingConnection.java | 7 ++++++-
.../cassandra/streaming/ConnectionHandler.java | 21 +++++++++++++-------
.../cassandra/streaming/StreamSession.java | 2 ++
5 files changed, 26 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index e7c997a..4b505f8 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.1.14
+ * Fix streaming_socket_timeout_in_ms not enforced (CASSANDRA-11286)
* Avoid dropping message too quickly due to missing unit conversion (CASSANDRA-11302)
* COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053)
* InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/conf/cassandra.yaml
----------------------------------------------------------------------
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index 1fa04e6..0da4800 100644
--- a/conf/cassandra.yaml
+++ b/conf/cassandra.yaml
@@ -630,10 +630,9 @@ request_timeout_in_ms: 10000
# and the times are synchronized between the nodes.
cross_node_timeout: false
-# Enable socket timeout for streaming operation.
-# When a timeout occurs during streaming, streaming is retried from the start
-# of the current file. This _can_ involve re-streaming an important amount of
-# data, so you should avoid setting the value too low.
+# Set socket timeout for streaming operation.
+# The stream session is failed if no data is received by any of the
+# participants within that period.
# Default value is 3600000, which means streams timeout after an hour.
# streaming_socket_timeout_in_ms: 3600000
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
index 1f98bc4..5ced786 100644
--- a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
+++ b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
@@ -27,6 +27,7 @@ import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.streaming.StreamResultFuture;
import org.apache.cassandra.streaming.messages.StreamInitMessage;
import org.apache.cassandra.streaming.messages.StreamMessage;
@@ -62,6 +63,10 @@ public class IncomingStreamingConnection extends Thread implements Closeable
DataInput input = new DataInputStream(socket.getInputStream());
StreamInitMessage init = StreamInitMessage.serializer.deserialize(input, version);
+ //Set SO_TIMEOUT on follower side
+ if (!init.isForOutgoing)
+ socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
+
// The initiator makes two connections, one for incoming and one for outgoing.
// The receiving side distinguish two connections by looking at StreamInitMessage#isForOutgoing.
// Note: we cannot use the same socket for incoming and outgoing streams because we want to
@@ -74,7 +79,7 @@ public class IncomingStreamingConnection extends Thread implements Closeable
close();
}
}
-
+
@Override
public void close()
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/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 ac267f9..52268b2 100644
--- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
+++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
@@ -105,12 +105,22 @@ public class ConnectionHandler
{
logger.debug("[Stream #{}] Closing stream connection handler on {}", session.planId(), session.peer);
- ListenableFuture<?> inClosed = incoming == null ? Futures.immediateFuture(null) : incoming.close();
- ListenableFuture<?> outClosed = outgoing == null ? Futures.immediateFuture(null) : outgoing.close();
+ ListenableFuture<?> inClosed = closeIncoming();
+ ListenableFuture<?> outClosed = closeOutgoing();
return Futures.allAsList(inClosed, outClosed);
}
+ public ListenableFuture<?> closeOutgoing()
+ {
+ return outgoing == null ? Futures.immediateFuture(null) : outgoing.close();
+ }
+
+ public ListenableFuture<?> closeIncoming()
+ {
+ return incoming == null ? Futures.immediateFuture(null) : incoming.close();
+ }
+
/**
* Enqueue messages to be sent.
*
@@ -165,11 +175,8 @@ public class ConnectionHandler
protected static ReadableByteChannel getReadChannel(Socket socket) throws IOException
{
- ReadableByteChannel in = socket.getChannel();
- // socket channel is null when encrypted(SSL)
- return in == null
- ? Channels.newChannel(socket.getInputStream())
- : in;
+ //we do this instead of socket.getChannel() so socketSoTimeout is respected
+ return Channels.newChannel(socket.getInputStream());
}
public void sendInitMessage(Socket socket, boolean isForOutgoing) throws IOException
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/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 98a6f1f..642e837 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -609,6 +609,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
else
{
state(State.WAIT_COMPLETE);
+ handler.closeIncoming();
}
}
@@ -696,6 +697,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
handler.sendMessage(new CompleteMessage());
completeSent = true;
state(State.WAIT_COMPLETE);
+ handler.closeOutgoing();
}
}
return completed;
[07/15] cassandra git commit: Merge branch 'cassandra-2.1' into
cassandra-2.2
Posted by yu...@apache.org.
Merge branch 'cassandra-2.1' into cassandra-2.2
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/b9ff7fe1
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/b9ff7fe1
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/b9ff7fe1
Branch: refs/heads/cassandra-2.2
Commit: b9ff7fe13740a21da25ed6414ea17fe5530652ea
Parents: 7e220bc 561000a
Author: Yuki Morishita <yu...@apache.org>
Authored: Thu Mar 10 13:00:09 2016 -0600
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 10 13:00:09 2016 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
conf/cassandra.yaml | 7 +++----
.../net/IncomingStreamingConnection.java | 7 ++++++-
.../cassandra/streaming/ConnectionHandler.java | 21 +++++++++++++-------
.../cassandra/streaming/StreamSession.java | 2 ++
5 files changed, 26 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index b9596d9,4b505f8..045b867
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,32 -1,5 +1,33 @@@
-2.1.14
+2.2.6
+ * Non-obsoleting compaction operations over compressed files can impose rate limit on normal reads (CASSANDRA-11301)
+ * Add missing newline at end of bin/cqlsh (CASSANDRA-11325)
+ * Fix AE in nodetool cfstats (backport CASSANDRA-10859) (CASSANDRA-11297)
+ * Unresolved hostname leads to replace being ignored (CASSANDRA-11210)
+ * Fix filtering on non-primary key columns for thrift static column families
+ (CASSANDRA-6377)
+ * Only log yaml config once, at startup (CASSANDRA-11217)
+ * Preserve order for preferred SSL cipher suites (CASSANDRA-11164)
+ * Reference leak with parallel repairs on the same table (CASSANDRA-11215)
+ * Range.compareTo() violates the contract of Comparable (CASSANDRA-11216)
+ * Avoid NPE when serializing ErrorMessage with null message (CASSANDRA-11167)
+ * Replacing an aggregate with a new version doesn't reset INITCOND (CASSANDRA-10840)
+ * (cqlsh) cqlsh cannot be called through symlink (CASSANDRA-11037)
+ * fix ohc and java-driver pom dependencies in build.xml (CASSANDRA-10793)
+ * Protect from keyspace dropped during repair (CASSANDRA-11065)
+ * Handle adding fields to a UDT in SELECT JSON and toJson() (CASSANDRA-11146)
+ * Better error message for cleanup (CASSANDRA-10991)
+ * cqlsh pg-style-strings broken if line ends with ';' (CASSANDRA-11123)
+ * Use cloned TokenMetadata in size estimates to avoid race against membership check
+ (CASSANDRA-10736)
+ * Always persist upsampled index summaries (CASSANDRA-10512)
+ * (cqlsh) Fix inconsistent auto-complete (CASSANDRA-10733)
+ * Make SELECT JSON and toJson() threadsafe (CASSANDRA-11048)
+ * Fix SELECT on tuple relations for mixed ASC/DESC clustering order (CASSANDRA-7281)
+ * (cqlsh) Support utf-8/cp65001 encoding on Windows (CASSANDRA-11030)
+ * Fix paging on DISTINCT queries repeats result when first row in partition changes
+ (CASSANDRA-10010)
+Merged from 2.1:
+ * Fix streaming_socket_timeout_in_ms not enforced (CASSANDRA-11286)
* Avoid dropping message too quickly due to missing unit conversion (CASSANDRA-11302)
* COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053)
* InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/conf/cassandra.yaml
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/ConnectionHandler.java
index 1ec7e1c,52268b2..d4662c7
--- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
+++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
@@@ -170,14 -175,10 +180,11 @@@ public class ConnectionHandle
protected static ReadableByteChannel getReadChannel(Socket socket) throws IOException
{
- ReadableByteChannel in = socket.getChannel();
- // socket channel is null when encrypted(SSL)
- return in == null
- ? Channels.newChannel(socket.getInputStream())
- : in;
+ //we do this instead of socket.getChannel() so socketSoTimeout is respected
+ return Channels.newChannel(socket.getInputStream());
}
+ @SuppressWarnings("resource")
public void sendInitMessage(Socket socket, boolean isForOutgoing) throws IOException
{
StreamInitMessage message = new StreamInitMessage(
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
[12/15] cassandra git commit: Merge branch 'cassandra-2.2' into
cassandra-3.0
Posted by yu...@apache.org.
Merge branch 'cassandra-2.2' into cassandra-3.0
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/c9d2b7ad
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/c9d2b7ad
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/c9d2b7ad
Branch: refs/heads/cassandra-3.0
Commit: c9d2b7addcfefb4cfa2bfcb96215153030d9d624
Parents: 232e12b b9ff7fe
Author: Yuki Morishita <yu...@apache.org>
Authored: Thu Mar 10 13:01:30 2016 -0600
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 10 13:01:30 2016 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
conf/cassandra.yaml | 7 +++----
.../net/IncomingStreamingConnection.java | 5 +++++
.../cassandra/streaming/ConnectionHandler.java | 21 +++++++++++++-------
.../cassandra/streaming/StreamSession.java | 2 ++
5 files changed, 25 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c9d2b7ad/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 7b93083,045b867..4c00994
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -56,8 -26,8 +56,9 @@@ Merged from 2.2
* (cqlsh) Support utf-8/cp65001 encoding on Windows (CASSANDRA-11030)
* Fix paging on DISTINCT queries repeats result when first row in partition changes
(CASSANDRA-10010)
+ * cqlsh: change default encoding to UTF-8 (CASSANDRA-11124)
Merged from 2.1:
+ * Fix streaming_socket_timeout_in_ms not enforced (CASSANDRA-11286)
* Avoid dropping message too quickly due to missing unit conversion (CASSANDRA-11302)
* COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053)
* InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c9d2b7ad/conf/cassandra.yaml
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c9d2b7ad/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
index e58d227,842676d..19e0671
--- a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
+++ b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
@@@ -25,8 -27,7 +25,9 @@@ import java.util.Set
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataInputPlus.DataInputStreamPlus;
+ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.streaming.StreamResultFuture;
import org.apache.cassandra.streaming.messages.StreamInitMessage;
import org.apache.cassandra.streaming.messages.StreamMessage;
@@@ -60,9 -60,13 +61,13 @@@ public class IncomingStreamingConnectio
if (version != StreamMessage.CURRENT_VERSION)
throw new IOException(String.format("Received stream using protocol version %d (my version %d). Terminating connection", version, MessagingService.current_version));
- DataInput input = new DataInputStream(socket.getInputStream());
+ DataInputPlus input = new DataInputStreamPlus(socket.getInputStream());
StreamInitMessage init = StreamInitMessage.serializer.deserialize(input, version);
+ //Set SO_TIMEOUT on follower side
+ if (!init.isForOutgoing)
+ socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
+
// The initiator makes two connections, one for incoming and one for outgoing.
// The receiving side distinguish two connections by looking at StreamInitMessage#isForOutgoing.
// Note: we cannot use the same socket for incoming and outgoing streams because we want to
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c9d2b7ad/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
[02/15] cassandra git commit: Fix streaming_socket_timeout_in_ms not
enforced
Posted by yu...@apache.org.
Fix streaming_socket_timeout_in_ms not enforced
Patch by Paulo Motta; reviewed by Yuki Morishita for CASSANDRA-11286
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/561000aa
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/561000aa
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/561000aa
Branch: refs/heads/cassandra-2.2
Commit: 561000aa3094699bab29766d9644ff50f6cb74f3
Parents: e94a2a0
Author: Paulo Motta <pa...@gmail.com>
Authored: Fri Feb 12 12:17:01 2016 -0300
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 10 12:54:24 2016 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
conf/cassandra.yaml | 7 +++----
.../net/IncomingStreamingConnection.java | 7 ++++++-
.../cassandra/streaming/ConnectionHandler.java | 21 +++++++++++++-------
.../cassandra/streaming/StreamSession.java | 2 ++
5 files changed, 26 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index e7c997a..4b505f8 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.1.14
+ * Fix streaming_socket_timeout_in_ms not enforced (CASSANDRA-11286)
* Avoid dropping message too quickly due to missing unit conversion (CASSANDRA-11302)
* COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053)
* InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/conf/cassandra.yaml
----------------------------------------------------------------------
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index 1fa04e6..0da4800 100644
--- a/conf/cassandra.yaml
+++ b/conf/cassandra.yaml
@@ -630,10 +630,9 @@ request_timeout_in_ms: 10000
# and the times are synchronized between the nodes.
cross_node_timeout: false
-# Enable socket timeout for streaming operation.
-# When a timeout occurs during streaming, streaming is retried from the start
-# of the current file. This _can_ involve re-streaming an important amount of
-# data, so you should avoid setting the value too low.
+# Set socket timeout for streaming operation.
+# The stream session is failed if no data is received by any of the
+# participants within that period.
# Default value is 3600000, which means streams timeout after an hour.
# streaming_socket_timeout_in_ms: 3600000
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
index 1f98bc4..5ced786 100644
--- a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
+++ b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
@@ -27,6 +27,7 @@ import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.streaming.StreamResultFuture;
import org.apache.cassandra.streaming.messages.StreamInitMessage;
import org.apache.cassandra.streaming.messages.StreamMessage;
@@ -62,6 +63,10 @@ public class IncomingStreamingConnection extends Thread implements Closeable
DataInput input = new DataInputStream(socket.getInputStream());
StreamInitMessage init = StreamInitMessage.serializer.deserialize(input, version);
+ //Set SO_TIMEOUT on follower side
+ if (!init.isForOutgoing)
+ socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
+
// The initiator makes two connections, one for incoming and one for outgoing.
// The receiving side distinguish two connections by looking at StreamInitMessage#isForOutgoing.
// Note: we cannot use the same socket for incoming and outgoing streams because we want to
@@ -74,7 +79,7 @@ public class IncomingStreamingConnection extends Thread implements Closeable
close();
}
}
-
+
@Override
public void close()
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/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 ac267f9..52268b2 100644
--- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
+++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
@@ -105,12 +105,22 @@ public class ConnectionHandler
{
logger.debug("[Stream #{}] Closing stream connection handler on {}", session.planId(), session.peer);
- ListenableFuture<?> inClosed = incoming == null ? Futures.immediateFuture(null) : incoming.close();
- ListenableFuture<?> outClosed = outgoing == null ? Futures.immediateFuture(null) : outgoing.close();
+ ListenableFuture<?> inClosed = closeIncoming();
+ ListenableFuture<?> outClosed = closeOutgoing();
return Futures.allAsList(inClosed, outClosed);
}
+ public ListenableFuture<?> closeOutgoing()
+ {
+ return outgoing == null ? Futures.immediateFuture(null) : outgoing.close();
+ }
+
+ public ListenableFuture<?> closeIncoming()
+ {
+ return incoming == null ? Futures.immediateFuture(null) : incoming.close();
+ }
+
/**
* Enqueue messages to be sent.
*
@@ -165,11 +175,8 @@ public class ConnectionHandler
protected static ReadableByteChannel getReadChannel(Socket socket) throws IOException
{
- ReadableByteChannel in = socket.getChannel();
- // socket channel is null when encrypted(SSL)
- return in == null
- ? Channels.newChannel(socket.getInputStream())
- : in;
+ //we do this instead of socket.getChannel() so socketSoTimeout is respected
+ return Channels.newChannel(socket.getInputStream());
}
public void sendInitMessage(Socket socket, boolean isForOutgoing) throws IOException
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/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 98a6f1f..642e837 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -609,6 +609,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
else
{
state(State.WAIT_COMPLETE);
+ handler.closeIncoming();
}
}
@@ -696,6 +697,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
handler.sendMessage(new CompleteMessage());
completeSent = true;
state(State.WAIT_COMPLETE);
+ handler.closeOutgoing();
}
}
return completed;
[14/15] cassandra git commit: Merge branch 'cassandra-3.0' into
cassandra-3.5
Posted by yu...@apache.org.
Merge branch 'cassandra-3.0' into cassandra-3.5
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/10d61d46
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/10d61d46
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/10d61d46
Branch: refs/heads/trunk
Commit: 10d61d46d36484ae0ae77108531528be5ba27619
Parents: f27ab29 c9d2b7a
Author: Yuki Morishita <yu...@apache.org>
Authored: Thu Mar 10 13:03:15 2016 -0600
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 10 13:03:15 2016 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
conf/cassandra.yaml | 7 +++----
.../net/IncomingStreamingConnection.java | 5 +++++
.../cassandra/streaming/ConnectionHandler.java | 21 +++++++++++++-------
.../cassandra/streaming/StreamSession.java | 2 ++
5 files changed, 25 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/10d61d46/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 68f5714,4c00994..292ebb5
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -88,9 -56,9 +88,10 @@@ Merged from 2.2
* (cqlsh) Support utf-8/cp65001 encoding on Windows (CASSANDRA-11030)
* Fix paging on DISTINCT queries repeats result when first row in partition changes
(CASSANDRA-10010)
+ * (cqlsh) Support timezone conversion using pytz (CASSANDRA-10397)
* cqlsh: change default encoding to UTF-8 (CASSANDRA-11124)
Merged from 2.1:
+ * Fix streaming_socket_timeout_in_ms not enforced (CASSANDRA-11286)
* Avoid dropping message too quickly due to missing unit conversion (CASSANDRA-11302)
* COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053)
* InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/10d61d46/conf/cassandra.yaml
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/10d61d46/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
[08/15] cassandra git commit: Merge branch 'cassandra-2.1' into
cassandra-2.2
Posted by yu...@apache.org.
Merge branch 'cassandra-2.1' into cassandra-2.2
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/b9ff7fe1
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/b9ff7fe1
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/b9ff7fe1
Branch: refs/heads/cassandra-3.0
Commit: b9ff7fe13740a21da25ed6414ea17fe5530652ea
Parents: 7e220bc 561000a
Author: Yuki Morishita <yu...@apache.org>
Authored: Thu Mar 10 13:00:09 2016 -0600
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 10 13:00:09 2016 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
conf/cassandra.yaml | 7 +++----
.../net/IncomingStreamingConnection.java | 7 ++++++-
.../cassandra/streaming/ConnectionHandler.java | 21 +++++++++++++-------
.../cassandra/streaming/StreamSession.java | 2 ++
5 files changed, 26 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index b9596d9,4b505f8..045b867
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,32 -1,5 +1,33 @@@
-2.1.14
+2.2.6
+ * Non-obsoleting compaction operations over compressed files can impose rate limit on normal reads (CASSANDRA-11301)
+ * Add missing newline at end of bin/cqlsh (CASSANDRA-11325)
+ * Fix AE in nodetool cfstats (backport CASSANDRA-10859) (CASSANDRA-11297)
+ * Unresolved hostname leads to replace being ignored (CASSANDRA-11210)
+ * Fix filtering on non-primary key columns for thrift static column families
+ (CASSANDRA-6377)
+ * Only log yaml config once, at startup (CASSANDRA-11217)
+ * Preserve order for preferred SSL cipher suites (CASSANDRA-11164)
+ * Reference leak with parallel repairs on the same table (CASSANDRA-11215)
+ * Range.compareTo() violates the contract of Comparable (CASSANDRA-11216)
+ * Avoid NPE when serializing ErrorMessage with null message (CASSANDRA-11167)
+ * Replacing an aggregate with a new version doesn't reset INITCOND (CASSANDRA-10840)
+ * (cqlsh) cqlsh cannot be called through symlink (CASSANDRA-11037)
+ * fix ohc and java-driver pom dependencies in build.xml (CASSANDRA-10793)
+ * Protect from keyspace dropped during repair (CASSANDRA-11065)
+ * Handle adding fields to a UDT in SELECT JSON and toJson() (CASSANDRA-11146)
+ * Better error message for cleanup (CASSANDRA-10991)
+ * cqlsh pg-style-strings broken if line ends with ';' (CASSANDRA-11123)
+ * Use cloned TokenMetadata in size estimates to avoid race against membership check
+ (CASSANDRA-10736)
+ * Always persist upsampled index summaries (CASSANDRA-10512)
+ * (cqlsh) Fix inconsistent auto-complete (CASSANDRA-10733)
+ * Make SELECT JSON and toJson() threadsafe (CASSANDRA-11048)
+ * Fix SELECT on tuple relations for mixed ASC/DESC clustering order (CASSANDRA-7281)
+ * (cqlsh) Support utf-8/cp65001 encoding on Windows (CASSANDRA-11030)
+ * Fix paging on DISTINCT queries repeats result when first row in partition changes
+ (CASSANDRA-10010)
+Merged from 2.1:
+ * Fix streaming_socket_timeout_in_ms not enforced (CASSANDRA-11286)
* Avoid dropping message too quickly due to missing unit conversion (CASSANDRA-11302)
* COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053)
* InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/conf/cassandra.yaml
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/ConnectionHandler.java
index 1ec7e1c,52268b2..d4662c7
--- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
+++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
@@@ -170,14 -175,10 +180,11 @@@ public class ConnectionHandle
protected static ReadableByteChannel getReadChannel(Socket socket) throws IOException
{
- ReadableByteChannel in = socket.getChannel();
- // socket channel is null when encrypted(SSL)
- return in == null
- ? Channels.newChannel(socket.getInputStream())
- : in;
+ //we do this instead of socket.getChannel() so socketSoTimeout is respected
+ return Channels.newChannel(socket.getInputStream());
}
+ @SuppressWarnings("resource")
public void sendInitMessage(Socket socket, boolean isForOutgoing) throws IOException
{
StreamInitMessage message = new StreamInitMessage(
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
[05/15] cassandra git commit: Fix streaming_socket_timeout_in_ms not
enforced
Posted by yu...@apache.org.
Fix streaming_socket_timeout_in_ms not enforced
Patch by Paulo Motta; reviewed by Yuki Morishita for CASSANDRA-11286
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/561000aa
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/561000aa
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/561000aa
Branch: refs/heads/cassandra-3.5
Commit: 561000aa3094699bab29766d9644ff50f6cb74f3
Parents: e94a2a0
Author: Paulo Motta <pa...@gmail.com>
Authored: Fri Feb 12 12:17:01 2016 -0300
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 10 12:54:24 2016 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
conf/cassandra.yaml | 7 +++----
.../net/IncomingStreamingConnection.java | 7 ++++++-
.../cassandra/streaming/ConnectionHandler.java | 21 +++++++++++++-------
.../cassandra/streaming/StreamSession.java | 2 ++
5 files changed, 26 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index e7c997a..4b505f8 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.1.14
+ * Fix streaming_socket_timeout_in_ms not enforced (CASSANDRA-11286)
* Avoid dropping message too quickly due to missing unit conversion (CASSANDRA-11302)
* COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053)
* InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/conf/cassandra.yaml
----------------------------------------------------------------------
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index 1fa04e6..0da4800 100644
--- a/conf/cassandra.yaml
+++ b/conf/cassandra.yaml
@@ -630,10 +630,9 @@ request_timeout_in_ms: 10000
# and the times are synchronized between the nodes.
cross_node_timeout: false
-# Enable socket timeout for streaming operation.
-# When a timeout occurs during streaming, streaming is retried from the start
-# of the current file. This _can_ involve re-streaming an important amount of
-# data, so you should avoid setting the value too low.
+# Set socket timeout for streaming operation.
+# The stream session is failed if no data is received by any of the
+# participants within that period.
# Default value is 3600000, which means streams timeout after an hour.
# streaming_socket_timeout_in_ms: 3600000
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
index 1f98bc4..5ced786 100644
--- a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
+++ b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
@@ -27,6 +27,7 @@ import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.streaming.StreamResultFuture;
import org.apache.cassandra.streaming.messages.StreamInitMessage;
import org.apache.cassandra.streaming.messages.StreamMessage;
@@ -62,6 +63,10 @@ public class IncomingStreamingConnection extends Thread implements Closeable
DataInput input = new DataInputStream(socket.getInputStream());
StreamInitMessage init = StreamInitMessage.serializer.deserialize(input, version);
+ //Set SO_TIMEOUT on follower side
+ if (!init.isForOutgoing)
+ socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
+
// The initiator makes two connections, one for incoming and one for outgoing.
// The receiving side distinguish two connections by looking at StreamInitMessage#isForOutgoing.
// Note: we cannot use the same socket for incoming and outgoing streams because we want to
@@ -74,7 +79,7 @@ public class IncomingStreamingConnection extends Thread implements Closeable
close();
}
}
-
+
@Override
public void close()
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/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 ac267f9..52268b2 100644
--- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
+++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
@@ -105,12 +105,22 @@ public class ConnectionHandler
{
logger.debug("[Stream #{}] Closing stream connection handler on {}", session.planId(), session.peer);
- ListenableFuture<?> inClosed = incoming == null ? Futures.immediateFuture(null) : incoming.close();
- ListenableFuture<?> outClosed = outgoing == null ? Futures.immediateFuture(null) : outgoing.close();
+ ListenableFuture<?> inClosed = closeIncoming();
+ ListenableFuture<?> outClosed = closeOutgoing();
return Futures.allAsList(inClosed, outClosed);
}
+ public ListenableFuture<?> closeOutgoing()
+ {
+ return outgoing == null ? Futures.immediateFuture(null) : outgoing.close();
+ }
+
+ public ListenableFuture<?> closeIncoming()
+ {
+ return incoming == null ? Futures.immediateFuture(null) : incoming.close();
+ }
+
/**
* Enqueue messages to be sent.
*
@@ -165,11 +175,8 @@ public class ConnectionHandler
protected static ReadableByteChannel getReadChannel(Socket socket) throws IOException
{
- ReadableByteChannel in = socket.getChannel();
- // socket channel is null when encrypted(SSL)
- return in == null
- ? Channels.newChannel(socket.getInputStream())
- : in;
+ //we do this instead of socket.getChannel() so socketSoTimeout is respected
+ return Channels.newChannel(socket.getInputStream());
}
public void sendInitMessage(Socket socket, boolean isForOutgoing) throws IOException
http://git-wip-us.apache.org/repos/asf/cassandra/blob/561000aa/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 98a6f1f..642e837 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -609,6 +609,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
else
{
state(State.WAIT_COMPLETE);
+ handler.closeIncoming();
}
}
@@ -696,6 +697,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
handler.sendMessage(new CompleteMessage());
completeSent = true;
state(State.WAIT_COMPLETE);
+ handler.closeOutgoing();
}
}
return completed;
[09/15] cassandra git commit: Merge branch 'cassandra-2.1' into
cassandra-2.2
Posted by yu...@apache.org.
Merge branch 'cassandra-2.1' into cassandra-2.2
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/b9ff7fe1
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/b9ff7fe1
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/b9ff7fe1
Branch: refs/heads/cassandra-3.5
Commit: b9ff7fe13740a21da25ed6414ea17fe5530652ea
Parents: 7e220bc 561000a
Author: Yuki Morishita <yu...@apache.org>
Authored: Thu Mar 10 13:00:09 2016 -0600
Committer: Yuki Morishita <yu...@apache.org>
Committed: Thu Mar 10 13:00:09 2016 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
conf/cassandra.yaml | 7 +++----
.../net/IncomingStreamingConnection.java | 7 ++++++-
.../cassandra/streaming/ConnectionHandler.java | 21 +++++++++++++-------
.../cassandra/streaming/StreamSession.java | 2 ++
5 files changed, 26 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index b9596d9,4b505f8..045b867
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,32 -1,5 +1,33 @@@
-2.1.14
+2.2.6
+ * Non-obsoleting compaction operations over compressed files can impose rate limit on normal reads (CASSANDRA-11301)
+ * Add missing newline at end of bin/cqlsh (CASSANDRA-11325)
+ * Fix AE in nodetool cfstats (backport CASSANDRA-10859) (CASSANDRA-11297)
+ * Unresolved hostname leads to replace being ignored (CASSANDRA-11210)
+ * Fix filtering on non-primary key columns for thrift static column families
+ (CASSANDRA-6377)
+ * Only log yaml config once, at startup (CASSANDRA-11217)
+ * Preserve order for preferred SSL cipher suites (CASSANDRA-11164)
+ * Reference leak with parallel repairs on the same table (CASSANDRA-11215)
+ * Range.compareTo() violates the contract of Comparable (CASSANDRA-11216)
+ * Avoid NPE when serializing ErrorMessage with null message (CASSANDRA-11167)
+ * Replacing an aggregate with a new version doesn't reset INITCOND (CASSANDRA-10840)
+ * (cqlsh) cqlsh cannot be called through symlink (CASSANDRA-11037)
+ * fix ohc and java-driver pom dependencies in build.xml (CASSANDRA-10793)
+ * Protect from keyspace dropped during repair (CASSANDRA-11065)
+ * Handle adding fields to a UDT in SELECT JSON and toJson() (CASSANDRA-11146)
+ * Better error message for cleanup (CASSANDRA-10991)
+ * cqlsh pg-style-strings broken if line ends with ';' (CASSANDRA-11123)
+ * Use cloned TokenMetadata in size estimates to avoid race against membership check
+ (CASSANDRA-10736)
+ * Always persist upsampled index summaries (CASSANDRA-10512)
+ * (cqlsh) Fix inconsistent auto-complete (CASSANDRA-10733)
+ * Make SELECT JSON and toJson() threadsafe (CASSANDRA-11048)
+ * Fix SELECT on tuple relations for mixed ASC/DESC clustering order (CASSANDRA-7281)
+ * (cqlsh) Support utf-8/cp65001 encoding on Windows (CASSANDRA-11030)
+ * Fix paging on DISTINCT queries repeats result when first row in partition changes
+ (CASSANDRA-10010)
+Merged from 2.1:
+ * Fix streaming_socket_timeout_in_ms not enforced (CASSANDRA-11286)
* Avoid dropping message too quickly due to missing unit conversion (CASSANDRA-11302)
* COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053)
* InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/conf/cassandra.yaml
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/ConnectionHandler.java
index 1ec7e1c,52268b2..d4662c7
--- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
+++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
@@@ -170,14 -175,10 +180,11 @@@ public class ConnectionHandle
protected static ReadableByteChannel getReadChannel(Socket socket) throws IOException
{
- ReadableByteChannel in = socket.getChannel();
- // socket channel is null when encrypted(SSL)
- return in == null
- ? Channels.newChannel(socket.getInputStream())
- : in;
+ //we do this instead of socket.getChannel() so socketSoTimeout is respected
+ return Channels.newChannel(socket.getInputStream());
}
+ @SuppressWarnings("resource")
public void sendInitMessage(Socket socket, boolean isForOutgoing) throws IOException
{
StreamInitMessage message = new StreamInitMessage(
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b9ff7fe1/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------