You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by sr...@apache.org on 2016/03/13 02:55:46 UTC
[1/5] storm git commit: netty loss of messages resolution
Repository: storm
Updated Branches:
refs/heads/master 71733add4 -> 5f818b9ad
netty loss of messages resolution
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/54b6ac49
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/54b6ac49
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/54b6ac49
Branch: refs/heads/master
Commit: 54b6ac4975582bb6b3ef3511369e52ad81db05b3
Parents: b477939
Author: Sanket <sc...@untilservice-lm>
Authored: Tue Mar 8 13:06:39 2016 -0600
Committer: Sanket <sc...@untilservice-lm>
Committed: Tue Mar 8 13:06:39 2016 -0600
----------------------------------------------------------------------
.../apache/storm/messaging/netty/Client.java | 38 ++++++++++++++++++++
1 file changed, 38 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/54b6ac49/storm-core/src/jvm/org/apache/storm/messaging/netty/Client.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/messaging/netty/Client.java b/storm-core/src/jvm/org/apache/storm/messaging/netty/Client.java
index 976b550..3e15d34 100644
--- a/storm-core/src/jvm/org/apache/storm/messaging/netty/Client.java
+++ b/storm-core/src/jvm/org/apache/storm/messaging/netty/Client.java
@@ -23,6 +23,7 @@ import java.util.Iterator;
import java.util.Collection;
import java.util.Map;
import java.util.HashMap;
+import java.util.Timer;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -74,6 +75,7 @@ public class Client extends ConnectionWithStatus implements IStatefulObject, ISa
private static final Logger LOG = LoggerFactory.getLogger(Client.class);
private static final String PREFIX = "Netty-Client-";
private static final long NO_DELAY_MS = 0L;
+ private static Timer timer;
private final Map stormConf;
private final StormBoundedExponentialBackoffRetry retryPolicy;
@@ -108,6 +110,13 @@ public class Client extends ConnectionWithStatus implements IStatefulObject, ISa
private final AtomicInteger messagesLost = new AtomicInteger(0);
/**
+ * Periodically checks for connected channel in order to avoid loss
+ * of messages
+ */
+ private final long CHANNEL_ALIVE_INTERVAL_MS = 30000L;
+
+
+ /**
* Number of messages buffered in memory.
*/
private final AtomicLong pendingMessages = new AtomicLong(0);
@@ -130,6 +139,10 @@ public class Client extends ConnectionWithStatus implements IStatefulObject, ISa
private final Object writeLock = new Object();
+ static {
+ timer = new Timer("Netty-ChannelAlive-Timer", true);
+ }
+
@SuppressWarnings("rawtypes")
Client(Map stormConf, ChannelFactory factory, HashedWheelTimer scheduler, String host, int port, Context context) {
this.stormConf = stormConf;
@@ -151,10 +164,35 @@ public class Client extends ConnectionWithStatus implements IStatefulObject, ISa
bootstrap = createClientBootstrap(factory, bufferSize, stormConf);
dstAddress = new InetSocketAddress(host, port);
dstAddressPrefixedName = prefixedName(dstAddress);
+ launchChannelAliveThread();
scheduleConnect(NO_DELAY_MS);
batcher = new MessageBuffer(messageBatchSize);
}
+ /**
+ * This thread helps us to check for channel connection periodically.
+ * This is performed just to know whether the destination address
+ * is alive or attempts to refresh connections if not alive. This
+ * solution is better than what we have now in case of a bad channel.
+ */
+ private void launchChannelAliveThread() {
+ // netty TimerTask is already defined and hence a fully
+ // qualified name
+ timer.schedule(new java.util.TimerTask() {
+ public void run() {
+ try {
+ LOG.debug("running timer task, address {}", dstAddress);
+ if(closing) {
+ this.cancel();
+ }
+ getConnectedChannel();
+ } catch (Exception exp) {
+ LOG.error("channel connection error {}", exp);
+ }
+ }
+ }, 0, CHANNEL_ALIVE_INTERVAL_MS);
+ }
+
private ClientBootstrap createClientBootstrap(ChannelFactory factory, int bufferSize, Map stormConf) {
ClientBootstrap bootstrap = new ClientBootstrap(factory);
bootstrap.setOption("tcpNoDelay", true);
[2/5] storm git commit: removed the timer object in static block
Posted by sr...@apache.org.
removed the timer object in static block
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/645e4a9d
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/645e4a9d
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/645e4a9d
Branch: refs/heads/master
Commit: 645e4a9d2271efcda897def11ffbb45a1ed5780d
Parents: 54b6ac4
Author: Sanket <sc...@untilservice-lm>
Authored: Tue Mar 8 16:34:49 2016 -0600
Committer: Sanket <sc...@untilservice-lm>
Committed: Tue Mar 8 16:34:49 2016 -0600
----------------------------------------------------------------------
.../jvm/org/apache/storm/messaging/netty/Client.java | 13 ++++++++-----
1 file changed, 8 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/645e4a9d/storm-core/src/jvm/org/apache/storm/messaging/netty/Client.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/messaging/netty/Client.java b/storm-core/src/jvm/org/apache/storm/messaging/netty/Client.java
index 3e15d34..77c2bf5 100644
--- a/storm-core/src/jvm/org/apache/storm/messaging/netty/Client.java
+++ b/storm-core/src/jvm/org/apache/storm/messaging/netty/Client.java
@@ -115,7 +115,6 @@ public class Client extends ConnectionWithStatus implements IStatefulObject, ISa
*/
private final long CHANNEL_ALIVE_INTERVAL_MS = 30000L;
-
/**
* Number of messages buffered in memory.
*/
@@ -139,10 +138,6 @@ public class Client extends ConnectionWithStatus implements IStatefulObject, ISa
private final Object writeLock = new Object();
- static {
- timer = new Timer("Netty-ChannelAlive-Timer", true);
- }
-
@SuppressWarnings("rawtypes")
Client(Map stormConf, ChannelFactory factory, HashedWheelTimer scheduler, String host, int port, Context context) {
this.stormConf = stormConf;
@@ -178,12 +173,20 @@ public class Client extends ConnectionWithStatus implements IStatefulObject, ISa
private void launchChannelAliveThread() {
// netty TimerTask is already defined and hence a fully
// qualified name
+ if (timer == null) {
+ synchronized (Client.class) {
+ if (timer == null) {
+ timer = new Timer("Netty-ChannelAlive-Timer", true);
+ }
+ }
+ }
timer.schedule(new java.util.TimerTask() {
public void run() {
try {
LOG.debug("running timer task, address {}", dstAddress);
if(closing) {
this.cancel();
+ return;
}
getConnectedChannel();
} catch (Exception exp) {
[4/5] storm git commit: Merge branch 'netty-best-effort' of
https://github.com/redsanket/storm into STORM-1609
Posted by sr...@apache.org.
Merge branch 'netty-best-effort' of https://github.com/redsanket/storm into STORM-1609
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/a908e884
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/a908e884
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/a908e884
Branch: refs/heads/master
Commit: a908e8841ccac74406aeb0f68b68871f74099c1a
Parents: 71733ad 072fcc8
Author: Sriharsha Chintalapani <ha...@hortonworks.com>
Authored: Sat Mar 12 17:47:10 2016 -0800
Committer: Sriharsha Chintalapani <ha...@hortonworks.com>
Committed: Sat Mar 12 17:47:10 2016 -0800
----------------------------------------------------------------------
.../apache/storm/messaging/netty/Client.java | 34 ++++++++++++++++++++
1 file changed, 34 insertions(+)
----------------------------------------------------------------------
[3/5] storm git commit: avoid race condition double check locking
Posted by sr...@apache.org.
avoid race condition double check locking
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/072fcc88
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/072fcc88
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/072fcc88
Branch: refs/heads/master
Commit: 072fcc8865d5e2ebc606ab5b88dad5ff0ef551d4
Parents: 645e4a9
Author: Sanket <sc...@untilservice-lm>
Authored: Wed Mar 9 12:07:52 2016 -0600
Committer: Sanket <sc...@untilservice-lm>
Committed: Wed Mar 9 12:07:52 2016 -0600
----------------------------------------------------------------------
.../src/jvm/org/apache/storm/messaging/netty/Client.java | 9 +--------
1 file changed, 1 insertion(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/072fcc88/storm-core/src/jvm/org/apache/storm/messaging/netty/Client.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/messaging/netty/Client.java b/storm-core/src/jvm/org/apache/storm/messaging/netty/Client.java
index 77c2bf5..035eb1b 100644
--- a/storm-core/src/jvm/org/apache/storm/messaging/netty/Client.java
+++ b/storm-core/src/jvm/org/apache/storm/messaging/netty/Client.java
@@ -75,7 +75,7 @@ public class Client extends ConnectionWithStatus implements IStatefulObject, ISa
private static final Logger LOG = LoggerFactory.getLogger(Client.class);
private static final String PREFIX = "Netty-Client-";
private static final long NO_DELAY_MS = 0L;
- private static Timer timer;
+ private static final Timer timer = new Timer("Netty-ChannelAlive-Timer", true);
private final Map stormConf;
private final StormBoundedExponentialBackoffRetry retryPolicy;
@@ -173,13 +173,6 @@ public class Client extends ConnectionWithStatus implements IStatefulObject, ISa
private void launchChannelAliveThread() {
// netty TimerTask is already defined and hence a fully
// qualified name
- if (timer == null) {
- synchronized (Client.class) {
- if (timer == null) {
- timer = new Timer("Netty-ChannelAlive-Timer", true);
- }
- }
- }
timer.schedule(new java.util.TimerTask() {
public void run() {
try {
[5/5] storm git commit: Added STORM-1609 to CHANGELOG.
Posted by sr...@apache.org.
Added STORM-1609 to CHANGELOG.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/5f818b9a
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/5f818b9a
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/5f818b9a
Branch: refs/heads/master
Commit: 5f818b9ad5c8f8b302720ecb68a84f15b267d7a2
Parents: a908e88
Author: Sriharsha Chintalapani <ha...@hortonworks.com>
Authored: Sat Mar 12 17:49:29 2016 -0800
Committer: Sriharsha Chintalapani <ha...@hortonworks.com>
Committed: Sat Mar 12 17:49:29 2016 -0800
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/5f818b9a/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3258f7f..4f31ec3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -54,6 +54,7 @@
* STORM-1521: When using Kerberos login from keytab with multiple bolts/executors ticket is not renewed in hbase bolt.
## 1.0.0
+ * STORM-1609: Netty Client is not best effort delivery on failed Connection
* STORM-1620: Update curator to fix CURATOR-209
* STORM-1469: Adding Plain Sasl Transport Plugin
* STORM-1588: Do not add event logger details if number of event loggers is zero