casReferenceId });
}
}
}
int totalNumberOfParallelDelegatesProcessingCas = 1; // default
CacheEntry cacheEntry = null;
CasStateEntry casStateEntry = null;
try {
casStateEntry = aController.getLocalCache().lookupEntry(casReferenceId);
cacheEntry = aController.getInProcessCache().getCacheEntryForCAS(casReferenceId);
if (cacheEntry != null) {
totalNumberOfParallelDelegatesProcessingCas = casStateEntry.getNumberOfParallelDelegates();
}
} catch (Exception e) {
System.out.println("Controller:" + aController.getComponentName() + " CAS:" + casReferenceId
+ " Not Found In Cache");
}
// Determine where to send the message
Endpoint endpoint = getDestination(aController, anErrorContext);
// If the error happened during a parallel step, treat the exception as response from the
// delegate
// When all responses from delegates are accounted for we allow the CAS to move on to the next
// step in the flow. Dont increment parallel delegate response count if a delegate was just
// disabled above. The count has been already incremented in handleAction() method of the
// AnalysisEngineController.
if ( casStateEntry != null
&& totalNumberOfParallelDelegatesProcessingCas > 1
&& (casStateEntry.howManyDelegatesResponded() < totalNumberOfParallelDelegatesProcessingCas)) {
casStateEntry.incrementHowManyDelegatesResponded();
}
if (aController instanceof AggregateAnalysisEngineController && t instanceof Exception) {
boolean flowControllerContinueFlag = false;
// if the deployment descriptor says no retries, dont care what the Flow Controller says
if (threshold != null && threshold.getContinueOnRetryFailure()) {
try {
// Consult Flow Controller to determine if it is ok to continue despite the error
flowControllerContinueFlag = ((AggregateAnalysisEngineController) aController)
.continueOnError(casReferenceId, key, (Exception) t);
} catch (Exception exc) {
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
"handleError", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
"UIMAEE_exception__WARNING", exc);
}
}
}
// By default return exception to the client. The exception will not be returned if the CAS is
// a subordinate and the flow controller is *not* configured to continue with bad CAS. In such
// case, the code below will mark the parent CAS as failed. When all child CASes of the parent
// CAS are accounted for, it will be returned to the client with an exception.
boolean doSendReplyToClient = true;
// Check if the caller has already decremented number of subordinates. This property is only
// set in the Aggregate's finalStep() method before the CAS is sent back to the client. If
// there was a problem sending the CAS to the client, we dont want to update the counter
// again. If an exception is reported elsewhere ( not in finalStep()), the default action is
// to decrement the number of subordinates associated with the parent CAS.
if (!flowControllerContinueFlag
&& !anErrorContext.containsKey(AsynchAEMessage.SkipSubordinateCountUpdate)) {
doSendReplyToClient = false;
// Check if the CAS is a subordinate (has parent CAS).
if (casStateEntry != null && casStateEntry.isSubordinate()) {
String parentCasReferenceId = casStateEntry.getInputCasReferenceId();
if (parentCasReferenceId != null) {
try {
CacheEntry parentCasCacheEntry = aController.getInProcessCache().getCacheEntryForCAS(
parentCasReferenceId);
parentCasStateEntry = aController.getLocalCache().lookupEntry(parentCasReferenceId);
String cmEndpointName = cacheEntry.getCasProducerKey();
String cmKey = ((AggregateAnalysisEngineController) aController)
.lookUpDelegateKey(cmEndpointName);
if (cmKey == null) {
cmKey = cacheEntry.getCasProducerKey();
}
Delegate delegateCM = ((AggregateAnalysisEngineController) aController)
.lookupDelegate(cmKey);
// The aggregate will return the input CAS when all child CASes are accounted for
synchronized (parentCasStateEntry) {
if (!parentCasStateEntry.isFailed()) {
CasStateEntry predecessorCas = parentCasStateEntry;
// Processing a failure of the child. Mark the parent CAS
// as failed. All child CASes will be dropped upon return
// from delegates. When all child CASes are dropped the
// aggregate will return an exception to the client containing
// the parent CAS id.
parentCasStateEntry.setFailed();
while (predecessorCas != null && predecessorCas.isSubordinate()) {
predecessorCas = aController.getLocalCache().lookupEntry(
predecessorCas.getInputCasReferenceId());
predecessorCas.setFailed();
}
predecessorCas.addThrowable(t);
// Stop Cas Multiplier
((AggregateAnalysisEngineController) aController).stopCasMultiplier(delegateCM,
parentCasCacheEntry.getCasReferenceId());
}
// Add the exception to the list of exceptions maintained by the parent CAS
parentCasStateEntry.addThrowable(t);
casStateEntry.setReplyReceived();
// Mark this CAS as failed
casStateEntry.setFailed();
if (parentCasStateEntry.getSubordinateCasInPlayCount() == 0
&& parentCasStateEntry.isPendingReply()) {
aController.process(parentCasCacheEntry.getCas(), parentCasCacheEntry
.getCasReferenceId());
} else {
aController.process(null, casStateEntry.getCasReferenceId());
}
}