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 2013/01/03 12:07:57 UTC
svn commit: r1428277 - in /zookeeper/bookkeeper/trunk: ./
hedwig-server/src/main/java/org/apache/hedwig/server/delivery/
hedwig-server/src/main/java/org/apache/hedwig/server/handlers/
hedwig-server/src/test/java/org/apache/hedwig/client/ hedwig-server/...
Author: ivank
Date: Thu Jan 3 11:07:56 2013
New Revision: 1428277
URL: http://svn.apache.org/viewvc?rev=1428277&view=rev
Log:
BOOKKEEPER-507: Race condition happens if closeSubscription and subscribe happened at the same time (in multiplexed client). (sijie via ivank)
Added:
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/client/TestSubAfterCloseSub.java
Modified:
zookeeper/bookkeeper/trunk/CHANGES.txt
zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/delivery/DeliveryManager.java
zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/delivery/FIFODeliveryManager.java
zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/handlers/SubscribeHandler.java
zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/StubDeliveryManager.java
Modified: zookeeper/bookkeeper/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/CHANGES.txt?rev=1428277&r1=1428276&r2=1428277&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/CHANGES.txt (original)
+++ zookeeper/bookkeeper/trunk/CHANGES.txt Thu Jan 3 11:07:56 2013
@@ -209,6 +209,8 @@ Trunk (unreleased changes)
BOOKKEEPER-458: Annoy BKReadException error when changing ledger. (jiannan via fpj)
+ BOOKKEEPER-507: Race condition happens if closeSubscription and subscribe happened at the same time (in multiplexed client). (sijie via ivank)
+
IMPROVEMENTS:
BOOKKEEPER-467: Allocate ports for testing dynamically (ivank)
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/delivery/DeliveryManager.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/delivery/DeliveryManager.java?rev=1428277&r1=1428276&r2=1428277&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/delivery/DeliveryManager.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/delivery/DeliveryManager.java Thu Jan 3 11:07:56 2013
@@ -27,11 +27,32 @@ import org.apache.hedwig.util.Callback;
public interface DeliveryManager {
public void start();
+ /**
+ * Start serving a given subscription.
+ *
+ * @param topic
+ * Topic Name
+ * @param subscriberId
+ * Subscriber Id
+ * @param preferences
+ * Subscription Preferences
+ * @param seqIdToStartFrom
+ * Message sequence id starting delivery from.
+ * @param endPoint
+ * End point to deliver messages to.
+ * @param filter
+ * Message filter used to filter messages before delivery.
+ * @param callback
+ * Callback instance.
+ * @param ctx
+ * Callback context.
+ */
public void startServingSubscription(ByteString topic, ByteString subscriberId,
SubscriptionPreferences preferences,
MessageSeqId seqIdToStartFrom,
DeliveryEndPoint endPoint,
- ServerMessageFilter filter);
+ ServerMessageFilter filter,
+ Callback<Void> callback, Object ctx);
/**
* Stop serving a given subscription.
@@ -40,6 +61,12 @@ public interface DeliveryManager {
* Topic Name
* @param subscriberId
* Subscriber Id
+ * @param event
+ * Subscription event indicating the reason to stop the subscriber.
+ * @param callback
+ * Callback instance.
+ * @param ctx
+ * Callback context.
*/
public void stopServingSubscriber(ByteString topic, ByteString subscriberId,
SubscriptionEvent event,
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/delivery/FIFODeliveryManager.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/delivery/FIFODeliveryManager.java?rev=1428277&r1=1428276&r2=1428277&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/delivery/FIFODeliveryManager.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/delivery/FIFODeliveryManager.java Thu Jan 3 11:07:56 2013
@@ -158,17 +158,22 @@ public class FIFODeliveryManager impleme
* @param filter
* Only messages passing this filter should be sent to this
* subscriber
+ * @param callback
+ * Callback instance
+ * @param ctx
+ * Callback context
*/
+ @Override
public void startServingSubscription(ByteString topic, ByteString subscriberId,
SubscriptionPreferences preferences,
MessageSeqId seqIdToStartFrom,
- DeliveryEndPoint endPoint, ServerMessageFilter filter) {
-
+ DeliveryEndPoint endPoint, ServerMessageFilter filter,
+ Callback<Void> callback, Object ctx) {
ActiveSubscriberState subscriber =
new ActiveSubscriberState(topic, subscriberId,
preferences,
seqIdToStartFrom.getLocalComponent() - 1,
- endPoint, filter);
+ endPoint, filter, callback, ctx);
enqueueWithoutFailure(subscriber);
}
@@ -376,7 +381,8 @@ public class FIFODeliveryManager impleme
boolean isThrottled = false;
final int messageWindowSize;
ServerMessageFilter filter;
- // TODO make use of these variables
+ Callback<Void> cb;
+ Object ctx;
final static int SEQ_ID_SLACK = 10;
@@ -384,7 +390,8 @@ public class FIFODeliveryManager impleme
SubscriptionPreferences preferences,
long lastLocalSeqIdDelivered,
DeliveryEndPoint deliveryEndPoint,
- ServerMessageFilter filter) {
+ ServerMessageFilter filter,
+ Callback<Void> cb, Object ctx) {
this.topic = topic;
this.subscriberId = subscriberId;
this.lastLocalSeqIdDelivered = lastLocalSeqIdDelivered;
@@ -401,6 +408,8 @@ public class FIFODeliveryManager impleme
messageWindowSize = UNLIMITED;
}
}
+ this.cb = cb;
+ this.ctx = ctx;
}
public void setNotConnected(SubscriptionEvent event) {
@@ -617,6 +626,8 @@ public class FIFODeliveryManager impleme
* {@link DeliveryManagerRequest} methods
*/
public void performRequest() {
+ // try the callback to tell it started to deliver the message
+ cb.operationFinished(ctx, (Void)null);
// Put this subscriber in the channel to subscriber mapping
ActiveSubscriberState prevSubscriber =
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/handlers/SubscribeHandler.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/handlers/SubscribeHandler.java?rev=1428277&r1=1428276&r2=1428277&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/handlers/SubscribeHandler.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/main/java/org/apache/hedwig/server/handlers/SubscribeHandler.java Thu Jan 3 11:07:56 2013
@@ -121,7 +121,7 @@ public class SubscribeHandler extends Ba
}
@Override
- public void operationFinished(Object ctx, SubscriptionData subData) {
+ public void operationFinished(Object ctx, final SubscriptionData subData) {
TopicSubscriber topicSub = new TopicSubscriber(topic, subscriberId);
synchronized (channel) {
@@ -185,27 +185,36 @@ public class SubscribeHandler extends Ba
.addListener(ChannelFutureListener.CLOSE);
return;
}
- // First write success and then tell the delivery manager,
- // otherwise the first message might go out before the response
- // to the subscribe
- SubscribeResponse.Builder subRespBuilder = SubscribeResponse.newBuilder()
- .setPreferences(subData.getPreferences());
- ResponseBody respBody = ResponseBody.newBuilder()
- .setSubscribeResponse(subRespBuilder).build();
- channel.write(PubSubResponseUtils.getSuccessResponse(request.getTxnId(), respBody));
- logger.info("Subscribe request (" + request.getTxnId() + ") for (topic:" + topic.toStringUtf8()
- + ", subscriber:" + subscriberId.toStringUtf8() + ") from channel " + channel.getRemoteAddress()
- + " succeed - its subscription data is " + SubscriptionStateUtils.toString(subData));
- subStats.updateLatency(MathUtils.now() - requestTime);
// want to start 1 ahead of the consume ptr
MessageSeqId lastConsumedSeqId = subData.getState().getMsgId();
MessageSeqId seqIdToStartFrom = MessageSeqId.newBuilder(lastConsumedSeqId).setLocalComponent(
lastConsumedSeqId.getLocalComponent() + 1).build();
deliveryMgr.startServingSubscription(topic, subscriberId,
- subData.getPreferences(),
- seqIdToStartFrom,
- new ChannelEndPoint(channel), filter);
+ subData.getPreferences(), seqIdToStartFrom, new ChannelEndPoint(channel), filter,
+ new Callback<Void>() {
+ @Override
+ public void operationFinished(Object ctx, Void result) {
+ // First write success and then tell the delivery manager,
+ // otherwise the first message might go out before the response
+ // to the subscribe
+ SubscribeResponse.Builder subRespBuilder = SubscribeResponse.newBuilder()
+ .setPreferences(subData.getPreferences());
+ ResponseBody respBody = ResponseBody.newBuilder()
+ .setSubscribeResponse(subRespBuilder).build();
+ channel.write(PubSubResponseUtils.getSuccessResponse(request.getTxnId(), respBody));
+ logger.info("Subscribe request (" + request.getTxnId() + ") for (topic:"
+ + topic.toStringUtf8() + ", subscriber:" + subscriberId.toStringUtf8()
+ + ") from channel " + channel.getRemoteAddress()
+ + " succeed - its subscription data is "
+ + SubscriptionStateUtils.toString(subData));
+ subStats.updateLatency(MathUtils.now() - requestTime);
+ }
+ @Override
+ public void operationFailed(Object ctx, PubSubException exception) {
+ // would not happened
+ }
+ }, null);
}
}, null);
Added: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/client/TestSubAfterCloseSub.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/client/TestSubAfterCloseSub.java?rev=1428277&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/client/TestSubAfterCloseSub.java (added)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/client/TestSubAfterCloseSub.java Thu Jan 3 11:07:56 2013
@@ -0,0 +1,194 @@
+/**
+ * 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.integration;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.google.protobuf.ByteString;
+import org.apache.hedwig.client.api.MessageHandler;
+import org.apache.hedwig.client.api.Subscriber;
+import org.apache.hedwig.client.conf.ClientConfiguration;
+import org.apache.hedwig.client.exceptions.InvalidSubscriberIdException;
+import org.apache.hedwig.client.exceptions.AlreadyStartDeliveryException;
+import org.apache.hedwig.client.HedwigClient;
+import org.apache.hedwig.client.api.Client;
+import org.apache.hedwig.client.api.Publisher;
+import org.apache.hedwig.client.api.Subscriber;
+import org.apache.hedwig.exceptions.PubSubException;
+import org.apache.hedwig.exceptions.PubSubException.ClientNotSubscribedException;
+import org.apache.hedwig.protocol.PubSubProtocol.Message;
+import org.apache.hedwig.protocol.PubSubProtocol.MessageSeqId;
+import org.apache.hedwig.protocol.PubSubProtocol.OperationType;
+import org.apache.hedwig.protocol.PubSubProtocol.ProtocolVersion;
+import org.apache.hedwig.protocol.PubSubProtocol.PubSubRequest;
+import org.apache.hedwig.protocol.PubSubProtocol.PubSubResponse;
+import org.apache.hedwig.protocol.PubSubProtocol.StartDeliveryRequest;
+import org.apache.hedwig.protocol.PubSubProtocol.StopDeliveryRequest;
+import org.apache.hedwig.protocol.PubSubProtocol.StatusCode;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach;
+import org.apache.hedwig.protoextensions.SubscriptionStateUtils;
+import org.apache.hedwig.server.HedwigHubTestBase;
+import org.apache.hedwig.server.netty.WriteRecordingChannel;
+import org.apache.hedwig.server.proxy.HedwigProxy;
+import org.apache.hedwig.server.proxy.ProxyConfiguration;
+import org.apache.hedwig.server.regions.HedwigHubClient;
+import org.apache.hedwig.util.Callback;
+import org.apache.hedwig.util.ConcurrencyUtils;
+import org.apache.hedwig.util.HedwigSocketAddress;
+import org.apache.bookkeeper.test.PortManager;
+import org.apache.hedwig.server.LoggingExceptionHandler;
+
+public class TestSubAfterCloseSub extends HedwigHubTestBase {
+
+ class TestClientConfiguration extends HubClientConfiguration {
+
+ boolean isSubscriptionChannelSharingEnabled;
+
+ TestClientConfiguration(boolean isSubscriptionChannelSharingEnabled) {
+ this.isSubscriptionChannelSharingEnabled = isSubscriptionChannelSharingEnabled;
+ }
+
+ @Override
+ public boolean isSubscriptionChannelSharingEnabled() {
+ return isSubscriptionChannelSharingEnabled;
+ }
+ }
+
+ private void sleepThread(final String name, final CountDownLatch wakeupLatch)
+ throws IOException {
+ Thread[] allthreads = new Thread[Thread.activeCount()];
+ Thread.enumerate(allthreads);
+ for (final Thread t : allthreads) {
+ if (t.getName().equals(name)) {
+ Thread sleeper = new Thread() {
+ @Override
+ public void run() {
+ try {
+ t.suspend();
+ wakeupLatch.await();
+ t.resume();
+ } catch (Exception e) {
+ logger.error("Error suspending thread " + name + " : ", e);
+ }
+ }
+ };
+ sleeper.start();
+ return;
+ }
+ }
+ throw new IOException("Could not find thread " + name);
+ }
+
+ /**
+ * {@link https://issues.apache.org/jira/browse/BOOKKEEPER-507}
+ */
+ @Test(timeout=15000)
+ public void testSubAfterCloseSubForSimpleClient() throws Exception {
+ runSubAfterCloseSubTest(false);
+ }
+
+ /**
+ * {@link https://issues.apache.org/jira/browse/BOOKKEEPER-507}
+ */
+ @Test(timeout=15000)
+ public void testSubAfterCloseSubForMultiplexClient() throws Exception {
+ runSubAfterCloseSubTest(true);
+ }
+
+ private void runSubAfterCloseSubTest(boolean sharedSubscriptionChannel) throws Exception {
+ HedwigClient client = new HedwigClient(new TestClientConfiguration(sharedSubscriptionChannel));
+ Publisher publisher = client.getPublisher();
+ final Subscriber subscriber = client.getSubscriber();
+
+ final ByteString topic = ByteString.copyFromUtf8("TestSubAfterCloseSub-" + sharedSubscriptionChannel);
+ final ByteString subid = ByteString.copyFromUtf8("mysub");
+
+ final CountDownLatch wakeupLatch = new CountDownLatch(1);
+ final CountDownLatch closeLatch = new CountDownLatch(1);
+ final CountDownLatch subLatch = new CountDownLatch(1);
+ final CountDownLatch deliverLatch = new CountDownLatch(1);
+
+ try {
+ subscriber.subscribe(topic, subid, CreateOrAttach.CREATE_OR_ATTACH);
+ sleepThread("DeliveryManagerThread", wakeupLatch);
+ subscriber.asyncCloseSubscription(topic, subid, new Callback<Void>() {
+ @Override
+ public void operationFinished(Object ctx, Void resultOfOperation) {
+ closeLatch.countDown();
+ }
+ @Override
+ public void operationFailed(Object ctx, PubSubException exception) {
+ logger.error("Closesub failed : ", exception);
+ }
+ }, null);
+ subscriber.asyncSubscribe(topic, subid, CreateOrAttach.ATTACH, new Callback<Void>() {
+ @Override
+ public void operationFinished(Object ctx, Void resultOfOperation) {
+ try {
+ subscriber.startDelivery(topic, subid, new MessageHandler() {
+ @Override
+ public void deliver(ByteString topic, ByteString subid, Message msg,
+ Callback<Void> callback, Object context) {
+ deliverLatch.countDown();
+ }
+ });
+ } catch (Exception cnse) {
+ logger.error("Failed to start delivery : ", cnse);
+ }
+ subLatch.countDown();
+ }
+ @Override
+ public void operationFailed(Object ctx, PubSubException exception) {
+ logger.error("Failed to subscriber : ", exception);
+ }
+ }, null);
+ // Make the delivery manager thread sleep for a while.
+ // Before {@link https://issues.apache.org/jira/browse/BOOKKEEPER-507},
+ // subscribe would succeed before closesub, while closesub would clear
+ // a successful subscription w/o notifying the client.
+ TimeUnit.SECONDS.sleep(2);
+ // wake up fifo delivery thread
+ wakeupLatch.countDown();
+ // wait close sub to succeed
+ assertTrue("Async close subscription should succeed.",
+ closeLatch.await(5, TimeUnit.SECONDS));
+ assertTrue("Subscribe should succeed.",
+ subLatch.await(5, TimeUnit.SECONDS));
+ // publish a message
+ publisher.publish(topic, Message.newBuilder().setBody(topic).build());
+ // wait for seconds to receive message
+ assertTrue("Message should be received through successful subscription.",
+ deliverLatch.await(5, TimeUnit.SECONDS));
+ } finally {
+ client.close();
+ }
+ }
+
+}
+
Modified: zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/StubDeliveryManager.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/StubDeliveryManager.java?rev=1428277&r1=1428276&r2=1428277&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/StubDeliveryManager.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-server/src/test/java/org/apache/hedwig/server/delivery/StubDeliveryManager.java Thu Jan 3 11:07:56 2013
@@ -58,9 +58,11 @@ public class StubDeliveryManager impleme
SubscriptionPreferences preferences,
MessageSeqId seqIdToStartFrom,
DeliveryEndPoint endPoint,
- ServerMessageFilter filter) {
+ ServerMessageFilter filter,
+ Callback<Void> cb, Object ctx) {
lastRequest.add(new StartServingRequest(topic, subscriberId, preferences,
seqIdToStartFrom, endPoint, filter));
+ cb.operationFinished(ctx, null);
}
@Override