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 2015/09/21 19:40:56 UTC
[11/24] storm git commit: Addressed PR feedbacks
Addressed PR feedbacks
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/1764b867
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/1764b867
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/1764b867
Branch: refs/heads/master
Commit: 1764b867b80b0e228e2f2b867d761c1e91155673
Parents: 35990fa
Author: Arun Mahadevan <ai...@hortonworks.com>
Authored: Mon Aug 24 13:24:46 2015 +0530
Committer: Arun Mahadevan <ai...@hortonworks.com>
Committed: Mon Aug 24 13:24:46 2015 +0530
----------------------------------------------------------------------
storm-core/src/clj/backtype/storm/cluster.clj | 3 +-
.../src/clj/backtype/storm/daemon/nimbus.clj | 3 +-
.../jvm/backtype/storm/generated/Nimbus.java | 11 +-
.../backtype/storm/generated/TopologyInfo.java | 105 ++++++++++---------
storm-core/src/py/storm/Nimbus.py | 8 ++
storm-core/src/py/storm/ttypes.py | 2 -
storm-core/src/storm.thrift | 6 +-
.../test/clj/backtype/storm/nimbus_test.clj | 10 +-
8 files changed, 90 insertions(+), 58 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/1764b867/storm-core/src/clj/backtype/storm/cluster.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/cluster.clj b/storm-core/src/clj/backtype/storm/cluster.clj
index dcb4fd5..d96cb98 100644
--- a/storm-core/src/clj/backtype/storm/cluster.clj
+++ b/storm-core/src/clj/backtype/storm/cluster.clj
@@ -240,8 +240,7 @@
(when cb
(cb id))))
-;; public for stubbing in nimbus_test
-(defn maybe-deserialize
+(defn- maybe-deserialize
[ser clazz]
(when ser
(Utils/deserialize ser clazz)))
http://git-wip-us.apache.org/repos/asf/storm/blob/1764b867/storm-core/src/clj/backtype/storm/daemon/nimbus.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/daemon/nimbus.clj b/storm-core/src/clj/backtype/storm/daemon/nimbus.clj
index c568e9e..9fe3928 100644
--- a/storm-core/src/clj/backtype/storm/daemon/nimbus.clj
+++ b/storm-core/src/clj/backtype/storm/daemon/nimbus.clj
@@ -1169,7 +1169,8 @@
(check-authorization! nimbus storm-name topology-conf "debug")
(when-not storm-id
(throw (NotAliveException. storm-name)))
- (log-message "Nimbus setting debug to " enable? " for storm-name " storm-name " storm-id " storm-id)
+ (log-message "Nimbus setting debug to " enable? " for storm-name '" storm-name "' storm-id '" storm-id "'"
+ (if (not (clojure.string/blank? component-id)) (str " component-id '" component-id "'")))
(locking (:submit-lock nimbus)
(.update-storm! storm-cluster-state storm-id storm-base-updates))))
http://git-wip-us.apache.org/repos/asf/storm/blob/1764b867/storm-core/src/jvm/backtype/storm/generated/Nimbus.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/generated/Nimbus.java b/storm-core/src/jvm/backtype/storm/generated/Nimbus.java
index 1c2b2f0..830d95e 100644
--- a/storm-core/src/jvm/backtype/storm/generated/Nimbus.java
+++ b/storm-core/src/jvm/backtype/storm/generated/Nimbus.java
@@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
-@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2015-8-7")
+@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2015-8-24")
public class Nimbus {
public interface Iface {
@@ -70,6 +70,15 @@ public class Nimbus {
public void rebalance(String name, RebalanceOptions options) throws NotAliveException, InvalidTopologyException, AuthorizationException, org.apache.thrift.TException;
+ /**
+ * Enable/disable logging the tuples generated in topology via an internal EventLogger bolt. The component name is optional
+ * and if null or empty, the debug flag will apply to the entire topology.
+ *
+ *
+ * @param name
+ * @param component
+ * @param enable
+ */
public void debug(String name, String component, boolean enable) throws NotAliveException, AuthorizationException, org.apache.thrift.TException;
public void uploadNewCredentials(String name, Credentials creds) throws NotAliveException, InvalidTopologyException, AuthorizationException, org.apache.thrift.TException;
http://git-wip-us.apache.org/repos/asf/storm/blob/1764b867/storm-core/src/jvm/backtype/storm/generated/TopologyInfo.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/generated/TopologyInfo.java b/storm-core/src/jvm/backtype/storm/generated/TopologyInfo.java
index e050f24..d9b7c8d 100644
--- a/storm-core/src/jvm/backtype/storm/generated/TopologyInfo.java
+++ b/storm-core/src/jvm/backtype/storm/generated/TopologyInfo.java
@@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
-@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2015-8-7")
+@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2015-8-24")
public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, TopologyInfo._Fields>, java.io.Serializable, Cloneable, Comparable<TopologyInfo> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("TopologyInfo");
@@ -77,7 +77,7 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
private List<ExecutorSummary> executors; // required
private String status; // required
private Map<String,List<ErrorInfo>> errors; // required
- private Map<String,Boolean> component_debug; // required
+ private Map<String,Boolean> component_debug; // optional
private String sched_status; // optional
private String owner; // optional
@@ -166,7 +166,7 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
// isset id assignments
private static final int __UPTIME_SECS_ISSET_ID = 0;
private byte __isset_bitfield = 0;
- private static final _Fields optionals[] = {_Fields.SCHED_STATUS,_Fields.OWNER};
+ private static final _Fields optionals[] = {_Fields.COMPONENT_DEBUG,_Fields.SCHED_STATUS,_Fields.OWNER};
public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
static {
Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
@@ -186,7 +186,7 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING),
new org.apache.thrift.meta_data.ListMetaData(org.apache.thrift.protocol.TType.LIST,
new org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT, ErrorInfo.class)))));
- tmpMap.put(_Fields.COMPONENT_DEBUG, new org.apache.thrift.meta_data.FieldMetaData("component_debug", org.apache.thrift.TFieldRequirementType.REQUIRED,
+ tmpMap.put(_Fields.COMPONENT_DEBUG, new org.apache.thrift.meta_data.FieldMetaData("component_debug", org.apache.thrift.TFieldRequirementType.OPTIONAL,
new org.apache.thrift.meta_data.MapMetaData(org.apache.thrift.protocol.TType.MAP,
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING),
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.BOOL))));
@@ -207,8 +207,7 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
int uptime_secs,
List<ExecutorSummary> executors,
String status,
- Map<String,List<ErrorInfo>> errors,
- Map<String,Boolean> component_debug)
+ Map<String,List<ErrorInfo>> errors)
{
this();
this.id = id;
@@ -218,7 +217,6 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
this.executors = executors;
this.status = status;
this.errors = errors;
- this.component_debug = component_debug;
}
/**
@@ -983,14 +981,16 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
sb.append(this.errors);
}
first = false;
- if (!first) sb.append(", ");
- sb.append("component_debug:");
- if (this.component_debug == null) {
- sb.append("null");
- } else {
- sb.append(this.component_debug);
+ if (is_set_component_debug()) {
+ if (!first) sb.append(", ");
+ sb.append("component_debug:");
+ if (this.component_debug == null) {
+ sb.append("null");
+ } else {
+ sb.append(this.component_debug);
+ }
+ first = false;
}
- first = false;
if (is_set_sched_status()) {
if (!first) sb.append(", ");
sb.append("sched_status:");
@@ -1041,10 +1041,6 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
throw new org.apache.thrift.protocol.TProtocolException("Required field 'errors' is unset! Struct:" + toString());
}
- if (!is_set_component_debug()) {
- throw new org.apache.thrift.protocol.TProtocolException("Required field 'component_debug' is unset! Struct:" + toString());
- }
-
// check for sub-struct validity
}
@@ -1266,17 +1262,19 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
oprot.writeFieldEnd();
}
if (struct.component_debug != null) {
- oprot.writeFieldBegin(COMPONENT_DEBUG_FIELD_DESC);
- {
- oprot.writeMapBegin(new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.BOOL, struct.component_debug.size()));
- for (Map.Entry<String, Boolean> _iter307 : struct.component_debug.entrySet())
+ if (struct.is_set_component_debug()) {
+ oprot.writeFieldBegin(COMPONENT_DEBUG_FIELD_DESC);
{
- oprot.writeString(_iter307.getKey());
- oprot.writeBool(_iter307.getValue());
+ oprot.writeMapBegin(new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.BOOL, struct.component_debug.size()));
+ for (Map.Entry<String, Boolean> _iter307 : struct.component_debug.entrySet())
+ {
+ oprot.writeString(_iter307.getKey());
+ oprot.writeBool(_iter307.getValue());
+ }
+ oprot.writeMapEnd();
}
- oprot.writeMapEnd();
+ oprot.writeFieldEnd();
}
- oprot.writeFieldEnd();
}
if (struct.sched_status != null) {
if (struct.is_set_sched_status()) {
@@ -1334,22 +1332,27 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
}
}
}
- {
- oprot.writeI32(struct.component_debug.size());
- for (Map.Entry<String, Boolean> _iter311 : struct.component_debug.entrySet())
- {
- oprot.writeString(_iter311.getKey());
- oprot.writeBool(_iter311.getValue());
- }
- }
BitSet optionals = new BitSet();
- if (struct.is_set_sched_status()) {
+ if (struct.is_set_component_debug()) {
optionals.set(0);
}
- if (struct.is_set_owner()) {
+ if (struct.is_set_sched_status()) {
optionals.set(1);
}
- oprot.writeBitSet(optionals, 2);
+ if (struct.is_set_owner()) {
+ optionals.set(2);
+ }
+ oprot.writeBitSet(optionals, 3);
+ if (struct.is_set_component_debug()) {
+ {
+ oprot.writeI32(struct.component_debug.size());
+ for (Map.Entry<String, Boolean> _iter311 : struct.component_debug.entrySet())
+ {
+ oprot.writeString(_iter311.getKey());
+ oprot.writeBool(_iter311.getValue());
+ }
+ }
+ }
if (struct.is_set_sched_status()) {
oprot.writeString(struct.sched_status);
}
@@ -1404,25 +1407,27 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
}
}
struct.set_errors_isSet(true);
- {
- org.apache.thrift.protocol.TMap _map322 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.BOOL, iprot.readI32());
- struct.component_debug = new HashMap<String,Boolean>(2*_map322.size);
- String _key323;
- boolean _val324;
- for (int _i325 = 0; _i325 < _map322.size; ++_i325)
+ BitSet incoming = iprot.readBitSet(3);
+ if (incoming.get(0)) {
{
- _key323 = iprot.readString();
- _val324 = iprot.readBool();
- struct.component_debug.put(_key323, _val324);
+ org.apache.thrift.protocol.TMap _map322 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.BOOL, iprot.readI32());
+ struct.component_debug = new HashMap<String,Boolean>(2*_map322.size);
+ String _key323;
+ boolean _val324;
+ for (int _i325 = 0; _i325 < _map322.size; ++_i325)
+ {
+ _key323 = iprot.readString();
+ _val324 = iprot.readBool();
+ struct.component_debug.put(_key323, _val324);
+ }
}
+ struct.set_component_debug_isSet(true);
}
- struct.set_component_debug_isSet(true);
- BitSet incoming = iprot.readBitSet(2);
- if (incoming.get(0)) {
+ if (incoming.get(1)) {
struct.sched_status = iprot.readString();
struct.set_sched_status_isSet(true);
}
- if (incoming.get(1)) {
+ if (incoming.get(2)) {
struct.owner = iprot.readString();
struct.set_owner_isSet(true);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/1764b867/storm-core/src/py/storm/Nimbus.py
----------------------------------------------------------------------
diff --git a/storm-core/src/py/storm/Nimbus.py b/storm-core/src/py/storm/Nimbus.py
index 2aff10d..8bbe8d0 100644
--- a/storm-core/src/py/storm/Nimbus.py
+++ b/storm-core/src/py/storm/Nimbus.py
@@ -96,6 +96,10 @@ class Iface:
def debug(self, name, component, enable):
"""
+ Enable/disable logging the tuples generated in topology via an internal EventLogger bolt. The component name is optional
+ and if null or empty, the debug flag will apply to the entire topology.
+
+
Parameters:
- name
- component
@@ -454,6 +458,10 @@ class Client(Iface):
def debug(self, name, component, enable):
"""
+ Enable/disable logging the tuples generated in topology via an internal EventLogger bolt. The component name is optional
+ and if null or empty, the debug flag will apply to the entire topology.
+
+
Parameters:
- name
- component
http://git-wip-us.apache.org/repos/asf/storm/blob/1764b867/storm-core/src/py/storm/ttypes.py
----------------------------------------------------------------------
diff --git a/storm-core/src/py/storm/ttypes.py b/storm-core/src/py/storm/ttypes.py
index fe9663e..2784dad 100644
--- a/storm-core/src/py/storm/ttypes.py
+++ b/storm-core/src/py/storm/ttypes.py
@@ -4254,8 +4254,6 @@ class TopologyInfo:
raise TProtocol.TProtocolException(message='Required field status is unset!')
if self.errors is None:
raise TProtocol.TProtocolException(message='Required field errors is unset!')
- if self.component_debug is None:
- raise TProtocol.TProtocolException(message='Required field component_debug is unset!')
return
http://git-wip-us.apache.org/repos/asf/storm/blob/1764b867/storm-core/src/storm.thrift
----------------------------------------------------------------------
diff --git a/storm-core/src/storm.thrift b/storm-core/src/storm.thrift
index c4417d4..8547b30 100644
--- a/storm-core/src/storm.thrift
+++ b/storm-core/src/storm.thrift
@@ -218,7 +218,7 @@ struct TopologyInfo {
4: required list<ExecutorSummary> executors;
5: required string status;
6: required map<string, list<ErrorInfo>> errors;
- 7: required map<string, bool> component_debug;
+ 7: optional map<string, bool> component_debug;
513: optional string sched_status;
514: optional string owner;
}
@@ -351,6 +351,10 @@ service Nimbus {
void activate(1: string name) throws (1: NotAliveException e, 2: AuthorizationException aze);
void deactivate(1: string name) throws (1: NotAliveException e, 2: AuthorizationException aze);
void rebalance(1: string name, 2: RebalanceOptions options) throws (1: NotAliveException e, 2: InvalidTopologyException ite, 3: AuthorizationException aze);
+ /**
+ * Enable/disable logging the tuples generated in topology via an internal EventLogger bolt. The component name is optional
+ * and if null or empty, the debug flag will apply to the entire topology.
+ **/
void debug(1: string name, 2: string component, 3: bool enable) throws (1: NotAliveException e, 2: AuthorizationException aze);
void uploadNewCredentials(1: string name, 2: Credentials creds) throws (1: NotAliveException e, 2: InvalidTopologyException ite, 3: AuthorizationException aze);
http://git-wip-us.apache.org/repos/asf/storm/blob/1764b867/storm-core/test/clj/backtype/storm/nimbus_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/backtype/storm/nimbus_test.clj b/storm-core/test/clj/backtype/storm/nimbus_test.clj
index 1f7258b..3d0978c 100644
--- a/storm-core/test/clj/backtype/storm/nimbus_test.clj
+++ b/storm-core/test/clj/backtype/storm/nimbus_test.clj
@@ -1224,7 +1224,7 @@
(.disconnect cluster-state)
))))
-(deftest test-debug-on
+(deftest test-debug-on-component
(with-local-cluster [cluster]
(let [nimbus (:nimbus cluster)
topology (thrift/mk-topology
@@ -1233,3 +1233,11 @@
(submit-local-topology nimbus "t1" {TOPOLOGY-WORKERS 1} topology)
(.debug nimbus "t1" "spout" true))))
+(deftest test-debug-on-global
+ (with-local-cluster [cluster]
+ (let [nimbus (:nimbus cluster)
+ topology (thrift/mk-topology
+ {"spout" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)}
+ {})]
+ (submit-local-topology nimbus "t1" {TOPOLOGY-WORKERS 1} topology)
+ (.debug nimbus "t1" "" true))))