You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by si...@apache.org on 2012/08/28 14:51:32 UTC
svn commit: r1378112 - in /zookeeper/bookkeeper/trunk: ./
hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/
hedwig-client/src/main/java/org/apache/hedwig/client/handlers/
hedwig-client/src/main/java/org/apache/hedwig/client/netty/
Author: sijie
Date: Tue Aug 28 12:51:31 2012
New Revision: 1378112
URL: http://svn.apache.org/viewvc?rev=1378112&view=rev
Log:
BOOKKEEPER-371: NPE in hedwig hub client causes hedwig hub to shut down. (Aniruddha via sijie)
Added:
zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/NoResponseHandlerException.java (with props)
Modified:
zookeeper/bookkeeper/trunk/CHANGES.txt
zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/MessageConsumeCallback.java
zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/SubscribeResponseHandler.java
zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/HedwigClientImpl.java
zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/HedwigPublisher.java
zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/HedwigSubscriber.java
zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/WriteCallback.java
Modified: zookeeper/bookkeeper/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/CHANGES.txt?rev=1378112&r1=1378111&r2=1378112&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/CHANGES.txt (original)
+++ zookeeper/bookkeeper/trunk/CHANGES.txt Tue Aug 28 12:51:31 2012
@@ -70,6 +70,8 @@ Trunk (unreleased changes)
BOOKKEEPER-376: LedgerManagers should consider 'underreplication' node as a special Znode (Uma via sijie)
+ BOOKKEEPER-371: NPE in hedwig hub client causes hedwig hub to shut down. (Aniruddha via sijie)
+
hedwig-client:
BOOKKEEPER-274: Hedwig cpp client library should not link to cppunit which is just used for test. (sijie via ivank)
Added: zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/NoResponseHandlerException.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/NoResponseHandlerException.java?rev=1378112&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/NoResponseHandlerException.java (added)
+++ zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/NoResponseHandlerException.java Tue Aug 28 12:51:31 2012
@@ -0,0 +1,34 @@
+/**
+ * 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.client.exceptions;
+
+/**
+ * This is a Hedwig client side exception thrown when it can't get the response
+ * handler from the channel pipeline responsible for a PubSubRequest.
+ */
+public class NoResponseHandlerException extends Exception {
+ private static final long serialVersionUID = 1L;
+
+ public NoResponseHandlerException(String message) {
+ super(message);
+ }
+
+ public NoResponseHandlerException(String message, Throwable t) {
+ super(message, t);
+ }
+}
Propchange: zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/exceptions/NoResponseHandlerException.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/MessageConsumeCallback.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/MessageConsumeCallback.java?rev=1378112&r1=1378111&r2=1378112&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/MessageConsumeCallback.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/MessageConsumeCallback.java Tue Aug 28 12:51:31 2012
@@ -19,6 +19,8 @@ package org.apache.hedwig.client.handler
import java.util.TimerTask;
+import org.apache.hedwig.client.exceptions.NoResponseHandlerException;
+import org.apache.hedwig.client.netty.ResponseHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.jboss.netty.channel.Channel;
@@ -61,8 +63,21 @@ public class MessageConsumeCallback impl
public void run() {
// Try to consume the message again
Channel topicSubscriberChannel = client.getSubscriber().getChannelForTopic(topicSubscriber);
- HedwigClientImpl.getResponseHandlerFromChannel(topicSubscriberChannel).getSubscribeResponseHandler()
- .asyncMessageConsume(messageConsumeData.msg);
+ ResponseHandler handler = null;
+ try {
+ handler = HedwigClientImpl.getResponseHandlerFromChannel(topicSubscriberChannel);
+ } catch (NoResponseHandlerException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No response handler found while invoking asyncMessageConsumed in the Message" +
+ " consume retry task ." , e);
+ }
+ // Explicitly close the channel
+ if (null != topicSubscriberChannel) {
+ topicSubscriberChannel.close();
+ }
+ return;
+ }
+ handler.getSubscribeResponseHandler().asyncMessageConsume(messageConsumeData.msg);
}
}
@@ -72,8 +87,20 @@ public class MessageConsumeCallback impl
// Message has been successfully consumed by the client app so callback
// to the ResponseHandler indicating that the message is consumed.
Channel topicSubscriberChannel = client.getSubscriber().getChannelForTopic(topicSubscriber);
- HedwigClientImpl.getResponseHandlerFromChannel(topicSubscriberChannel).getSubscribeResponseHandler()
- .messageConsumed(messageConsumeData.msg);
+ ResponseHandler handler = null;
+ try {
+ handler = HedwigClientImpl.getResponseHandlerFromChannel(topicSubscriberChannel);
+ } catch (NoResponseHandlerException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No response handler found while invoking messageConsumed." , e);
+ }
+ // Explicitly close the channel
+ if (null != topicSubscriberChannel) {
+ topicSubscriberChannel.close();
+ }
+ return;
+ }
+ handler.getSubscribeResponseHandler().messageConsumed(messageConsumeData.msg);
}
public void operationFailed(Object ctx, PubSubException exception) {
Modified: zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/SubscribeResponseHandler.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/SubscribeResponseHandler.java?rev=1378112&r1=1378111&r2=1378112&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/SubscribeResponseHandler.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/handlers/SubscribeResponseHandler.java Tue Aug 28 12:51:31 2012
@@ -23,7 +23,7 @@ import java.util.LinkedList;
import java.util.Queue;
import java.util.Set;
-import org.jboss.netty.channel.ChannelFuture;
+import org.apache.hedwig.client.exceptions.NoResponseHandlerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.jboss.netty.channel.Channel;
@@ -90,7 +90,14 @@ public class SubscribeResponseHandler {
// If this was not a successful response to the Subscribe request, we
// won't be using the Netty Channel created so just close it.
if (!response.getStatusCode().equals(StatusCode.SUCCESS)) {
- HedwigClientImpl.getResponseHandlerFromChannel(channel).handleChannelClosedExplicitly();
+ try {
+ HedwigClientImpl.getResponseHandlerFromChannel(channel).handleChannelClosedExplicitly();
+ } catch (NoResponseHandlerException e) {
+ // Log an error. But should we also return and not process anything further?
+ logger.error("No response handler found while trying to close channel explicitly while handling a " +
+ "failed subscription response.", e);
+ // Continue closing the channel because this is an unexpected event and state should be reset.
+ }
channel.close();
}
Modified: zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/HedwigClientImpl.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/HedwigClientImpl.java?rev=1378112&r1=1378111&r2=1378112&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/HedwigClientImpl.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/HedwigClientImpl.java Tue Aug 28 12:51:31 2012
@@ -27,6 +27,7 @@ import java.util.concurrent.ConcurrentMa
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
+import org.apache.hedwig.client.exceptions.NoResponseHandlerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.jboss.netty.bootstrap.ClientBootstrap;
@@ -177,7 +178,13 @@ public class HedwigClientImpl implements
// channels in HedwigPublisher.host2Channel. This stores the
// channels used for Publish and Unsubscribe requests.
for (Channel channel : pub.host2Channel.values()) {
- ResponseHandler responseHandler = getResponseHandlerFromChannel(channel);
+ ResponseHandler responseHandler = null;
+ try {
+ responseHandler = getResponseHandlerFromChannel(channel);
+ } catch (NoResponseHandlerException e) {
+ logger.warn("No response handler found for channel" + channel + " in the retry timeout task.", e);
+ continue;
+ }
for (PubSubData pubSubData : responseHandler.txn2PubSubData.values()) {
checkPubSubDataToTimeOut(pubSubData, responseHandler, curTime, timeoutInterval);
}
@@ -186,7 +193,13 @@ public class HedwigClientImpl implements
// HedwigSubscriber.topicSubscriber2Channel. This stores the
// channels used exclusively for Subscribe requests.
for (Channel channel : sub.topicSubscriber2Channel.values()) {
- ResponseHandler responseHandler = getResponseHandlerFromChannel(channel);
+ ResponseHandler responseHandler = null;
+ try {
+ responseHandler = getResponseHandlerFromChannel(channel);
+ } catch (NoResponseHandlerException e) {
+ logger.warn("No response handler found for channel" + channel + " in the retry timeout task.", e);
+ continue;
+ }
for (PubSubData pubSubData : responseHandler.txn2PubSubData.values()) {
checkPubSubDataToTimeOut(pubSubData, responseHandler, curTime, timeoutInterval);
}
@@ -320,9 +333,17 @@ public class HedwigClientImpl implements
* @param channel
* Channel we are retrieving the ResponseHandler instance for
* @return ResponseHandler Instance tied to the Channel's Pipeline
+ * @throws NoResponseHandlerException if the response handler found for the channel is null.
*/
- public static ResponseHandler getResponseHandlerFromChannel(Channel channel) {
- return (ResponseHandler) channel.getPipeline().getLast();
+ public static ResponseHandler getResponseHandlerFromChannel(Channel channel) throws NoResponseHandlerException {
+ if (null == channel) {
+ throw new NoResponseHandlerException("Received a null value for the channel. Cannot retrieve the response handler");
+ }
+ ResponseHandler handler = (ResponseHandler) channel.getPipeline().getLast();
+ if (null == handler) {
+ throw new NoResponseHandlerException("Could not retrieve the response handler from the channel's pipeline.");
+ }
+ return handler;
}
// Public getter for entries in the topic2Host Map.
Modified: zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/HedwigPublisher.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/HedwigPublisher.java?rev=1378112&r1=1378111&r2=1378112&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/HedwigPublisher.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/HedwigPublisher.java Tue Aug 28 12:51:31 2012
@@ -21,6 +21,7 @@ import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import org.apache.hedwig.client.exceptions.NoResponseHandlerException;
import org.apache.hedwig.protocol.PubSubProtocol;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -202,7 +203,15 @@ public class HedwigPublisher implements
// Before we do the write, store this information into the
// ResponseHandler so when the server responds, we know what
// appropriate Callback Data to invoke for the given txn ID.
- HedwigClientImpl.getResponseHandlerFromChannel(channel).txn2PubSubData.put(txnId, pubSubData);
+ try {
+ HedwigClientImpl.getResponseHandlerFromChannel(channel).txn2PubSubData.put(txnId, pubSubData);
+ } catch (NoResponseHandlerException e) {
+ logger.error("No response handler found while storing the publish callback.");
+ // Callback on pubsubdata indicating failure.
+ pubSubData.getCallback().operationFailed(pubSubData.context, new CouldNotConnectException("No response " +
+ "handler found while attempting to publish. So could not connect."));
+ return;
+ }
// Finally, write the Publish request through the Channel.
if (logger.isDebugEnabled())
@@ -239,7 +248,11 @@ public class HedwigPublisher implements
// topic. Close these redundant channels as they won't be used.
if (logger.isDebugEnabled())
logger.debug("Channel mapping to host: " + host + " already exists so no need to store it.");
- HedwigClientImpl.getResponseHandlerFromChannel(channel).handleChannelClosedExplicitly();
+ try {
+ HedwigClientImpl.getResponseHandlerFromChannel(channel).handleChannelClosedExplicitly();
+ } catch (NoResponseHandlerException e) {
+ logger.error("Could not get response handler while closing channel.");
+ }
channel.close();
}
}
@@ -256,7 +269,11 @@ public class HedwigPublisher implements
closed = true;
}
for (Channel channel : host2Channel.values()) {
- client.getResponseHandlerFromChannel(channel).handleChannelClosedExplicitly();
+ try {
+ client.getResponseHandlerFromChannel(channel).handleChannelClosedExplicitly();
+ } catch (NoResponseHandlerException e) {
+ logger.error("No response handler while trying explicitly close Publisher channel " + channel);
+ }
channel.close().awaitUninterruptibly();
}
host2Channel.clear();
Modified: zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/HedwigSubscriber.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/HedwigSubscriber.java?rev=1378112&r1=1378111&r2=1378112&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/HedwigSubscriber.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/HedwigSubscriber.java Tue Aug 28 12:51:31 2012
@@ -22,6 +22,7 @@ import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import org.apache.hedwig.client.exceptions.NoResponseHandlerException;
import org.apache.hedwig.protocol.PubSubProtocol;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -446,7 +447,15 @@ public class HedwigSubscriber implements
// Before we do the write, store this information into the
// ResponseHandler so when the server responds, we know what
// appropriate Callback Data to invoke for the given txn ID.
- HedwigClientImpl.getResponseHandlerFromChannel(channel).txn2PubSubData.put(txnId, pubSubData);
+ try {
+ HedwigClientImpl.getResponseHandlerFromChannel(channel).txn2PubSubData.put(txnId, pubSubData);
+ } catch (Exception e) {
+ logger.error("No response handler found while storing the subscribe callback.");
+ // Call operationFailed on the pubsubdata callback to indicate failure
+ pubSubData.getCallback().operationFailed(pubSubData.context, new CouldNotConnectException("No response " +
+ "handler found while attempting to subscribe."));
+ return;
+ }
// Finally, write the Subscribe request through the Channel.
if (logger.isDebugEnabled())
@@ -579,8 +588,30 @@ public class HedwigSubscriber implements
}
}
}
- HedwigClientImpl.getResponseHandlerFromChannel(topicSubscriberChannel).getSubscribeResponseHandler()
- .setMessageHandler(messageHandler);
+ try {
+ HedwigClientImpl.getResponseHandlerFromChannel(topicSubscriberChannel).getSubscribeResponseHandler()
+ .setMessageHandler(messageHandler);
+ } catch (NoResponseHandlerException e) {
+ // We did not find a response handler. So remove this subscription handler and throw an exception.
+ topicSubscriber2MessageHandler.remove(topicSubscriber);
+ asyncCloseSubscription(topic, subscriberId, new Callback<Void>() {
+ @Override
+ public void operationFinished(Object ctx, Void resultOfOperation) {
+ logger.warn("Closed subscription for topic " + topic.toStringUtf8() + " and subscriber " +
+ subscriberId.toStringUtf8());
+ }
+
+ @Override
+ public void operationFailed(Object ctx, PubSubException exception) {
+ logger.warn("Error while closing subscription for topic " + topic.toStringUtf8() + " and subscriber " +
+ subscriberId.toStringUtf8());
+ }
+ }, null);
+
+ // We should tell the client to resubscribe.
+ throw new ClientNotSubscribedException("Closed subscription for topic " + topic.toStringUtf8() + " and" +
+ "subscriber Id " + subscriberId.toStringUtf8());
+ }
// Now make the TopicSubscriber Channel readable (it is set to not be
// readable when the initial subscription is done). Note that this is an
// asynchronous call. If this fails (not likely), the futureListener
@@ -616,8 +647,13 @@ public class HedwigSubscriber implements
// Unregister the MessageHandler for the subscribe Channel's
// Response Handler.
Channel topicSubscriberChannel = topicSubscriber2Channel.get(topicSubscriber);
- HedwigClientImpl.getResponseHandlerFromChannel(topicSubscriberChannel).getSubscribeResponseHandler()
- .setMessageHandler(null);
+ try {
+ HedwigClientImpl.getResponseHandlerFromChannel(topicSubscriberChannel).getSubscribeResponseHandler()
+ .setMessageHandler(null);
+ } catch (NoResponseHandlerException e) {
+ // Here it's okay if we can't set the response handler's message handler to null. We should just remove it.
+ logger.warn("Could not set message handler to null for subscription channel " + topicSubscriberChannel + ", ignoring.");
+ }
this.topicSubscriber2MessageHandler.remove(topicSubscriber);
// Now make the TopicSubscriber channel not-readable. This will buffer
// up messages if any are sent from the server. Note that this is an
@@ -671,7 +707,13 @@ public class HedwigSubscriber implements
Channel channel = topicSubscriber2Channel.get(topicSubscriber);
topicSubscriber2Channel.remove(topicSubscriber);
// Close the subscribe channel asynchronously.
- HedwigClientImpl.getResponseHandlerFromChannel(channel).handleChannelClosedExplicitly();
+ try {
+ HedwigClientImpl.getResponseHandlerFromChannel(channel).handleChannelClosedExplicitly();
+ } catch (NoResponseHandlerException e) {
+ // Don't close the channel if you can't find the handler.
+ logger.warn("No response handler found, so could not close subscription channel " + channel);
+ }
+ // We still close the channel as this is an unexpected event and should be handled as one.
ChannelFuture future = channel.close();
future.addListener(new ChannelFutureListener() {
@Override
@@ -725,7 +767,11 @@ public class HedwigSubscriber implements
// Close all of the open Channels.
for (Channel channel : topicSubscriber2Channel.values()) {
- client.getResponseHandlerFromChannel(channel).handleChannelClosedExplicitly();
+ try {
+ client.getResponseHandlerFromChannel(channel).handleChannelClosedExplicitly();
+ } catch (NoResponseHandlerException e) {
+ logger.error("No response handler found while trying to close subscription channel.");
+ }
channel.close().awaitUninterruptibly();
}
topicSubscriber2Channel.clear();
Modified: zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/WriteCallback.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/WriteCallback.java?rev=1378112&r1=1378111&r2=1378112&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/WriteCallback.java (original)
+++ zookeeper/bookkeeper/trunk/hedwig-client/src/main/java/org/apache/hedwig/client/netty/WriteCallback.java Tue Aug 28 12:51:31 2012
@@ -20,6 +20,7 @@ package org.apache.hedwig.client.netty;
import java.net.InetSocketAddress;
import java.util.LinkedList;
+import org.apache.hedwig.client.exceptions.NoResponseHandlerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.jboss.netty.channel.ChannelFuture;
@@ -65,7 +66,12 @@ public class WriteCallback implements Ch
// the saved txnId to PubSubData in the ResponseHandler. These
// requests will not receive an ack response from the server
// so there is no point storing that information there anymore.
- HedwigClientImpl.getResponseHandlerFromChannel(future.getChannel()).txn2PubSubData.remove(pubSubData.txnId);
+ try {
+ HedwigClientImpl.getResponseHandlerFromChannel(future.getChannel()).txn2PubSubData.remove(pubSubData.txnId);
+ } catch (NoResponseHandlerException e) {
+ // We just couldn't remove the transaction ID's mapping. The handler was null, so this has been reset anyway.
+ logger.warn("Could not find response handler to remove txnId mapping to pubsub data. Ignoring.");
+ }
future.getChannel().close();