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/17 17:32:24 UTC
svn commit: r1374321 [3/3] - in /zookeeper/bookkeeper/trunk: ./
bookkeeper-server/ hedwig-client/src/main/cpp/lib/
hedwig-client/src/main/cpp/test/
hedwig-client/src/main/java/org/apache/hedwig/client/netty/
hedwig-protocol/ hedwig-protocol/src/main/ja...
Modified: zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/java/org/apache/hedwig/protoextensions/SubscriptionStateUtils.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/java/org/apache/hedwig/protoextensions/SubscriptionStateUtils.java?rev=1374321&r1=1374320&r2=1374321&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/java/org/apache/hedwig/protoextensions/SubscriptionStateUtils.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/java/org/apache/hedwig/protoextensions/SubscriptionStateUtils.java Fri Aug 17 15:32:23 2012
@@ -17,25 +17,84 @@
*/
package org.apache.hedwig.protoextensions;
+import java.util.HashMap;
+import java.util.Map;
+
import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import org.apache.hedwig.protocol.PubSubProtocol;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionData;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionPreferences;
import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
public class SubscriptionStateUtils {
+ static final Logger logger = LoggerFactory.getLogger(SubscriptionStateUtils.class);
+
// For now, to differentiate hub subscribers from local ones, the
// subscriberId will be prepended with a hard-coded prefix. Local
// subscribers will validate that the subscriberId used cannot start with
// this prefix. This is only used internally by the hub subscribers.
public static final String HUB_SUBSCRIBER_PREFIX = "__";
+ public static SubscriptionData parseSubscriptionData(byte[] data)
+ throws InvalidProtocolBufferException {
+ try {
+ return SubscriptionData.parseFrom(data);
+ } catch (InvalidProtocolBufferException ex) {
+ logger.info("Failed to parse data as SubscriptionData. Fall backward to parse it as SubscriptionState for backward compatability.");
+ // backward compability
+ SubscriptionState state = SubscriptionState.parseFrom(data);
+ return SubscriptionData.newBuilder().setState(state).build();
+ }
+ }
+
+ public static String toString(SubscriptionData data) {
+ StringBuilder sb = new StringBuilder();
+ if (data.hasState()) {
+ sb.append("State : { ").append(toString(data.getState())).append(" };");
+ }
+ if (data.hasPreferences()) {
+ sb.append("Preferences : { ").append(toString(data.getPreferences())).append(" };");
+ }
+ return sb.toString();
+ }
+
public static String toString(SubscriptionState state) {
StringBuilder sb = new StringBuilder();
sb.append("consumeSeqId: " + MessageIdUtils.msgIdToReadableString(state.getMsgId()));
return sb.toString();
}
+ public static String toString(SubscriptionPreferences preferences) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("System Preferences : [");
+ if (preferences.hasMessageBound()) {
+ sb.append("(messageBound=").append(preferences.getMessageBound())
+ .append(")");
+ }
+ sb.append("]");
+ if (preferences.hasOptions()) {
+ sb.append(", Customized Preferences : [");
+ sb.append(MapUtils.toString(preferences.getOptions()));
+ sb.append("]");
+ }
+ return sb.toString();
+ }
+
public static boolean isHubSubscriber(ByteString subscriberId) {
return subscriberId.toStringUtf8().startsWith(HUB_SUBSCRIBER_PREFIX);
}
+ public static Map<String, ByteString> buildUserOptions(SubscriptionPreferences preferences) {
+ if (preferences.hasOptions()) {
+ return MapUtils.buildMap(preferences.getOptions());
+ } else {
+ return new HashMap<String, ByteString>();
+ }
+ }
+
}
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=1374321&r1=1374320&r2=1374321&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/protobuf/PubSubProtocol.proto (original)
+++ zookeeper/bookkeeper/trunk/hedwig-protocol/src/main/protobuf/PubSubProtocol.proto Fri Aug 17 15:32:23 2012
@@ -23,6 +23,15 @@ enum ProtocolVersion{
VERSION_ONE = 1;
}
+// common structure to store header or properties
+message Map {
+ message Entry {
+ optional string key = 1;
+ optional bytes value = 2;
+ }
+ repeated Entry entries = 1;
+}
+
/*
* this is the structure that will be serialized
*/
@@ -84,24 +93,46 @@ message PublishRequest{
required Message msg = 2;
}
+// record all preferences for a subscription,
+// would be serialized to be stored in meta store
+message SubscriptionPreferences {
+ // user customized subscription options
+ optional Map options = 1;
+
+ ///
+ /// system defined options
+ ///
+
+ // message bound
+ optional uint32 messageBound = 2;
+}
+
message SubscribeRequest{
required bytes subscriberId = 2;
-
- enum CreateOrAttach{
- CREATE = 0;
- ATTACH = 1;
- CREATE_OR_ATTACH = 2;
- };
- optional CreateOrAttach createOrAttach = 3 [default = CREATE_OR_ATTACH];
-
- // wait for cross-regional subscriptions to be established before returning
- optional bool synchronous = 4 [default = false];
- optional uint32 messageBound = 5;
+
+ enum CreateOrAttach{
+ CREATE = 0;
+ ATTACH = 1;
+ CREATE_OR_ATTACH = 2;
+ };
+ optional CreateOrAttach createOrAttach = 3 [default = CREATE_OR_ATTACH];
+
+ // wait for cross-regional subscriptions to be established before returning
+ optional bool synchronous = 4 [default = false];
+ // @Deprecated. set message bound in SubscriptionPreferences
+ optional uint32 messageBound = 5;
+
+ // subscription options
+ optional SubscriptionPreferences preferences = 6;
}
+// used in client only
+// options are stored in SubscriptionPreferences structure
message SubscriptionOptions {
optional SubscribeRequest.CreateOrAttach createOrAttach = 2 [default = CREATE_OR_ATTACH];
optional uint32 messageBound = 3 [default = 0];
+ // user customized subscription options
+ optional Map options = 4;
}
message ConsumeRequest{
@@ -189,9 +220,18 @@ enum StatusCode{
message SubscriptionState {
required MessageSeqId msgId = 1;
+ // @Deprecated.
+ // It is a bad idea to put fields that don't change frequently
+ // together with fields that change frequently
+ // so move it to subscription preferences structure
optional uint32 messageBound = 2;
}
+message SubscriptionData {
+ optional SubscriptionState state = 1;
+ optional SubscriptionPreferences preferences = 2;
+}
+
message LedgerRange{
required uint64 ledgerId = 1;
optional MessageSeqId endSeqIdIncluded = 2;
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/admin/HedwigAdmin.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/admin/HedwigAdmin.java?rev=1374321&r1=1374320&r2=1374321&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/admin/HedwigAdmin.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/admin/HedwigAdmin.java Fri Aug 17 15:32:23 2012
@@ -38,14 +38,12 @@ import org.apache.hedwig.protocol.PubSub
import org.apache.hedwig.protocol.PubSubProtocol.LedgerRanges;
import org.apache.hedwig.protocol.PubSubProtocol.Message;
import org.apache.hedwig.protocol.PubSubProtocol.MessageSeqId;
-import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionState;
-import org.apache.hedwig.protoextensions.SubscriptionStateUtils;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionData;
import org.apache.hedwig.server.common.ServerConfiguration;
import org.apache.hedwig.server.meta.MetadataManagerFactory;
import org.apache.hedwig.server.meta.SubscriptionDataManager;
import org.apache.hedwig.server.meta.TopicOwnershipManager;
import org.apache.hedwig.server.meta.TopicPersistenceManager;
-import org.apache.hedwig.server.subscriptions.InMemorySubscriptionState;
import org.apache.hedwig.server.topics.HubInfo;
import org.apache.hedwig.server.topics.HubLoad;
import org.apache.hedwig.util.Callback;
@@ -446,16 +444,14 @@ public class HedwigAdmin {
* @return subscriptions of a topic
* @throws Exception
*/
- public Map<ByteString, SubscriptionState> getTopicSubscriptions(ByteString topic)
+ public Map<ByteString, SubscriptionData> getTopicSubscriptions(ByteString topic)
throws Exception {
- Map<ByteString, SubscriptionState> states =
- new HashMap<ByteString, SubscriptionState>();
- final SyncObj<Map<ByteString, InMemorySubscriptionState>> syncObj =
- new SyncObj<Map<ByteString, InMemorySubscriptionState>>();
- sdm.readSubscriptions(topic, new Callback<Map<ByteString, InMemorySubscriptionState>>() {
+ final SyncObj<Map<ByteString, SubscriptionData>> syncObj =
+ new SyncObj<Map<ByteString, SubscriptionData>>();
+ sdm.readSubscriptions(topic, new Callback<Map<ByteString, SubscriptionData>>() {
@Override
- public void operationFinished(Object ctx, Map<ByteString, InMemorySubscriptionState> result) {
+ public void operationFinished(Object ctx, Map<ByteString, SubscriptionData> result) {
syncObj.success(result);
}
@Override
@@ -470,17 +466,7 @@ public class HedwigAdmin {
throw syncObj.exception;
}
- Map<ByteString, InMemorySubscriptionState> subStats = syncObj.value;
-
- if (null == subStats) {
- return states;
- }
-
- for (Map.Entry<ByteString, InMemorySubscriptionState> entry : subStats.entrySet()) {
- states.put(entry.getKey(), entry.getValue().getSubscriptionState());
- }
-
- return states;
+ return syncObj.value;
}
/**
@@ -493,11 +479,11 @@ public class HedwigAdmin {
* @return subscription state
* @throws Exception
*/
- public SubscriptionState getSubscription(ByteString topic, ByteString subscriber) throws Exception {
- final SyncObj<SubscriptionState> syncObj = new SyncObj<SubscriptionState>();
- sdm.readSubscriptionState(topic, subscriber, new Callback<SubscriptionState>() {
+ public SubscriptionData getSubscription(ByteString topic, ByteString subscriber) throws Exception {
+ final SyncObj<SubscriptionData> syncObj = new SyncObj<SubscriptionData>();
+ sdm.readSubscriptionData(topic, subscriber, new Callback<SubscriptionData>() {
@Override
- public void operationFinished(Object ctx, SubscriptionState result) {
+ public void operationFinished(Object ctx, SubscriptionData result) {
syncObj.success(result);
}
@Override
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/admin/console/HedwigConsole.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/admin/console/HedwigConsole.java?rev=1374321&r1=1374320&r2=1374321&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/admin/console/HedwigConsole.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/admin/console/HedwigConsole.java Fri Aug 17 15:32:23 2012
@@ -49,7 +49,7 @@ import org.apache.hedwig.protocol.PubSub
import org.apache.hedwig.protocol.PubSubProtocol.Message;
import org.apache.hedwig.protocol.PubSubProtocol.MessageSeqId;
import org.apache.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach;
-import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionState;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionData;
import org.apache.hedwig.protoextensions.SubscriptionStateUtils;
import org.apache.hedwig.server.common.ServerConfiguration;
import org.apache.hedwig.server.topics.HubInfo;
@@ -393,8 +393,9 @@ public class HedwigConsole {
return false;
}
long lastConsumedId = 0;
- SubscriptionState state = admin.getSubscription(ByteString.copyFromUtf8(args[1]), ByteString.copyFromUtf8(args[2]));
- if (null == state) {
+ SubscriptionData subData = admin.getSubscription(ByteString.copyFromUtf8(args[1]),
+ ByteString.copyFromUtf8(args[2]));
+ if (null == subData) {
System.err.println("Failed to read subscription for topic: " + args[1]
+ " subscriber: " + args[2]);
return true;
@@ -605,7 +606,7 @@ public class HedwigConsole {
ByteString btopic = ByteString.copyFromUtf8(topic);
HubInfo owner = admin.getTopicOwner(btopic);
List<LedgerRange> ranges = admin.getTopicLedgers(btopic);
- Map<ByteString, SubscriptionState> states = admin.getTopicSubscriptions(btopic);
+ Map<ByteString, SubscriptionData> states = admin.getTopicSubscriptions(btopic);
System.out.println("===== Topic Information : " + topic + " =====");
System.out.println();
@@ -646,13 +647,13 @@ public class HedwigConsole {
System.out.println();
}
- private void printTopicSubscriptions(Map<ByteString, SubscriptionState> states) {
+ private void printTopicSubscriptions(Map<ByteString, SubscriptionData> states) {
System.out.println(">>> Subscription Info <<<");
if (0 == states.size()) {
System.out.println("No subscriber.");
return;
}
- for (Map.Entry<ByteString, SubscriptionState> entry : states.entrySet()) {
+ for (Map.Entry<ByteString, SubscriptionData> entry : states.entrySet()) {
System.out.println("Subscriber " + entry.getKey().toStringUtf8() + " : "
+ SubscriptionStateUtils.toString(entry.getValue()));
}
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/admin/console/ReadTopic.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/admin/console/ReadTopic.java?rev=1374321&r1=1374320&r2=1374321&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/admin/console/ReadTopic.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/admin/console/ReadTopic.java Fri Aug 17 15:32:23 2012
@@ -39,7 +39,7 @@ import org.apache.hedwig.protocol.PubSub
import org.apache.hedwig.protocol.PubSubProtocol.Message;
import org.apache.hedwig.protocol.PubSubProtocol.MessageSeqId;
import org.apache.hedwig.protocol.PubSubProtocol.RegionSpecificSeqId;
-import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionState;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionData;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
@@ -142,13 +142,13 @@ public class ReadTopic {
}
protected int getLeastSubscription() throws Exception {
- Map<ByteString, SubscriptionState> states = admin.getTopicSubscriptions(topic);
+ Map<ByteString, SubscriptionData> states = admin.getTopicSubscriptions(topic);
if (states.isEmpty()) {
return RC_NOSUBSCRIBERS;
}
- for (Map.Entry<ByteString, SubscriptionState> entry : states.entrySet()) {
- SubscriptionState state = entry.getValue();
- long localMsgId = state.getMsgId().getLocalComponent();
+ for (Map.Entry<ByteString, SubscriptionData> entry : states.entrySet()) {
+ SubscriptionData state = entry.getValue();
+ long localMsgId = state.getState().getMsgId().getLocalComponent();
if (localMsgId < leastConsumedSeqId) {
leastConsumedSeqId = localMsgId;
}
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/SubscriptionDataManager.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/SubscriptionDataManager.java?rev=1374321&r1=1374320&r2=1374321&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/SubscriptionDataManager.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/meta/SubscriptionDataManager.java Fri Aug 17 15:32:23 2012
@@ -23,7 +23,7 @@ import java.util.Map;
import com.google.protobuf.ByteString;
import org.apache.hedwig.exceptions.PubSubException;
-import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionState;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionData;
import org.apache.hedwig.server.subscriptions.InMemorySubscriptionState;
import org.apache.hedwig.util.Callback;
@@ -33,14 +33,14 @@ import org.apache.hedwig.util.Callback;
public interface SubscriptionDataManager extends Closeable {
/**
- * Create subscription state.
+ * Create subscription data.
*
* @param topic
* Topic name
* @param subscriberId
* Subscriber id
- * @param state
- * Subscription state
+ * @param data
+ * Subscription data
* @param callback
* Callback when subscription state created.
* {@link PubSubException.SubscriptionStateExistsException} is returned when subscription state
@@ -48,18 +48,29 @@ public interface SubscriptionDataManager
* @param ctx
* Context of the callback
*/
- public void createSubscriptionState(ByteString topic, ByteString subscriberId, SubscriptionState state,
- Callback<Void> callback, Object ctx);
+ public void createSubscriptionData(ByteString topic, ByteString subscriberId, SubscriptionData data,
+ Callback<Void> callback, Object ctx);
+
+ /**
+ * Whether the metadata manager supports partial update.
+ *
+ * @return true if the metadata manager supports partial update.
+ * otherwise, return false.
+ */
+ public boolean isPartialUpdateSupported();
/**
- * Update subscription state.
+ * Update subscription data.
*
* @param topic
* Topic name
* @param subscriberId
* Subscriber id
- * @param state
- * Subscription state
+ * @param dataToUpdate
+ * Subscription data to update. So it is a partial data, which contains
+ * the part of data to update. The implementation should not replace
+ * existing subscription data with <i>dataToUpdate</i> directly.
+ * E.g. if there is only state in it, you should update state only.
* @param callback
* Callback when subscription state updated.
* {@link PubSubException.NoSubscriptionStateException} is returned when no subscription state
@@ -67,11 +78,28 @@ public interface SubscriptionDataManager
* @param ctx
* Context of the callback
*/
- public void updateSubscriptionState(ByteString topic, ByteString subscriberId, SubscriptionState state,
+ public void updateSubscriptionData(ByteString topic, ByteString subscriberId, SubscriptionData dataToUpdate,
+ Callback<Void> callback, Object ctx);
+
+ /**
+ * Replace subscription data.
+ *
+ * @param topic
+ * Topic name
+ * @param subscriberId
+ * Subscriber id
+ * @param dataToReplace
+ * Subscription data to replace.
+ * @param callback
+ * Callback when subscription state updated.
+ * @param ctx
+ * Context of the callback
+ */
+ public void replaceSubscriptionData(ByteString topic, ByteString subscriberId, SubscriptionData dataToReplace,
Callback<Void> callback, Object ctx);
/**
- * Remove subscription state.
+ * Remove subscription data.
*
* @param topic
* Topic name
@@ -84,24 +112,24 @@ public interface SubscriptionDataManager
* @param ctx
* Context of the callback
*/
- public void deleteSubscriptionState(ByteString topic, ByteString subscriberId,
- Callback<Void> callback, Object ctx);
+ public void deleteSubscriptionData(ByteString topic, ByteString subscriberId,
+ Callback<Void> callback, Object ctx);
/**
- * Read subscription state.
+ * Read subscription data.
*
* @param topic
* Topic Name
* @param subscriberId
* Subscriber id
* @param callback
- * Callback when subscription state read.
- * Null is returned when no subscription state is found.
+ * Callback when subscription data read.
+ * Null is returned when no subscription data is found.
* @param ctx
* Context of the callback
*/
- public void readSubscriptionState(ByteString topic, ByteString subscriberId,
- Callback<SubscriptionState> callback, Object ctx);
+ public void readSubscriptionData(ByteString topic, ByteString subscriberId,
+ Callback<SubscriptionData> callback, Object ctx);
/**
* Read all subscriptions of a topic.
@@ -113,6 +141,6 @@ public interface SubscriptionDataManager
* @param ctx
* Contxt of the callback
*/
- public void readSubscriptions(ByteString topic, Callback<Map<ByteString, InMemorySubscriptionState>> cb,
+ public void readSubscriptions(ByteString topic, Callback<Map<ByteString, SubscriptionData>> cb,
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=1374321&r1=1374320&r2=1374321&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 Fri Aug 17 15:32:23 2012
@@ -41,10 +41,10 @@ import org.apache.bookkeeper.meta.ZkVers
import org.apache.hedwig.exceptions.PubSubException;
import org.apache.hedwig.protocol.PubSubProtocol.LedgerRanges;
import org.apache.hedwig.protocol.PubSubProtocol.StatusCode;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionData;
import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionState;
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;
@@ -377,9 +377,14 @@ public class ZkMetadataManagerFactory ex
}
@Override
- public void createSubscriptionState(final ByteString topic, final ByteString subscriberId, final SubscriptionState state,
- final Callback<Void> callback, final Object ctx) {
- ZkUtils.createFullPathOptimistic(zk, topicSubscriberPath(topic, subscriberId), state.toByteArray(),
+ public boolean isPartialUpdateSupported() {
+ return false;
+ }
+
+ @Override
+ public void createSubscriptionData(final ByteString topic, final ByteString subscriberId, final SubscriptionData data,
+ final Callback<Void> callback, final Object ctx) {
+ ZkUtils.createFullPathOptimistic(zk, topicSubscriberPath(topic, subscriberId), data.toByteArray(),
Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, new SafeAsyncZKCallback.StringCallback() {
@Override
@@ -393,8 +398,8 @@ public class ZkMetadataManagerFactory ex
} else if (rc == Code.OK.intValue()) {
if (logger.isDebugEnabled()) {
logger.debug("Successfully recorded subscription for topic: " + topic.toStringUtf8()
- + " subscriberId: " + subscriberId.toStringUtf8() + " state: "
- + SubscriptionStateUtils.toString(state));
+ + " subscriberId: " + subscriberId.toStringUtf8() + " data: "
+ + SubscriptionStateUtils.toString(data));
}
callback.operationFinished(ctx, null);
} else {
@@ -408,9 +413,15 @@ public class ZkMetadataManagerFactory ex
}
@Override
- public void updateSubscriptionState(final ByteString topic, final ByteString subscriberId, final SubscriptionState state,
+ public void updateSubscriptionData(final ByteString topic, final ByteString subscriberId, final SubscriptionData data,
+ final Callback<Void> callback, final Object ctx) {
+ throw new UnsupportedOperationException("ZooKeeper based metadata manager doesn't support partial update!");
+ }
+
+ @Override
+ public void replaceSubscriptionData(final ByteString topic, final ByteString subscriberId, final SubscriptionData data,
final Callback<Void> callback, final Object ctx) {
- zk.setData(topicSubscriberPath(topic, subscriberId), state.toByteArray(), -1,
+ zk.setData(topicSubscriberPath(topic, subscriberId), data.toByteArray(), -1,
new SafeAsyncZKCallback.StatCallback() {
@Override
public void safeProcessResult(int rc, String path, Object ctx, Stat stat) {
@@ -423,14 +434,14 @@ public class ZkMetadataManagerFactory ex
} else if (rc != Code.OK.intValue()) {
KeeperException e = ZkUtils.logErrorAndCreateZKException("Topic: " + topic.toStringUtf8()
+ " subscriberId: " + subscriberId.toStringUtf8()
- + " could not set subscription state: " + SubscriptionStateUtils.toString(state),
+ + " could not set subscription data: " + SubscriptionStateUtils.toString(data),
path, rc);
callback.operationFailed(ctx, new PubSubException.ServiceDownException(e));
} else {
if (logger.isDebugEnabled()) {
logger.debug("Successfully updated subscription for topic: " + topic.toStringUtf8()
- + " subscriberId: " + subscriberId.toStringUtf8() + " state: "
- + SubscriptionStateUtils.toString(state));
+ + " subscriberId: " + subscriberId.toStringUtf8() + " data: "
+ + SubscriptionStateUtils.toString(data));
}
callback.operationFinished(ctx, null);
@@ -440,8 +451,8 @@ public class ZkMetadataManagerFactory ex
}
@Override
- public void deleteSubscriptionState(final ByteString topic, final ByteString subscriberId,
- final Callback<Void> callback, Object ctx) {
+ public void deleteSubscriptionData(final ByteString topic, final ByteString subscriberId,
+ final Callback<Void> callback, Object ctx) {
zk.delete(topicSubscriberPath(topic, subscriberId), -1, new SafeAsyncZKCallback.VoidCallback() {
@Override
public void safeProcessResult(int rc, String path, Object ctx) {
@@ -469,8 +480,8 @@ public class ZkMetadataManagerFactory ex
}
@Override
- public void readSubscriptionState(final ByteString topic, final ByteString subscriberId,
- final Callback<SubscriptionState> callback, final Object ctx) {
+ public void readSubscriptionData(final ByteString topic, final ByteString subscriberId,
+ final Callback<SubscriptionData> callback, final Object ctx) {
zk.getData(topicSubscriberPath(topic, subscriberId), false, new SafeAsyncZKCallback.DataCallback() {
@Override
public void safeProcessResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
@@ -486,11 +497,11 @@ public class ZkMetadataManagerFactory ex
return;
}
- SubscriptionState state;
+ SubscriptionData subData;
try {
- state = SubscriptionState.parseFrom(data);
+ subData = SubscriptionStateUtils.parseSubscriptionData(data);
} catch (InvalidProtocolBufferException ex) {
- String msg = "Failed to deserialize state for topic: " + topic.toStringUtf8()
+ String msg = "Failed to deserialize subscription data for topic: " + topic.toStringUtf8()
+ " subscriberId: " + subscriberId.toStringUtf8();
logger.error(msg, ex);
callback.operationFailed(ctx, new PubSubException.UnexpectedConditionException(msg));
@@ -500,16 +511,16 @@ public class ZkMetadataManagerFactory ex
if (logger.isDebugEnabled()) {
logger.debug("Found subscription while acquiring topic: " + topic.toStringUtf8()
+ " subscriberId: " + subscriberId.toStringUtf8()
- + "state: " + SubscriptionStateUtils.toString(state));
+ + " data: " + SubscriptionStateUtils.toString(subData));
}
- callback.operationFinished(ctx, state);
+ callback.operationFinished(ctx, subData);
}
}, ctx);
}
@Override
public void readSubscriptions(final ByteString topic,
- final Callback<Map<ByteString, InMemorySubscriptionState>> cb, final Object ctx) {
+ final Callback<Map<ByteString, SubscriptionData>> cb, final Object ctx) {
String topicSubscribersPath = topicSubscribersPath(new StringBuilder(), topic).toString();
zk.getChildren(topicSubscribersPath, false, new SafeAsyncZKCallback.ChildrenCallback() {
@Override
@@ -522,7 +533,7 @@ public class ZkMetadataManagerFactory ex
return;
}
- final Map<ByteString, InMemorySubscriptionState> topicSubs = new ConcurrentHashMap<ByteString, InMemorySubscriptionState>();
+ final Map<ByteString, SubscriptionData> topicSubs = new ConcurrentHashMap<ByteString, SubscriptionData>();
if (rc == Code.NONODE.intValue() || children.size() == 0) {
if (logger.isDebugEnabled()) {
@@ -556,12 +567,11 @@ public class ZkMetadataManagerFactory ex
return;
}
- SubscriptionState state;
-
+ SubscriptionData subData;
try {
- state = SubscriptionState.parseFrom(data);
+ subData = SubscriptionStateUtils.parseSubscriptionData(data);
} catch (InvalidProtocolBufferException ex) {
- String msg = "Failed to deserialize state for topic: " + topic.toStringUtf8()
+ String msg = "Failed to deserialize subscription data for topic: " + topic.toStringUtf8()
+ " subscriberId: " + subscriberId.toStringUtf8();
logger.error(msg, ex);
reportFailure(new PubSubException.UnexpectedConditionException(msg));
@@ -571,10 +581,10 @@ public class ZkMetadataManagerFactory ex
if (logger.isDebugEnabled()) {
logger.debug("Found subscription while acquiring topic: " + topic.toStringUtf8()
+ " subscriberId: " + child + "state: "
- + SubscriptionStateUtils.toString(state));
+ + SubscriptionStateUtils.toString(subData));
}
- topicSubs.put(subscriberId, new InMemorySubscriptionState(state));
+ topicSubs.put(subscriberId, subData);
if (count.incrementAndGet() == children.size()) {
assert topicSubs.size() == count.get();
cb.operationFinished(ctx, topicSubs);
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/subscriptions/AbstractSubscriptionManager.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/subscriptions/AbstractSubscriptionManager.java?rev=1374321&r1=1374320&r2=1374321&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/subscriptions/AbstractSubscriptionManager.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/subscriptions/AbstractSubscriptionManager.java Fri Aug 17 15:32:23 2012
@@ -34,6 +34,8 @@ import com.google.protobuf.ByteString;
import org.apache.hedwig.exceptions.PubSubException;
import org.apache.hedwig.protocol.PubSubProtocol.MessageSeqId;
import org.apache.hedwig.protocol.PubSubProtocol.SubscribeRequest;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionData;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionPreferences;
import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionState;
import org.apache.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach;
import org.apache.hedwig.protoextensions.MessageIdUtils;
@@ -124,7 +126,7 @@ public abstract class AbstractSubscripti
for (InMemorySubscriptionState curSubscription : topicSubscriptions.values()) {
if (curSubscription.getSubscriptionState().getMsgId().getLocalComponent() < minConsumedMessage)
minConsumedMessage = curSubscription.getSubscriptionState().getMsgId().getLocalComponent();
- hasBound = hasBound && curSubscription.getSubscriptionState().hasMessageBound();
+ hasBound = hasBound && curSubscription.getSubscriptionPreferences().hasMessageBound();
}
boolean callPersistenceManager = true;
// Don't call the PersistenceManager if nobody is subscribed to
@@ -286,8 +288,7 @@ public abstract class AbstractSubscripti
for (Entry<ByteString, InMemorySubscriptionState> entry : states.entrySet()) {
InMemorySubscriptionState memState = entry.getValue();
if (memState.setLastConsumeSeqIdImmediately()) {
- updateSubscriptionState(topic, entry.getKey(), memState.getSubscriptionState(),
- mcb, ctx);
+ updateSubscriptionState(topic, entry.getKey(), memState, mcb, ctx);
} else {
mcb.operationFinished(ctx, null);
}
@@ -332,7 +333,7 @@ public abstract class AbstractSubscripti
}
final ByteString subscriberId = subRequest.getSubscriberId();
- InMemorySubscriptionState subscriptionState = topicSubscriptions.get(subscriberId);
+ final InMemorySubscriptionState subscriptionState = topicSubscriptions.get(subscriberId);
CreateOrAttach createOrAttach = subRequest.getCreateOrAttach();
if (subscriptionState != null) {
@@ -346,11 +347,41 @@ public abstract class AbstractSubscripti
return;
}
+ // Subscription existed before, check whether new preferences provided
+ // if new preferences provided, merged the subscription data and updated them
+ // TODO: needs ACL mechanism when changing preferences
+ if (subRequest.hasPreferences() &&
+ subscriptionState.updatePreferences(subRequest.getPreferences())) {
+ updateSubscriptionPreferences(topic, subscriberId, subscriptionState, new Callback<Void>() {
+ @Override
+ public void operationFailed(Object ctx, PubSubException exception) {
+ cb.operationFailed(ctx, exception);
+ }
+
+ @Override
+ public void operationFinished(Object ctx, Void resultOfOperation) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Topic: " + topic.toStringUtf8() + " subscriberId: " + subscriberId.toStringUtf8()
+ + " attaching to subscription with state: "
+ + SubscriptionStateUtils.toString(subscriptionState.getSubscriptionState())
+ + ", with preferences: "
+ + SubscriptionStateUtils.toString(subscriptionState.getSubscriptionPreferences()));
+ }
+ // update message bound if necessary
+ updateMessageBound(topic);
+ cb.operationFinished(ctx, subscriptionState.getLastConsumeSeqId());
+ }
+ }, ctx);
+ return;
+ }
+
// otherwise just attach
if (logger.isDebugEnabled()) {
logger.debug("Topic: " + topic.toStringUtf8() + " subscriberId: " + subscriberId.toStringUtf8()
+ " attaching to subscription with state: "
- + SubscriptionStateUtils.toString(subscriptionState.getSubscriptionState()));
+ + SubscriptionStateUtils.toString(subscriptionState.getSubscriptionState())
+ + ", with preferences: "
+ + SubscriptionStateUtils.toString(subscriptionState.getSubscriptionPreferences()));
}
cb.operationFinished(ctx, subscriptionState.getLastConsumeSeqId());
@@ -368,12 +399,24 @@ public abstract class AbstractSubscripti
// now the hard case, this is a brand new subscription, must record
SubscriptionState.Builder stateBuilder = SubscriptionState.newBuilder().setMsgId(consumeSeqId);
+
+ SubscriptionPreferences.Builder preferencesBuilder;
+ if (subRequest.hasPreferences()) {
+ preferencesBuilder = SubscriptionPreferences.newBuilder(subRequest.getPreferences());
+ } else {
+ preferencesBuilder = SubscriptionPreferences.newBuilder();
+ }
+
+ // backward compability
if (subRequest.hasMessageBound()) {
- stateBuilder = stateBuilder.setMessageBound(subRequest.getMessageBound());
+ preferencesBuilder = preferencesBuilder.setMessageBound(subRequest.getMessageBound());
}
- final SubscriptionState newState = stateBuilder.build();
- createSubscriptionState(topic, subscriberId, newState, new Callback<Void>() {
+ SubscriptionData.Builder subDataBuilder =
+ SubscriptionData.newBuilder().setState(stateBuilder).setPreferences(preferencesBuilder);
+ final SubscriptionData subData = subDataBuilder.build();
+
+ createSubscriptionData(topic, subscriberId, subData, new Callback<Void>() {
@Override
public void operationFailed(Object ctx, PubSubException exception) {
cb.operationFailed(ctx, exception);
@@ -388,7 +431,7 @@ public abstract class AbstractSubscripti
logger.error("subscription for subscriber " + subscriberId.toStringUtf8() + " to topic "
+ topic.toStringUtf8() + " failed due to failed listener callback", exception);
// should remove subscription when synchronized cross-region subscription failed
- deleteSubscriptionState(topic, subscriberId, new Callback<Void>() {
+ deleteSubscriptionData(topic, subscriberId, new Callback<Void>() {
@Override
public void operationFinished(Object context,
Void resultOfOperation) {
@@ -418,7 +461,7 @@ public abstract class AbstractSubscripti
@Override
public void operationFinished(Object ctx, Void resultOfOperation) {
- topicSubscriptions.put(subscriberId, new InMemorySubscriptionState(newState));
+ topicSubscriptions.put(subscriberId, new InMemorySubscriptionState(subData));
updateMessageBound(topic);
@@ -446,11 +489,11 @@ public abstract class AbstractSubscripti
}
int maxBound = Integer.MIN_VALUE;
for (Map.Entry<ByteString, InMemorySubscriptionState> e : topicSubscriptions.entrySet()) {
- if (!e.getValue().getSubscriptionState().hasMessageBound()) {
+ if (!e.getValue().getSubscriptionPreferences().hasMessageBound()) {
maxBound = Integer.MIN_VALUE;
break;
} else {
- maxBound = Math.max(maxBound, e.getValue().getSubscriptionState().getMessageBound());
+ maxBound = Math.max(maxBound, e.getValue().getSubscriptionPreferences().getMessageBound());
}
}
if (maxBound == Integer.MIN_VALUE) {
@@ -492,7 +535,7 @@ public abstract class AbstractSubscripti
}
if (subState.setLastConsumeSeqId(consumeSeqId, cfg.getConsumeInterval())) {
- updateSubscriptionState(topic, subscriberId, subState.getSubscriptionState(), cb, ctx);
+ updateSubscriptionState(topic, subscriberId, subState, cb, ctx);
} else {
if (logger.isDebugEnabled()) {
logger.debug("Only advanced consume pointer in memory, will persist later, topic: "
@@ -534,7 +577,7 @@ public abstract class AbstractSubscripti
return;
}
- deleteSubscriptionState(topic, subscriberId, new Callback<Void>() {
+ deleteSubscriptionData(topic, subscriberId, new Callback<Void>() {
@Override
public void operationFailed(Object ctx, PubSubException exception) {
cb.operationFailed(ctx, exception);
@@ -600,13 +643,44 @@ public abstract class AbstractSubscripti
}
}
- protected abstract void createSubscriptionState(final ByteString topic, ByteString subscriberId,
- SubscriptionState state, Callback<Void> callback, Object ctx);
+ private void updateSubscriptionState(ByteString topic, ByteString subscriberId,
+ InMemorySubscriptionState state,
+ Callback<Void> callback, Object ctx) {
+ SubscriptionData subData;
+ if (isPartialUpdateSupported()) {
+ subData = SubscriptionData.newBuilder().setState(state.getSubscriptionState()).build();
+ updateSubscriptionData(topic, subscriberId, subData, callback, ctx);
+ } else {
+ subData = state.toSubscriptionData();
+ replaceSubscriptionData(topic, subscriberId, subData, callback, ctx);
+ }
+ }
+
+ private void updateSubscriptionPreferences(ByteString topic, ByteString subscriberId,
+ InMemorySubscriptionState state,
+ Callback<Void> callback, Object ctx) {
+ SubscriptionData subData;
+ if (isPartialUpdateSupported()) {
+ subData = SubscriptionData.newBuilder().setPreferences(state.getSubscriptionPreferences()).build();
+ updateSubscriptionData(topic, subscriberId, subData, callback, ctx);
+ } else {
+ subData = state.toSubscriptionData();
+ replaceSubscriptionData(topic, subscriberId, subData, callback, ctx);
+ }
+ }
+
+ protected abstract boolean isPartialUpdateSupported();
+
+ protected abstract void createSubscriptionData(final ByteString topic, ByteString subscriberId,
+ SubscriptionData data, Callback<Void> callback, Object ctx);
+
+ protected abstract void updateSubscriptionData(ByteString topic, ByteString subscriberId, SubscriptionData data,
+ Callback<Void> callback, Object ctx);
- protected abstract void updateSubscriptionState(ByteString topic, ByteString subscriberId, SubscriptionState state,
+ protected abstract void replaceSubscriptionData(ByteString topic, ByteString subscriberId, SubscriptionData data,
Callback<Void> callback, Object ctx);
- protected abstract void deleteSubscriptionState(ByteString topic, ByteString subscriberId, Callback<Void> callback,
+ protected abstract void deleteSubscriptionData(ByteString topic, ByteString subscriberId, Callback<Void> callback,
Object ctx);
}
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/subscriptions/InMemorySubscriptionManager.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/subscriptions/InMemorySubscriptionManager.java?rev=1374321&r1=1374320&r2=1374321&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/subscriptions/InMemorySubscriptionManager.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/subscriptions/InMemorySubscriptionManager.java Fri Aug 17 15:32:23 2012
@@ -22,7 +22,7 @@ import java.util.concurrent.ConcurrentHa
import java.util.concurrent.ScheduledExecutorService;
import com.google.protobuf.ByteString;
-import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionState;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionData;
import org.apache.hedwig.server.common.ServerConfiguration;
import org.apache.hedwig.server.persistence.PersistenceManager;
import org.apache.hedwig.server.topics.TopicManager;
@@ -35,21 +35,32 @@ public class InMemorySubscriptionManager
}
@Override
- protected void createSubscriptionState(ByteString topic, ByteString subscriberId, SubscriptionState state,
+ protected void createSubscriptionData(ByteString topic, ByteString subscriberId, SubscriptionData subData,
Callback<Void> callback, Object ctx) {
// nothing to do, in-memory info is already recorded by base class
callback.operationFinished(ctx, null);
}
@Override
- protected void deleteSubscriptionState(ByteString topic, ByteString subscriberId, Callback<Void> callback,
- Object ctx) {
+ protected void deleteSubscriptionData(ByteString topic, ByteString subscriberId, Callback<Void> callback,
+ Object ctx) {
// nothing to do, in-memory info is already deleted by base class
callback.operationFinished(ctx, null);
}
@Override
- protected void updateSubscriptionState(ByteString topic, ByteString subscriberId, SubscriptionState state,
+ protected boolean isPartialUpdateSupported() {
+ return false;
+ }
+
+ @Override
+ protected void updateSubscriptionData(ByteString topic, ByteString subscriberId, SubscriptionData data,
+ Callback<Void> callback, Object ctx) {
+ throw new UnsupportedOperationException("Doesn't support partial update");
+ }
+
+ @Override
+ protected void replaceSubscriptionData(ByteString topic, ByteString subscriberId, SubscriptionData data,
Callback<Void> callback, Object ctx) {
// nothing to do, in-memory info is already updated by base class
callback.operationFinished(ctx, null);
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/subscriptions/InMemorySubscriptionState.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/subscriptions/InMemorySubscriptionState.java?rev=1374321&r1=1374320&r2=1374321&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/subscriptions/InMemorySubscriptionState.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/subscriptions/InMemorySubscriptionState.java Fri Aug 17 15:32:23 2012
@@ -17,26 +17,56 @@
*/
package org.apache.hedwig.server.subscriptions;
+import java.util.HashMap;
+import java.util.Map;
+
+import com.google.protobuf.ByteString;
+
import org.apache.hedwig.protocol.PubSubProtocol.MessageSeqId;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionData;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionPreferences;
import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionState;
+import org.apache.hedwig.protoextensions.MapUtils;
+import org.apache.hedwig.protoextensions.SubscriptionStateUtils;
public class InMemorySubscriptionState {
SubscriptionState subscriptionState;
+ SubscriptionPreferences subscriptionPreferences;
MessageSeqId lastConsumeSeqId;
- public InMemorySubscriptionState(SubscriptionState subscriptionState, MessageSeqId lastConsumeSeqId) {
- this.subscriptionState = subscriptionState;
+ public InMemorySubscriptionState(SubscriptionData subscriptionData, MessageSeqId lastConsumeSeqId) {
+ this.subscriptionState = subscriptionData.getState();
+ if (subscriptionData.hasPreferences()) {
+ this.subscriptionPreferences = subscriptionData.getPreferences();
+ } else {
+ // set initial subscription preferences
+ SubscriptionPreferences.Builder prefsBuilder = SubscriptionPreferences.newBuilder();
+ // progate the old system preferences from subscription state to preferences
+ prefsBuilder.setMessageBound(subscriptionState.getMessageBound());
+ this.subscriptionPreferences = prefsBuilder.build();
+
+ }
this.lastConsumeSeqId = lastConsumeSeqId;
}
- public InMemorySubscriptionState(SubscriptionState subscriptionState) {
- this(subscriptionState, subscriptionState.getMsgId());
+ public InMemorySubscriptionState(SubscriptionData subscriptionData) {
+ this(subscriptionData, subscriptionData.getState().getMsgId());
+ }
+
+ public SubscriptionData toSubscriptionData() {
+ return SubscriptionData.newBuilder().setState(subscriptionState)
+ .setPreferences(subscriptionPreferences)
+ .build();
}
public SubscriptionState getSubscriptionState() {
return subscriptionState;
}
+ public SubscriptionPreferences getSubscriptionPreferences() {
+ return subscriptionPreferences;
+ }
+
public MessageSeqId getLastConsumeSeqId() {
return lastConsumeSeqId;
}
@@ -82,4 +112,51 @@ public class InMemorySubscriptionState {
return true;
}
+ /**
+ * Update preferences.
+ *
+ * @return true if preferences is updated, which needs to be persisted, false otherwise.
+ */
+ public boolean updatePreferences(SubscriptionPreferences preferences) {
+ boolean changed = false;
+ SubscriptionPreferences.Builder newPreferencesBuilder = SubscriptionPreferences.newBuilder(subscriptionPreferences);
+ if (preferences.hasMessageBound()) {
+ if (!subscriptionPreferences.hasMessageBound() ||
+ subscriptionPreferences.getMessageBound() != preferences.getMessageBound()) {
+ newPreferencesBuilder.setMessageBound(preferences.getMessageBound());
+ changed = true;
+ }
+ }
+ if (preferences.hasOptions()) {
+ Map<String, ByteString> userOptions = SubscriptionStateUtils.buildUserOptions(subscriptionPreferences);
+ Map<String, ByteString> optUpdates = SubscriptionStateUtils.buildUserOptions(preferences);
+ boolean optChanged = false;
+ for (Map.Entry<String, ByteString> entry : optUpdates.entrySet()) {
+ String key = entry.getKey();
+ if (userOptions.containsKey(key)) {
+ if (null == entry.getValue()) {
+ userOptions.remove(key);
+ optChanged = true;
+ } else {
+ if (!entry.getValue().equals(userOptions.get(key))) {
+ userOptions.put(key, entry.getValue());
+ optChanged = true;
+ }
+ }
+ } else {
+ userOptions.put(key, entry.getValue());
+ optChanged = true;
+ }
+ }
+ if (optChanged) {
+ changed = true;
+ newPreferencesBuilder.setOptions(MapUtils.buildMapBuilder(userOptions));
+ }
+ }
+ if (changed) {
+ subscriptionPreferences = newPreferencesBuilder.build();
+ }
+ return changed;
+ }
+
}
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/subscriptions/MMSubscriptionManager.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/subscriptions/MMSubscriptionManager.java?rev=1374321&r1=1374320&r2=1374321&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/subscriptions/MMSubscriptionManager.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/subscriptions/MMSubscriptionManager.java Fri Aug 17 15:32:23 2012
@@ -19,10 +19,12 @@ package org.apache.hedwig.server.subscri
import java.io.IOException;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import com.google.protobuf.ByteString;
-import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionState;
+import org.apache.hedwig.exceptions.PubSubException;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionData;
import org.apache.hedwig.server.common.ServerConfiguration;
import org.apache.hedwig.server.delivery.DeliveryManager;
import org.apache.hedwig.server.meta.MetadataManagerFactory;
@@ -49,25 +51,49 @@ public class MMSubscriptionManager exten
@Override
protected void readSubscriptions(final ByteString topic,
final Callback<Map<ByteString, InMemorySubscriptionState>> cb, final Object ctx) {
- subManager.readSubscriptions(topic, cb, ctx);
+ subManager.readSubscriptions(topic, new Callback<Map<ByteString, SubscriptionData>>() {
+ @Override
+ public void operationFailed(Object ctx, PubSubException pse) {
+ cb.operationFailed(ctx, pse);
+ }
+ @Override
+ public void operationFinished(Object ctx, Map<ByteString, SubscriptionData> subs) {
+ Map<ByteString, InMemorySubscriptionState> results = new ConcurrentHashMap<ByteString, InMemorySubscriptionState>();
+ for (Map.Entry<ByteString, SubscriptionData> subEntry : subs.entrySet()) {
+ results.put(subEntry.getKey(), new InMemorySubscriptionState(subEntry.getValue()));
+ }
+ cb.operationFinished(ctx, results);
+ }
+ }, ctx);
}
@Override
- protected void createSubscriptionState(final ByteString topic, final ByteString subscriberId,
- final SubscriptionState state, final Callback<Void> callback, final Object ctx) {
- subManager.createSubscriptionState(topic, subscriberId, state, callback, ctx);
+ protected boolean isPartialUpdateSupported() {
+ return subManager.isPartialUpdateSupported();
}
@Override
- protected void updateSubscriptionState(final ByteString topic, final ByteString subscriberId,
- final SubscriptionState state, final Callback<Void> callback, final Object ctx) {
- subManager.updateSubscriptionState(topic, subscriberId, state, callback, ctx);
+ protected void createSubscriptionData(final ByteString topic, final ByteString subscriberId,
+ final SubscriptionData subData, final Callback<Void> callback, final Object ctx) {
+ subManager.createSubscriptionData(topic, subscriberId, subData, callback, ctx);
}
@Override
- protected void deleteSubscriptionState(final ByteString topic, final ByteString subscriberId,
- final Callback<Void> callback, final Object ctx) {
- subManager.deleteSubscriptionState(topic, subscriberId, callback, ctx);
+ protected void replaceSubscriptionData(final ByteString topic, final ByteString subscriberId,
+ final SubscriptionData subData, final Callback<Void> callback, final Object ctx) {
+ subManager.replaceSubscriptionData(topic, subscriberId, subData, callback, ctx);
+ }
+
+ @Override
+ protected void updateSubscriptionData(final ByteString topic, final ByteString subscriberId,
+ final SubscriptionData subData, final Callback<Void> callback, final Object ctx) {
+ subManager.updateSubscriptionData(topic, subscriberId, subData, callback, ctx);
+ }
+
+ @Override
+ protected void deleteSubscriptionData(final ByteString topic, final ByteString subscriberId,
+ final Callback<Void> callback, final Object ctx) {
+ subManager.deleteSubscriptionData(topic, subscriberId, callback, ctx);
}
@Override
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/TestBackwardCompat.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/TestBackwardCompat.java?rev=1374321&r1=1374320&r2=1374321&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/TestBackwardCompat.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/TestBackwardCompat.java Fri Aug 17 15:32:23 2012
@@ -43,6 +43,7 @@ public class TestBackwardCompat extends
private static Logger logger = LoggerFactory.getLogger(TestBackwardCompat.class);
+ static final int CONSUMEINTERVAL = 5;
static ZooKeeperUtil zkUtil = new ZooKeeperUtil();
static class BookKeeperCluster400 {
@@ -257,6 +258,10 @@ public class TestBackwardCompat extends
Server410(final String zkHosts) {
conf = new org.apache.hw_v4_1_0.hedwig.server.common.ServerConfiguration() {
@Override
+ public int getConsumeInterval() {
+ return CONSUMEINTERVAL;
+ }
+ @Override
public String getZkHost() {
return zkHosts;
}
@@ -281,8 +286,52 @@ public class TestBackwardCompat extends
org.apache.hw_v4_1_0.hedwig.client.api.Publisher publisher;
org.apache.hw_v4_1_0.hedwig.client.api.Subscriber subscriber;
+ class IntMessageHandler implements org.apache.hw_v4_1_0.hedwig.client.api.MessageHandler {
+ ByteString topic;
+ ByteString subId;
+ int next;
+
+ CountDownLatch latch;
+
+ IntMessageHandler(ByteString t, ByteString s, int start, int num) {
+ this.topic = t;
+ this.subId = s;
+ this.next = start;
+ this.latch = new CountDownLatch(num);
+ }
+
+ @Override
+ public void deliver(ByteString t, ByteString s,
+ org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.Message msg,
+ org.apache.hw_v4_1_0.hedwig.util.Callback<Void> callback, Object context) {
+ if (!t.equals(topic) || !s.equals(subId)) {
+ return;
+ }
+ int num = Integer.parseInt(msg.getBody().toStringUtf8());
+ if (num == next) {
+ latch.countDown();
+ ++next;
+ }
+ callback.operationFinished(context, null);
+ }
+
+ public boolean await(long timeout, TimeUnit unit)
+ throws InterruptedException {
+ return latch.await(timeout, unit);
+ }
+ }
+
Client410() {
- conf = new org.apache.hw_v4_1_0.hedwig.client.conf.ClientConfiguration();
+ conf = new org.apache.hw_v4_1_0.hedwig.client.conf.ClientConfiguration() {
+ @Override
+ public boolean isAutoSendConsumeMessageEnabled() {
+ return true;
+ }
+ @Override
+ public int getConsumedMessagesBufferSize() {
+ return 1;
+ }
+ };
client = new org.apache.hw_v4_1_0.hedwig.client.HedwigClient(conf);
publisher = client.getPublisher();
subscriber = client.getSubscriber();
@@ -302,6 +351,79 @@ public class TestBackwardCompat extends
publisher.publish(topic, message);
return null;
}
+
+ void publishInts(ByteString topic, int start, int num) throws Exception {
+ for (int i=0; i<num; i++) {
+ org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.Message msg =
+ org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.Message.newBuilder().setBody(ByteString.copyFromUtf8("" + (start+i))).build();
+ publisher.publish(topic, msg);
+ }
+ }
+
+ void sendXExpectLastY(ByteString topic, ByteString subid, final int x, final int y)
+ throws Exception {
+ for (int i=0; i<x; i++) {
+ publisher.publish(topic, org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.Message.newBuilder().setBody(
+ ByteString.copyFromUtf8(String.valueOf(i))).build());
+ }
+ subscriber.subscribe(topic, subid, org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.ATTACH);
+
+ final AtomicInteger expected = new AtomicInteger(x - y);
+ final CountDownLatch latch = new CountDownLatch(1);
+ subscriber.startDelivery(topic, subid, new org.apache.hw_v4_1_0.hedwig.client.api.MessageHandler() {
+ @Override
+ synchronized public void deliver(ByteString topic, ByteString subscriberId,
+ org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.Message msg,
+ org.apache.hw_v4_1_0.hedwig.util.Callback<Void> callback, Object context) {
+ try {
+ int value = Integer.valueOf(msg.getBody().toStringUtf8());
+ if (value == expected.get()) {
+ expected.incrementAndGet();
+ } else {
+ logger.error("Did not receive expected value, expected {}, got {}",
+ expected.get(), value);
+ expected.set(0);
+ latch.countDown();
+ }
+ if (expected.get() == x) {
+ latch.countDown();
+ }
+ callback.operationFinished(context, null);
+ } catch (Exception e) {
+ logger.error("Received bad message", e);
+ latch.countDown();
+ }
+ }
+ });
+ assertTrue("Timed out waiting for messages Y is " + y + " expected is currently "
+ + expected.get(), latch.await(10, TimeUnit.SECONDS));
+ assertEquals("Should be expected message with " + x, x, expected.get());
+ subscriber.stopDelivery(topic, subid);
+ subscriber.closeSubscription(topic, subid);
+ }
+
+ void subscribe(ByteString topic, ByteString subscriberId) throws Exception {
+ org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscriptionOptions options =
+ org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscriptionOptions.newBuilder()
+ .setCreateOrAttach(org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.CREATE_OR_ATTACH).build();
+ subscribe(topic, subscriberId, options);
+ }
+
+ void subscribe(ByteString topic, ByteString subscriberId,
+ org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscriptionOptions options) throws Exception {
+ subscriber.subscribe(topic, subscriberId, options);
+ }
+
+ void closeSubscription(ByteString topic, ByteString subscriberId) throws Exception {
+ subscriber.closeSubscription(topic, subscriberId);
+ }
+
+ void receiveInts(ByteString topic, ByteString subscriberId, int start, int num) throws Exception {
+ IntMessageHandler msgHandler = new IntMessageHandler(topic, subscriberId, start, num);
+ subscriber.startDelivery(topic, subscriberId, msgHandler);
+ msgHandler.await(num, TimeUnit.SECONDS);
+ subscriber.stopDelivery(topic, subscriberId);
+ }
}
/**
@@ -387,6 +509,11 @@ public class TestBackwardCompat extends
ServerCurrent(final String zkHosts) {
conf = new org.apache.hedwig.server.common.ServerConfiguration() {
@Override
+ public int getConsumeInterval() {
+ return CONSUMEINTERVAL;
+ }
+
+ @Override
public String getZkHost() {
return zkHosts;
}
@@ -676,4 +803,119 @@ public class TestBackwardCompat extends
bkc410.stop();
}
+ /**
+ * Test compatability between version 4.1.0 and the current version.
+ *
+ * A current server could read subscription data recorded by 4.1.0 server.
+ */
+ @Test
+ public void testSubscriptionDataCompat410() throws Exception {
+ ByteString topic = ByteString.copyFromUtf8("TestCompat410");
+ ByteString sub410 = ByteString.copyFromUtf8("sub410");
+ ByteString subcur = ByteString.copyFromUtf8("subcur");
+
+ // start bookkeeper 410
+ BookKeeperCluster410 bkc410 = new BookKeeperCluster410(3);
+ bkc410.start();
+
+ // start 410 server
+ Server410 s410 = new Server410(zkUtil.getZooKeeperConnectString());
+ s410.start();
+
+ Client410 c410 = new Client410();
+ c410.subscribe(topic, sub410);
+ c410.closeSubscription(topic, sub410);
+
+ ClientCurrent ccur = new ClientCurrent();
+ ccur.subscribe(topic, subcur);
+ ccur.closeSubscription(topic, subcur);
+
+ // publish messages using old client
+ c410.publishInts(topic, 0, 10);
+ // stop 410 server
+ s410.stop();
+
+ // start current server
+ ServerCurrent scur = new ServerCurrent(zkUtil.getZooKeeperConnectString());
+ scur.start();
+
+ c410.subscribe(topic, sub410);
+ c410.receiveInts(topic, sub410, 0, 10);
+
+ ccur.subscribe(topic, subcur);
+ ccur.receiveInts(topic, subcur, 0, 10);
+
+ // publish messages using current client
+ ccur.publishInts(topic, 10, 10);
+
+ c410.receiveInts(topic, sub410, 10, 10);
+ ccur.receiveInts(topic, subcur, 10, 10);
+
+ // stop current server
+ scur.stop();
+
+ c410.close();
+ ccur.close();
+
+ // stop bookkeeper cluster
+ bkc410.stop();
+ }
+
+ /**
+ * Test compatability between version 4.1.0 and the current version.
+ *
+ * A 4.1.0 client could not update message bound, while current could do it.
+ */
+ @Test
+ public void testUpdateMessageBoundCompat410() throws Exception {
+ ByteString topic = ByteString.copyFromUtf8("TestUpdateMessageBoundCompat410");
+ ByteString subid = ByteString.copyFromUtf8("mysub");
+
+ // start bookkeeper
+ BookKeeperClusterCurrent bkccur= new BookKeeperClusterCurrent(3);
+ bkccur.start();
+
+ // start hub server
+ ServerCurrent scur = new ServerCurrent(zkUtil.getZooKeeperConnectString());
+ scur.start();
+
+ org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions options5cur =
+ org.apache.hedwig.protocol.PubSubProtocol.SubscriptionOptions.newBuilder()
+ .setCreateOrAttach(org.apache.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.CREATE_OR_ATTACH)
+ .setMessageBound(5).build();
+ org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscriptionOptions options5v410 =
+ org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscriptionOptions.newBuilder()
+ .setCreateOrAttach(org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.CREATE_OR_ATTACH)
+ .setMessageBound(5).build();
+ org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscriptionOptions options20v410 =
+ org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscriptionOptions.newBuilder()
+ .setCreateOrAttach(org.apache.hw_v4_1_0.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach.CREATE_OR_ATTACH)
+ .setMessageBound(20).build();
+
+ Client410 c410 = new Client410();
+ c410.subscribe(topic, subid, options20v410);
+ c410.closeSubscription(topic, subid);
+ c410.sendXExpectLastY(topic, subid, 50, 20);
+
+ c410.subscribe(topic, subid, options5v410);
+ c410.closeSubscription(topic, subid);
+ // the message bound isn't updated.
+ c410.sendXExpectLastY(topic, subid, 50, 20);
+
+ ClientCurrent ccur = new ClientCurrent();
+ ccur.subscribe(topic, subid, options5cur);
+ ccur.closeSubscription(topic, subid);
+ // the message bound should be updated.
+ c410.sendXExpectLastY(topic, subid, 50, 5);
+
+ // stop current server
+ scur.stop();
+
+ c410.close();
+ ccur.close();
+
+ // stop bookkeeper cluster
+ bkccur.stop();
+ }
+
}
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestMetadataManager.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestMetadataManager.java?rev=1374321&r1=1374320&r2=1374321&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestMetadataManager.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/meta/TestMetadataManager.java Fri Aug 17 15:32:23 2012
@@ -32,8 +32,8 @@ import org.apache.hedwig.exceptions.PubS
import org.apache.hedwig.protocol.PubSubProtocol.LedgerRanges;
import org.apache.hedwig.protocol.PubSubProtocol.LedgerRange;
import org.apache.hedwig.protocol.PubSubProtocol.MessageSeqId;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionData;
import org.apache.hedwig.protocol.PubSubProtocol.SubscriptionState;
-import org.apache.hedwig.server.subscriptions.InMemorySubscriptionState;
import org.apache.hedwig.server.topics.HubInfo;
import org.apache.hedwig.server.meta.MetadataManagerFactory;
import org.apache.hedwig.util.Either;
@@ -228,36 +228,41 @@ public class TestMetadataManager extends
}
@Test
- public void testSubscriptionState() throws Exception {
+ public void testSubscriptionData() throws Exception {
SubscriptionDataManager subManager = metadataManagerFactory.newSubscriptionDataManager();
- ByteString topic = ByteString.copyFromUtf8("testSubscriptionState");
+ ByteString topic = ByteString.copyFromUtf8("testSubscriptionData");
ByteString subid = ByteString.copyFromUtf8("mysub");
StubCallback<Void> callback = new StubCallback<Void>();
- StubCallback<SubscriptionState> readCallback = new StubCallback<SubscriptionState>();
- StubCallback<Map<ByteString, InMemorySubscriptionState>> subsCallback
- = new StubCallback<Map<ByteString, InMemorySubscriptionState>>();
+ StubCallback<SubscriptionData> readCallback = new StubCallback<SubscriptionData>();
+ StubCallback<Map<ByteString, SubscriptionData>> subsCallback
+ = new StubCallback<Map<ByteString, SubscriptionData>>();
- subManager.readSubscriptionState(topic, subid, readCallback, null);
- Either<SubscriptionState, PubSubException> readRes = readCallback.queue.take();
+ subManager.readSubscriptionData(topic, subid, readCallback, null);
+ Either<SubscriptionData, PubSubException> readRes = readCallback.queue.take();
Assert.assertEquals("Found inconsistent subscription state", null, readRes.left());
Assert.assertEquals("Should not fail with PubSubException", null, readRes.right());
// read non-existed subscription state
subManager.readSubscriptions(topic, subsCallback, null);
- Either<Map<ByteString, InMemorySubscriptionState>, PubSubException> res = subsCallback.queue.take();
+ Either<Map<ByteString, SubscriptionData>, PubSubException> res = subsCallback.queue.take();
Assert.assertEquals("Found more than 0 subscribers", 0, res.left().size());
Assert.assertEquals("Should not fail with PubSubException", null, res.right());
// update non-existed subscription state
- subManager.updateSubscriptionState(topic, subid, SubscriptionState.getDefaultInstance(),
- callback, null);
+ if (subManager.isPartialUpdateSupported()) {
+ subManager.updateSubscriptionData(topic, subid, SubscriptionData.getDefaultInstance(),
+ callback, null);
+ } else {
+ subManager.replaceSubscriptionData(topic, subid, SubscriptionData.getDefaultInstance(),
+ callback, null);
+ }
Assert.assertTrue("Should fail to update a non-existed subscriber with PubSubException",
callback.queue.take().right() instanceof PubSubException.NoSubscriptionStateException);
// delete non-existed subscription state
- subManager.deleteSubscriptionState(topic, subid, callback, null);
+ subManager.deleteSubscriptionData(topic, subid, callback, null);
Assert.assertTrue("Should fail to delete a non-existed subscriber with PubSubException",
callback.queue.take().right() instanceof PubSubException.NoSubscriptionStateException);
@@ -266,10 +271,11 @@ public class TestMetadataManager extends
builder.setLocalComponent(seqId);
MessageSeqId msgId = builder.build();
- SubscriptionState state = SubscriptionState.newBuilder(SubscriptionState.getDefaultInstance()).setMsgId(msgId).build();
+ SubscriptionState.Builder stateBuilder = SubscriptionState.newBuilder(SubscriptionState.getDefaultInstance()).setMsgId(msgId);
+ SubscriptionData data = SubscriptionData.newBuilder().setState(stateBuilder).build();
// create a subscription state
- subManager.createSubscriptionState(topic, subid, state, callback, null);
+ subManager.createSubscriptionData(topic, subid, data, callback, null);
Assert.assertEquals("Should not fail with PubSubException",
null, callback.queue.take().right());
@@ -278,11 +284,11 @@ public class TestMetadataManager extends
res = subsCallback.queue.take();
Assert.assertEquals("Should find just 1 subscriber", 1, res.left().size());
Assert.assertEquals("Should not fail with PubSubException", null, res.right());
- InMemorySubscriptionState imss = res.left().get(subid);
+ SubscriptionData imss = res.left().get(subid);
Assert.assertEquals("Found inconsistent subscription state",
- state, imss.getSubscriptionState());
+ data, imss);
Assert.assertEquals("Found inconsistent last consumed seq id",
- seqId, imss.getLastConsumeSeqId().getLocalComponent());
+ seqId, imss.getState().getMsgId().getLocalComponent());
// move consume seq id
seqId = 99;
@@ -290,16 +296,21 @@ public class TestMetadataManager extends
builder.setLocalComponent(seqId);
msgId = builder.build();
- state = SubscriptionState.newBuilder(state).setMsgId(msgId).build();
+ stateBuilder = SubscriptionState.newBuilder(data.getState()).setMsgId(msgId);
+ data = SubscriptionData.newBuilder().setState(stateBuilder).build();
// update subscription state
- subManager.updateSubscriptionState(topic, subid, state, callback, null);
+ if (subManager.isPartialUpdateSupported()) {
+ subManager.updateSubscriptionData(topic, subid, data, callback, null);
+ } else {
+ subManager.replaceSubscriptionData(topic, subid, data, callback, null);
+ }
Assert.assertEquals("Fail to update a subscription state", null, callback.queue.take().right());
// read subscription state
- subManager.readSubscriptionState(topic, subid, readCallback, null);
+ subManager.readSubscriptionData(topic, subid, readCallback, null);
Assert.assertEquals("Found inconsistent subscription state",
- state, readCallback.queue.take().left());
+ data, readCallback.queue.take().left());
// read subscriptions again
subManager.readSubscriptions(topic, subsCallback, null);
@@ -308,11 +319,11 @@ public class TestMetadataManager extends
Assert.assertEquals("Should not fail with PubSubException", null, res.right());
imss = res.left().get(subid);
Assert.assertEquals("Found inconsistent subscription state",
- state, imss.getSubscriptionState());
+ data, imss);
Assert.assertEquals("Found inconsistent last consumed seq id",
- seqId, imss.getLastConsumeSeqId().getLocalComponent());
+ seqId, imss.getState().getMsgId().getLocalComponent());
- subManager.deleteSubscriptionState(topic, subid, callback, null);
+ subManager.deleteSubscriptionData(topic, subid, callback, null);
Assert.assertEquals("Fail to delete an existed subscriber", null, callback.queue.take().right());
// read subscription states again
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/MessageBoundedPersistenceTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/MessageBoundedPersistenceTest.java?rev=1374321&r1=1374320&r2=1374321&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/MessageBoundedPersistenceTest.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/persistence/MessageBoundedPersistenceTest.java Fri Aug 17 15:32:23 2012
@@ -191,6 +191,45 @@ public class MessageBoundedPersistenceTe
}
@Test
+ public void testUpdateMessageBound() throws Exception {
+ ByteString topic = ByteString.copyFromUtf8("UpdateMessageBound");
+
+ Client client = new HedwigClient(new ClientConfiguration());
+ Publisher pub = client.getPublisher();
+ Subscriber sub = client.getSubscriber();
+
+ SubscriptionOptions options5 = SubscriptionOptions.newBuilder()
+ .setCreateOrAttach(CreateOrAttach.CREATE_OR_ATTACH).setMessageBound(5).build();
+ SubscriptionOptions options20 = SubscriptionOptions.newBuilder()
+ .setCreateOrAttach(CreateOrAttach.CREATE_OR_ATTACH).setMessageBound(20).build();
+ SubscriptionOptions options10 = SubscriptionOptions.newBuilder()
+ .setCreateOrAttach(CreateOrAttach.CREATE_OR_ATTACH).setMessageBound(10).build();
+
+ ByteString subid = ByteString.copyFromUtf8("updateSubId");
+
+ sub.subscribe(topic, subid, options5);
+ sub.closeSubscription(topic, subid);
+ sendXExpectLastY(pub, sub, topic, subid, 50, 5);
+
+ // update bound to 20
+ sub.subscribe(topic, subid, options20);
+ sub.closeSubscription(topic, subid);
+ sendXExpectLastY(pub, sub, topic, subid, 50, 20);
+
+ // update bound to 10
+ sub.subscribe(topic, subid, options10);
+ sub.closeSubscription(topic, subid);
+ sendXExpectLastY(pub, sub, topic, subid, 50, 10);
+
+ // message bound is not provided, no update
+ sub.subscribe(topic, subid, CreateOrAttach.CREATE_OR_ATTACH);
+ sub.closeSubscription(topic, subid);
+ sendXExpectLastY(pub, sub, topic, subid, 50, 10);
+
+ client.close();
+ }
+
+ @Test
public void testLedgerGC() throws Exception {
Client client = new HedwigClient(new MessageBoundClientConfiguration());
Publisher pub = client.getPublisher();