You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2017/01/24 03:42:45 UTC
[1/3] storm git commit: STORM-2298 Don't kill Nimbus when
ClusterMetricsConsumer is failed to initialize
Repository: storm
Updated Branches:
refs/heads/master 7e02b71ea -> 1811273f1
STORM-2298 Don't kill Nimbus when ClusterMetricsConsumer is failed to initialize
* don't propagate errors, but just log error to guide reconfiguration / check the system
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/2e9e9d63
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/2e9e9d63
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/2e9e9d63
Branch: refs/heads/master
Commit: 2e9e9d63f8bd5a712f73b28a083b50c1df11c9d1
Parents: bb533a9
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Tue Jan 17 23:51:27 2017 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Thu Jan 19 10:33:32 2017 +0900
----------------------------------------------------------------------
.../metric/ClusterMetricsConsumerExecutor.java | 31 ++++-
.../ClusterMetricsConsumerExecutorTest.java | 133 +++++++++++++++++++
2 files changed, 158 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/2e9e9d63/storm-core/src/jvm/org/apache/storm/metric/ClusterMetricsConsumerExecutor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/metric/ClusterMetricsConsumerExecutor.java b/storm-core/src/jvm/org/apache/storm/metric/ClusterMetricsConsumerExecutor.java
index 1102ded..2bf31c4 100644
--- a/storm-core/src/jvm/org/apache/storm/metric/ClusterMetricsConsumerExecutor.java
+++ b/storm-core/src/jvm/org/apache/storm/metric/ClusterMetricsConsumerExecutor.java
@@ -17,7 +17,6 @@
*/
package org.apache.storm.metric;
-import org.apache.storm.Config;
import org.apache.storm.metric.api.DataPoint;
import org.apache.storm.metric.api.IClusterMetricsConsumer;
import org.slf4j.Logger;
@@ -27,6 +26,10 @@ import java.util.Collection;
public class ClusterMetricsConsumerExecutor {
public static final Logger LOG = LoggerFactory.getLogger(ClusterMetricsConsumerExecutor.class);
+ private static final String ERROR_MESSAGE_PREPARATION_CLUSTER_METRICS_CONSUMER_FAILED =
+ "Preparation of Cluster Metrics Consumer failed. " +
+ "Please check your configuration and/or corresponding systems and relaunch Nimbus. " +
+ "Skipping handle metrics.";
private IClusterMetricsConsumer metricsConsumer;
private String consumerClassName;
@@ -40,15 +43,24 @@ public class ClusterMetricsConsumerExecutor {
public void prepare() {
try {
metricsConsumer = (IClusterMetricsConsumer)Class.forName(consumerClassName).newInstance();
+ metricsConsumer.prepare(registrationArgument);
} catch (Exception e) {
- throw new RuntimeException("Could not instantiate a class listed in config under section " +
- Config.STORM_CLUSTER_METRICS_CONSUMER_REGISTER + " with fully qualified name " + consumerClassName, e);
- }
+ LOG.error("Could not instantiate or prepare Cluster Metrics Consumer with fully qualified name " +
+ consumerClassName, e);
- metricsConsumer.prepare(registrationArgument);
+ if (metricsConsumer != null) {
+ metricsConsumer.cleanup();
+ }
+ metricsConsumer = null;
+ }
}
public void handleDataPoints(final IClusterMetricsConsumer.ClusterInfo clusterInfo, final Collection<DataPoint> dataPoints) {
+ if (metricsConsumer == null) {
+ LOG.error(ERROR_MESSAGE_PREPARATION_CLUSTER_METRICS_CONSUMER_FAILED);
+ return;
+ }
+
try {
metricsConsumer.handleDataPoints(clusterInfo, dataPoints);
} catch (Throwable e) {
@@ -57,6 +69,11 @@ public class ClusterMetricsConsumerExecutor {
}
public void handleDataPoints(final IClusterMetricsConsumer.SupervisorInfo supervisorInfo, final Collection<DataPoint> dataPoints) {
+ if (metricsConsumer == null) {
+ LOG.error(ERROR_MESSAGE_PREPARATION_CLUSTER_METRICS_CONSUMER_FAILED);
+ return;
+ }
+
try {
metricsConsumer.handleDataPoints(supervisorInfo, dataPoints);
} catch (Throwable e) {
@@ -65,6 +82,8 @@ public class ClusterMetricsConsumerExecutor {
}
public void cleanup() {
- metricsConsumer.cleanup();
+ if (metricsConsumer != null) {
+ metricsConsumer.cleanup();
+ }
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/2e9e9d63/storm-core/test/jvm/org/apache/storm/metric/ClusterMetricsConsumerExecutorTest.java
----------------------------------------------------------------------
diff --git a/storm-core/test/jvm/org/apache/storm/metric/ClusterMetricsConsumerExecutorTest.java b/storm-core/test/jvm/org/apache/storm/metric/ClusterMetricsConsumerExecutorTest.java
new file mode 100644
index 0000000..b8b3afd
--- /dev/null
+++ b/storm-core/test/jvm/org/apache/storm/metric/ClusterMetricsConsumerExecutorTest.java
@@ -0,0 +1,133 @@
+/**
+ * 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.metric;
+
+import org.apache.storm.metric.api.DataPoint;
+import org.apache.storm.metric.api.IClusterMetricsConsumer;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Collection;
+import java.util.Collections;
+
+import static org.mockito.Mockito.mock;
+
+public class ClusterMetricsConsumerExecutorTest {
+
+ @Before
+ public void setUp() {
+ MockFailingClusterMetricsConsumer.resetAllCounts();
+ }
+
+ @Test
+ public void testPrepareDoesNotThrowExceptionWhenInitializingClusterMetricsConsumerIsFailing() throws Exception {
+ ClusterMetricsConsumerExecutor sut = new ClusterMetricsConsumerExecutor(
+ MockFailingClusterMetricsConsumer.class.getName(), 2);
+
+ // it shouldn't propagate any exceptions
+ sut.prepare();
+ sut.prepare();
+
+ Assert.assertEquals(2, MockFailingClusterMetricsConsumer.getPrepareCallCount());
+ }
+
+ @Test
+ public void testHandleDataPointsWithClusterMetricsShouldSkipHandlingMetricsIfFailedBefore() throws Exception {
+ ClusterMetricsConsumerExecutor sut = new ClusterMetricsConsumerExecutor(
+ MockFailingClusterMetricsConsumer.class.getName(), 2);
+
+ // below calls shouldn't propagate any exceptions
+ sut.prepare();
+
+ // no specific reason to mock... this is one of easiest ways to make dummy instance
+ sut.handleDataPoints(mock(IClusterMetricsConsumer.ClusterInfo.class), Collections.emptyList());
+
+ Assert.assertEquals(1, MockFailingClusterMetricsConsumer.getPrepareCallCount());
+ Assert.assertEquals(0, MockFailingClusterMetricsConsumer.getHandleDataPointsWithClusterInfoCallCount());
+ }
+
+ @Test
+ public void testHandleDataPointsWithSupervisorMetricsShouldRetryInitializingClusterMetricsConsumerIfFailedBefore() throws Exception {
+ ClusterMetricsConsumerExecutor sut = new ClusterMetricsConsumerExecutor(
+ MockFailingClusterMetricsConsumer.class.getName(), 2);
+
+ // below calls shouldn't propagate any exceptions
+ sut.prepare();
+
+ // no specific reason to mock... this is one of easiest ways to make dummy instance
+ sut.handleDataPoints(mock(IClusterMetricsConsumer.SupervisorInfo.class), Collections.emptyList());
+
+ Assert.assertEquals(1, MockFailingClusterMetricsConsumer.getPrepareCallCount());
+ Assert.assertEquals(0, MockFailingClusterMetricsConsumer.getHandleDataPointsWithSupervisorInfoCallCount());
+ }
+
+ public static class MockFailingClusterMetricsConsumer implements IClusterMetricsConsumer {
+ private static int prepareCallCount = 0;
+ private static int handleDataPointsWithClusterInfoCallCount = 0;
+ private static int handleDataPointsWithSupervisorInfoCallCount = 0;
+ private static int cleanupCallCount = 0;
+
+ @Override
+ public void prepare(Object registrationArgument) {
+ prepareCallCount++;
+
+ throw new RuntimeException("prepare failing...");
+ }
+
+ @Override
+ public void handleDataPoints(ClusterInfo clusterInfo, Collection<DataPoint> dataPoints) {
+ handleDataPointsWithClusterInfoCallCount++;
+ }
+
+ @Override
+ public void handleDataPoints(SupervisorInfo supervisorInfo, Collection<DataPoint> dataPoints) {
+ handleDataPointsWithSupervisorInfoCallCount++;
+ }
+
+ @Override
+ public void cleanup() {
+ cleanupCallCount++;
+ }
+
+ public static int getPrepareCallCount() {
+ return prepareCallCount;
+ }
+
+ public static int getHandleDataPointsWithClusterInfoCallCount() {
+ return handleDataPointsWithClusterInfoCallCount;
+ }
+
+ public static int getHandleDataPointsWithSupervisorInfoCallCount() {
+ return handleDataPointsWithSupervisorInfoCallCount;
+ }
+
+ public static int getCleanupCallCount() {
+ return cleanupCallCount;
+ }
+
+ public static void resetAllCounts() {
+ prepareCallCount = 0;
+ handleDataPointsWithClusterInfoCallCount = 0;
+ handleDataPointsWithSupervisorInfoCallCount = 0;
+ cleanupCallCount = 0;
+ }
+ }
+
+}
\ No newline at end of file
[3/3] storm git commit: STORM-2298: CHANGELOG
Posted by ka...@apache.org.
STORM-2298: CHANGELOG
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/1811273f
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/1811273f
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/1811273f
Branch: refs/heads/master
Commit: 1811273f18e79430167af709c23dbb9c9e2f7f7d
Parents: e760b66
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Tue Jan 24 12:42:35 2017 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Tue Jan 24 12:42:35 2017 +0900
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/1811273f/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6ec955f..94a20d2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -194,6 +194,7 @@
* STORM-1769: Added a test to check local nimbus with notifier plugin
## 1.1.0
+ * STORM-2298: Don't kill Nimbus when ClusterMetricsConsumer is failed to initialize
* STORM-2301: [storm-cassandra] upgrade cassandra driver to 3.1.2
* STORM-1446: Compile the Calcite logical plan to Storm Trident logical plan
* STORM-2303: [storm-opentsdb] Fix list invariant issue for JDK 7
[2/3] storm git commit: Merge branch 'STORM-2298'
Posted by ka...@apache.org.
Merge branch 'STORM-2298'
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/e760b668
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/e760b668
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/e760b668
Branch: refs/heads/master
Commit: e760b66853fb01035d2c1372710d2a19712f2f0c
Parents: 7e02b71 2e9e9d6
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Tue Jan 24 12:05:32 2017 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Tue Jan 24 12:05:32 2017 +0900
----------------------------------------------------------------------
.../metric/ClusterMetricsConsumerExecutor.java | 31 ++++-
.../ClusterMetricsConsumerExecutorTest.java | 133 +++++++++++++++++++
2 files changed, 158 insertions(+), 6 deletions(-)
----------------------------------------------------------------------