/*
 * 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.uima.aae.error.handler;

import java.net.ConnectException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

import org.apache.uima.UIMAFramework;
import org.apache.uima.aae.UIMAEE_Constants;
import org.apache.uima.aae.InProcessCache.CacheEntry;
import org.apache.uima.aae.controller.AggregateAnalysisEngineController;
import org.apache.uima.aae.controller.AnalysisEngineController;
import org.apache.uima.aae.controller.Controller;
import org.apache.uima.aae.controller.Endpoint;
import org.apache.uima.aae.controller.LocalCache.CasStateEntry;
import org.apache.uima.aae.delegate.Delegate;
import org.apache.uima.aae.error.AsynchAEException;
import org.apache.uima.aae.error.ErrorContext;
import org.apache.uima.aae.error.ErrorHandler;
import org.apache.uima.aae.error.ErrorHandlerBase;
import org.apache.uima.aae.error.ExpiredMessageException;
import org.apache.uima.aae.error.InvalidMessageException;
import org.apache.uima.aae.error.MessageTimeoutException;
import org.apache.uima.aae.error.Threshold;
import org.apache.uima.aae.error.UimaEEServiceException;
import org.apache.uima.aae.jmx.ServiceErrors;
import org.apache.uima.aae.message.AsynchAEMessage;
import org.apache.uima.aae.monitor.Monitor;
import org.apache.uima.aae.spi.transport.UimaMessage;
import org.apache.uima.aae.spi.transport.UimaTransport;
import org.apache.uima.util.Level;

public class ProcessCasErrorHandler extends ErrorHandlerBase implements ErrorHandler {
  private static final Class CLASS_NAME = ProcessCasErrorHandler.class;

  private Map delegateMap = null;

  private Object monitor = new Object();

  public ProcessCasErrorHandler() {
    delegateMap = new HashMap();
  }

  /*
   * Copy map provided by dd2spring but create unique Threshold objects if necessary (UIMA-1623)
   */
  public ProcessCasErrorHandler(Map aDelegateMap) {
    delegateMap = aDelegateMap;
    for (Map.Entry<String, Threshold> entry : (Set<Map.Entry<String, Threshold>>)delegateMap.entrySet()) {
      entry.setValue(entry.getValue().initialize());
    }
  }

  private Endpoint getDestination(AnalysisEngineController aController, ErrorContext anErrorContext) {
    Endpoint endpoint = null;
    String casReferenceId = (String) anErrorContext.get(AsynchAEMessage.CasReference);
    if (aController instanceof AggregateAnalysisEngineController) {
      endpoint = ((AggregateAnalysisEngineController) aController).getMessageOrigin(casReferenceId);

      // Remove the entry from the Message Origin Map since it is no longer needed. The CAS will be
      // dropped as soon as the exception is sent up to the client.
      if (endpoint != null && aController.isTopLevelComponent()) {
        ((AggregateAnalysisEngineController) aController).removeMessageOrigin(casReferenceId);
      }
    } else if (anErrorContext.containsKey(AsynchAEMessage.Endpoint)) {
      endpoint = (Endpoint) anErrorContext.get(AsynchAEMessage.Endpoint);
    }
    return endpoint;
  }

  private boolean isDisabled(AggregateAnalysisEngineController aController, String aDelegateKey) {
    return aController.isDelegateDisabled(aDelegateKey);
  }

  private boolean ignoreError(Throwable t, ErrorContext anErrorContext, boolean isClient) {
    // Ignores Invalid Messages, expired messages and ConnectExceptions IFF a connection
    // to a client cannot be established. Clients can be killed in the middle of a run
    // and that should not be an error.
    if (t instanceof InvalidMessageException || t instanceof ExpiredMessageException
            || (isClient && t.getCause() != null && t.getCause() instanceof ConnectException)) {
      return true;
    }
    return false;
  }

  private void sendExceptionToClient(Throwable t, String aCasReferenceId, Endpoint anEndpoint,
          AnalysisEngineController aController) throws Exception {
    // Notify the parent of the exception
    if (anEndpoint != null && aCasReferenceId != null && !anEndpoint.isCasMultiplier()) {
      try {
        if (!anEndpoint.isRemote()) {
          anEndpoint.setReplyEndpoint(true);
          UimaTransport vmTransport = aController.getTransport(anEndpoint.getEndpoint());
          UimaMessage message = vmTransport.produceMessage(AsynchAEMessage.Process,
                  AsynchAEMessage.Response, aController.getName());
          message.addIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.Exception);
          message.addStringProperty(AsynchAEMessage.CasReference, aCasReferenceId);

          Throwable wrapper = null;
          if (!(t instanceof UimaEEServiceException)) {
            // Strip off AsyncAEException and replace with UimaEEServiceException
            if (t instanceof AsynchAEException && t.getCause() != null) {
              wrapper = new UimaEEServiceException(t.getCause());
            } else {
              wrapper = new UimaEEServiceException(t);
            }
          }
          if (wrapper == null) {
            message.addObjectProperty(AsynchAEMessage.Cargo, t);
          } else {
            message.addObjectProperty(AsynchAEMessage.Cargo, wrapper);
          }
          if (!aController.isStopped()) {
            vmTransport.getUimaMessageDispatcher(anEndpoint.getEndpoint()).dispatch(message);
            aController.dropStats(aCasReferenceId, aController.getName());
          }
        } else {
          CasStateEntry stateEntry = null;
          String parentCasReferenceId = null;
          try {
            stateEntry = aController.getLocalCache().lookupEntry(aCasReferenceId);
            if (stateEntry != null && stateEntry.isSubordinate()) {
              CasStateEntry topParentEntry = aController.getLocalCache().getTopCasAncestor(
                      aCasReferenceId);
              parentCasReferenceId = topParentEntry.getCasReferenceId();
            }
          } catch (Exception e) {
          }

          if (!aController.isStopped()) {
            aController.getOutputChannel().sendReply(t, aCasReferenceId, parentCasReferenceId,
                    anEndpoint, AsynchAEMessage.Process);
          }
        }
      } catch (Exception e) {
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
          if ( aController != null ) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                    "sendExceptionToParent", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                    "UIMAEE_service_exception_WARNING", aController.getComponentName());
          }

          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                  "sendExceptionToParent", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_exception__WARNING", e);
        }
      }
    }

  }

  private boolean isClient(Endpoint anEndpoint, AnalysisEngineController aController,
          String aCasReferenceId) {
    Endpoint clientEndpoint = null;

    if (aController.isTopLevelComponent()) {
      clientEndpoint = aController.getInProcessCache().getEndpoint(null, aCasReferenceId);
    } else if (aController instanceof AggregateAnalysisEngineController) {
      clientEndpoint = ((AggregateAnalysisEngineController) aController)
              .getMessageOrigin(aCasReferenceId);
    }
    if (anEndpoint != null && clientEndpoint != null) {
      return anEndpoint.getEndpoint().equalsIgnoreCase(clientEndpoint.getEndpoint());
    }
    return false;
  }

  public boolean handleError(Throwable t, ErrorContext anErrorContext,
          AnalysisEngineController aController) {
    org.apache.uima.aae.controller.LocalCache.CasStateEntry parentCasStateEntry = null;
    String delegateKey = null;
    if (!isHandlerForError(anErrorContext, AsynchAEMessage.Process)) {
      return false;
    }

    String casReferenceId = null;
    if (anErrorContext.containsKey(AsynchAEMessage.CasReference)) {
      casReferenceId = (String) anErrorContext.get(AsynchAEMessage.CasReference);
    } else {
      return true; // No CAS, nothing to do
    }
    boolean isRequest = false;

    if (anErrorContext.containsKey(AsynchAEMessage.MessageType)) {
      int msgType = ((Integer) anErrorContext.get(AsynchAEMessage.MessageType)).intValue();
      if (msgType == AsynchAEMessage.Request) {
        isRequest = true;
      }
    }

    // Determine if the exception occured while sending a reply to the client
    boolean isEndpointTheClient = isClient((Endpoint) anErrorContext.get(AsynchAEMessage.Endpoint),
            aController, casReferenceId);

    if (ignoreError(t, anErrorContext, isEndpointTheClient)) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, getClass().getName(), "handleError",
                UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_ignore_error__INFO",
                new Object[] { aController.getComponentName(), t.getClass().getName() });
      }
      if (casReferenceId != null) {
          CasStateEntry casStateEntry = aController.getLocalCache().lookupEntry(casReferenceId);
          //	Cleanup if the CAS has no children. If it does, the cleanup will be done when all
          //	child CASes are processed.
          if ( casStateEntry != null && casStateEntry.getSubordinateCasInPlayCount() == 0) {
      	  // Cleanup resources associated with a CAS and then release the CAS
            try {
              if (aController instanceof AggregateAnalysisEngineController) {
                ((AggregateAnalysisEngineController) aController).dropFlow(casReferenceId, true);
                ((AggregateAnalysisEngineController) aController).removeMessageOrigin(casReferenceId);
              }
              aController.dropStats(casReferenceId, aController.getName());
            } catch (Exception e) {
              // Throwing this CAS away, ignore exception
            } finally {
              if (aController.isTopLevelComponent()) {
                aController.dropCAS(casReferenceId, true);
              }
            }
        }
      }

      return true; // handled here. This message will not processed
    }

    // Log the exception unless context says otherwise. Currently failures on
    // send are silent
    if (!anErrorContext.silentHandling()
            && UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
      if ( aController != null ) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                "handleError", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_service_exception_WARNING", aController.getComponentName());
      }

      UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), "handleError",
              UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_exception__WARNING", t);
    }

    String key = "";
    Threshold threshold = null;
    boolean delegateDisabled = false;
    Delegate delegate = null;
    // R E T R Y
    // Do retry first if this an Aggregate Controller
    if (!isEndpointTheClient && aController instanceof AggregateAnalysisEngineController) {
      Endpoint endpoint = null;

      if (anErrorContext.get(AsynchAEMessage.Endpoint) != null) {
        endpoint = (Endpoint) anErrorContext.get(AsynchAEMessage.Endpoint);
        key = ((AggregateAnalysisEngineController) aController).lookUpDelegateKey(endpoint
                .getEndpoint());
        delegate = ((AggregateAnalysisEngineController) aController).lookupDelegate(key);
      }
      threshold = super.getThreshold(endpoint, delegateMap, aController);
      if (endpoint != null) {
        // key =
        // ((AggregateAnalysisEngineController)aController).lookUpDelegateKey(endpoint.getEndpoint());
        delegateDisabled = ((AggregateAnalysisEngineController) aController)
                .isDelegateDisabled(key);
        if (threshold != null && threshold.getMaxRetries() > 0 && !delegateDisabled) {
          // If max retry count is not reached, send the last command again and return true
          if (super.retryLastCommand(AsynchAEMessage.Process, endpoint, aController, key,
                  threshold, anErrorContext)) {
            if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
              UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, getClass().getName(),
                      "handleError", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                      "UIMAEE_retry_cas__FINE",
                      new Object[] { aController.getComponentName(), key, casReferenceId });
            }
            return true; // Command has been retried. Done here.
          }
        } else if (threshold == null) {
          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.CONFIG)) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.CONFIG, getClass().getName(),
                    "handleError", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                    "UIMAEE_no_threshold_for_endpoint__CONFIG",
                    new Object[] { aController.getComponentName(), "Process", key });
          }
        }
        if (delegate != null) {
          // Received reply from the delegate. Remove the CAS from the
          // delegate's list of CASes pending reply
          // Delegate delegate =
          // ((AggregateAnalysisEngineController)aController).lookupDelegate(key);
          delegate.removeCasFromOutstandingList(casReferenceId);
        }
      } else {
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, getClass().getName(),
                  "handleError", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_no_endpoint_provided__INFO",
                  new Object[] { aController.getComponentName() });
        }
      }
    } else {
      if (delegateMap != null && delegateMap.containsKey(key)) {
        threshold = (Threshold) delegateMap.get(key);
      }
    }

    if (key != null && key.trim().length() > 0) {
      // Retries either exceeded or not configured for retry
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, getClass().getName(), "handleError",
                UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_cas_retries_exceeded__FINE",
                new Object[] { aController.getComponentName(), key, casReferenceId });
      }
    }
    boolean disabledDueToExceededThreshold = false;

    // Dont increment errors for destinations that are clients of this service.
    if (key != null && !aController.isStopped() && (isRequest || !isEndpointTheClient)) {
      synchronized (monitor) {
        // Dont increment errors for delegates that have been already disabled
        if (!delegateDisabled) {
          // Process Error Count is only incremented after retries are done.
          super.incrementStatistic(aController.getMonitor(), key, Monitor.ProcessErrorCount);
          super.incrementStatistic(aController.getMonitor(), key, Monitor.TotalProcessErrorCount);
          aController.getServiceErrors().incrementProcessErrors();
          if (aController instanceof AggregateAnalysisEngineController
                  && anErrorContext.get(AsynchAEMessage.Endpoint) != null) {
            Endpoint endpoint = (Endpoint) anErrorContext.get(AsynchAEMessage.Endpoint);
            if (endpoint.isRemote()) {
              ServiceErrors serviceErrs = ((AggregateAnalysisEngineController) aController)
                      .getDelegateServiceErrors(key);
              if (serviceErrs != null) {
                serviceErrs.incrementProcessErrors();
              }
            }
          }

          /***
           * if (threshold != null && threshold.getThreshold() > 0 &&
           * super.exceedsThresholdWithinWindow(aController.getMonitor(), Monitor.ProcessErrorCount,
           * key, threshold) )
           */

          long procCount = aController.getMonitor().getLongNumericStatistic(key,
                  Monitor.ProcessCount).getValue();
          if (threshold != null && threshold.exceededWindow(procCount)) {
            if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
              UIMAFramework.getLogger(CLASS_NAME).logrb(
                      Level.INFO,
                      getClass().getName(),
                      "handleError",
                      UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                      "UIMAEE_process_cas_exceeded_threshold__INFO",
                      new Object[] { aController.getComponentName(), key, casReferenceId,
                          threshold.getThreshold(), threshold.getAction() });
            }
            // Add new property to skip handling of CASes in pending lists. Those CASes
            // will be handled later in this method, once we complete processing of the CAS
            // that caused the exception currently being processed. During handling of the
            // CASes in pending state, this error handler is called for each CAS to force
            // its timeout.
            disabledDueToExceededThreshold = ErrorHandler.DISABLE.equalsIgnoreCase(threshold
                    .getAction());
            if (disabledDueToExceededThreshold) {
              delegateKey = key;
              anErrorContext.add(AsynchAEMessage.SkipPendingLists, "true");
            }
            if (ErrorHandler.TERMINATE.equalsIgnoreCase(threshold.getAction())) {
              anErrorContext.add(ErrorContext.THROWABLE_ERROR, t);
              if (casReferenceId != null) {
                try {
                  CasStateEntry casStateEntry = aController.getLocalCache().lookupEntry(
                          casReferenceId);
                  if (casStateEntry != null && casStateEntry.isSubordinate()) {
                    CasStateEntry parenCasStateEntry = aController.getLocalCache()
                            .getTopCasAncestor(casReferenceId);
                    // Replace Cas Id with the parent Cas Id
                    anErrorContext.remove(AsynchAEMessage.CasReference);
                    anErrorContext.add(AsynchAEMessage.CasReference, parenCasStateEntry
                            .getCasReferenceId());
                  }
                } catch (Exception e) {
                }
              }

            }
            aController.takeAction(threshold.getAction(), key, anErrorContext);
            if (ErrorHandler.CONTINUE.equalsIgnoreCase(threshold.getAction())) {
              //  The following is only true if the Action=Continue and the FlowController is configured
              //  to continue on Exception in a delegate with a given key. The exception is provided
              //  in the anErrorContext. The controller, while handling exception in takeAction(), will set 
              //  ErrorContext.ERROR_HANDLED property iff the FlowController says to continue.
              if (anErrorContext.containsKey(ErrorContext.ERROR_HANDLED) && (Boolean)anErrorContext.get(ErrorContext.ERROR_HANDLED) == true )
                //  The FlowController indicated to continue. The process() method was already called. No
                //  need to do anything else. At this point the error has been handled.
                return true;
            }
          }
        } else {
          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, getClass().getName(),
                    "handleError", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                    "UIMAEE_delegate_already_disabled__INFO",
                    new Object[] { aController.getComponentName(), key, casReferenceId });
          }
        }
      }

    } else {
      Endpoint endpt = (Endpoint) anErrorContext.get(AsynchAEMessage.Endpoint);
      if (endpt != null) {
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(
                  Level.INFO,
                  getClass().getName(),
                  "handleError",
                  UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_process_exception__INFO",
                  new Object[] { aController.getComponentName(), endpt.getEndpoint(),
                      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) {
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
                    "handleError", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                    "UIMAEE_parent_cas_notin_cache__INFO", new Object[] { aController.getComponentName(), casReferenceId });
        }
    }
    // 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)) {
            if ( aController != null ) {
              UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                      "handleError", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                      "UIMAEE_service_exception_WARNING", aController.getComponentName());
            }

            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());
                }
              }
            } catch (Exception ex) {
              // Input CAS doesnt exist. Nothing to update, move on
            }
          }
        } else if (casStateEntry != null) { // input CAS
          casStateEntry.setFailed();
          casStateEntry.addThrowable(t);
          aController.process(null, casStateEntry.getCasReferenceId());
        }
        return true;
      }

      if (threshold != null && flowControllerContinueFlag) {
        // The Exception has been almost fully handled. Check if the delegate was disabled above.
        // If it was, we need to force timeout on all CASes in pending state associated with that
        // delegate.
        if (disabledDueToExceededThreshold && delegateKey != null) {
          aController.forceTimeoutOnPendingCases(delegateKey);
        }
        if (totalNumberOfParallelDelegatesProcessingCas == 1
                || (casStateEntry.howManyDelegatesResponded() == totalNumberOfParallelDelegatesProcessingCas)) {
          aController.process(aController.getInProcessCache().getCasByReference(casReferenceId),
                  casReferenceId);
        }
        // Dont send request to release the CAS to remote CM. This will happen in the final step. We
        // are continuing
        // despite the error here.

        return true;
      } else if (doSendReplyToClient) {
        try {
          // Send exception to the client if the exception happened while processing input CAS
          // Child CASes that cause exceptions will be dropped, their parent CAS will be marked
          // as failed and it will be returned back to the client in the final step once all child
          // CASes are accounted for and dropped.
          if (casStateEntry != null && !casStateEntry.isSubordinate()
                  && deliverExceptionToClient(t)) {
            sendExceptionToClient(t, casReferenceId, endpoint, aController);
          }
        } catch (Exception e) {
          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
            if ( aController != null ) {
              UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                      "handleError", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                      "UIMAEE_service_exception_WARNING", aController.getComponentName());
            }

            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                    "handleError", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                    "UIMAEE_exception__WARNING", e);
          }
        }
      }

      // Now check if the CAS origin is a remote CAS Multiplier. If so, send a request to release
      // the CAS. Remote
      // CAS Multipliers must be "gated" to prevent flooding the Aggregate queue with too many
      // CASes. There is
      // an explicit protocol between an Aggregate AS and its CM. The Aggregate AS sends a request
      // to free a CAS
      // in the CM whenever the Aggregate has capacity to process more CASes. Here we are recovering
      // from an
      // an error but we still need to send a request to free the CAS in the remote CM to prevent a
      // hang in the CM
      try {
        // First check if the current controller is the one that first produced the CAS
        if (cacheEntry != null
                && aController.getName().equalsIgnoreCase(cacheEntry.getCasProducerAggregateName())) {
          // Fetch the key of the Cas Multiplier
          String casProducerKey = cacheEntry.getCasProducerKey();
          if (casProducerKey != null) {
            // Create an endpoint object from the key. This object will be cloned from the Endpoint
            // object
            // defined in the spring configuration file.
            Endpoint cmEndpoint = ((AggregateAnalysisEngineController) aController).lookUpEndpoint(
                    casProducerKey, true);

            // CAS reference id will be different if the CAS originated from a remote Cas
            // Multiplier.
            // String cmCasReferenceId = cacheEntry.getRemoteCMCasReferenceId();
            // If the Cas Multiplier is remote send a request to free a CAS with a given cas id
            if (cmEndpoint != null && cmEndpoint.isCasMultiplier() && cmEndpoint.isRemote()) {
              cmEndpoint.setReplyEndpoint(true);
              cmEndpoint.setIsCasMultiplier(true);
              cmEndpoint.setFreeCasEndpoint(true);
              aController.getOutputChannel().sendRequest(AsynchAEMessage.ReleaseCAS,
                      cacheEntry.getCasReferenceId(), cmEndpoint);
            }
          }
        }
      } catch (Exception e) {
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
          if ( aController != null ) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                    "handleError", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                    "UIMAEE_service_exception_WARNING", aController.getComponentName());
          }

          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                  "handleError", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_exception__WARNING", e);
        }
      }
    } else // Primitive Controller
    {
      try {
        if (deliverExceptionToClient(t)) {
          sendExceptionToClient(t, casReferenceId, endpoint, aController);
        }
      } catch (Exception e) {
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
          if ( aController != null ) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                    "handleError", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                    "UIMAEE_service_exception_WARNING", aController.getComponentName());
          }

          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                  "handleError", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_exception__WARNING", e);
        }
      }
    }
    // The Exception has been almost fully handled. Check if the delegate was disabled above.
    // If it was, we need to force timeout on all CASes in pending state associated with that
    // delegate.
    if (disabledDueToExceededThreshold && delegateKey != null) {
      aController.forceTimeoutOnPendingCases(delegateKey);
    }

    try {
      // Only top level component can Drop the CAS.
      if (aController.isTopLevelComponent()) {
        if (totalNumberOfParallelDelegatesProcessingCas == 1
                || (casStateEntry.howManyDelegatesResponded() == totalNumberOfParallelDelegatesProcessingCas)) {
          aController.takeAction(ErrorHandler.DROPCAS, key, anErrorContext);
        }
      }
      if (casReferenceId != null && aController instanceof AggregateAnalysisEngineController) {
        if (parentCasStateEntry != null && parentCasStateEntry.getSubordinateCasInPlayCount() == 0
                && parentCasStateEntry.isPendingReply()) {
          ((AggregateAnalysisEngineController) aController).finalStep(parentCasStateEntry
                  .getFinalStep(), parentCasStateEntry.getCasReferenceId());
        }
        // Cleanup state information from local caches
        ((AggregateAnalysisEngineController) aController).dropFlow(casReferenceId, true);
        ((AggregateAnalysisEngineController) aController).removeMessageOrigin(casReferenceId);
      }

      aController.dropStats(casReferenceId, aController.getName());
    } catch (Exception e) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
        if ( aController != null ) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                  "handleError", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_service_exception_WARNING", aController.getComponentName());
        }

        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                "handleError", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_exception__WARNING", e);
      }
    }

    return true;
  }

  private boolean deliverExceptionToClient(Throwable t) {
    // Dont send TimeOutExceptions to client
    if (t instanceof MessageTimeoutException || t instanceof UimaEEServiceException
            && t.getCause() != null && t.getCause() instanceof MessageTimeoutException) {
      return false;
    }
    return true;
  }
}