Skip to content
Browse files

added rebalance thrift definition

  • Loading branch information...
1 parent cf7a37b commit 277d4e636776c183988953af874016b103168cc3 @nathanmarz committed Dec 17, 2011
View
2 src/clj/backtype/storm/bootstrap.clj
@@ -31,7 +31,7 @@
InvalidTopologyException ClusterSummary TopologyInfo
TopologySummary TaskSummary TaskStats TaskSpecificStats
SpoutStats BoltStats ErrorInfo SupervisorSummary
- KillOptions JavaObject JavaObjectArg]))
+ KillOptions RebalanceOptions JavaObject JavaObjectArg]))
(import (quote [backtype.storm.daemon.common StormBase Assignment
TaskInfo SupervisorInfo WorkerHeartbeat TaskHeartbeat]))
(import (quote [backtype.storm.grouping CustomStreamGrouping]))
View
12 src/clj/backtype/storm/daemon/nimbus.clj
@@ -137,7 +137,9 @@
(locking (:submit-lock nimbus)
(let [[event & event-args] (if (keyword? event) [event] event)
status (topology-status nimbus storm-id)]
- (if status ; handles the case where event was scheduled but has been removed
+ ;; handles the case where event was scheduled but topology has been removed
+ (if-not status
+ (log-message "Cannot apply event " event " to " storm-id " because topology no longer exists")
(let [get-event (fn [m e]
(if (contains? m e)
(m e)
@@ -647,6 +649,14 @@
(transition-name! nimbus storm-name [:kill wait-amt] true)
))
+ (^void rebalance [this ^String storm-name ^RebalanceOptions options]
+ (check-storm-active! nimbus storm-name true)
+ (let [wait-amt (if (.is_set_wait_secs options)
+ (.get_wait_secs options)
+ )]
+ (transition-name! nimbus storm-name [:rebalance wait-amt] true)
+ ))
+
(activate [this storm-name]
(transition-name! nimbus storm-name :activate true)
)
View
780 src/jvm/backtype/storm/generated/Nimbus.java
@@ -35,6 +35,8 @@
public void deactivate(String name) throws NotAliveException, org.apache.thrift7.TException;
+ public void rebalance(String name, RebalanceOptions options) throws NotAliveException, org.apache.thrift7.TException;
+
public String beginFileUpload() throws org.apache.thrift7.TException;
public void uploadChunk(String location, ByteBuffer chunk) throws org.apache.thrift7.TException;
@@ -67,6 +69,8 @@
public void deactivate(String name, org.apache.thrift7.async.AsyncMethodCallback<AsyncClient.deactivate_call> resultHandler) throws org.apache.thrift7.TException;
+ public void rebalance(String name, RebalanceOptions options, org.apache.thrift7.async.AsyncMethodCallback<AsyncClient.rebalance_call> resultHandler) throws org.apache.thrift7.TException;
+
public void beginFileUpload(org.apache.thrift7.async.AsyncMethodCallback<AsyncClient.beginFileUpload_call> resultHandler) throws org.apache.thrift7.TException;
public void uploadChunk(String location, ByteBuffer chunk, org.apache.thrift7.async.AsyncMethodCallback<AsyncClient.uploadChunk_call> resultHandler) throws org.apache.thrift7.TException;
@@ -229,6 +233,30 @@ public void recv_deactivate() throws NotAliveException, org.apache.thrift7.TExce
return;
}
+ public void rebalance(String name, RebalanceOptions options) throws NotAliveException, org.apache.thrift7.TException
+ {
+ send_rebalance(name, options);
+ recv_rebalance();
+ }
+
+ public void send_rebalance(String name, RebalanceOptions options) throws org.apache.thrift7.TException
+ {
+ rebalance_args args = new rebalance_args();
+ args.set_name(name);
+ args.set_options(options);
+ sendBase("rebalance", args);
+ }
+
+ public void recv_rebalance() throws NotAliveException, org.apache.thrift7.TException
+ {
+ rebalance_result result = new rebalance_result();
+ receiveBase(result, "rebalance");
+ if (result.e != null) {
+ throw result.e;
+ }
+ return;
+ }
+
public String beginFileUpload() throws org.apache.thrift7.TException
{
send_beginFileUpload();
@@ -628,6 +656,41 @@ public void getResult() throws NotAliveException, org.apache.thrift7.TException
}
}
+ public void rebalance(String name, RebalanceOptions options, org.apache.thrift7.async.AsyncMethodCallback<rebalance_call> resultHandler) throws org.apache.thrift7.TException {
+ checkReady();
+ rebalance_call method_call = new rebalance_call(name, options, resultHandler, this, ___protocolFactory, ___transport);
+ this.___currentMethod = method_call;
+ ___manager.call(method_call);
+ }
+
+ public static class rebalance_call extends org.apache.thrift7.async.TAsyncMethodCall {
+ private String name;
+ private RebalanceOptions options;
+ public rebalance_call(String name, RebalanceOptions options, org.apache.thrift7.async.AsyncMethodCallback<rebalance_call> resultHandler, org.apache.thrift7.async.TAsyncClient client, org.apache.thrift7.protocol.TProtocolFactory protocolFactory, org.apache.thrift7.transport.TNonblockingTransport transport) throws org.apache.thrift7.TException {
+ super(client, protocolFactory, transport, resultHandler, false);
+ this.name = name;
+ this.options = options;
+ }
+
+ public void write_args(org.apache.thrift7.protocol.TProtocol prot) throws org.apache.thrift7.TException {
+ prot.writeMessageBegin(new org.apache.thrift7.protocol.TMessage("rebalance", org.apache.thrift7.protocol.TMessageType.CALL, 0));
+ rebalance_args args = new rebalance_args();
+ args.set_name(name);
+ args.set_options(options);
+ args.write(prot);
+ prot.writeMessageEnd();
+ }
+
+ public void getResult() throws NotAliveException, org.apache.thrift7.TException {
+ if (getState() != org.apache.thrift7.async.TAsyncMethodCall.State.RESPONSE_READ) {
+ throw new IllegalStateException("Method call not finished!");
+ }
+ org.apache.thrift7.transport.TMemoryInputTransport memoryTransport = new org.apache.thrift7.transport.TMemoryInputTransport(getFrameBuffer().array());
+ org.apache.thrift7.protocol.TProtocol prot = client.getProtocolFactory().getProtocol(memoryTransport);
+ (new Client(prot)).recv_rebalance();
+ }
+ }
+
public void beginFileUpload(org.apache.thrift7.async.AsyncMethodCallback<beginFileUpload_call> resultHandler) throws org.apache.thrift7.TException {
checkReady();
beginFileUpload_call method_call = new beginFileUpload_call(resultHandler, this, ___protocolFactory, ___transport);
@@ -931,6 +994,7 @@ protected Processor(I iface, Map<String, org.apache.thrift7.ProcessFunction<I,
processMap.put("killTopologyWithOpts", new killTopologyWithOpts());
processMap.put("activate", new activate());
processMap.put("deactivate", new deactivate());
+ processMap.put("rebalance", new rebalance());
processMap.put("beginFileUpload", new beginFileUpload());
processMap.put("uploadChunk", new uploadChunk());
processMap.put("finishFileUpload", new finishFileUpload());
@@ -1045,6 +1109,26 @@ protected deactivate_result getResult(I iface, deactivate_args args) throws org.
}
}
+ private static class rebalance<I extends Iface> extends org.apache.thrift7.ProcessFunction<I, rebalance_args> {
+ public rebalance() {
+ super("rebalance");
+ }
+
+ protected rebalance_args getEmptyArgsInstance() {
+ return new rebalance_args();
+ }
+
+ protected rebalance_result getResult(I iface, rebalance_args args) throws org.apache.thrift7.TException {
+ rebalance_result result = new rebalance_result();
+ try {
+ iface.rebalance(args.name, args.options);
+ } catch (NotAliveException e) {
+ result.e = e;
+ }
+ return result;
+ }
+ }
+
private static class beginFileUpload<I extends Iface> extends org.apache.thrift7.ProcessFunction<I, beginFileUpload_args> {
public beginFileUpload() {
super("beginFileUpload");
@@ -4680,6 +4764,702 @@ private void readObject(java.io.ObjectInputStream in) throws java.io.IOException
}
+ public static class rebalance_args implements org.apache.thrift7.TBase<rebalance_args, rebalance_args._Fields>, java.io.Serializable, Cloneable {
+ private static final org.apache.thrift7.protocol.TStruct STRUCT_DESC = new org.apache.thrift7.protocol.TStruct("rebalance_args");
+
+ private static final org.apache.thrift7.protocol.TField NAME_FIELD_DESC = new org.apache.thrift7.protocol.TField("name", org.apache.thrift7.protocol.TType.STRING, (short)1);
+ private static final org.apache.thrift7.protocol.TField OPTIONS_FIELD_DESC = new org.apache.thrift7.protocol.TField("options", org.apache.thrift7.protocol.TType.STRUCT, (short)2);
+
+ private String name; // required
+ private RebalanceOptions options; // required
+
+ /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
+ public enum _Fields implements org.apache.thrift7.TFieldIdEnum {
+ NAME((short)1, "name"),
+ OPTIONS((short)2, "options");
+
+ private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
+
+ static {
+ for (_Fields field : EnumSet.allOf(_Fields.class)) {
+ byName.put(field.getFieldName(), field);
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, or null if its not found.
+ */
+ public static _Fields findByThriftId(int fieldId) {
+ switch(fieldId) {
+ case 1: // NAME
+ return NAME;
+ case 2: // OPTIONS
+ return OPTIONS;
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, throwing an exception
+ * if it is not found.
+ */
+ public static _Fields findByThriftIdOrThrow(int fieldId) {
+ _Fields fields = findByThriftId(fieldId);
+ if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+ return fields;
+ }
+
+ /**
+ * Find the _Fields constant that matches name, or null if its not found.
+ */
+ public static _Fields findByName(String name) {
+ return byName.get(name);
+ }
+
+ private final short _thriftId;
+ private final String _fieldName;
+
+ _Fields(short thriftId, String fieldName) {
+ _thriftId = thriftId;
+ _fieldName = fieldName;
+ }
+
+ public short getThriftFieldId() {
+ return _thriftId;
+ }
+
+ public String getFieldName() {
+ return _fieldName;
+ }
+ }
+
+ // isset id assignments
+
+ public static final Map<_Fields, org.apache.thrift7.meta_data.FieldMetaData> metaDataMap;
+ static {
+ Map<_Fields, org.apache.thrift7.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift7.meta_data.FieldMetaData>(_Fields.class);
+ tmpMap.put(_Fields.NAME, new org.apache.thrift7.meta_data.FieldMetaData("name", org.apache.thrift7.TFieldRequirementType.DEFAULT,
+ new org.apache.thrift7.meta_data.FieldValueMetaData(org.apache.thrift7.protocol.TType.STRING)));
+ tmpMap.put(_Fields.OPTIONS, new org.apache.thrift7.meta_data.FieldMetaData("options", org.apache.thrift7.TFieldRequirementType.DEFAULT,
+ new org.apache.thrift7.meta_data.StructMetaData(org.apache.thrift7.protocol.TType.STRUCT, RebalanceOptions.class)));
+ metaDataMap = Collections.unmodifiableMap(tmpMap);
+ org.apache.thrift7.meta_data.FieldMetaData.addStructMetaDataMap(rebalance_args.class, metaDataMap);
+ }
+
+ public rebalance_args() {
+ }
+
+ public rebalance_args(
+ String name,
+ RebalanceOptions options)
+ {
+ this();
+ this.name = name;
+ this.options = options;
+ }
+
+ /**
+ * Performs a deep copy on <i>other</i>.
+ */
+ public rebalance_args(rebalance_args other) {
+ if (other.is_set_name()) {
+ this.name = other.name;
+ }
+ if (other.is_set_options()) {
+ this.options = new RebalanceOptions(other.options);
+ }
+ }
+
+ public rebalance_args deepCopy() {
+ return new rebalance_args(this);
+ }
+
+ @Override
+ public void clear() {
+ this.name = null;
+ this.options = null;
+ }
+
+ public String get_name() {
+ return this.name;
+ }
+
+ public void set_name(String name) {
+ this.name = name;
+ }
+
+ public void unset_name() {
+ this.name = null;
+ }
+
+ /** Returns true if field name is set (has been assigned a value) and false otherwise */
+ public boolean is_set_name() {
+ return this.name != null;
+ }
+
+ public void set_name_isSet(boolean value) {
+ if (!value) {
+ this.name = null;
+ }
+ }
+
+ public RebalanceOptions get_options() {
+ return this.options;
+ }
+
+ public void set_options(RebalanceOptions options) {
+ this.options = options;
+ }
+
+ public void unset_options() {
+ this.options = null;
+ }
+
+ /** Returns true if field options is set (has been assigned a value) and false otherwise */
+ public boolean is_set_options() {
+ return this.options != null;
+ }
+
+ public void set_options_isSet(boolean value) {
+ if (!value) {
+ this.options = null;
+ }
+ }
+
+ public void setFieldValue(_Fields field, Object value) {
+ switch (field) {
+ case NAME:
+ if (value == null) {
+ unset_name();
+ } else {
+ set_name((String)value);
+ }
+ break;
+
+ case OPTIONS:
+ if (value == null) {
+ unset_options();
+ } else {
+ set_options((RebalanceOptions)value);
+ }
+ break;
+
+ }
+ }
+
+ public Object getFieldValue(_Fields field) {
+ switch (field) {
+ case NAME:
+ return get_name();
+
+ case OPTIONS:
+ return get_options();
+
+ }
+ throw new IllegalStateException();
+ }
+
+ /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
+ public boolean isSet(_Fields field) {
+ if (field == null) {
+ throw new IllegalArgumentException();
+ }
+
+ switch (field) {
+ case NAME:
+ return is_set_name();
+ case OPTIONS:
+ return is_set_options();
+ }
+ throw new IllegalStateException();
+ }
+
+ @Override
+ public boolean equals(Object that) {
+ if (that == null)
+ return false;
+ if (that instanceof rebalance_args)
+ return this.equals((rebalance_args)that);
+ return false;
+ }
+
+ public boolean equals(rebalance_args that) {
+ if (that == null)
+ return false;
+
+ boolean this_present_name = true && this.is_set_name();
+ boolean that_present_name = true && that.is_set_name();
+ if (this_present_name || that_present_name) {
+ if (!(this_present_name && that_present_name))
+ return false;
+ if (!this.name.equals(that.name))
+ return false;
+ }
+
+ boolean this_present_options = true && this.is_set_options();
+ boolean that_present_options = true && that.is_set_options();
+ if (this_present_options || that_present_options) {
+ if (!(this_present_options && that_present_options))
+ return false;
+ if (!this.options.equals(that.options))
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ HashCodeBuilder builder = new HashCodeBuilder();
+
+ boolean present_name = true && (is_set_name());
+ builder.append(present_name);
+ if (present_name)
+ builder.append(name);
+
+ boolean present_options = true && (is_set_options());
+ builder.append(present_options);
+ if (present_options)
+ builder.append(options);
+
+ return builder.toHashCode();
+ }
+
+ public int compareTo(rebalance_args other) {
+ if (!getClass().equals(other.getClass())) {
+ return getClass().getName().compareTo(other.getClass().getName());
+ }
+
+ int lastComparison = 0;
+ rebalance_args typedOther = (rebalance_args)other;
+
+ lastComparison = Boolean.valueOf(is_set_name()).compareTo(typedOther.is_set_name());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (is_set_name()) {
+ lastComparison = org.apache.thrift7.TBaseHelper.compareTo(this.name, typedOther.name);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ lastComparison = Boolean.valueOf(is_set_options()).compareTo(typedOther.is_set_options());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (is_set_options()) {
+ lastComparison = org.apache.thrift7.TBaseHelper.compareTo(this.options, typedOther.options);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ return 0;
+ }
+
+ public _Fields fieldForId(int fieldId) {
+ return _Fields.findByThriftId(fieldId);
+ }
+
+ public void read(org.apache.thrift7.protocol.TProtocol iprot) throws org.apache.thrift7.TException {
+ org.apache.thrift7.protocol.TField field;
+ iprot.readStructBegin();
+ while (true)
+ {
+ field = iprot.readFieldBegin();
+ if (field.type == org.apache.thrift7.protocol.TType.STOP) {
+ break;
+ }
+ switch (field.id) {
+ case 1: // NAME
+ if (field.type == org.apache.thrift7.protocol.TType.STRING) {
+ this.name = iprot.readString();
+ } else {
+ org.apache.thrift7.protocol.TProtocolUtil.skip(iprot, field.type);
+ }
+ break;
+ case 2: // OPTIONS
+ if (field.type == org.apache.thrift7.protocol.TType.STRUCT) {
+ this.options = new RebalanceOptions();
+ this.options.read(iprot);
+ } else {
+ org.apache.thrift7.protocol.TProtocolUtil.skip(iprot, field.type);
+ }
+ break;
+ default:
+ org.apache.thrift7.protocol.TProtocolUtil.skip(iprot, field.type);
+ }
+ iprot.readFieldEnd();
+ }
+ iprot.readStructEnd();
+ validate();
+ }
+
+ public void write(org.apache.thrift7.protocol.TProtocol oprot) throws org.apache.thrift7.TException {
+ validate();
+
+ oprot.writeStructBegin(STRUCT_DESC);
+ if (this.name != null) {
+ oprot.writeFieldBegin(NAME_FIELD_DESC);
+ oprot.writeString(this.name);
+ oprot.writeFieldEnd();
+ }
+ if (this.options != null) {
+ oprot.writeFieldBegin(OPTIONS_FIELD_DESC);
+ this.options.write(oprot);
+ oprot.writeFieldEnd();
+ }
+ oprot.writeFieldStop();
+ oprot.writeStructEnd();
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("rebalance_args(");
+ boolean first = true;
+
+ sb.append("name:");
+ if (this.name == null) {
+ sb.append("null");
+ } else {
+ sb.append(this.name);
+ }
+ first = false;
+ if (!first) sb.append(", ");
+ sb.append("options:");
+ if (this.options == null) {
+ sb.append("null");
+ } else {
+ sb.append(this.options);
+ }
+ first = false;
+ sb.append(")");
+ return sb.toString();
+ }
+
+ public void validate() throws org.apache.thrift7.TException {
+ // check for required fields
+ }
+
+ private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+ try {
+ write(new org.apache.thrift7.protocol.TCompactProtocol(new org.apache.thrift7.transport.TIOStreamTransport(out)));
+ } catch (org.apache.thrift7.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
+ try {
+ read(new org.apache.thrift7.protocol.TCompactProtocol(new org.apache.thrift7.transport.TIOStreamTransport(in)));
+ } catch (org.apache.thrift7.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ }
+
+ public static class rebalance_result implements org.apache.thrift7.TBase<rebalance_result, rebalance_result._Fields>, java.io.Serializable, Cloneable {
+ private static final org.apache.thrift7.protocol.TStruct STRUCT_DESC = new org.apache.thrift7.protocol.TStruct("rebalance_result");
+
+ private static final org.apache.thrift7.protocol.TField E_FIELD_DESC = new org.apache.thrift7.protocol.TField("e", org.apache.thrift7.protocol.TType.STRUCT, (short)1);
+
+ private NotAliveException e; // required
+
+ /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
+ public enum _Fields implements org.apache.thrift7.TFieldIdEnum {
+ E((short)1, "e");
+
+ private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
+
+ static {
+ for (_Fields field : EnumSet.allOf(_Fields.class)) {
+ byName.put(field.getFieldName(), field);
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, or null if its not found.
+ */
+ public static _Fields findByThriftId(int fieldId) {
+ switch(fieldId) {
+ case 1: // E
+ return E;
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, throwing an exception
+ * if it is not found.
+ */
+ public static _Fields findByThriftIdOrThrow(int fieldId) {
+ _Fields fields = findByThriftId(fieldId);
+ if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+ return fields;
+ }
+
+ /**
+ * Find the _Fields constant that matches name, or null if its not found.
+ */
+ public static _Fields findByName(String name) {
+ return byName.get(name);
+ }
+
+ private final short _thriftId;
+ private final String _fieldName;
+
+ _Fields(short thriftId, String fieldName) {
+ _thriftId = thriftId;
+ _fieldName = fieldName;
+ }
+
+ public short getThriftFieldId() {
+ return _thriftId;
+ }
+
+ public String getFieldName() {
+ return _fieldName;
+ }
+ }
+
+ // isset id assignments
+
+ public static final Map<_Fields, org.apache.thrift7.meta_data.FieldMetaData> metaDataMap;
+ static {
+ Map<_Fields, org.apache.thrift7.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift7.meta_data.FieldMetaData>(_Fields.class);
+ tmpMap.put(_Fields.E, new org.apache.thrift7.meta_data.FieldMetaData("e", org.apache.thrift7.TFieldRequirementType.DEFAULT,
+ new org.apache.thrift7.meta_data.FieldValueMetaData(org.apache.thrift7.protocol.TType.STRUCT)));
+ metaDataMap = Collections.unmodifiableMap(tmpMap);
+ org.apache.thrift7.meta_data.FieldMetaData.addStructMetaDataMap(rebalance_result.class, metaDataMap);
+ }
+
+ public rebalance_result() {
+ }
+
+ public rebalance_result(
+ NotAliveException e)
+ {
+ this();
+ this.e = e;
+ }
+
+ /**
+ * Performs a deep copy on <i>other</i>.
+ */
+ public rebalance_result(rebalance_result other) {
+ if (other.is_set_e()) {
+ this.e = new NotAliveException(other.e);
+ }
+ }
+
+ public rebalance_result deepCopy() {
+ return new rebalance_result(this);
+ }
+
+ @Override
+ public void clear() {
+ this.e = null;
+ }
+
+ public NotAliveException get_e() {
+ return this.e;
+ }
+
+ public void set_e(NotAliveException e) {
+ this.e = e;
+ }
+
+ public void unset_e() {
+ this.e = null;
+ }
+
+ /** Returns true if field e is set (has been assigned a value) and false otherwise */
+ public boolean is_set_e() {
+ return this.e != null;
+ }
+
+ public void set_e_isSet(boolean value) {
+ if (!value) {
+ this.e = null;
+ }
+ }
+
+ public void setFieldValue(_Fields field, Object value) {
+ switch (field) {
+ case E:
+ if (value == null) {
+ unset_e();
+ } else {
+ set_e((NotAliveException)value);
+ }
+ break;
+
+ }
+ }
+
+ public Object getFieldValue(_Fields field) {
+ switch (field) {
+ case E:
+ return get_e();
+
+ }
+ throw new IllegalStateException();
+ }
+
+ /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
+ public boolean isSet(_Fields field) {
+ if (field == null) {
+ throw new IllegalArgumentException();
+ }
+
+ switch (field) {
+ case E:
+ return is_set_e();
+ }
+ throw new IllegalStateException();
+ }
+
+ @Override
+ public boolean equals(Object that) {
+ if (that == null)
+ return false;
+ if (that instanceof rebalance_result)
+ return this.equals((rebalance_result)that);
+ return false;
+ }
+
+ public boolean equals(rebalance_result that) {
+ if (that == null)
+ return false;
+
+ boolean this_present_e = true && this.is_set_e();
+ boolean that_present_e = true && that.is_set_e();
+ if (this_present_e || that_present_e) {
+ if (!(this_present_e && that_present_e))
+ return false;
+ if (!this.e.equals(that.e))
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ HashCodeBuilder builder = new HashCodeBuilder();
+
+ boolean present_e = true && (is_set_e());
+ builder.append(present_e);
+ if (present_e)
+ builder.append(e);
+
+ return builder.toHashCode();
+ }
+
+ public int compareTo(rebalance_result other) {
+ if (!getClass().equals(other.getClass())) {
+ return getClass().getName().compareTo(other.getClass().getName());
+ }
+
+ int lastComparison = 0;
+ rebalance_result typedOther = (rebalance_result)other;
+
+ lastComparison = Boolean.valueOf(is_set_e()).compareTo(typedOther.is_set_e());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (is_set_e()) {
+ lastComparison = org.apache.thrift7.TBaseHelper.compareTo(this.e, typedOther.e);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ return 0;
+ }
+
+ public _Fields fieldForId(int fieldId) {
+ return _Fields.findByThriftId(fieldId);
+ }
+
+ public void read(org.apache.thrift7.protocol.TProtocol iprot) throws org.apache.thrift7.TException {
+ org.apache.thrift7.protocol.TField field;
+ iprot.readStructBegin();
+ while (true)
+ {
+ field = iprot.readFieldBegin();
+ if (field.type == org.apache.thrift7.protocol.TType.STOP) {
+ break;
+ }
+ switch (field.id) {
+ case 1: // E
+ if (field.type == org.apache.thrift7.protocol.TType.STRUCT) {
+ this.e = new NotAliveException();
+ this.e.read(iprot);
+ } else {
+ org.apache.thrift7.protocol.TProtocolUtil.skip(iprot, field.type);
+ }
+ break;
+ default:
+ org.apache.thrift7.protocol.TProtocolUtil.skip(iprot, field.type);
+ }
+ iprot.readFieldEnd();
+ }
+ iprot.readStructEnd();
+ validate();
+ }
+
+ public void write(org.apache.thrift7.protocol.TProtocol oprot) throws org.apache.thrift7.TException {
+ oprot.writeStructBegin(STRUCT_DESC);
+
+ if (this.is_set_e()) {
+ oprot.writeFieldBegin(E_FIELD_DESC);
+ this.e.write(oprot);
+ oprot.writeFieldEnd();
+ }
+ oprot.writeFieldStop();
+ oprot.writeStructEnd();
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("rebalance_result(");
+ boolean first = true;
+
+ sb.append("e:");
+ if (this.e == null) {
+ sb.append("null");
+ } else {
+ sb.append(this.e);
+ }
+ first = false;
+ sb.append(")");
+ return sb.toString();
+ }
+
+ public void validate() throws org.apache.thrift7.TException {
+ // check for required fields
+ }
+
+ private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+ try {
+ write(new org.apache.thrift7.protocol.TCompactProtocol(new org.apache.thrift7.transport.TIOStreamTransport(out)));
+ } catch (org.apache.thrift7.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
+ try {
+ read(new org.apache.thrift7.protocol.TCompactProtocol(new org.apache.thrift7.transport.TIOStreamTransport(in)));
+ } catch (org.apache.thrift7.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ }
+
public static class beginFileUpload_args implements org.apache.thrift7.TBase<beginFileUpload_args, beginFileUpload_args._Fields>, java.io.Serializable, Cloneable {
private static final org.apache.thrift7.protocol.TStruct STRUCT_DESC = new org.apache.thrift7.protocol.TStruct("beginFileUpload_args");
View
320 src/jvm/backtype/storm/generated/RebalanceOptions.java
@@ -0,0 +1,320 @@
+/**
+ * Autogenerated by Thrift Compiler (0.7.0)
+ *
+ * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
+ */
+package backtype.storm.generated;
+
+import org.apache.commons.lang.builder.HashCodeBuilder;
+import java.util.List;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.EnumMap;
+import java.util.Set;
+import java.util.HashSet;
+import java.util.EnumSet;
+import java.util.Collections;
+import java.util.BitSet;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RebalanceOptions implements org.apache.thrift7.TBase<RebalanceOptions, RebalanceOptions._Fields>, java.io.Serializable, Cloneable {
+ private static final org.apache.thrift7.protocol.TStruct STRUCT_DESC = new org.apache.thrift7.protocol.TStruct("RebalanceOptions");
+
+ private static final org.apache.thrift7.protocol.TField WAIT_SECS_FIELD_DESC = new org.apache.thrift7.protocol.TField("wait_secs", org.apache.thrift7.protocol.TType.I32, (short)1);
+
+ private int wait_secs; // required
+
+ /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
+ public enum _Fields implements org.apache.thrift7.TFieldIdEnum {
+ WAIT_SECS((short)1, "wait_secs");
+
+ private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
+
+ static {
+ for (_Fields field : EnumSet.allOf(_Fields.class)) {
+ byName.put(field.getFieldName(), field);
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, or null if its not found.
+ */
+ public static _Fields findByThriftId(int fieldId) {
+ switch(fieldId) {
+ case 1: // WAIT_SECS
+ return WAIT_SECS;
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, throwing an exception
+ * if it is not found.
+ */
+ public static _Fields findByThriftIdOrThrow(int fieldId) {
+ _Fields fields = findByThriftId(fieldId);
+ if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+ return fields;
+ }
+
+ /**
+ * Find the _Fields constant that matches name, or null if its not found.
+ */
+ public static _Fields findByName(String name) {
+ return byName.get(name);
+ }
+
+ private final short _thriftId;
+ private final String _fieldName;
+
+ _Fields(short thriftId, String fieldName) {
+ _thriftId = thriftId;
+ _fieldName = fieldName;
+ }
+
+ public short getThriftFieldId() {
+ return _thriftId;
+ }
+
+ public String getFieldName() {
+ return _fieldName;
+ }
+ }
+
+ // isset id assignments
+ private static final int __WAIT_SECS_ISSET_ID = 0;
+ private BitSet __isset_bit_vector = new BitSet(1);
+
+ public static final Map<_Fields, org.apache.thrift7.meta_data.FieldMetaData> metaDataMap;
+ static {
+ Map<_Fields, org.apache.thrift7.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift7.meta_data.FieldMetaData>(_Fields.class);
+ tmpMap.put(_Fields.WAIT_SECS, new org.apache.thrift7.meta_data.FieldMetaData("wait_secs", org.apache.thrift7.TFieldRequirementType.OPTIONAL,
+ new org.apache.thrift7.meta_data.FieldValueMetaData(org.apache.thrift7.protocol.TType.I32)));
+ metaDataMap = Collections.unmodifiableMap(tmpMap);
+ org.apache.thrift7.meta_data.FieldMetaData.addStructMetaDataMap(RebalanceOptions.class, metaDataMap);
+ }
+
+ public RebalanceOptions() {
+ }
+
+ /**
+ * Performs a deep copy on <i>other</i>.
+ */
+ public RebalanceOptions(RebalanceOptions other) {
+ __isset_bit_vector.clear();
+ __isset_bit_vector.or(other.__isset_bit_vector);
+ this.wait_secs = other.wait_secs;
+ }
+
+ public RebalanceOptions deepCopy() {
+ return new RebalanceOptions(this);
+ }
+
+ @Override
+ public void clear() {
+ set_wait_secs_isSet(false);
+ this.wait_secs = 0;
+ }
+
+ public int get_wait_secs() {
+ return this.wait_secs;
+ }
+
+ public void set_wait_secs(int wait_secs) {
+ this.wait_secs = wait_secs;
+ set_wait_secs_isSet(true);
+ }
+
+ public void unset_wait_secs() {
+ __isset_bit_vector.clear(__WAIT_SECS_ISSET_ID);
+ }
+
+ /** Returns true if field wait_secs is set (has been assigned a value) and false otherwise */
+ public boolean is_set_wait_secs() {
+ return __isset_bit_vector.get(__WAIT_SECS_ISSET_ID);
+ }
+
+ public void set_wait_secs_isSet(boolean value) {
+ __isset_bit_vector.set(__WAIT_SECS_ISSET_ID, value);
+ }
+
+ public void setFieldValue(_Fields field, Object value) {
+ switch (field) {
+ case WAIT_SECS:
+ if (value == null) {
+ unset_wait_secs();
+ } else {
+ set_wait_secs((Integer)value);
+ }
+ break;
+
+ }
+ }
+
+ public Object getFieldValue(_Fields field) {
+ switch (field) {
+ case WAIT_SECS:
+ return Integer.valueOf(get_wait_secs());
+
+ }
+ throw new IllegalStateException();
+ }
+
+ /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
+ public boolean isSet(_Fields field) {
+ if (field == null) {
+ throw new IllegalArgumentException();
+ }
+
+ switch (field) {
+ case WAIT_SECS:
+ return is_set_wait_secs();
+ }
+ throw new IllegalStateException();
+ }
+
+ @Override
+ public boolean equals(Object that) {
+ if (that == null)
+ return false;
+ if (that instanceof RebalanceOptions)
+ return this.equals((RebalanceOptions)that);
+ return false;
+ }
+
+ public boolean equals(RebalanceOptions that) {
+ if (that == null)
+ return false;
+
+ boolean this_present_wait_secs = true && this.is_set_wait_secs();
+ boolean that_present_wait_secs = true && that.is_set_wait_secs();
+ if (this_present_wait_secs || that_present_wait_secs) {
+ if (!(this_present_wait_secs && that_present_wait_secs))
+ return false;
+ if (this.wait_secs != that.wait_secs)
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ HashCodeBuilder builder = new HashCodeBuilder();
+
+ boolean present_wait_secs = true && (is_set_wait_secs());
+ builder.append(present_wait_secs);
+ if (present_wait_secs)
+ builder.append(wait_secs);
+
+ return builder.toHashCode();
+ }
+
+ public int compareTo(RebalanceOptions other) {
+ if (!getClass().equals(other.getClass())) {
+ return getClass().getName().compareTo(other.getClass().getName());
+ }
+
+ int lastComparison = 0;
+ RebalanceOptions typedOther = (RebalanceOptions)other;
+
+ lastComparison = Boolean.valueOf(is_set_wait_secs()).compareTo(typedOther.is_set_wait_secs());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (is_set_wait_secs()) {
+ lastComparison = org.apache.thrift7.TBaseHelper.compareTo(this.wait_secs, typedOther.wait_secs);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ return 0;
+ }
+
+ public _Fields fieldForId(int fieldId) {
+ return _Fields.findByThriftId(fieldId);
+ }
+
+ public void read(org.apache.thrift7.protocol.TProtocol iprot) throws org.apache.thrift7.TException {
+ org.apache.thrift7.protocol.TField field;
+ iprot.readStructBegin();
+ while (true)
+ {
+ field = iprot.readFieldBegin();
+ if (field.type == org.apache.thrift7.protocol.TType.STOP) {
+ break;
+ }
+ switch (field.id) {
+ case 1: // WAIT_SECS
+ if (field.type == org.apache.thrift7.protocol.TType.I32) {
+ this.wait_secs = iprot.readI32();
+ set_wait_secs_isSet(true);
+ } else {
+ org.apache.thrift7.protocol.TProtocolUtil.skip(iprot, field.type);
+ }
+ break;
+ default:
+ org.apache.thrift7.protocol.TProtocolUtil.skip(iprot, field.type);
+ }
+ iprot.readFieldEnd();
+ }
+ iprot.readStructEnd();
+ validate();
+ }
+
+ public void write(org.apache.thrift7.protocol.TProtocol oprot) throws org.apache.thrift7.TException {
+ validate();
+
+ oprot.writeStructBegin(STRUCT_DESC);
+ if (is_set_wait_secs()) {
+ oprot.writeFieldBegin(WAIT_SECS_FIELD_DESC);
+ oprot.writeI32(this.wait_secs);
+ oprot.writeFieldEnd();
+ }
+ oprot.writeFieldStop();
+ oprot.writeStructEnd();
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("RebalanceOptions(");
+ boolean first = true;
+
+ if (is_set_wait_secs()) {
+ sb.append("wait_secs:");
+ sb.append(this.wait_secs);
+ first = false;
+ }
+ sb.append(")");
+ return sb.toString();
+ }
+
+ public void validate() throws org.apache.thrift7.TException {
+ // check for required fields
+ }
+
+ private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+ try {
+ write(new org.apache.thrift7.protocol.TCompactProtocol(new org.apache.thrift7.transport.TIOStreamTransport(out)));
+ } catch (org.apache.thrift7.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
+ try {
+ // it doesn't seem like you should have to do this, but java serialization is wacky, and doesn't call the default constructor.
+ __isset_bit_vector = new BitSet(1);
+ read(new org.apache.thrift7.protocol.TCompactProtocol(new org.apache.thrift7.transport.TIOStreamTransport(in)));
+ } catch (org.apache.thrift7.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+}
+
View
7 src/py/storm/Nimbus-remote
@@ -26,6 +26,7 @@ if len(sys.argv) <= 1 or sys.argv[1] == '--help':
print ' void killTopologyWithOpts(string name, KillOptions options)'
print ' void activate(string name)'
print ' void deactivate(string name)'
+ print ' void rebalance(string name, RebalanceOptions options)'
print ' string beginFileUpload()'
print ' void uploadChunk(string location, string chunk)'
print ' void finishFileUpload(string location)'
@@ -115,6 +116,12 @@ elif cmd == 'deactivate':
sys.exit(1)
pp.pprint(client.deactivate(args[0],))
+elif cmd == 'rebalance':
+ if len(args) != 2:
+ print 'rebalance requires 2 args'
+ sys.exit(1)
+ pp.pprint(client.rebalance(args[0],eval(args[1]),))
+
elif cmd == 'beginFileUpload':
if len(args) != 0:
print 'beginFileUpload requires 0 args'
View
189 src/py/storm/Nimbus.py
@@ -55,6 +55,14 @@ def deactivate(self, name):
"""
pass
+ def rebalance(self, name, options):
+ """
+ Parameters:
+ - name
+ - options
+ """
+ pass
+
def beginFileUpload(self, ):
pass
@@ -279,6 +287,38 @@ def recv_deactivate(self, ):
raise result.e
return
+ def rebalance(self, name, options):
+ """
+ Parameters:
+ - name
+ - options
+ """
+ self.send_rebalance(name, options)
+ self.recv_rebalance()
+
+ def send_rebalance(self, name, options):
+ self._oprot.writeMessageBegin('rebalance', TMessageType.CALL, self._seqid)
+ args = rebalance_args()
+ args.name = name
+ args.options = options
+ args.write(self._oprot)
+ self._oprot.writeMessageEnd()
+ self._oprot.trans.flush()
+
+ def recv_rebalance(self, ):
+ (fname, mtype, rseqid) = self._iprot.readMessageBegin()
+ if mtype == TMessageType.EXCEPTION:
+ x = TApplicationException()
+ x.read(self._iprot)
+ self._iprot.readMessageEnd()
+ raise x
+ result = rebalance_result()
+ result.read(self._iprot)
+ self._iprot.readMessageEnd()
+ if result.e is not None:
+ raise result.e
+ return
+
def beginFileUpload(self, ):
self.send_beginFileUpload()
return self.recv_beginFileUpload()
@@ -553,6 +593,7 @@ def __init__(self, handler):
self._processMap["killTopologyWithOpts"] = Processor.process_killTopologyWithOpts
self._processMap["activate"] = Processor.process_activate
self._processMap["deactivate"] = Processor.process_deactivate
+ self._processMap["rebalance"] = Processor.process_rebalance
self._processMap["beginFileUpload"] = Processor.process_beginFileUpload
self._processMap["uploadChunk"] = Processor.process_uploadChunk
self._processMap["finishFileUpload"] = Processor.process_finishFileUpload
@@ -650,6 +691,20 @@ def process_deactivate(self, seqid, iprot, oprot):
oprot.writeMessageEnd()
oprot.trans.flush()
+ def process_rebalance(self, seqid, iprot, oprot):
+ args = rebalance_args()
+ args.read(iprot)
+ iprot.readMessageEnd()
+ result = rebalance_result()
+ try:
+ self._handler.rebalance(args.name, args.options)
+ except NotAliveException, e:
+ result.e = e
+ oprot.writeMessageBegin("rebalance", TMessageType.REPLY, seqid)
+ result.write(oprot)
+ oprot.writeMessageEnd()
+ oprot.trans.flush()
+
def process_beginFileUpload(self, seqid, iprot, oprot):
args = beginFileUpload_args()
args.read(iprot)
@@ -1429,6 +1484,140 @@ def __eq__(self, other):
def __ne__(self, other):
return not (self == other)
+class rebalance_args:
+ """
+ Attributes:
+ - name
+ - options
+ """
+
+ thrift_spec = (
+ None, # 0
+ (1, TType.STRING, 'name', None, None, ), # 1
+ (2, TType.STRUCT, 'options', (RebalanceOptions, RebalanceOptions.thrift_spec), None, ), # 2
+ )
+
+ def __init__(self, name=None, options=None,):
+ self.name = name
+ self.options = options
+
+ def read(self, iprot):
+ if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None:
+ fastbinary.decode_binary(self, iprot.trans, (self.__class__, self.thrift_spec))
+ return
+ iprot.readStructBegin()
+ while True:
+ (fname, ftype, fid) = iprot.readFieldBegin()
+ if ftype == TType.STOP:
+ break
+ if fid == 1:
+ if ftype == TType.STRING:
+ self.name = iprot.readString().decode('utf-8')
+ else:
+ iprot.skip(ftype)
+ elif fid == 2:
+ if ftype == TType.STRUCT:
+ self.options = RebalanceOptions()
+ self.options.read(iprot)
+ else:
+ iprot.skip(ftype)
+ else:
+ iprot.skip(ftype)
+ iprot.readFieldEnd()
+ iprot.readStructEnd()
+
+ def write(self, oprot):
+ if oprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and self.thrift_spec is not None and fastbinary is not None:
+ oprot.trans.write(fastbinary.encode_binary(self, (self.__class__, self.thrift_spec)))
+ return
+ oprot.writeStructBegin('rebalance_args')
+ if self.name is not None:
+ oprot.writeFieldBegin('name', TType.STRING, 1)
+ oprot.writeString(self.name.encode('utf-8'))
+ oprot.writeFieldEnd()
+ if self.options is not None:
+ oprot.writeFieldBegin('options', TType.STRUCT, 2)
+ self.options.write(oprot)
+ oprot.writeFieldEnd()
+ oprot.writeFieldStop()
+ oprot.writeStructEnd()
+
+ def validate(self):
+ return
+
+
+ def __repr__(self):
+ L = ['%s=%r' % (key, value)
+ for key, value in self.__dict__.iteritems()]
+ return '%s(%s)' % (self.__class__.__name__, ', '.join(L))
+
+ def __eq__(self, other):
+ return isinstance(other, self.__class__) and self.__dict__ == other.__dict__
+
+ def __ne__(self, other):
+ return not (self == other)
+
+class rebalance_result:
+ """
+ Attributes:
+ - e
+ """
+
+ thrift_spec = (
+ None, # 0
+ (1, TType.STRUCT, 'e', (NotAliveException, NotAliveException.thrift_spec), None, ), # 1
+ )
+
+ def __init__(self, e=None,):
+ self.e = e
+
+ def read(self, iprot):
+ if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None:
+ fastbinary.decode_binary(self, iprot.trans, (self.__class__, self.thrift_spec))
+ return
+ iprot.readStructBegin()
+ while True:
+ (fname, ftype, fid) = iprot.readFieldBegin()
+ if ftype == TType.STOP:
+ break
+ if fid == 1:
+ if ftype == TType.STRUCT:
+ self.e = NotAliveException()
+ self.e.read(iprot)
+ else:
+ iprot.skip(ftype)
+ else:
+ iprot.skip(ftype)
+ iprot.readFieldEnd()
+ iprot.readStructEnd()
+
+ def write(self, oprot):
+ if oprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and self.thrift_spec is not None and fastbinary is not None:
+ oprot.trans.write(fastbinary.encode_binary(self, (self.__class__, self.thrift_spec)))
+ return
+ oprot.writeStructBegin('rebalance_result')
+ if self.e is not None:
+ oprot.writeFieldBegin('e', TType.STRUCT, 1)
+ self.e.write(oprot)
+ oprot.writeFieldEnd()
+ oprot.writeFieldStop()
+ oprot.writeStructEnd()
+
+ def validate(self):
+ return
+
+
+ def __repr__(self):
+ L = ['%s=%r' % (key, value)
+ for key, value in self.__dict__.iteritems()]
+ return '%s(%s)' % (self.__class__.__name__, ', '.join(L))
+
+ def __eq__(self, other):
+ return isinstance(other, self.__class__) and self.__dict__ == other.__dict__
+
+ def __ne__(self, other):
+ return not (self == other)
+
class beginFileUpload_args:
thrift_spec = (
View
60 src/py/storm/ttypes.py
@@ -2671,6 +2671,66 @@ def __eq__(self, other):
def __ne__(self, other):
return not (self == other)
+class RebalanceOptions:
+ """
+ Attributes:
+ - wait_secs
+ """
+
+ thrift_spec = (
+ None, # 0
+ (1, TType.I32, 'wait_secs', None, None, ), # 1
+ )
+
+ def __init__(self, wait_secs=None,):
+ self.wait_secs = wait_secs
+
+ def read(self, iprot):
+ if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None:
+ fastbinary.decode_binary(self, iprot.trans, (self.__class__, self.thrift_spec))
+ return
+ iprot.readStructBegin()
+ while True:
+ (fname, ftype, fid) = iprot.readFieldBegin()
+ if ftype == TType.STOP:
+ break
+ if fid == 1:
+ if ftype == TType.I32:
+ self.wait_secs = iprot.readI32();
+ else:
+ iprot.skip(ftype)
+ else:
+ iprot.skip(ftype)
+ iprot.readFieldEnd()
+ iprot.readStructEnd()
+
+ def write(self, oprot):
+ if oprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and self.thrift_spec is not None and fastbinary is not None:
+ oprot.trans.write(fastbinary.encode_binary(self, (self.__class__, self.thrift_spec)))
+ return
+ oprot.writeStructBegin('RebalanceOptions')
+ if self.wait_secs is not None:
+ oprot.writeFieldBegin('wait_secs', TType.I32, 1)
+ oprot.writeI32(self.wait_secs)
+ oprot.writeFieldEnd()
+ oprot.writeFieldStop()
+ oprot.writeStructEnd()
+
+ def validate(self):
+ return
+
+
+ def __repr__(self):
+ L = ['%s=%r' % (key, value)
+ for key, value in self.__dict__.iteritems()]
+ return '%s(%s)' % (self.__class__.__name__, ', '.join(L))
+
+ def __eq__(self, other):
+ return isinstance(other, self.__class__) and self.__dict__ == other.__dict__
+
+ def __ne__(self, other):
+ return not (self == other)
+
class DRPCRequest:
"""
Attributes:
View
6 src/storm.thrift
@@ -171,12 +171,18 @@ struct KillOptions {
1: optional i32 wait_secs;
}
+struct RebalanceOptions {
+ 1: optional i32 wait_secs;
+}
+
+
service Nimbus {
void submitTopology(1: string name, 2: string uploadedJarLocation, 3: string jsonConf, 4: StormTopology topology) throws (1: AlreadyAliveException e, 2: InvalidTopologyException ite);
void killTopology(1: string name) throws (1: NotAliveException e);
void killTopologyWithOpts(1: string name, 2: KillOptions options) throws (1: NotAliveException e);
void activate(1: string name) throws (1: NotAliveException e);
void deactivate(1: string name) throws (1: NotAliveException e);
+ void rebalance(1: string name, 2: RebalanceOptions options) throws (1: NotAliveException e);
// need to add functions for asking about status of storms, what nodes they're running on, looking at task logs

0 comments on commit 277d4e6

Please sign in to comment.
Something went wrong with that request. Please try again.