You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by iv...@apache.org on 2012/08/06 13:01:37 UTC
svn commit: r1369778 [1/2] - in /zookeeper/bookkeeper/trunk: ./
hedwig-protocol/src/main/java/org/apache/hedwig/exceptions/
hedwig-protocol/src/main/java/org/apache/hedwig/protocol/
hedwig-protocol/src/main/protobuf/ hedwig-server/src/main/java/org/apa...
Author: ivank
Date: Mon Aug 6 11:01:36 2012
New Revision: 1369778
URL: http://svn.apache.org/viewvc?rev=1369778&view=rev
Log:
BOOKKEEPER-259: Create a topic manager using versioned write for leader election (sijie via ivank)
Added:
zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/TopicOwnershipManager.java
zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/HubInfo.java
zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/HubLoad.java
zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/HubServerManager.java
zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/MMTopicManager.java
zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/ZkHubServerManager.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/topics/TestHubInfo.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/topics/TestHubLoad.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/topics/TestMMTopicManager.java
- copied, changed from r1369767, zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/topics/TestZkTopicManager.java
Modified:
zookeeper/bookkeeper/trunk/CHANGES.txt
zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/java/org/apache/hedwig/exceptions/PubSubException.java
zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/java/org/apache/hedwig/protocol/PubSubProtocol.java
zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/protobuf/PubSubProtocol.proto
zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/common/ServerConfiguration.java
zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/MetadataManagerFactory.java
zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/ZkMetadataManagerFactory.java
zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/netty/PubSubServer.java
zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/ZkTopicManager.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestMetadataManager.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestMetadataManagerFactory.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/topics/TestZkTopicManager.java
Modified: zookeeper/bookkeeper/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/CHANGES.txt?rev=1369778&r1=1369777&r2=1369778&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/CHANGES.txt (original)
+++ zookeeper/bookkeeper/trunk/CHANGES.txt Mon Aug 6 11:01:36 2012
@@ -64,6 +64,8 @@ Trunk (unreleased changes)
BOOKKEEPER-343: Failed to register hedwig JMX beans in test cases (sijie via ivank)
+ BOOKKEEPER-259: Create a topic manager using versioned write for leader election (sijie via ivank)
+
IMPROVEMENTS:
bookkeeper-server:
Modified: zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/java/org/apache/hedwig/exceptions/PubSubException.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/java/org/apache/hedwig/exceptions/PubSubException.java?rev=1369778&r1=1369777&r2=1369778&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/java/org/apache/hedwig/exceptions/PubSubException.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/java/org/apache/hedwig/exceptions/PubSubException.java Mon Aug 6 11:01:36 2012
@@ -62,6 +62,10 @@ public abstract class PubSubException ex
return new NoSubscriptionStateException(msg);
} else if (code == StatusCode.SUBSCRIPTION_STATE_EXISTS) {
return new SubscriptionStateExistsException(msg);
+ } else if (code == StatusCode.NO_TOPIC_OWNER_INFO) {
+ return new NoTopicOwnerInfoException(msg);
+ } else if (code == StatusCode.TOPIC_OWNER_INFO_EXISTS) {
+ return new TopicOwnerInfoExistsException(msg);
}
/*
* Insert new ones here
@@ -162,6 +166,18 @@ public abstract class PubSubException ex
}
}
+ public static class NoTopicOwnerInfoException extends PubSubException {
+ public NoTopicOwnerInfoException(String msg) {
+ super(StatusCode.NO_TOPIC_OWNER_INFO, msg);
+ }
+ }
+
+ public static class TopicOwnerInfoExistsException extends PubSubException {
+ public TopicOwnerInfoExistsException(String msg) {
+ super(StatusCode.TOPIC_OWNER_INFO_EXISTS, msg);
+ }
+ }
+
/*
* Insert new ones here
*/
Modified: zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/java/org/apache/hedwig/protocol/PubSubProtocol.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/java/org/apache/hedwig/protocol/PubSubProtocol.java?rev=1369778&r1=1369777&r2=1369778&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/java/org/apache/hedwig/protocol/PubSubProtocol.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/java/org/apache/hedwig/protocol/PubSubProtocol.java Mon Aug 6 11:01:36 2012
@@ -172,8 +172,10 @@ public final class PubSubProtocol {
TOPIC_PERSISTENCE_INFO_EXISTS(12, 522),
NO_SUBSCRIPTION_STATE(13, 523),
SUBSCRIPTION_STATE_EXISTS(14, 524),
- UNEXPECTED_CONDITION(15, 600),
- COMPOSITE(16, 700),
+ NO_TOPIC_OWNER_INFO(15, 525),
+ TOPIC_OWNER_INFO_EXISTS(16, 526),
+ UNEXPECTED_CONDITION(17, 600),
+ COMPOSITE(18, 700),
;
public static final int SUCCESS_VALUE = 0;
@@ -191,6 +193,8 @@ public final class PubSubProtocol {
public static final int TOPIC_PERSISTENCE_INFO_EXISTS_VALUE = 522;
public static final int NO_SUBSCRIPTION_STATE_VALUE = 523;
public static final int SUBSCRIPTION_STATE_EXISTS_VALUE = 524;
+ public static final int NO_TOPIC_OWNER_INFO_VALUE = 525;
+ public static final int TOPIC_OWNER_INFO_EXISTS_VALUE = 526;
public static final int UNEXPECTED_CONDITION_VALUE = 600;
public static final int COMPOSITE_VALUE = 700;
@@ -214,6 +218,8 @@ public final class PubSubProtocol {
case 522: return TOPIC_PERSISTENCE_INFO_EXISTS;
case 523: return NO_SUBSCRIPTION_STATE;
case 524: return SUBSCRIPTION_STATE_EXISTS;
+ case 525: return NO_TOPIC_OWNER_INFO;
+ case 526: return TOPIC_OWNER_INFO_EXISTS;
case 600: return UNEXPECTED_CONDITION;
case 700: return COMPOSITE;
default: return null;
@@ -246,7 +252,7 @@ public final class PubSubProtocol {
}
private static final StatusCode[] VALUES = {
- SUCCESS, MALFORMED_REQUEST, NO_SUCH_TOPIC, CLIENT_ALREADY_SUBSCRIBED, CLIENT_NOT_SUBSCRIBED, COULD_NOT_CONNECT, TOPIC_BUSY, NOT_RESPONSIBLE_FOR_TOPIC, SERVICE_DOWN, UNCERTAIN_STATE, BAD_VERSION, NO_TOPIC_PERSISTENCE_INFO, TOPIC_PERSISTENCE_INFO_EXISTS, NO_SUBSCRIPTION_STATE, SUBSCRIPTION_STATE_EXISTS, UNEXPECTED_CONDITION, COMPOSITE,
+ SUCCESS, MALFORMED_REQUEST, NO_SUCH_TOPIC, CLIENT_ALREADY_SUBSCRIBED, CLIENT_NOT_SUBSCRIBED, COULD_NOT_CONNECT, TOPIC_BUSY, NOT_RESPONSIBLE_FOR_TOPIC, SERVICE_DOWN, UNCERTAIN_STATE, BAD_VERSION, NO_TOPIC_PERSISTENCE_INFO, TOPIC_PERSISTENCE_INFO_EXISTS, NO_SUBSCRIPTION_STATE, SUBSCRIPTION_STATE_EXISTS, NO_TOPIC_OWNER_INFO, TOPIC_OWNER_INFO_EXISTS, UNEXPECTED_CONDITION, COMPOSITE,
};
public static StatusCode valueOf(
@@ -10427,6 +10433,800 @@ public final class PubSubProtocol {
// @@protoc_insertion_point(class_scope:Hedwig.ManagerMeta)
}
+ public interface HubInfoDataOrBuilder
+ extends com.google.protobuf.MessageOrBuilder {
+
+ // required string hostname = 2;
+ boolean hasHostname();
+ String getHostname();
+
+ // required uint64 czxid = 3;
+ boolean hasCzxid();
+ long getCzxid();
+ }
+ public static final class HubInfoData extends
+ com.google.protobuf.GeneratedMessage
+ implements HubInfoDataOrBuilder {
+ // Use HubInfoData.newBuilder() to construct.
+ private HubInfoData(Builder builder) {
+ super(builder);
+ }
+ private HubInfoData(boolean noInit) {}
+
+ private static final HubInfoData defaultInstance;
+ public static HubInfoData getDefaultInstance() {
+ return defaultInstance;
+ }
+
+ public HubInfoData getDefaultInstanceForType() {
+ return defaultInstance;
+ }
+
+ public static final com.google.protobuf.Descriptors.Descriptor
+ getDescriptor() {
+ return org.apache.hedwig.protocol.PubSubProtocol.internal_static_Hedwig_HubInfoData_descriptor;
+ }
+
+ protected com.google.protobuf.GeneratedMessage.FieldAccessorTable
+ internalGetFieldAccessorTable() {
+ return org.apache.hedwig.protocol.PubSubProtocol.internal_static_Hedwig_HubInfoData_fieldAccessorTable;
+ }
+
+ private int bitField0_;
+ // required string hostname = 2;
+ public static final int HOSTNAME_FIELD_NUMBER = 2;
+ private java.lang.Object hostname_;
+ public boolean hasHostname() {
+ return ((bitField0_ & 0x00000001) == 0x00000001);
+ }
+ public String getHostname() {
+ java.lang.Object ref = hostname_;
+ if (ref instanceof String) {
+ return (String) ref;
+ } else {
+ com.google.protobuf.ByteString bs =
+ (com.google.protobuf.ByteString) ref;
+ String s = bs.toStringUtf8();
+ if (com.google.protobuf.Internal.isValidUtf8(bs)) {
+ hostname_ = s;
+ }
+ return s;
+ }
+ }
+ private com.google.protobuf.ByteString getHostnameBytes() {
+ java.lang.Object ref = hostname_;
+ if (ref instanceof String) {
+ com.google.protobuf.ByteString b =
+ com.google.protobuf.ByteString.copyFromUtf8((String) ref);
+ hostname_ = b;
+ return b;
+ } else {
+ return (com.google.protobuf.ByteString) ref;
+ }
+ }
+
+ // required uint64 czxid = 3;
+ public static final int CZXID_FIELD_NUMBER = 3;
+ private long czxid_;
+ public boolean hasCzxid() {
+ return ((bitField0_ & 0x00000002) == 0x00000002);
+ }
+ public long getCzxid() {
+ return czxid_;
+ }
+
+ private void initFields() {
+ hostname_ = "";
+ czxid_ = 0L;
+ }
+ private byte memoizedIsInitialized = -1;
+ public final boolean isInitialized() {
+ byte isInitialized = memoizedIsInitialized;
+ if (isInitialized != -1) return isInitialized == 1;
+
+ if (!hasHostname()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
+ if (!hasCzxid()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
+ memoizedIsInitialized = 1;
+ return true;
+ }
+
+ public void writeTo(com.google.protobuf.CodedOutputStream output)
+ throws java.io.IOException {
+ getSerializedSize();
+ if (((bitField0_ & 0x00000001) == 0x00000001)) {
+ output.writeBytes(2, getHostnameBytes());
+ }
+ if (((bitField0_ & 0x00000002) == 0x00000002)) {
+ output.writeUInt64(3, czxid_);
+ }
+ getUnknownFields().writeTo(output);
+ }
+
+ private int memoizedSerializedSize = -1;
+ public int getSerializedSize() {
+ int size = memoizedSerializedSize;
+ if (size != -1) return size;
+
+ size = 0;
+ if (((bitField0_ & 0x00000001) == 0x00000001)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeBytesSize(2, getHostnameBytes());
+ }
+ if (((bitField0_ & 0x00000002) == 0x00000002)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeUInt64Size(3, czxid_);
+ }
+ size += getUnknownFields().getSerializedSize();
+ memoizedSerializedSize = size;
+ return size;
+ }
+
+ private static final long serialVersionUID = 0L;
+ @java.lang.Override
+ protected java.lang.Object writeReplace()
+ throws java.io.ObjectStreamException {
+ return super.writeReplace();
+ }
+
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubInfoData parseFrom(
+ com.google.protobuf.ByteString data)
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ return newBuilder().mergeFrom(data).buildParsed();
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubInfoData parseFrom(
+ com.google.protobuf.ByteString data,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ return newBuilder().mergeFrom(data, extensionRegistry)
+ .buildParsed();
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubInfoData parseFrom(byte[] data)
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ return newBuilder().mergeFrom(data).buildParsed();
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubInfoData parseFrom(
+ byte[] data,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ return newBuilder().mergeFrom(data, extensionRegistry)
+ .buildParsed();
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubInfoData parseFrom(java.io.InputStream input)
+ throws java.io.IOException {
+ return newBuilder().mergeFrom(input).buildParsed();
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubInfoData parseFrom(
+ java.io.InputStream input,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return newBuilder().mergeFrom(input, extensionRegistry)
+ .buildParsed();
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubInfoData parseDelimitedFrom(java.io.InputStream input)
+ throws java.io.IOException {
+ Builder builder = newBuilder();
+ if (builder.mergeDelimitedFrom(input)) {
+ return builder.buildParsed();
+ } else {
+ return null;
+ }
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubInfoData parseDelimitedFrom(
+ java.io.InputStream input,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ Builder builder = newBuilder();
+ if (builder.mergeDelimitedFrom(input, extensionRegistry)) {
+ return builder.buildParsed();
+ } else {
+ return null;
+ }
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubInfoData parseFrom(
+ com.google.protobuf.CodedInputStream input)
+ throws java.io.IOException {
+ return newBuilder().mergeFrom(input).buildParsed();
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubInfoData parseFrom(
+ com.google.protobuf.CodedInputStream input,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return newBuilder().mergeFrom(input, extensionRegistry)
+ .buildParsed();
+ }
+
+ public static Builder newBuilder() { return Builder.create(); }
+ public Builder newBuilderForType() { return newBuilder(); }
+ public static Builder newBuilder(org.apache.hedwig.protocol.PubSubProtocol.HubInfoData prototype) {
+ return newBuilder().mergeFrom(prototype);
+ }
+ public Builder toBuilder() { return newBuilder(this); }
+
+ @java.lang.Override
+ protected Builder newBuilderForType(
+ com.google.protobuf.GeneratedMessage.BuilderParent parent) {
+ Builder builder = new Builder(parent);
+ return builder;
+ }
+ public static final class Builder extends
+ com.google.protobuf.GeneratedMessage.Builder<Builder>
+ implements org.apache.hedwig.protocol.PubSubProtocol.HubInfoDataOrBuilder {
+ public static final com.google.protobuf.Descriptors.Descriptor
+ getDescriptor() {
+ return org.apache.hedwig.protocol.PubSubProtocol.internal_static_Hedwig_HubInfoData_descriptor;
+ }
+
+ protected com.google.protobuf.GeneratedMessage.FieldAccessorTable
+ internalGetFieldAccessorTable() {
+ return org.apache.hedwig.protocol.PubSubProtocol.internal_static_Hedwig_HubInfoData_fieldAccessorTable;
+ }
+
+ // Construct using org.apache.hedwig.protocol.PubSubProtocol.HubInfoData.newBuilder()
+ private Builder() {
+ maybeForceBuilderInitialization();
+ }
+
+ private Builder(BuilderParent parent) {
+ super(parent);
+ maybeForceBuilderInitialization();
+ }
+ private void maybeForceBuilderInitialization() {
+ if (com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders) {
+ }
+ }
+ private static Builder create() {
+ return new Builder();
+ }
+
+ public Builder clear() {
+ super.clear();
+ hostname_ = "";
+ bitField0_ = (bitField0_ & ~0x00000001);
+ czxid_ = 0L;
+ bitField0_ = (bitField0_ & ~0x00000002);
+ return this;
+ }
+
+ public Builder clone() {
+ return create().mergeFrom(buildPartial());
+ }
+
+ public com.google.protobuf.Descriptors.Descriptor
+ getDescriptorForType() {
+ return org.apache.hedwig.protocol.PubSubProtocol.HubInfoData.getDescriptor();
+ }
+
+ public org.apache.hedwig.protocol.PubSubProtocol.HubInfoData getDefaultInstanceForType() {
+ return org.apache.hedwig.protocol.PubSubProtocol.HubInfoData.getDefaultInstance();
+ }
+
+ public org.apache.hedwig.protocol.PubSubProtocol.HubInfoData build() {
+ org.apache.hedwig.protocol.PubSubProtocol.HubInfoData result = buildPartial();
+ if (!result.isInitialized()) {
+ throw newUninitializedMessageException(result);
+ }
+ return result;
+ }
+
+ private org.apache.hedwig.protocol.PubSubProtocol.HubInfoData buildParsed()
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ org.apache.hedwig.protocol.PubSubProtocol.HubInfoData result = buildPartial();
+ if (!result.isInitialized()) {
+ throw newUninitializedMessageException(
+ result).asInvalidProtocolBufferException();
+ }
+ return result;
+ }
+
+ public org.apache.hedwig.protocol.PubSubProtocol.HubInfoData buildPartial() {
+ org.apache.hedwig.protocol.PubSubProtocol.HubInfoData result = new org.apache.hedwig.protocol.PubSubProtocol.HubInfoData(this);
+ int from_bitField0_ = bitField0_;
+ int to_bitField0_ = 0;
+ if (((from_bitField0_ & 0x00000001) == 0x00000001)) {
+ to_bitField0_ |= 0x00000001;
+ }
+ result.hostname_ = hostname_;
+ if (((from_bitField0_ & 0x00000002) == 0x00000002)) {
+ to_bitField0_ |= 0x00000002;
+ }
+ result.czxid_ = czxid_;
+ result.bitField0_ = to_bitField0_;
+ onBuilt();
+ return result;
+ }
+
+ public Builder mergeFrom(com.google.protobuf.Message other) {
+ if (other instanceof org.apache.hedwig.protocol.PubSubProtocol.HubInfoData) {
+ return mergeFrom((org.apache.hedwig.protocol.PubSubProtocol.HubInfoData)other);
+ } else {
+ super.mergeFrom(other);
+ return this;
+ }
+ }
+
+ public Builder mergeFrom(org.apache.hedwig.protocol.PubSubProtocol.HubInfoData other) {
+ if (other == org.apache.hedwig.protocol.PubSubProtocol.HubInfoData.getDefaultInstance()) return this;
+ if (other.hasHostname()) {
+ setHostname(other.getHostname());
+ }
+ if (other.hasCzxid()) {
+ setCzxid(other.getCzxid());
+ }
+ this.mergeUnknownFields(other.getUnknownFields());
+ return this;
+ }
+
+ public final boolean isInitialized() {
+ if (!hasHostname()) {
+
+ return false;
+ }
+ if (!hasCzxid()) {
+
+ return false;
+ }
+ return true;
+ }
+
+ public Builder mergeFrom(
+ com.google.protobuf.CodedInputStream input,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ com.google.protobuf.UnknownFieldSet.Builder unknownFields =
+ com.google.protobuf.UnknownFieldSet.newBuilder(
+ this.getUnknownFields());
+ while (true) {
+ int tag = input.readTag();
+ switch (tag) {
+ case 0:
+ this.setUnknownFields(unknownFields.build());
+ onChanged();
+ return this;
+ default: {
+ if (!parseUnknownField(input, unknownFields,
+ extensionRegistry, tag)) {
+ this.setUnknownFields(unknownFields.build());
+ onChanged();
+ return this;
+ }
+ break;
+ }
+ case 18: {
+ bitField0_ |= 0x00000001;
+ hostname_ = input.readBytes();
+ break;
+ }
+ case 24: {
+ bitField0_ |= 0x00000002;
+ czxid_ = input.readUInt64();
+ break;
+ }
+ }
+ }
+ }
+
+ private int bitField0_;
+
+ // required string hostname = 2;
+ private java.lang.Object hostname_ = "";
+ public boolean hasHostname() {
+ return ((bitField0_ & 0x00000001) == 0x00000001);
+ }
+ public String getHostname() {
+ java.lang.Object ref = hostname_;
+ if (!(ref instanceof String)) {
+ String s = ((com.google.protobuf.ByteString) ref).toStringUtf8();
+ hostname_ = s;
+ return s;
+ } else {
+ return (String) ref;
+ }
+ }
+ public Builder setHostname(String value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ bitField0_ |= 0x00000001;
+ hostname_ = value;
+ onChanged();
+ return this;
+ }
+ public Builder clearHostname() {
+ bitField0_ = (bitField0_ & ~0x00000001);
+ hostname_ = getDefaultInstance().getHostname();
+ onChanged();
+ return this;
+ }
+ void setHostname(com.google.protobuf.ByteString value) {
+ bitField0_ |= 0x00000001;
+ hostname_ = value;
+ onChanged();
+ }
+
+ // required uint64 czxid = 3;
+ private long czxid_ ;
+ public boolean hasCzxid() {
+ return ((bitField0_ & 0x00000002) == 0x00000002);
+ }
+ public long getCzxid() {
+ return czxid_;
+ }
+ public Builder setCzxid(long value) {
+ bitField0_ |= 0x00000002;
+ czxid_ = value;
+ onChanged();
+ return this;
+ }
+ public Builder clearCzxid() {
+ bitField0_ = (bitField0_ & ~0x00000002);
+ czxid_ = 0L;
+ onChanged();
+ return this;
+ }
+
+ // @@protoc_insertion_point(builder_scope:Hedwig.HubInfoData)
+ }
+
+ static {
+ defaultInstance = new HubInfoData(true);
+ defaultInstance.initFields();
+ }
+
+ // @@protoc_insertion_point(class_scope:Hedwig.HubInfoData)
+ }
+
+ public interface HubLoadDataOrBuilder
+ extends com.google.protobuf.MessageOrBuilder {
+
+ // required uint64 numTopics = 2;
+ boolean hasNumTopics();
+ long getNumTopics();
+ }
+ public static final class HubLoadData extends
+ com.google.protobuf.GeneratedMessage
+ implements HubLoadDataOrBuilder {
+ // Use HubLoadData.newBuilder() to construct.
+ private HubLoadData(Builder builder) {
+ super(builder);
+ }
+ private HubLoadData(boolean noInit) {}
+
+ private static final HubLoadData defaultInstance;
+ public static HubLoadData getDefaultInstance() {
+ return defaultInstance;
+ }
+
+ public HubLoadData getDefaultInstanceForType() {
+ return defaultInstance;
+ }
+
+ public static final com.google.protobuf.Descriptors.Descriptor
+ getDescriptor() {
+ return org.apache.hedwig.protocol.PubSubProtocol.internal_static_Hedwig_HubLoadData_descriptor;
+ }
+
+ protected com.google.protobuf.GeneratedMessage.FieldAccessorTable
+ internalGetFieldAccessorTable() {
+ return org.apache.hedwig.protocol.PubSubProtocol.internal_static_Hedwig_HubLoadData_fieldAccessorTable;
+ }
+
+ private int bitField0_;
+ // required uint64 numTopics = 2;
+ public static final int NUMTOPICS_FIELD_NUMBER = 2;
+ private long numTopics_;
+ public boolean hasNumTopics() {
+ return ((bitField0_ & 0x00000001) == 0x00000001);
+ }
+ public long getNumTopics() {
+ return numTopics_;
+ }
+
+ private void initFields() {
+ numTopics_ = 0L;
+ }
+ private byte memoizedIsInitialized = -1;
+ public final boolean isInitialized() {
+ byte isInitialized = memoizedIsInitialized;
+ if (isInitialized != -1) return isInitialized == 1;
+
+ if (!hasNumTopics()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
+ memoizedIsInitialized = 1;
+ return true;
+ }
+
+ public void writeTo(com.google.protobuf.CodedOutputStream output)
+ throws java.io.IOException {
+ getSerializedSize();
+ if (((bitField0_ & 0x00000001) == 0x00000001)) {
+ output.writeUInt64(2, numTopics_);
+ }
+ getUnknownFields().writeTo(output);
+ }
+
+ private int memoizedSerializedSize = -1;
+ public int getSerializedSize() {
+ int size = memoizedSerializedSize;
+ if (size != -1) return size;
+
+ size = 0;
+ if (((bitField0_ & 0x00000001) == 0x00000001)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeUInt64Size(2, numTopics_);
+ }
+ size += getUnknownFields().getSerializedSize();
+ memoizedSerializedSize = size;
+ return size;
+ }
+
+ private static final long serialVersionUID = 0L;
+ @java.lang.Override
+ protected java.lang.Object writeReplace()
+ throws java.io.ObjectStreamException {
+ return super.writeReplace();
+ }
+
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubLoadData parseFrom(
+ com.google.protobuf.ByteString data)
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ return newBuilder().mergeFrom(data).buildParsed();
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubLoadData parseFrom(
+ com.google.protobuf.ByteString data,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ return newBuilder().mergeFrom(data, extensionRegistry)
+ .buildParsed();
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubLoadData parseFrom(byte[] data)
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ return newBuilder().mergeFrom(data).buildParsed();
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubLoadData parseFrom(
+ byte[] data,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ return newBuilder().mergeFrom(data, extensionRegistry)
+ .buildParsed();
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubLoadData parseFrom(java.io.InputStream input)
+ throws java.io.IOException {
+ return newBuilder().mergeFrom(input).buildParsed();
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubLoadData parseFrom(
+ java.io.InputStream input,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return newBuilder().mergeFrom(input, extensionRegistry)
+ .buildParsed();
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubLoadData parseDelimitedFrom(java.io.InputStream input)
+ throws java.io.IOException {
+ Builder builder = newBuilder();
+ if (builder.mergeDelimitedFrom(input)) {
+ return builder.buildParsed();
+ } else {
+ return null;
+ }
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubLoadData parseDelimitedFrom(
+ java.io.InputStream input,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ Builder builder = newBuilder();
+ if (builder.mergeDelimitedFrom(input, extensionRegistry)) {
+ return builder.buildParsed();
+ } else {
+ return null;
+ }
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubLoadData parseFrom(
+ com.google.protobuf.CodedInputStream input)
+ throws java.io.IOException {
+ return newBuilder().mergeFrom(input).buildParsed();
+ }
+ public static org.apache.hedwig.protocol.PubSubProtocol.HubLoadData parseFrom(
+ com.google.protobuf.CodedInputStream input,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return newBuilder().mergeFrom(input, extensionRegistry)
+ .buildParsed();
+ }
+
+ public static Builder newBuilder() { return Builder.create(); }
+ public Builder newBuilderForType() { return newBuilder(); }
+ public static Builder newBuilder(org.apache.hedwig.protocol.PubSubProtocol.HubLoadData prototype) {
+ return newBuilder().mergeFrom(prototype);
+ }
+ public Builder toBuilder() { return newBuilder(this); }
+
+ @java.lang.Override
+ protected Builder newBuilderForType(
+ com.google.protobuf.GeneratedMessage.BuilderParent parent) {
+ Builder builder = new Builder(parent);
+ return builder;
+ }
+ public static final class Builder extends
+ com.google.protobuf.GeneratedMessage.Builder<Builder>
+ implements org.apache.hedwig.protocol.PubSubProtocol.HubLoadDataOrBuilder {
+ public static final com.google.protobuf.Descriptors.Descriptor
+ getDescriptor() {
+ return org.apache.hedwig.protocol.PubSubProtocol.internal_static_Hedwig_HubLoadData_descriptor;
+ }
+
+ protected com.google.protobuf.GeneratedMessage.FieldAccessorTable
+ internalGetFieldAccessorTable() {
+ return org.apache.hedwig.protocol.PubSubProtocol.internal_static_Hedwig_HubLoadData_fieldAccessorTable;
+ }
+
+ // Construct using org.apache.hedwig.protocol.PubSubProtocol.HubLoadData.newBuilder()
+ private Builder() {
+ maybeForceBuilderInitialization();
+ }
+
+ private Builder(BuilderParent parent) {
+ super(parent);
+ maybeForceBuilderInitialization();
+ }
+ private void maybeForceBuilderInitialization() {
+ if (com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders) {
+ }
+ }
+ private static Builder create() {
+ return new Builder();
+ }
+
+ public Builder clear() {
+ super.clear();
+ numTopics_ = 0L;
+ bitField0_ = (bitField0_ & ~0x00000001);
+ return this;
+ }
+
+ public Builder clone() {
+ return create().mergeFrom(buildPartial());
+ }
+
+ public com.google.protobuf.Descriptors.Descriptor
+ getDescriptorForType() {
+ return org.apache.hedwig.protocol.PubSubProtocol.HubLoadData.getDescriptor();
+ }
+
+ public org.apache.hedwig.protocol.PubSubProtocol.HubLoadData getDefaultInstanceForType() {
+ return org.apache.hedwig.protocol.PubSubProtocol.HubLoadData.getDefaultInstance();
+ }
+
+ public org.apache.hedwig.protocol.PubSubProtocol.HubLoadData build() {
+ org.apache.hedwig.protocol.PubSubProtocol.HubLoadData result = buildPartial();
+ if (!result.isInitialized()) {
+ throw newUninitializedMessageException(result);
+ }
+ return result;
+ }
+
+ private org.apache.hedwig.protocol.PubSubProtocol.HubLoadData buildParsed()
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ org.apache.hedwig.protocol.PubSubProtocol.HubLoadData result = buildPartial();
+ if (!result.isInitialized()) {
+ throw newUninitializedMessageException(
+ result).asInvalidProtocolBufferException();
+ }
+ return result;
+ }
+
+ public org.apache.hedwig.protocol.PubSubProtocol.HubLoadData buildPartial() {
+ org.apache.hedwig.protocol.PubSubProtocol.HubLoadData result = new org.apache.hedwig.protocol.PubSubProtocol.HubLoadData(this);
+ int from_bitField0_ = bitField0_;
+ int to_bitField0_ = 0;
+ if (((from_bitField0_ & 0x00000001) == 0x00000001)) {
+ to_bitField0_ |= 0x00000001;
+ }
+ result.numTopics_ = numTopics_;
+ result.bitField0_ = to_bitField0_;
+ onBuilt();
+ return result;
+ }
+
+ public Builder mergeFrom(com.google.protobuf.Message other) {
+ if (other instanceof org.apache.hedwig.protocol.PubSubProtocol.HubLoadData) {
+ return mergeFrom((org.apache.hedwig.protocol.PubSubProtocol.HubLoadData)other);
+ } else {
+ super.mergeFrom(other);
+ return this;
+ }
+ }
+
+ public Builder mergeFrom(org.apache.hedwig.protocol.PubSubProtocol.HubLoadData other) {
+ if (other == org.apache.hedwig.protocol.PubSubProtocol.HubLoadData.getDefaultInstance()) return this;
+ if (other.hasNumTopics()) {
+ setNumTopics(other.getNumTopics());
+ }
+ this.mergeUnknownFields(other.getUnknownFields());
+ return this;
+ }
+
+ public final boolean isInitialized() {
+ if (!hasNumTopics()) {
+
+ return false;
+ }
+ return true;
+ }
+
+ public Builder mergeFrom(
+ com.google.protobuf.CodedInputStream input,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ com.google.protobuf.UnknownFieldSet.Builder unknownFields =
+ com.google.protobuf.UnknownFieldSet.newBuilder(
+ this.getUnknownFields());
+ while (true) {
+ int tag = input.readTag();
+ switch (tag) {
+ case 0:
+ this.setUnknownFields(unknownFields.build());
+ onChanged();
+ return this;
+ default: {
+ if (!parseUnknownField(input, unknownFields,
+ extensionRegistry, tag)) {
+ this.setUnknownFields(unknownFields.build());
+ onChanged();
+ return this;
+ }
+ break;
+ }
+ case 16: {
+ bitField0_ |= 0x00000001;
+ numTopics_ = input.readUInt64();
+ break;
+ }
+ }
+ }
+ }
+
+ private int bitField0_;
+
+ // required uint64 numTopics = 2;
+ private long numTopics_ ;
+ public boolean hasNumTopics() {
+ return ((bitField0_ & 0x00000001) == 0x00000001);
+ }
+ public long getNumTopics() {
+ return numTopics_;
+ }
+ public Builder setNumTopics(long value) {
+ bitField0_ |= 0x00000001;
+ numTopics_ = value;
+ onChanged();
+ return this;
+ }
+ public Builder clearNumTopics() {
+ bitField0_ = (bitField0_ & ~0x00000001);
+ numTopics_ = 0L;
+ onChanged();
+ return this;
+ }
+
+ // @@protoc_insertion_point(builder_scope:Hedwig.HubLoadData)
+ }
+
+ static {
+ defaultInstance = new HubLoadData(true);
+ defaultInstance.initFields();
+ }
+
+ // @@protoc_insertion_point(class_scope:Hedwig.HubLoadData)
+ }
+
private static com.google.protobuf.Descriptors.Descriptor
internal_static_Hedwig_Message_descriptor;
private static
@@ -10517,6 +11317,16 @@ public final class PubSubProtocol {
private static
com.google.protobuf.GeneratedMessage.FieldAccessorTable
internal_static_Hedwig_ManagerMeta_fieldAccessorTable;
+ private static com.google.protobuf.Descriptors.Descriptor
+ internal_static_Hedwig_HubInfoData_descriptor;
+ private static
+ com.google.protobuf.GeneratedMessage.FieldAccessorTable
+ internal_static_Hedwig_HubInfoData_fieldAccessorTable;
+ private static com.google.protobuf.Descriptors.Descriptor
+ internal_static_Hedwig_HubLoadData_descriptor;
+ private static
+ com.google.protobuf.GeneratedMessage.FieldAccessorTable
+ internal_static_Hedwig_HubLoadData_fieldAccessorTable;
public static com.google.protobuf.Descriptors.FileDescriptor
getDescriptor() {
@@ -10577,23 +11387,26 @@ public final class PubSubProtocol {
"eqIdIncluded\030\002 \001(\0132\024.Hedwig.MessageSeqId" +
"\"3\n\014LedgerRanges\022#\n\006ranges\030\001 \003(\0132\023.Hedwi",
"g.LedgerRange\":\n\013ManagerMeta\022\023\n\013managerI" +
- "mpl\030\002 \002(\t\022\026\n\016managerVersion\030\003 \002(\r*\"\n\017Pro" +
- "tocolVersion\022\017\n\013VERSION_ONE\020\001*p\n\rOperati" +
- "onType\022\013\n\007PUBLISH\020\000\022\r\n\tSUBSCRIBE\020\001\022\013\n\007CO" +
- "NSUME\020\002\022\017\n\013UNSUBSCRIBE\020\003\022\022\n\016START_DELIVE" +
- "RY\020\004\022\021\n\rSTOP_DELIVERY\020\005*\260\003\n\nStatusCode\022\013" +
- "\n\007SUCCESS\020\000\022\026\n\021MALFORMED_REQUEST\020\221\003\022\022\n\rN" +
- "O_SUCH_TOPIC\020\222\003\022\036\n\031CLIENT_ALREADY_SUBSCR" +
- "IBED\020\223\003\022\032\n\025CLIENT_NOT_SUBSCRIBED\020\224\003\022\026\n\021C" +
- "OULD_NOT_CONNECT\020\225\003\022\017\n\nTOPIC_BUSY\020\226\003\022\036\n\031",
- "NOT_RESPONSIBLE_FOR_TOPIC\020\365\003\022\021\n\014SERVICE_" +
- "DOWN\020\366\003\022\024\n\017UNCERTAIN_STATE\020\367\003\022\020\n\013BAD_VER" +
- "SION\020\210\004\022\036\n\031NO_TOPIC_PERSISTENCE_INFO\020\211\004\022" +
- "\"\n\035TOPIC_PERSISTENCE_INFO_EXISTS\020\212\004\022\032\n\025N" +
- "O_SUBSCRIPTION_STATE\020\213\004\022\036\n\031SUBSCRIPTION_" +
- "STATE_EXISTS\020\214\004\022\031\n\024UNEXPECTED_CONDITION\020" +
- "\330\004\022\016\n\tCOMPOSITE\020\274\005B\036\n\032org.apache.hedwig." +
- "protocolH\001"
+ "mpl\030\002 \002(\t\022\026\n\016managerVersion\030\003 \002(\r\".\n\013Hub" +
+ "InfoData\022\020\n\010hostname\030\002 \002(\t\022\r\n\005czxid\030\003 \002(" +
+ "\004\" \n\013HubLoadData\022\021\n\tnumTopics\030\002 \002(\004*\"\n\017P" +
+ "rotocolVersion\022\017\n\013VERSION_ONE\020\001*p\n\rOpera" +
+ "tionType\022\013\n\007PUBLISH\020\000\022\r\n\tSUBSCRIBE\020\001\022\013\n\007" +
+ "CONSUME\020\002\022\017\n\013UNSUBSCRIBE\020\003\022\022\n\016START_DELI" +
+ "VERY\020\004\022\021\n\rSTOP_DELIVERY\020\005*\350\003\n\nStatusCode" +
+ "\022\013\n\007SUCCESS\020\000\022\026\n\021MALFORMED_REQUEST\020\221\003\022\022\n" +
+ "\rNO_SUCH_TOPIC\020\222\003\022\036\n\031CLIENT_ALREADY_SUBS",
+ "CRIBED\020\223\003\022\032\n\025CLIENT_NOT_SUBSCRIBED\020\224\003\022\026\n" +
+ "\021COULD_NOT_CONNECT\020\225\003\022\017\n\nTOPIC_BUSY\020\226\003\022\036" +
+ "\n\031NOT_RESPONSIBLE_FOR_TOPIC\020\365\003\022\021\n\014SERVIC" +
+ "E_DOWN\020\366\003\022\024\n\017UNCERTAIN_STATE\020\367\003\022\020\n\013BAD_V" +
+ "ERSION\020\210\004\022\036\n\031NO_TOPIC_PERSISTENCE_INFO\020\211" +
+ "\004\022\"\n\035TOPIC_PERSISTENCE_INFO_EXISTS\020\212\004\022\032\n" +
+ "\025NO_SUBSCRIPTION_STATE\020\213\004\022\036\n\031SUBSCRIPTIO" +
+ "N_STATE_EXISTS\020\214\004\022\030\n\023NO_TOPIC_OWNER_INFO" +
+ "\020\215\004\022\034\n\027TOPIC_OWNER_INFO_EXISTS\020\216\004\022\031\n\024UNE" +
+ "XPECTED_CONDITION\020\330\004\022\016\n\tCOMPOSITE\020\274\005B\036\n\032",
+ "org.apache.hedwig.protocolH\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -10744,6 +11557,22 @@ public final class PubSubProtocol {
new java.lang.String[] { "ManagerImpl", "ManagerVersion", },
org.apache.hedwig.protocol.PubSubProtocol.ManagerMeta.class,
org.apache.hedwig.protocol.PubSubProtocol.ManagerMeta.Builder.class);
+ internal_static_Hedwig_HubInfoData_descriptor =
+ getDescriptor().getMessageTypes().get(18);
+ internal_static_Hedwig_HubInfoData_fieldAccessorTable = new
+ com.google.protobuf.GeneratedMessage.FieldAccessorTable(
+ internal_static_Hedwig_HubInfoData_descriptor,
+ new java.lang.String[] { "Hostname", "Czxid", },
+ org.apache.hedwig.protocol.PubSubProtocol.HubInfoData.class,
+ org.apache.hedwig.protocol.PubSubProtocol.HubInfoData.Builder.class);
+ internal_static_Hedwig_HubLoadData_descriptor =
+ getDescriptor().getMessageTypes().get(19);
+ internal_static_Hedwig_HubLoadData_fieldAccessorTable = new
+ com.google.protobuf.GeneratedMessage.FieldAccessorTable(
+ internal_static_Hedwig_HubLoadData_descriptor,
+ new java.lang.String[] { "NumTopics", },
+ org.apache.hedwig.protocol.PubSubProtocol.HubLoadData.class,
+ org.apache.hedwig.protocol.PubSubProtocol.HubLoadData.Builder.class);
return null;
}
};
Modified: zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/protobuf/PubSubProtocol.proto
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/protobuf/PubSubProtocol.proto?rev=1369778&r1=1369777&r2=1369778&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/protobuf/PubSubProtocol.proto (original)
+++ zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/protobuf/PubSubProtocol.proto Mon Aug 6 11:01:36 2012
@@ -175,6 +175,8 @@ enum StatusCode{
TOPIC_PERSISTENCE_INFO_EXISTS = 522;
NO_SUBSCRIPTION_STATE = 523;
SUBSCRIPTION_STATE_EXISTS = 524;
+ NO_TOPIC_OWNER_INFO = 525;
+ TOPIC_OWNER_INFO_EXISTS = 526;
//For all unexpected error conditions
UNEXPECTED_CONDITION = 600;
@@ -203,3 +205,12 @@ message ManagerMeta {
required string managerImpl = 2;
required uint32 managerVersion = 3;
}
+
+message HubInfoData {
+ required string hostname = 2;
+ required uint64 czxid = 3;
+}
+
+message HubLoadData {
+ required uint64 numTopics = 2;
+}
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/common/ServerConfiguration.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/common/ServerConfiguration.java?rev=1369778&r1=1369777&r2=1369778&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/common/ServerConfiguration.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/common/ServerConfiguration.java Mon Aug 6 11:01:36 2012
@@ -63,6 +63,7 @@ public class ServerConfiguration extends
protected final static String RETRY_REMOTE_SUBSCRIBE_THREAD_RUN_INTERVAL = "retry_remote_subscribe_thread_run_interval";
// manager related settings
+ protected final static String METADATA_MANAGER_BASED_TOPIC_MANAGER_ENABLED = "metadata_manager_based_topic_manager_enabled";
protected final static String METADATA_MANAGER_FACTORY_CLASS = "metadata_manager_factory_class";
private static ClassLoader defaultLoader;
@@ -322,6 +323,15 @@ public class ServerConfiguration extends
}
/**
+ * Whether enable metadata manager based topic manager.
+ *
+ * @return true if enabled metadata manager based topic manager.
+ */
+ public boolean isMetadataManagerBasedTopicManagerEnabled() {
+ return conf.getBoolean(METADATA_MANAGER_BASED_TOPIC_MANAGER_ENABLED, false);
+ }
+
+ /**
* Get metadata manager factory class.
*
* @return manager class
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/MetadataManagerFactory.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/MetadataManagerFactory.java?rev=1369778&r1=1369777&r2=1369778&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/MetadataManagerFactory.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/MetadataManagerFactory.java Mon Aug 6 11:01:36 2012
@@ -81,6 +81,13 @@ public abstract class MetadataManagerFac
public abstract SubscriptionDataManager newSubscriptionDataManager();
/**
+ * Create topic ownership manager.
+ *
+ * @return topic ownership manager.
+ */
+ public abstract TopicOwnershipManager newTopicOwnershipManager();
+
+ /**
* Create new Metadata Manager Factory.
*
* @param conf
Added: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/TopicOwnershipManager.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/TopicOwnershipManager.java?rev=1369778&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/TopicOwnershipManager.java (added)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/TopicOwnershipManager.java Mon Aug 6 11:01:36 2012
@@ -0,0 +1,100 @@
+/**
+ * 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.hedwig.server.meta;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Map;
+
+import com.google.protobuf.ByteString;
+
+import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
+import org.apache.hedwig.protocol.PubSubProtocol.LedgerRanges;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionState;
+import org.apache.hedwig.server.common.ServerConfiguration;
+import org.apache.hedwig.server.subscriptions.InMemorySubscriptionState;
+import org.apache.hedwig.server.topics.HubInfo;
+import org.apache.hedwig.util.Callback;
+import org.apache.zookeeper.ZooKeeper;
+
+/**
+ * Manage topic ownership
+ */
+public interface TopicOwnershipManager extends Closeable {
+
+ /**
+ * Read owner information of a topic.
+ *
+ * @param topic
+ * Topic Name
+ * @param callback
+ * Callback to return hub info. If there is no owner info, return null;
+ * If there is data but not valid owner info, return a Versioned object with null hub info;
+ * If there is valid owner info, return versioned hub info.
+ * @param ctx
+ * Context of the callback
+ */
+ public void readOwnerInfo(ByteString topic, Callback<Versioned<HubInfo>> callback, Object ctx);
+
+ /**
+ * Write owner info for a specified topic.
+ * A new owner info would be created if there is no one existed before.
+ *
+ * @param topic
+ * Topic Name
+ * @param owner
+ * Owner hub info
+ * @param version
+ * Current version of owner info
+ * If <code>version</code> is null, create owner info.
+ * {@link PubSubException.TopicOwnerInfoExistsException} is returned when
+ * owner info existed before.
+ * If <code>version</code> is not null, the owner info is updated only when
+ * provided version equals to its current version.
+ * {@link PubSubException.BadVersionException} is returned when version doesn't match,
+ * {@link PubSubException.NoTopicOwnerInfoException} is returned when no owner info
+ * found to update.
+ * @param callback
+ * Callback when owner info updated. New version would be returned if succeed to write.
+ * @param ctx
+ * Context of the callback
+ */
+ public void writeOwnerInfo(ByteString topic, HubInfo owner, Version version,
+ Callback<Version> callback, Object ctx);
+
+ /**
+ * Delete owner info for a specified topic.
+ *
+ * @param topic
+ * Topic Name
+ * @param version
+ * Current version of owner info
+ * If <code>version</code> is null, delete owner info no matter its current version.
+ * If <code>version</code> is not null, the owner info is deleted only when
+ * provided version equals to its current version.
+ * @param callback
+ * Callback when owner info deleted.
+ * {@link PubSubException.NoTopicOwnerInfoException} is returned when no owner info.
+ * {@link PubSubException.BadVersionException} is returned when version doesn't match.
+ * @param ctx
+ * Context of the callback.
+ */
+ public void deleteOwnerInfo(ByteString topic, Version version,
+ Callback<Void> callback, Object ctx);
+}
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/ZkMetadataManagerFactory.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/ZkMetadataManagerFactory.java?rev=1369778&r1=1369777&r2=1369778&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/ZkMetadataManagerFactory.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/ZkMetadataManagerFactory.java Mon Aug 6 11:01:36 2012
@@ -44,6 +44,7 @@ import org.apache.hedwig.protocol.PubSub
import org.apache.hedwig.protoextensions.SubscriptionStateUtils;
import org.apache.hedwig.server.common.ServerConfiguration;
import org.apache.hedwig.server.subscriptions.InMemorySubscriptionState;
+import org.apache.hedwig.server.topics.HubInfo;
import org.apache.hedwig.util.Callback;
import org.apache.hedwig.zookeeper.SafeAsyncZKCallback;
import org.apache.hedwig.zookeeper.ZkUtils;
@@ -94,6 +95,11 @@ public class ZkMetadataManagerFactory ex
return new ZkSubscriptionDataManagerImpl(cfg, zk);
}
+ @Override
+ public TopicOwnershipManager newTopicOwnershipManager() {
+ return new ZkTopicOwnershipManagerImpl(cfg, zk);
+ }
+
/**
* ZooKeeper based topic persistence manager.
*/
@@ -556,4 +562,177 @@ public class ZkMetadataManagerFactory ex
}, ctx);
}
}
+
+ /**
+ * ZooKeeper base topic ownership manager.
+ */
+ static class ZkTopicOwnershipManagerImpl implements TopicOwnershipManager {
+
+ ZooKeeper zk;
+ ServerConfiguration cfg;
+
+ ZkTopicOwnershipManagerImpl(ServerConfiguration conf, ZooKeeper zk) {
+ this.cfg = conf;
+ this.zk = zk;
+ }
+
+ @Override
+ public void close() throws IOException {
+ // do nothing in zookeeper based impl
+ }
+
+ /**
+ * Return znode path to store topic owner.
+ *
+ * @param topic
+ * Topic Name
+ * @return znode path to store topic owner.
+ */
+ String hubPath(ByteString topic) {
+ return cfg.getZkTopicPath(new StringBuilder(), topic).append("/hub").toString();
+ }
+
+ @Override
+ public void readOwnerInfo(final ByteString topic, final Callback<Versioned<HubInfo>> callback, Object ctx) {
+ String ownerPath = hubPath(topic);
+ zk.getData(ownerPath, false, new SafeAsyncZKCallback.DataCallback() {
+ @Override
+ public void safeProcessResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
+ if (Code.NONODE.intValue() == rc) {
+ callback.operationFinished(ctx, null);
+ return;
+ }
+
+ if (Code.OK.intValue() != rc) {
+ KeeperException e = ZkUtils.logErrorAndCreateZKException("Could not read ownership for topic: "
+ + topic.toStringUtf8(), path, rc);
+ callback.operationFailed(ctx, new PubSubException.ServiceDownException(e));
+ return;
+ }
+ HubInfo owner = null;
+ try {
+ owner = HubInfo.parse(new String(data));
+ } catch (HubInfo.InvalidHubInfoException ihie) {
+ logger.warn("Failed to parse hub info for topic " + topic.toStringUtf8() + " : ", ihie);
+ }
+ int version = stat.getVersion();
+ callback.operationFinished(ctx, new Versioned<HubInfo>(owner, new ZkVersion(version)));
+ return;
+ }
+ }, ctx);
+ }
+
+ @Override
+ public void writeOwnerInfo(final ByteString topic, final HubInfo owner, final Version version,
+ final Callback<Version> callback, Object ctx) {
+ if (null == version) {
+ createOwnerInfo(topic, owner, callback, ctx);
+ return;
+ }
+
+ if (!(version instanceof ZkVersion)) {
+ callback.operationFailed(ctx, new PubSubException.UnexpectedConditionException(
+ "Invalid version provided to update owner info for topic " + topic.toStringUtf8()));
+ return;
+ }
+
+ int znodeVersion = ((ZkVersion)version).getZnodeVersion();
+ zk.setData(hubPath(topic), owner.toString().getBytes(), znodeVersion,
+ new SafeAsyncZKCallback.StatCallback() {
+ @Override
+ public void safeProcessResult(int rc, String path, Object ctx, Stat stat) {
+ if (rc == Code.NONODE.intValue()) {
+ // no node
+ callback.operationFailed(ctx, PubSubException.create(StatusCode.NO_TOPIC_OWNER_INFO,
+ "No owner info found for topic " + topic.toStringUtf8()));
+ return;
+ } else if (rc == Code.BadVersion) {
+ // bad version
+ callback.operationFailed(ctx, PubSubException.create(StatusCode.BAD_VERSION,
+ "Bad version provided to update owner info of topic " + topic.toStringUtf8()));
+ return;
+ } else if (Code.OK.intValue() == rc) {
+ callback.operationFinished(ctx, new ZkVersion(stat.getVersion()));
+ return;
+ } else {
+ KeeperException e = ZkUtils.logErrorAndCreateZKException(
+ "Failed to update ownership of topic " + topic.toStringUtf8() +
+ " to " + owner, path, rc);
+ callback.operationFailed(ctx, new PubSubException.ServiceDownException(e));
+ return;
+ }
+ }
+ }, ctx);
+ }
+
+ protected void createOwnerInfo(final ByteString topic, final HubInfo owner,
+ final Callback<Version> callback, Object ctx) {
+ String ownerPath = hubPath(topic);
+ ZkUtils.createFullPathOptimistic(zk, ownerPath, owner.toString().getBytes(), Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT, new SafeAsyncZKCallback.StringCallback() {
+ @Override
+ public void safeProcessResult(int rc, String path, Object ctx, String name) {
+ if (Code.OK.intValue() == rc) {
+ // assume the initial version is 0
+ callback.operationFinished(ctx, new ZkVersion(0));
+ return;
+ } else if (Code.NODEEXISTS.intValue() == rc) {
+ // node existed
+ callback.operationFailed(ctx, PubSubException.create(StatusCode.TOPIC_OWNER_INFO_EXISTS,
+ "Owner info of topic " + topic.toStringUtf8() + " existed."));
+ return;
+ } else {
+ KeeperException e = ZkUtils.logErrorAndCreateZKException(
+ "Failed to create znode for ownership of topic: "
+ + topic.toStringUtf8(), path, rc);
+ callback.operationFailed(ctx, new PubSubException.ServiceDownException(e));
+ return;
+ }
+ }
+ }, ctx);
+ }
+
+ @Override
+ public void deleteOwnerInfo(final ByteString topic, final Version version,
+ final Callback<Void> callback, Object ctx) {
+ int znodeVersion = -1;
+ if (null != version) {
+ if (!(version instanceof ZkVersion)) {
+ callback.operationFailed(ctx, new PubSubException.UnexpectedConditionException(
+ "Invalid version provided to delete owner info for topic " + topic.toStringUtf8()));
+ return;
+ } else {
+ znodeVersion = ((ZkVersion)version).getZnodeVersion();
+ }
+ }
+
+ zk.delete(hubPath(topic), znodeVersion, new SafeAsyncZKCallback.VoidCallback() {
+ @Override
+ public void safeProcessResult(int rc, String path, Object ctx) {
+ if (Code.OK.intValue() == rc) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Successfully deleted owner info for topic " + topic.toStringUtf8() + ".");
+ }
+ callback.operationFinished(ctx, null);
+ return;
+ } else if (Code.NONODE.intValue() == rc) {
+ // no node
+ callback.operationFailed(ctx, PubSubException.create(StatusCode.NO_TOPIC_OWNER_INFO,
+ "No owner info found for topic " + topic.toStringUtf8()));
+ return;
+ } else if (Code.BadVersion == rc) {
+ // bad version
+ callback.operationFailed(ctx, PubSubException.create(StatusCode.BAD_VERSION,
+ "Bad version provided to delete owner info of topic " + topic.toStringUtf8()));
+ return;
+ } else {
+ KeeperException e = ZkUtils.logErrorAndCreateZKException(
+ "Failed to delete owner info for topic "
+ + topic.toStringUtf8(), path, rc);
+ callback.operationFailed(ctx, new PubSubException.ServiceDownException(e));
+ }
+ }
+ }, ctx);
+ }
+ }
}
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/netty/PubSubServer.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/netty/PubSubServer.java?rev=1369778&r1=1369777&r2=1369778&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/netty/PubSubServer.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/netty/PubSubServer.java Mon Aug 6 11:01:36 2012
@@ -64,6 +64,7 @@ import org.apache.hedwig.server.handlers
import org.apache.hedwig.server.handlers.UnsubscribeHandler;
import org.apache.hedwig.server.jmx.HedwigMBeanRegistry;
import org.apache.hedwig.server.meta.MetadataManagerFactory;
+import org.apache.hedwig.server.meta.ZkMetadataManagerFactory;
import org.apache.hedwig.server.persistence.BookkeeperPersistenceManager;
import org.apache.hedwig.server.persistence.LocalDBPersistenceManager;
import org.apache.hedwig.server.persistence.PersistenceManager;
@@ -75,6 +76,7 @@ import org.apache.hedwig.server.ssl.SslS
import org.apache.hedwig.server.subscriptions.InMemorySubscriptionManager;
import org.apache.hedwig.server.subscriptions.SubscriptionManager;
import org.apache.hedwig.server.subscriptions.MMSubscriptionManager;
+import org.apache.hedwig.server.topics.MMTopicManager;
import org.apache.hedwig.server.topics.TopicManager;
import org.apache.hedwig.server.topics.TrivialOwnAllTopicManager;
import org.apache.hedwig.server.topics.ZkTopicManager;
@@ -195,9 +197,18 @@ public class PubSubServer {
tm = new TrivialOwnAllTopicManager(conf, scheduler);
} else {
try {
- tm = new ZkTopicManager(zk, conf, scheduler);
+ if (conf.isMetadataManagerBasedTopicManagerEnabled()) {
+ tm = new MMTopicManager(conf, zk, mm, scheduler);
+ } else {
+ if (!(mm instanceof ZkMetadataManagerFactory)) {
+ throw new IOException("Uses " + mm.getClass().getName() + " to store hedwig metadata, "
+ + "but uses zookeeper ephemeral znodes to store topic ownership. "
+ + "Check your configuration as this could lead to scalability issues.");
+ }
+ tm = new ZkTopicManager(zk, conf, scheduler);
+ }
} catch (PubSubException e) {
- logger.error("Could not instantiate zk-topic manager", e);
+ logger.error("Could not instantiate TopicOwnershipManager based topic manager", e);
throw new IOException(e);
}
}
Added: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/HubInfo.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/HubInfo.java?rev=1369778&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/HubInfo.java (added)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/HubInfo.java Mon Aug 6 11:01:36 2012
@@ -0,0 +1,162 @@
+package org.apache.hedwig.server.topics;
+
+/*
+ *
+ * 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.
+ *
+ */
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.StringReader;
+
+import org.apache.hedwig.protocol.PubSubProtocol.HubInfoData;
+import org.apache.hedwig.util.HedwigSocketAddress;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.TextFormat;
+
+/**
+ * Info identifies a hub server.
+ */
+public class HubInfo {
+
+ public static class InvalidHubInfoException extends Exception {
+ public InvalidHubInfoException(String msg) {
+ super(msg);
+ }
+
+ public InvalidHubInfoException(String msg, Throwable t) {
+ super(msg, t);
+ }
+ }
+
+ // address identify a hub server
+ final HedwigSocketAddress addr;
+ // its znode czxid
+ final long czxid;
+ // protobuf encoded hub info data to be serialized
+ HubInfoData hubInfoData;
+
+ public HubInfo(HedwigSocketAddress addr, long czxid) {
+ this(addr, czxid, null);
+ }
+
+ protected HubInfo(HedwigSocketAddress addr, long czxid,
+ HubInfoData data) {
+ this.addr = addr;
+ this.czxid = czxid;
+ this.hubInfoData = data;
+ }
+
+ public HedwigSocketAddress getAddress() {
+ return addr;
+ }
+
+ public long getZxid() {
+ return czxid;
+ }
+
+ private synchronized HubInfoData getHubInfoData() {
+ if (null == hubInfoData) {
+ hubInfoData = HubInfoData.newBuilder().setHostname(addr.toString())
+ .setCzxid(czxid).build();
+ }
+ return hubInfoData;
+ }
+
+ @Override
+ public String toString() {
+ return TextFormat.printToString(getHubInfoData());
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (null == o) {
+ return false;
+ }
+ if (!(o instanceof HubInfo)) {
+ return false;
+ }
+ HubInfo other = (HubInfo)o;
+ if (null == addr) {
+ if (null == other.addr) {
+ return true;
+ } else {
+ return czxid == other.czxid;
+ }
+ } else {
+ if (addr.equals(other.addr)) {
+ return czxid == other.czxid;
+ } else {
+ return false;
+ }
+ }
+ }
+
+ @Override
+ public int hashCode() {
+ return addr.hashCode();
+ }
+
+ /**
+ * Parse hub info from a string.
+ *
+ * @param hubInfoStr
+ * String representation of hub info
+ * @return hub info
+ * @throws InvalidHubInfoException when <code>hubInfoStr</code> is not a valid
+ * string representation of hub info.
+ */
+ public static HubInfo parse(String hubInfoStr) throws InvalidHubInfoException {
+ // it is not protobuf encoded hub info, it might be generated by ZkTopicManager
+ if (!hubInfoStr.startsWith("hostname")) {
+ final HedwigSocketAddress owner;
+ try {
+ owner = new HedwigSocketAddress(hubInfoStr);
+ } catch (Exception e) {
+ throw new InvalidHubInfoException("Corrupted hub server address : " + hubInfoStr, e);
+ }
+ return new HubInfo(owner, 0L);
+ }
+
+ // it is a protobuf encoded hub info.
+ HubInfoData hubInfoData;
+
+ try {
+ BufferedReader reader = new BufferedReader(
+ new StringReader(hubInfoStr));
+ HubInfoData.Builder dataBuilder = HubInfoData.newBuilder();
+ TextFormat.merge(reader, dataBuilder);
+ hubInfoData = dataBuilder.build();
+ } catch (InvalidProtocolBufferException ipbe) {
+ throw new InvalidHubInfoException("Corrupted hub info : " + hubInfoStr, ipbe);
+ } catch (IOException ie) {
+ throw new InvalidHubInfoException("Corrupted hub info : " + hubInfoStr, ie);
+ }
+
+ final HedwigSocketAddress owner;
+ try {
+ owner = new HedwigSocketAddress(hubInfoData.getHostname().trim());
+ } catch (Exception e) {
+ throw new InvalidHubInfoException("Corrupted hub server address : " + hubInfoData.getHostname(), e);
+ }
+ long ownerZxid = hubInfoData.getCzxid();
+ return new HubInfo(owner, ownerZxid, hubInfoData);
+ }
+}
Added: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/HubLoad.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/HubLoad.java?rev=1369778&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/HubLoad.java (added)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/HubLoad.java Mon Aug 6 11:01:36 2012
@@ -0,0 +1,130 @@
+/*
+ *
+ * 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.hedwig.server.topics;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.StringReader;
+
+import org.apache.hedwig.protocol.PubSubProtocol.HubLoadData;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.TextFormat;
+
+/**
+ * This class encapsulates metrics for determining the load on a hub server.
+ */
+class HubLoad implements Comparable<HubLoad> {
+
+ public static final HubLoad MAX_LOAD = new HubLoad(Long.MAX_VALUE);
+ public static final HubLoad MIN_LOAD = new HubLoad(0);
+
+ public static class InvalidHubLoadException extends Exception {
+ public InvalidHubLoadException(String msg) {
+ super(msg);
+ }
+
+ public InvalidHubLoadException(String msg, Throwable t) {
+ super(msg, t);
+ }
+ }
+
+ // how many topics that a hub server serves
+ long numTopics;
+
+ public HubLoad(long num) {
+ this.numTopics = num;
+ }
+
+ public HubLoad(HubLoadData data) {
+ this.numTopics = data.getNumTopics();
+ }
+
+ public HubLoad setNumTopics(long numTopics) {
+ this.numTopics = numTopics;
+ return this;
+ }
+
+ public HubLoadData toHubLoadData() {
+ return HubLoadData.newBuilder().setNumTopics(numTopics).build();
+ }
+
+ @Override
+ public String toString() {
+ return TextFormat.printToString(toHubLoadData());
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (null == o ||
+ !(o instanceof HubLoad)) {
+ return false;
+ }
+ return 0 == compareTo((HubLoad)o);
+ }
+
+ @Override
+ public int compareTo(HubLoad other) {
+ return numTopics > other.numTopics ?
+ 1 : (numTopics < other.numTopics ? -1 : 0);
+ }
+
+ @Override
+ public int hashCode() {
+ return (int)numTopics;
+ }
+
+ /**
+ * Parse hub load from a string.
+ *
+ * @param hubLoadStr
+ * String representation of hub load
+ * @return hub load
+ * @throws InvalidHubLoadException when <code>hubLoadStr</code> is not a valid
+ * string representation of hub load.
+ */
+ public static HubLoad parse(String hubLoadStr) throws InvalidHubLoadException {
+ // it is no protobuf encoded hub info, it might be generated by ZkTopicManager
+ if (!hubLoadStr.startsWith("numTopics")) {
+ try {
+ long numTopics = Long.parseLong(hubLoadStr, 10);
+ return new HubLoad(numTopics);
+ } catch (NumberFormatException nfe) {
+ throw new InvalidHubLoadException("Corrupted hub load data : " + hubLoadStr, nfe);
+ }
+ }
+ // it it a protobuf encoded hub load data.
+ HubLoadData hubLoadData;
+ try {
+ BufferedReader reader = new BufferedReader(
+ new StringReader(hubLoadStr));
+ HubLoadData.Builder dataBuilder = HubLoadData.newBuilder();
+ TextFormat.merge(reader, dataBuilder);
+ hubLoadData = dataBuilder.build();
+ } catch (InvalidProtocolBufferException ipbe) {
+ throw new InvalidHubLoadException("Corrupted hub load data : " + hubLoadStr, ipbe);
+ } catch (IOException ie) {
+ throw new InvalidHubLoadException("Corrupted hub load data : " + hubLoadStr, ie);
+ }
+
+ return new HubLoad(hubLoadData);
+ }
+}
Added: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/HubServerManager.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/HubServerManager.java?rev=1369778&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/HubServerManager.java (added)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/topics/HubServerManager.java Mon Aug 6 11:01:36 2012
@@ -0,0 +1,108 @@
+/**
+ * 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.hedwig.server.topics;
+
+import java.io.IOException;
+
+import org.apache.hedwig.util.Callback;
+
+/**
+ * The HubServerManager class manages info about hub servers.
+ */
+interface HubServerManager {
+
+ static interface ManagerListener {
+
+ /**
+ * Server manager is suspended if encountering some transient errors.
+ * {@link #onResume()} would be called if those errors could be fixed.
+ * {@link #onShutdown()} would be called if those errors could not be fixed.
+ */
+ public void onSuspend();
+
+ /**
+ * Server manager is resumed after fixing some transient errors.
+ */
+ public void onResume();
+
+ /**
+ * Server manager had to shutdown due to unrecoverable errors.
+ */
+ public void onShutdown();
+ }
+
+ /**
+ * Register a listener to listen events of server manager
+ *
+ * @param listener
+ * Server Manager Listener
+ */
+ public void registerListener(ManagerListener listener);
+
+ /**
+ * Register itself to the cluster.
+ *
+ * @param selfLoad
+ * Self load data
+ * @param callback
+ * Callback when itself registered.
+ * @param ctx
+ * Callback context.
+ */
+ public void registerSelf(HubLoad selfLoad, Callback<HubInfo> callback, Object ctx);
+
+ /**
+ * Unregister itself from the cluster.
+ */
+ public void unregisterSelf() throws IOException;
+
+ /**
+ * Uploading self server load data.
+ *
+ * It is an asynchrounous call which should not block other operations.
+ * Currently we don't need to care about whether it succeed or not.
+ *
+ * @param selfLoad
+ * Hub server load data.
+ */
+ public void uploadSelfLoadData(HubLoad selfLoad);
+
+ /**
+ * Check whether a hub server is alive as the id
+ *
+ * @param hub
+ * Hub id to identify a lifecycle of a hub server
+ * @param callback
+ * Callback of check result. If the hub server is still
+ * alive as the provided id <code>hub</code>, return true.
+ * Otherwise return false.
+ * @param ctx
+ * Callback context
+ */
+ public void isHubAlive(HubInfo hub, Callback<Boolean> callback, Object ctx);
+
+ /**
+ * Choose a least loaded hub server from available hub servers.
+ *
+ * @param callback
+ * Callback to return least loaded hub server.
+ * @param ctx
+ * Callback context.
+ */
+ public void chooseLeastLoadedHub(Callback<HubInfo> callback, Object ctx);
+}