You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@uima.apache.org by cw...@apache.org on 2013/11/01 21:28:44 UTC
svn commit: r1538051 - in /uima/uima-as/trunk:
uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/
uimaj-as-jms/src/main/java/org/apache/uima/adapter/jms/client/
Author: cwiklik
Date: Fri Nov 1 20:28:43 2013
New Revision: 1538051
URL: http://svn.apache.org/r1538051
Log:
UIMA-3393 Improved detection of NPE and provided better handling
Modified:
uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/ActiveMQMessageSender.java
uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngine_impl.java
uima/uima-as/trunk/uimaj-as-jms/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngineCommon_impl.java
Modified: uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/ActiveMQMessageSender.java
URL: http://svn.apache.org/viewvc/uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/ActiveMQMessageSender.java?rev=1538051&r1=1538050&r2=1538051&view=diff
==============================================================================
--- uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/ActiveMQMessageSender.java (original)
+++ uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/ActiveMQMessageSender.java Fri Nov 1 20:28:43 2013
@@ -36,6 +36,7 @@ import org.apache.activemq.command.Activ
import org.apache.uima.UIMAFramework;
import org.apache.uima.aae.UIMAEE_Constants;
import org.apache.uima.adapter.jms.JmsConstants;
+import org.apache.uima.adapter.jms.client.BaseUIMAAsynchronousEngineCommon_impl.SharedConnection;
import org.apache.uima.util.Level;
/**
@@ -157,9 +158,12 @@ public class ActiveMQMessageSender exten
public MessageProducer getMessageProducer() {
if ( engine.running && engine.producerInitialized == false ) {
try {
- setConnection(engine.lookupConnection(getBrokerURL()).getConnection());
- initializeProducer();
- engine.producerInitialized = true;
+ SharedConnection con = engine.lookupConnection(getBrokerURL());
+ if ( con != null ) {
+ setConnection(con.getConnection());
+ initializeProducer();
+ engine.producerInitialized = true;
+ }
} catch( Exception e) {
e.printStackTrace();
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
Modified: uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngine_impl.java
URL: http://svn.apache.org/viewvc/uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngine_impl.java?rev=1538051&r1=1538050&r2=1538051&view=diff
==============================================================================
--- uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngine_impl.java (original)
+++ uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngine_impl.java Fri Nov 1 20:28:43 2013
@@ -224,7 +224,7 @@ public class BaseUIMAAsynchronousEngine_
}
private void stopConnection() {
SharedConnection sharedConnection;
- if ((sharedConnection = lookupConnection(brokerURI)) != null) {
+ if ( brokerURI != null && (sharedConnection = lookupConnection(brokerURI)) != null) {
// Remove a client from registry
sharedConnection.unregisterClient(this);
ActiveMQConnection amqc = (ActiveMQConnection)sharedConnection.getConnection();
@@ -422,7 +422,7 @@ public class BaseUIMAAsynchronousEngine_
((ActiveMQConnection) aConnection).setPrefetchPolicy(prefetchPolicy);
}
- private SharedConnection validateConnection(String aBrokerURI) throws Exception {
+ protected SharedConnection validateConnection(String aBrokerURI) throws Exception {
// checks if a sharedConnection exists and if not creates a new one
return createSharedConnection(aBrokerURI);
}
Modified: uima/uima-as/trunk/uimaj-as-jms/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngineCommon_impl.java
URL: http://svn.apache.org/viewvc/uima/uima-as/trunk/uimaj-as-jms/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngineCommon_impl.java?rev=1538051&r1=1538050&r2=1538051&view=diff
==============================================================================
--- uima/uima-as/trunk/uimaj-as-jms/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngineCommon_impl.java (original)
+++ uima/uima-as/trunk/uimaj-as-jms/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngineCommon_impl.java Fri Nov 1 20:28:43 2013
@@ -26,6 +26,7 @@ import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.BlockingQueue;
@@ -61,7 +62,6 @@ import org.apache.uima.aae.client.UimaAS
import org.apache.uima.aae.client.UimaAsBaseCallbackListener;
import org.apache.uima.aae.client.UimaAsynchronousEngine;
import org.apache.uima.aae.controller.AggregateAnalysisEngineController_impl;
-import org.apache.uima.aae.controller.Endpoint;
import org.apache.uima.aae.delegate.Delegate;
import org.apache.uima.aae.delegate.Delegate.DelegateEntry;
import org.apache.uima.aae.error.AsynchAEException;
@@ -84,18 +84,17 @@ import org.apache.uima.cas.CAS;
import org.apache.uima.cas.SerialFormat;
import org.apache.uima.cas.impl.AllowPreexistingFS;
import org.apache.uima.cas.impl.BinaryCasSerDes6;
+import org.apache.uima.cas.impl.BinaryCasSerDes6.ReuseInfo;
import org.apache.uima.cas.impl.CASImpl;
import org.apache.uima.cas.impl.Serialization;
import org.apache.uima.cas.impl.TypeSystemImpl;
import org.apache.uima.cas.impl.XmiSerializationSharedData;
-import org.apache.uima.cas.impl.BinaryCasSerDes6.ReuseInfo;
import org.apache.uima.collection.CollectionReader;
import org.apache.uima.collection.EntityProcessStatus;
import org.apache.uima.jms.error.handler.BrokerConnectionException;
import org.apache.uima.resource.ResourceInitializationException;
import org.apache.uima.resource.ResourceProcessException;
import org.apache.uima.resource.metadata.ProcessingResourceMetaData;
-import org.apache.uima.resourceSpecifier.factory.SerializationStrategy;
import org.apache.uima.util.Level;
import org.apache.uima.util.ProcessTrace;
import org.apache.uima.util.XMLInputSource;
@@ -255,6 +254,8 @@ public abstract class BaseUIMAAsynchrono
abstract protected void initializeConsumer(String aBrokerURI, Connection connection) throws Exception;
+ abstract protected SharedConnection validateConnection(String aBrokerURI) throws Exception;
+
// enables/disable timer per CAS. Defaul is to use single timer for
// all outstanding CASes
protected volatile boolean timerPerCAS=false;
@@ -908,7 +909,7 @@ public abstract class BaseUIMAAsynchrono
SharedConnection sharedConnection = lookupConnection(getBrokerURI());
// Send Ping to service as getMeta request
- if ( !serviceDelegate.isAwaitingPingReply() && sharedConnection.isOpen() ) {
+ if ( sharedConnection != null && !serviceDelegate.isAwaitingPingReply() && sharedConnection.isOpen() ) {
serviceDelegate.setAwaitingPingReply();
// Add the cas to a list of CASes pending reply. Also start the timer if necessary
// serviceDelegate.addCasToOutstandingList(requestToCache.getCasReferenceId());
@@ -953,7 +954,7 @@ public abstract class BaseUIMAAsynchrono
}
}
SharedConnection sharedConnection = lookupConnection(getBrokerURI());
- if ( !sharedConnection.isOpen() ) {
+ if ( sharedConnection != null && !sharedConnection.isOpen() ) {
if (requestToCache != null && !requestToCache.isSynchronousInvocation() && aCAS != null ) {
aCAS.release();
}
@@ -2847,7 +2848,7 @@ public abstract class BaseUIMAAsynchrono
SharedConnection sharedConnection;
if ( !connectionOpen() ) {
sharedConnection = lookupConnection(getBrokerURI());
- while ( running ) {
+ while ( sharedConnection != null && running ) {
// blocks until connection is refreshed
try {
sharedConnection.retryConnectionUntilSuccessfull();
@@ -2888,12 +2889,67 @@ public abstract class BaseUIMAAsynchrono
protected SharedConnection lookupConnection(String brokerUrl) {
+ SharedConnection sharedConnection = null;
+ try {
+ if ( sharedConnections.containsKey(brokerUrl) ) {
+ return sharedConnections.get(brokerUrl);
+ } else {
+ boolean first = true;
+ while( running ) {
+
+ try {
+ producerInitialized = false;
+ sharedConnection = validateConnection(brokerUrl);
+ sharedConnection.retryConnectionUntilSuccessfull();
+ break;
+ } catch( Exception e) {
+ Object monitor = new Object();
+ try {
+ synchronized(monitor) {
+ if ( first ) {
+ first = false;
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), "onException",
+ JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_retrying_jms_connection__WARNING",
+ new Object[] { brokerUrl });
+ }
+ }
+ monitor.wait(5000); // retry connection every 5 secs
+ }
+ } catch( Exception ex) {
+ }
+ }
+ } // while
+ }
+ } catch( Exception exx ) {
+
+ if ( brokerUrl == null ) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), "onException",
+ JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_invalid_broker_url__WARNING",
+ new Object[] { });
+ Thread.dumpStack();
+ }
+ StringBuffer sb = new StringBuffer();
+ for( Entry<String, SharedConnection> conn : sharedConnections.entrySet()) {
+ sb.append("-- Shared Connection broker:"+conn.getKey()+" State:"+conn.getValue().getState()+"\n");
+ }
+ if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), "onException",
+ JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_shared_connections__INFO",
+ new Object[] { sb.toString()});
+ }
+ throw new RuntimeException("Invalid State");
+ }
+
+ return sharedConnection;
+ /*
if ( brokerUrl != null ) {
if ( sharedConnections.containsKey(brokerUrl) ) {
return sharedConnections.get(brokerUrl);
}
}
return null;
+ */
}
// This class is used to share JMS Connection by all instances of UIMA AS