You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by zhuoliu <gi...@git.apache.org> on 2016/02/25 00:01:26 UTC
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
GitHub user zhuoliu opened a pull request:
https://github.com/apache/storm/pull/1149
[STORM-1574] Better handle backpressure thread exception & clear dir.
1. Better handle exception.
2. Remove topo dir during kill.
3. Add a unit test.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/zhuoliu/storm 1574
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/1149.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1149
----
commit ee1a51993a4244dd6a3a112f70db0c8d74770557
Author: zhuol <zh...@yahoo-inc.com>
Date: 2016-02-24T22:56:39Z
[STORM-1574] Better handle backpressure thread exception and clear topo dir.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/storm/pull/1149
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by wuchong <gi...@git.apache.org>.
Github user wuchong commented on a diff in the pull request:
https://github.com/apache/storm/pull/1149#discussion_r54043672
--- Diff: storm-core/test/jvm/org/apache/storm/utils/WorkerBackpressureThreadTest.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.storm.utils;
+
+import java.util.concurrent.atomic.AtomicLong;
+import org.junit.Assert;
+import org.junit.Test;
+import junit.framework.TestCase;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class WorkerBackpressureThreadTest extends TestCase {
+ private static final Logger LOG = LoggerFactory.getLogger(WorkerBackpressureThreadTest.class);
+
+ @Test
+ public void testNormalEvent() throws Exception {
+ Object trigger = new Object();
+ AtomicLong workerData = new AtomicLong(0);
+ WorkerBackpressureCallback callback = new WorkerBackpressureCallback() {
+ @Override
+ public void onEvent(Object obj) {
+ ((AtomicLong) obj).getAndDecrement();
+ }
+ };
+ WorkerBackpressureThread workerBackpressureThread = new WorkerBackpressureThread(trigger, workerData, callback);
+ workerBackpressureThread.start();
+ Thread.sleep(100);
+ WorkerBackpressureThread.notifyBackpressureChecker(trigger);
+ Thread.sleep(100);
+ Assert.assertNotEquals("Check the calling times of backpressure events, should not be 0. ",
+ workerData.get(), 0);
+ }
+
+ @Test
+ public void testThrowRuntimeExceptionEvent() throws Exception {
+ Object trigger = new Object();
+ Object workerData = new Object();
+ WorkerBackpressureCallback callback = new WorkerBackpressureCallback() {
+ @Override
+ public void onEvent(Object obj) {
+ throw new RuntimeException();
+ }
+ };
+ WorkerBackpressureThread workerBackpressureThread = new WorkerBackpressureThread(trigger, workerData, callback);
+ workerBackpressureThread.start();
+ Thread.sleep(100);
+ WorkerBackpressureThread.notifyBackpressureChecker(trigger);
+ Thread.sleep(100);
--- End diff --
The CI failed maybe because the sleep time is too short.
I set the sleep time > 500 , and the workerBackpressureThread is dead as expect.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1149#discussion_r54049182
--- Diff: storm-core/src/jvm/org/apache/storm/utils/WorkerBackpressureThread.java ---
@@ -43,17 +48,33 @@ static public void notifyBackpressureChecker(Object trigger) {
}
}
+ public void terminate() {
+ running = false;
+ }
+
public void run() {
- try {
- while (true) {
+ while (running) {
+ try {
synchronized(trigger) {
trigger.wait(100);
}
callback.onEvent(workerData); // check all executors and update zk backpressure throttle for the worker if needed
+ } catch (InterruptedException interEx) {
+ LOG.info("WorkerBackpressureThread gets interrupted! Ignoring Exception: ", interEx);
}
- } catch (Exception e) {
- throw new RuntimeException(e);
}
}
}
+class BackpressureUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(BackpressureUncaughtExceptionHandler.class);
+ @Override
+ public void uncaughtException(Thread t, Throwable e) {
+ try {
+ Utils.handleUncaughtException(e);
+ } catch (Error error) {
+ LOG.info("Received error in WorkerBackpressureThread.. terminating the worker...");
--- End diff --
This should be LOG.error and error can be passed as second argument.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by kishorvpatil <gi...@git.apache.org>.
Github user kishorvpatil commented on the pull request:
https://github.com/apache/storm/pull/1149#issuecomment-189025035
LGTM. +1
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1149#discussion_r54110183
--- Diff: storm-core/src/jvm/org/apache/storm/utils/WorkerBackpressureThread.java ---
@@ -43,17 +48,33 @@ static public void notifyBackpressureChecker(Object trigger) {
}
}
+ public void terminate() {
+ running = false;
+ }
+
public void run() {
- try {
- while (true) {
+ while (running) {
+ try {
synchronized(trigger) {
trigger.wait(100);
}
callback.onEvent(workerData); // check all executors and update zk backpressure throttle for the worker if needed
+ } catch (InterruptedException interEx) {
+ LOG.info("WorkerBackpressureThread gets interrupted! Ignoring Exception: ", interEx);
}
- } catch (Exception e) {
- throw new RuntimeException(e);
}
}
}
+class BackpressureUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(BackpressureUncaughtExceptionHandler.class);
+ @Override
+ public void uncaughtException(Thread t, Throwable e) {
+ try {
+ Utils.handleUncaughtException(e);
--- End diff --
This ignores all RuntimeExceptions. This is fine for what it was originally written for, which was around thread pools. But in this case any RuntimeException is going to still cause the backpressure thread to exit.
@kishorvpatil would you be OK if we renamed handeUncaughtException to be more descriptive, because it is deceptive right now, and I think will cause bug while we translate code to java. Can we call it eatRuntimeExitOnOOM.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1149#discussion_r54107806
--- Diff: storm-core/src/jvm/org/apache/storm/utils/WorkerBackpressureThread.java ---
@@ -43,17 +48,33 @@ static public void notifyBackpressureChecker(Object trigger) {
}
}
+ public void terminate() {
+ running = false;
+ }
--- End diff --
Could we do an interrupt and a join after this?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on the pull request:
https://github.com/apache/storm/pull/1149#issuecomment-189022796
@zhuoliu yes you will need a separate pull request for 1.x
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by zhuoliu <gi...@git.apache.org>.
Github user zhuoliu commented on a diff in the pull request:
https://github.com/apache/storm/pull/1149#discussion_r54172980
--- Diff: storm-core/src/clj/org/apache/storm/daemon/worker.clj ---
@@ -155,7 +155,10 @@
;; update the worker's backpressure flag to zookeeper only when it has changed
(log-debug "BP " @(:backpressure worker) " WAS " prev-backpressure-flag)
(when (not= prev-backpressure-flag @(:backpressure worker))
- (.workerBackpressure storm-cluster-state storm-id assignment-id port @(:backpressure worker)))
+ (try
+ (.workerBackpressure storm-cluster-state storm-id assignment-id (long port) @(:backpressure worker))
--- End diff --
(:port worker) returns an Integer. Will cause an exception if not turning to long. See also Storm-1578.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by zhuoliu <gi...@git.apache.org>.
Github user zhuoliu commented on the pull request:
https://github.com/apache/storm/pull/1149#issuecomment-188898763
@revans2 Updated the exception handling.
Filed a JIRA for the Utils.uncaughtExceptionHandler renaming here:
https://issues.apache.org/jira/browse/STORM-1577
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by zhuoliu <gi...@git.apache.org>.
Github user zhuoliu commented on the pull request:
https://github.com/apache/storm/pull/1149#issuecomment-188939301
Since we will ignore all exception with onEvent in worker.clj, we should not need to test RuntimeException here. So delete it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by hustfxj <gi...@git.apache.org>.
Github user hustfxj commented on the pull request:
https://github.com/apache/storm/pull/1149#issuecomment-188537244
+1 It looks good to me
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on the pull request:
https://github.com/apache/storm/pull/1149#issuecomment-188931704
The changes look good, but I would like to see the tests updated still. I am not sure if they will still pass without some changes. Specifically around the uncaught exception handler. It would be nice to have an API so we can mock that out.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by zhuoliu <gi...@git.apache.org>.
Github user zhuoliu commented on the pull request:
https://github.com/apache/storm/pull/1149#issuecomment-189018954
Also, I need to put up another pull request for 1.x.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1149#discussion_r54107946
--- Diff: storm-core/src/jvm/org/apache/storm/utils/WorkerBackpressureThread.java ---
@@ -43,17 +48,33 @@ static public void notifyBackpressureChecker(Object trigger) {
}
}
+ public void terminate() {
+ running = false;
+ }
--- End diff --
We could then ignore the interrupted exception in run as it would just be used as a trigger to wake up the running thread.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by zhuoliu <gi...@git.apache.org>.
Github user zhuoliu commented on the pull request:
https://github.com/apache/storm/pull/1149#issuecomment-189009158
@revans2 updated the test with timeout, and fixed the Cast exception (Int/Long) in Storm-1578, also manual test works fine.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on the pull request:
https://github.com/apache/storm/pull/1149#issuecomment-189022383
+1
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by wuchong <gi...@git.apache.org>.
Github user wuchong commented on the pull request:
https://github.com/apache/storm/pull/1149#issuecomment-188582664
+1
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1149#discussion_r54109478
--- Diff: storm-core/test/jvm/org/apache/storm/utils/WorkerBackpressureThreadTest.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.storm.utils;
+
+import java.util.concurrent.atomic.AtomicLong;
+import org.junit.Assert;
+import org.junit.Test;
+import junit.framework.TestCase;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class WorkerBackpressureThreadTest extends TestCase {
+ private static final Logger LOG = LoggerFactory.getLogger(WorkerBackpressureThreadTest.class);
+
+ @Test
+ public void testNormalEvent() throws Exception {
+ Object trigger = new Object();
+ AtomicLong workerData = new AtomicLong(0);
+ WorkerBackpressureCallback callback = new WorkerBackpressureCallback() {
+ @Override
+ public void onEvent(Object obj) {
+ ((AtomicLong) obj).getAndDecrement();
+ }
+ };
+ WorkerBackpressureThread workerBackpressureThread = new WorkerBackpressureThread(trigger, workerData, callback);
+ workerBackpressureThread.start();
+ Thread.sleep(100);
+ WorkerBackpressureThread.notifyBackpressureChecker(trigger);
+ Thread.sleep(100);
+ Assert.assertNotEquals("Check the calling times of backpressure events, should not be 0. ",
+ workerData.get(), 0);
+ }
+
+ @Test
+ public void testThrowRuntimeExceptionEvent() throws Exception {
+ Object trigger = new Object();
+ Object workerData = new Object();
+ WorkerBackpressureCallback callback = new WorkerBackpressureCallback() {
+ @Override
+ public void onEvent(Object obj) {
+ throw new RuntimeException();
+ }
+ };
+ WorkerBackpressureThread workerBackpressureThread = new WorkerBackpressureThread(trigger, workerData, callback);
+ workerBackpressureThread.start();
+ Thread.sleep(100);
+ WorkerBackpressureThread.notifyBackpressureChecker(trigger);
+ Thread.sleep(100);
--- End diff --
I really hate hard coded sleeps in tests. Could we please replace this with something like
```
workerBackpressureThread.start();
long start = System.currentTimeMillis();
while (!workerBackpressureThread.isAlive()) {
assertTrue("Timeout", (System.currentTimeMillis() - start) < 1000);
Thread.sleep(100);
}
WorkerBackpressureThread.notifyBackpressureChecker(trigger);
long start = System.currentTimeMillis();
while (workerBackpressureThread.isAlive()) {
assertTrue("Timeout", (System.currentTimeMillis() - start) < 1000);
Thread.sleep(100);
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on the pull request:
https://github.com/apache/storm/pull/1149#issuecomment-188972526
I still would like to see the hard coded sleeps replaced by polling with a timeout.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1574] Better handle backpressure thread...
Posted by hustfxj <gi...@git.apache.org>.
Github user hustfxj commented on the pull request:
https://github.com/apache/storm/pull/1149#issuecomment-188543268
Some Tests failed. It prompts tests can't find remove-backpressure!. I guess it generate some questions when it merge with master.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---