UIMA Asynchronous Scaleout

Terms used in describing AS capabilities include:

It is useful to compare and contrast the approaches and capabilities of AS and CPM.

The design goals for AS are:

When UIMA is used, it is called using Application APIs. A typical top-level driver has this basic flow:

A "run", in turn, consists of 3 parts:

Initialize is called by the framework when the instance is created. The other methods need to be called by the driver. collectionProcessComplete should be called when the driver determines that it is finished sending input CASes for processing using the process() method. reinitialize() can be called if needed, after changing parameter settings, to get the co-located components to reinitialize.

JMX (Java Management Extensions) are used for monitoring and controlling an AS application. As of release 2.3.0, extensive monitoring facilities have been implemented; these are described in a separate chapter on Chapter 5, Monitoring and Tuning. The only controlling facility provided is to stop a service.

In addition, a configurable Monitoring program is provided which works with the JMX provided measurements and aggregates and samples these over specified intervals, and creates monitoring entries in the UIMA log, for tuning purposes. You can use this to detect overloaded and/or idle services; see the Chapter 5, Monitoring and Tuning chapter for details.

To call a UIMA AS Service from Document Analyzer or any other base UIMA application, use a descriptor such as the following:

<customResourceSpecifier xmlns="http://uima.apache.org/resourceSpecifier">
  <resourceClassName>
    org.apache.uima.aae.jms_adapter.JmsAnalysisEngineServiceAdapter
  </resourceClassName>
  <parameters>
    <parameter name="brokerURL" 
               value="tcp://uima17.watson.ibm.com:61616"/>
    <parameter name="endpoint" 
               value="uima.as.RoomDateMeetingDetectorAggregateQueue"/>
    <parameter name="timeout" 
               value="xxx"/>
    <parameter name="getmetatimeout" 
               value="yyy"/>               
  </parameters>
</customResourceSpecifier>

The resourceClassName must be set exactly as shown. Set the brokerURL and endpoint parameters to the appropriate values for the UIMA AS Service you want to call. These are the same settings you would use in a deployment descriptor to specify the location of a remote delegate. Note that this is a synchronous adapter, which processes one CAS at a time, so it will not take advantage of the scalability that UIMA AS provides. To process more than one CAS at a time, you must use the Asynchronous UIMA AS Client API Chapter 4, Asynchronous Scaleout Application Interface.

For more information on the customResourceSpecifier see Section 2.8, “Custom Resource Specifiers” in UIMA References.

Running UIMA AS applications involves deploying (starting) UIMA AS services, perhaps over a wide area network, perhaps on many machines. UIMA AS as a few preliminary tools to help. These include the ability of the Chapter 4, Asynchronous Scaleout Application Interface to deploy UIMA AS services (limited to deployment within the same JVM), and scripts such as deployAsyncService that start up a UIMA AS Service.

deployAsyncService has a facility that launches a keyboard listener after starting, which listens for a "s" or "q" keystroke. The "s" stops the service immediately, and the "q" quiesces the service, letting any in-process work finish before stopping.

JMX beans for services include a control option to stop the service.

This chapter describes error configuration for AS components.

The AS framework manages a collection of component parts written by users (user code) which can throw exceptions. In addition, the AS framework can run timers when sending commands to user code which can create timeouts.

An AS component is either an AS aggregate or an AS primitive. AS aggregates can have multiple levels of aggregation; error configuration is done for each level of aggregation. The rest of this chapter focuses on the error configuration one level at a time (either for one particular level in an aggregate hierarchy, or for an AS primitive).

There is a small number of commands which can be sent to an AS component. When a component returns the result, if an error occurs, an error result is returned instead of the normal result.

Configuration and support is provided for three classes of errors:

The second and third class of errors is only possible for AS aggregates.

When errors happen, the framework provides a standard set of configurable actions. See Section 2.8, “Commands and allowed actions” for a summary table of the actions available in different situations.

Components managed by AS may generate errors when they are sent a command. The error is associated with the command that was running to produce the error.

There are three incoming message commands supported by the AS framework:

Error handling actions are associated with these various commands. Some error handling actions make sense only if there is an associated CAS object, and are therefore only allowed with the processCas command.

When an error happens, it is either "recovered", or not; only errors from delegates of an AS aggregate can be recovered. Recovery may be achieved by retrying the request or by skipping the delegate.

Commands normally return results; however if an non-recoverable error occurs, the command returns an error result instead.

For AS aggregates, each level in aggregate hierarchy can be configured to try to recover the error. If a particular AS aggregate level does not recover, the error is sent up to the next level of the hierarchy (or to the calling client, if a top level). The error result is updated to reflect the actions the framework takes for this error.

Non-recovered errors can optionally have an associated "Terminate" or "Disable" action (see below), triggered by the error when a threshold is reached. "Disable" applies to the delegate that generated the error while "Terminate" applies to the aggregate and any co-located aggregates it is contained within.

Figure 2.1. Basic error handling chain for AS Aggregates for errors from delegates

The basic error handling chain starts with an error, and can attempt to recover using retry. If this fails (or is not configured), the error count is incremented and checked against the thresholds for this delegate. If the threshold has been reached the specified action is taken, disabling the delegate or terminating the entire component. If the Terminate error is not taken, recovery by the Continue action can be attempted. If that fails, an error result is returned to the caller.

For AS primitives, only the Terminate action is available, and Retry and Continue do not apply.

Figure 2.2. Basic error handling chain for AS Primitives

Error results are returned instead of a CAS, if an error occurs and is not recovered.

If the application uses the synchronous sendAndReceive() method, an Error Result is passed back to the client API in the form of a Java exception. The particular exception varies, depending on many factors, but will include a complete stack trace beginning with the cause of the error. If the application uses an asynchronous API, the exception is wrapped in a EntityProcessStatus object and delivered via a callback listener registered by the application. See section 4.4 Status Callback Listener for details.

When errors occur in delegates, the aggregate containing them can attempt to recover. There are two basic recovery actions: retrying the same command and continuing past (skipping) the failing component.

Every command sent to a delegate can have an associated (configurable) timeout. If the timeout occurs before the delegate responds, the framework creates an error representing the timeout.

When errors occur in delegates, retry is attempted (if configured), some number of times. If that fails, error counts are incremented and thresholds examined for Terminate/Disable actions. If not reached, or if the action is Disable, Continue is attempted (if configured); if Continue fails, the error is not recovered, and the aggregate returns the error result from the delegate to the aggregate's caller. On Terminate, the error is returned to the caller.

2.5.1. Aggregate Error Actions

This section describes in more detail the error actions valid only for AS aggregates.

2.5.1.1. Retry

Retry is an action which re-sends the same command that failed back to the input queue of the delegate. (Note: It may be picked up by a different instance of that delegate, which may have a better chance of succeeding.) The framework will keep a copy of the CAS sent to remote components in order to have it to send again if a retry is required.

Retry is not allowed for colocated delegates.

The "collectionProcessComplete" command is never retried.

Retry is done some number of times, as specified in the deployment descriptor.

2.5.1.2. Disable Action

Figure 2.3. Disable action

When this action is taken the framework marks the particular delegate causing the error as "disabled" so it will be skipped in subsequent calls. The framework calls the flow controller, telling it to remove the particular delegate from the flow.

2.5.1.3. Continue Option on Delegate Process CAS Failures

For processCas commands, the Continue action causes the framework to give the flow controller object for that CAS the error details, and ask the flow controller if processing can continue. If the flow controller returns "true", the flow controller will be called asking for the next step; if "false", the framework stops processing the CAS, returning an error to the client reply queue, or just returning the CAS to the casPool of the CAS multiplier which created it.

For "collectionProcessComplete" commands, Continue means to ignore the error, and continue as if the collectionProcessComplete call had returned normally.

This action is not allowed for the getMetadata command.

The Terminate and Disable actions are conditioned by testing against a threshold.

Thresholds are specified as two numbers - an error count and a window. The threshold is reached if the number of errors occurring within the window size is equal to or greater than "the error count". A value of 0 disables the error threshold so no action can be taken. A window of 0 means no window, i.e. all errors are counted

Errors associated with the processCas command are the only ones that are counted in the threshold measurements.

When this action is taken the service represented by this component sends an error reply and then terminates, disconnecting itself as a listener from its input queue, and cleaning itself up (releasing resources, etc.). During cleanup, the component analysis engine's destroy method is called.

Figure 2.4. Terminate action

If the threshold is not exceeded, the error counts associated with the threshold are incremented.

All of the allowed actions are optional, and default to not being done, except for getMetadata being sent to a delegate that is remote - this has a default timeout of 1 minute.

Here's a table of the allowed actions, by command. In this table, the Retry, Continue, and Disable actions apply to the particular Delegate associated with the error; the Terminate action applies to the entire component.

The framework returns an Error Result to the caller for errors that are not recovered.

The rationale for providing the terminate action for primitive services is that only the service can know that it is no longer capable of continued operation. Consider a scaled component with multiple service instances, where one of them goes "bad" and starts throwing exceptions: the clients of this service have no way of stopping new requests from being delivered to this bad service instance. The teminate action allows the bad service to remove itself from further processing; this could allow life cycle management to restart a new instance.

Each deployment descriptor describes one service, associated with a single UIMA descriptor (aggregate or primitive), and describes the deployment of those UIMA components that are co-located, together with specifications of connections to those subcomponents that are remote.

The deployment descriptor is used to augment information contained in an analysis engine descriptor. It adds information concerning

The application can include both Java and non-Java components; the deployment descriptors are slightly different for non-Java components.

Each deployment descriptor describes components associated with one UIMA descriptor. The basic structure of a Deployment Descriptor is as follows:

<analysisEngineDeploymentDescription 
      xmlns="http://uima.apache.org/resourceSpecifier">

  <!-- the standard (optional) header -->
  <name>[String]</name>
  <description>[String]</description>
  <version>[String]</version>
  <vendor>[String]</vendor>

  <deployment protocol="jms" provider="activemq">

    <casPool numberOfCASes="xxx" initialFsHeapSize="nnn"/>

    <service>         <!-- must have only 1 -->

      <!-- 0 or more of the following -->
      <!-- name required, value optional -->
      <custom name="..." value="..."/>

      <inputQueue .../>

      <topDescriptor .../> 

      <environmentVariables .../>  <!--optional -->

      <analysisEngine key="key name" async="[true/false]"
             internalReplyQueueScaleout="nn1" <!-- optional -->
             inputQueueScaleout="nn2">        <!-- optional -->

        <scaleout numberOfInstances="1"/>       <!-- optional -->
                                                <!-- optional --> 
        <casMultiplier 
             poolSize="5"                       <!-- optional --> 
             initialFsHeapSize="nnn"            <!-- optional -->
             processParentLast="[true/false]"/> <!-- optional -->
             
        <asyncPrimitiveErrorConfiguration .../> <!-- optional -->

        <delegates>    <!-- optional, only for aggregates -->
                                       <!-- 0 or more -->
          <analysisEngine key="key name" async="[true/false]"
               internalReplyQueueScaleout="nn1"
               inputQueueScaleout="nn2">
  
                ...    <!-- optional nested specifications -->
          </analysisEngine>
                . . .
          <remoteAnalysisEngine key="key name"  <!-- 0 or more -->
               remoteReplyQueueScaleout="nn1">  <!-- optional -->
  
            <!-- next is either required or must be omitted -->
            <casMultiplier 
                    poolSize="5" 
                    initialFsHeapSize="nnn"
                    processParentLast="[true/false]"/> <!-- optional -->
                    
            <inputQueue ... />
            <serializer method="xmi"/>
            <asyncAggregateErrorConfiguration ... />
          </remoteAnalysisEngine>
                . . .
        </delegates>
      </analysisEngine>
    </service> 
  </deployment>
</analysisEngineDeploymentDescription>

This element specifies information for managing CAS pools. Having more CASes in the pools enables more AS components to run at the same time. For instance, if your application had four components, but one was slow, you might deploy 10 instances of the slow component. To get all 10 instances working on CASes simultaneously, your CAS pool should be at least 10 CASes. The casPool size should be small enough to avoid paging.

The initialFsHeapSize attribute is optional, and allows setting the size of the initial CAS Feature Structure heap. This number is specified in bytes, and the default is approximately 2 megabytes for Java top-level services, and 40 kilobytes for C++ top level services. The heap grows as needed; this parameter is useful for those cases where the expected heap size is much smaller than the default.

This section is required and specifies the deployment information for the service.

The <custom> element(s) are optional. Each one, if specified, requires a name parameter, and can have an optional value parameter. They are intended to provide additional information needed for particular kinds of deployment.

The following lists the things that can be specified here.

The inputQueue element is required. It identifies the input queue for the service.

<inputQueue brokerURL="tcp://x.y.z:portnumber"
    endpoint="queue_name"
    prefetch="1"/>

The brokerURL attribute is optional. When omitted, a default value of tcp://localhost:61616 will be used. A different brokerURL can be provided as an override when launching a service. Consult README that provides an example of brokerURL override. The queue broker address includes a protocol specification, which should be set to either "tcp", or "http". The brokerURL attribute specifies the queue broker URL, typically its network address and port. .

The http protocol is similar to the tcp protocol, but is preferred for wide-area-network connections where there may be firewall issues, as it supports http tunnelling.

The queue name is used to uniquely identify a queue belonging to a particular broker.

The prefetch attribute controls prefetching of messages for an instance of the service. It can be 0 - which disables prefetching. This is useful in some realtime applications for reducing latency. In this case, when a new request arrives, any available instance will take the request; if prefetching was set above 0, the request might be prefetched by a busy service. The default value if not specified is 0.

Each service must indicate some analysis engine to run, using this element.

<topDescriptor>
  <import location="..." /> <!-- or name="..." -->
</topDescriptor>

This is the standard UIMA import element. Imports can be by name or by location; see Section 2.2, “Imports” in UIMA References.

This element is optional, and provides a way to set environment variables.

Components written in C++ can be run as a top level service. These components are launched in a separate process, and by default, all the environment variables of the launching process are passed to the new process. This element allows the environment variables of the new process to be augmented.

<environmentVariables>
<!-- one or more of the following element -->
<environmentVariable name="xxx">value goes here</environmentVariable>
</environmentVariables>

Usually, the value will replace any existing value. As a special exception, for the environment variables used as the PATH (for Windows) or LD_LIBRARY_PATH (for Linux) or DYLD_LIBRARY_PATH (for MacOS), the value will be "prepended" with a path separator character appropriate for the platform, to any existing value.

This is used to describe an element which is an analysis engine. It is optional and only needed if the defaults are being overridden. The async attribute is only used for aggregates, and specifies that this aggregate will be run asynchronously (with input queues in front of all of its delegates) or not. If not specified, the async property defaults to "false" except in the case where the deployment descriptor includes the <delegates> element, when it defaults to "true". If you specify async="false", then it is an error to specify any <delegates> in the deployment descriptor.

The key attribute must have as its value the key name used in the containing aggregate descriptor to uniquely identify this delegate. Since the top level aggregate is not contained in another aggregate, this can be omitted for that element. Deployment information is matched to delegates using the key name specified in the aggregate descriptor to identify the delegate.

<analysisEngine key="key name" async="true"
             internalReplyQueueScaleout="nn1"  <!-- optional -->
             inputQueueScaleout="nn2">         <!-- optional -->

  <scaleout numberOfInstances="1"/>        <!-- optional  -->
  <!-- casMultiplier is either required, or must be omitted-->
  <casMultiplier 
       poolSize="5"                        <!-- optional  -->
       initialFsHeapSize="nn"              <!-- optional  -->   
       processParentLast="[true/false]"/>  <!-- optional  -->             

    <!-- next two are optional, but only one allowed -->
  <asyncAggregateErrorConfiguration .../>  <!-- optional  -->
  <asyncPrimitiveErrorConfiguration .../>  <!-- optional  -->

  <delegates>                              <!-- optional  -->
    <analysisEngine key="key name" ...>    <!-- 0 or more -->
            ...       <!-- optional nested specifications -->
    </analysisEngine>
            . . . 
    <remoteAnalysisEngine key="key name"   <!-- 0 or more -->
             remoteReplyQueueScaleout="nn1"> <!-- optional -->

      <!-- next is either required or must be omitted -->
      <casMultiplier 
           poolSize="5"                        <!-- optional  -->
           initialFsHeapSize="nnn"             <!-- optional  -->
           processParentLast="[true/false]"/>  <!-- optional  -->             
      
      <inputQueue ... />
      <serializer method="[xmi|binary]"/>              <!-- optional -->
      <asyncAggregateErrorConfiguration .../> <!-- optional -->
    </remoteAnalysisEngine>
            . . . 
  </delegates>                . . .        
</analysisEngine>

<analysisEngine> is used to specify deployment details for an analysis engine. It is optional, and if omitted, defaults will be used: The analysis engine will be run synchronously (processing only one CAS at a time), with a scaleout of 1, using the default error configuration.

The attributes internalReplyQueueScaleout and inputQueueScaleout only have meaning and are allowed when async="true" is specified (which in turn can only be set true for aggregates) or is the default ( which happens when the aggregate has delegate deployment options specified in the deployment descriptor). These attributes default to 1. For asynchronous aggregates, they control the number of threads used to do the work of the aggregate outside of running the delegates. This work can include one or more of the following:

These attributes provide a way to scale out this work on multi-core machines, if these tasks become a bottleneck.

Note that if an aggregates flow controller specifies that the first delegate the CAS should flow to is a remote, the work of serializing the CAS to that remote is done using the inputQueue thread, and the scaleout parameter that would apply would be the inputQueueScaleout. For subsequent delegates, the work is done on the internalReplyQueueScaleout threads.

The <scaleout ...> element specifies, for co-located primitive or non-AS aggregates (async="false") at the bottom of an aggregate tree, how many replicated instances are created.

The <casMultiplier> element inside an <analysisEngine> element is required if the analysis engine component is a CAS multiplier, and is an error if specified for other components. It specifies for CAS multipliers the size of the pool of CASes used by that CAS multiplier for generating extra CASes.

The initialFsHeapSize attribute on the <casMultiplier> element is optional, and allows setting the size of the initial CAS Feature Structure heap for CASes in this pool. This number is specified in bytes, and the default is approximately 2 megabytes for Java top-level services, and 40 kilobytes for C++ top level services. The heap grows as needed; this parameter is useful for those cases where the expected heap size is much smaller than the default.

The processParentLast attribute on the <casMultiplier> element is optional, and specifies processing order of an input CAS relative to its children. If true, a flow of an input CAS will be suspended after it is returned from a Cas Multiplier delegate until all its child CASes have finished processing. If false, an input CAS can be processed in parallel with its children.

The <remoteAnalysisEngine> elements are used to specify that the delegate is not co-located, and how to connect to it. The remoteReplyQueueScaleout is optional; if not specified it defaults to 1. This scaleout is the number of threads that will be used to do the work of the containing aggregate when replies are returned from this remote delegate. This work is described above. It may be useful to set this to > 1 if, for instance, there are many CASes coming back from a remote delegate (perhaps the remote is a CAS Multiplier), and each one has to be deserialized.

The <serializer> element describes what method of serialization to use. This element is optional and it may be set to either binary or xmi. If omitted, xmi serialization will be used by default. Xmi serialization can be quite verbose and produce large output for CASes containing many annotations; on the plus side, it supports serialization between components where the type systems may not be exactly identical (for instance, they could be different subsets of larger, common type systems). Binary serialization produces a smaller output size and is more efficient; on the minus side, it requires that the type systems for both components have exactly the same type and feature codes - which in practice means that the type systems have to be identical. Also, the binary serialization format is new with 2.3.0 release, and is not always available. For example, C++ services do not (currently) support this format.

The <inputQueue> element specifies the remote's input queue. The casMultiplier element inside a remoteAnalysisEngine element is only specified if the remote component is a CAS Multiplier, and it specifies the size of a pool of CASes kept to receive the new CASes from the remote component, and the initial size of those CASes. Its poolSize must be equal to or larger than the casMultiplier poolSize specified for that remote component.

Services may be running on nodes with firewalls, where the only port open is the one for http. In this case, you can use the http protocol.

The <asyncPrimitiveErrorConfiguration> element is only allowed within a top-level analysis engine specification (that is, one that is not a delegate of another, containing analysis engine).

Error Configuration descriptors can be included directly in the deployment descriptors, or they may use the <import> mechanism to import another file having the specification.

For AS Aggregates, the configuration applicable to delegates goes in <asyncAggregateErrorConfiguration> elements for the delegate.

For AS Primitives, there is one <asyncPrimitiveErrorConfiguration> element that configures threshold-based termination. The other kinds of error configuration are not applicable for AS Primitives.

See Chapter 2, Error Handling for Asynchronous Scaleout for a complete overview of error handling.

The Error Configuration descriptor for AS Aggregates is as follows; note that all the elements are optional:

<asyncAggregateErrorConfiguration 
      xmlns="http://uima.apache.org/resourceSpecifier">

  <!-- the standard (optional) header -->
  <name>[String]</name>
  <description>[String]</description>
  <version>[String]</version>
  <vendor>[String]</vendor>

  <import ... />  <!-- optional -->   

  <getMetadataErrors
          maxRetries="n" 
          timeout="xxx_milliseconds"
          errorAction="disable|terminate"/> 

  <processCasErrors
          maxRetries="n" 
          timeout="xxx_milliseconds"
          continueOnRetryFailure="true|false"
          thresholdCount="xxx"
          thresholdWindow="yyy"
          thresholdAction="disable|terminate"/>

  <collectionProcessCompleteErrors
          timeout="xxx_milliseconds"
          additionalErrorAction="disable|terminate"/>

</asyncAggregateErrorConfiguration>

For an AS Primitive, the <asyncPrimitiveErrorConfiguration> element appears at the top level, and has this form:

<asyncPrimitiveErrorConfiguration 
      xmlns="http://uima.apache.org/resourceSpecifier">

  <!-- the standard (optional) header -->
  <name>[String]</name>
  <description>[String]</description>
  <version>[String]</version>
  <vendor>[String]</vendor>

  <import ... />  <!-- optional -->   

  <processCasErrors
          thresholdCount="xxx"
          thresholdWindow="yyy"
          thresholdAction="terminate"/>

  <collectionProcessCompleteErrors
          additionalErrorAction="terminate"/>
          
</asyncPrimitiveErrorConfiguration>

The maxRetries attribute specifies the maximum number of retries to do. If this is set to 0 (the default), no retries are done.

The continueOnRetryFailure attribute, if set to 'true' causes the framework to ask the aggregate's flow controller if the processing for the CAS can continue. If this attribute is 'false' or if the flow controller indicates it cannot continue, further processing on the CAS is stopped and an error is returned from the aggregate. Warning: there are some conditions in the current implementation where this is not yet being done; this is a known issue.

The timeout attribute specifies the timeout values used when sending commands to the delegates. The units are milliseconds and a value of 0 has the special meaning of no timeout.

The thresholdCount and thresholdWindow attributes specify the threshold at which the thresholdAction is taken. If xxx errors occur within a window of size yyy, the framework takes the specified action of either disabling this delegate, or terminating the containing AS Aggregate (or if not an AS Aggregate, terminating the AS Primitive). A thresholdCount of 0 (the default) has the special meaning of no threshold, i.e. errors ignored, and a thresholdWindow of 0 (the default) means no window, i.e. all errors counted.

An action of 'disable' applies to the specified delegate, removing it from the flow so the containing aggregate will no longer send it commands. The 'terminate' action applies to the entire service containing this component, disconnecting it from its input queue and shutting it down. Note that when disabling, the framework asks the flow controller to remove the delegate from the flow, but if the flow controller cannot reasonably operate without this component it can convert the action to 'terminate' by throwing an AnalysisEngineProcessException.FLOW_CANNOT_CONTINUE_AFTER_REMOVE exception.

Note that the only action for an AS Primitive on getMetadata failure is to terminate, and this is always the case, so it is not listed as an configuration option. This is also the default action for an AS Aggregate getMetadata failure.

If the <errorConfiguration> element is omitted, or if some sub elements of this are omitted, the following defaults are used:

The Asynchronous Client API provides Java applications the capability to connect to and make requests to UIMA-AS services. ProcessCas and CollectionProcessingComplete requests are supported.

It provides four kinds of capabilities:

An application can use this API to prepare and send each CAS to a service one at a time, or alternately can use a UIMA collection reader to prepare the CASes to be delivered.

The application normally provides a listener class to receive asynchronous replies. For individual CAS requests a synchronous sendAndReceive call is available. As an alternative for this synchronous call, instead of using this client API, the standard UIMA Analysis Engine APIs can be used with an analysis engine instantiated from a JMS Service Descriptor. See Section 1.7, “JMS Service Descriptor”.

As a convenience, the Asynchronous Client API can also be used to deploy (i.e., "start") services. Java services deployed by the API are instantiated in the same JVM. Logging for all UIMA components in the same JVM are merged; class names and thread IDs can be used to distinguish log entries from different services. All services in the JVM can be monitored by a single JMX console. Native C++ UIMA services can be called from the JVM via the JNI or optionally be launched as separate processes on the same machine. In either case logging and JMX monitoring for native services are integrated with the other UIMA components in the JVM.

An application developer's starting point for accessing UIMA-AS services is the UimaAsynchronousEngine Interface. For each service an application wants to use, it must instantiate a client object:

UimaAsynchronousEngine uimaAsEngine = 
        new BaseUIMAAsynchronousEngine_impl();

The following is a short introduction to some important methods on this class.

The application context map is used to pass initialization parameters. These parameters are itemized below.

Asynchronous events are returned to applications via methods in classes registered to the Client API object with addStatusCallbackListener(). These classes must extend the class org.apache.uima.aae.client.UimaAsBaseCallbackListener.

Errors are delivered to the callback listeners as an EntityProcessStatus or UimaASProcessStatus object. These objects provide the methods:

Services can be undeployed from a client object as follows:

// create Map to hold required parameters
Map<String,Object> appCtx = new HashMap<String,Object>();
appCtx.put(UimaAsynchronousEngine.DD2SpringXsltFilePath, 
           System.getenv("UIMA_HOME") + "/bin/dd2spring.xsl");
appCtx.put(UimaAsynchronousEngine.SaxonClasspath, 
           "file:" + System.getenv("UIMA_HOME") + "/saxon/saxon8.jar");
String id = uimaAsEngine.deploy(service, appCtx);
uimaAsEngine.undeploy(id);
    

The Client API has a built in recovery strategy to handle cases where a broker fails or becomes unreachable, and then, later becomes available again.

Before sending a new request to a broker, the client checks the state of its connection. If the connection has failed, the client enters a loop where it will attemp to reconnect every 5 seconds. One message is logged to notify this is happening. The recovery attempt stops when the the connection is recovered, or when all UIMA AS clients that are sharing this failed connection, terminate.

During the recovery attempt, any CASes that are submitted via the client APIs will fail or timeout. If the application uses the sendAndReceive() synchronous API, the failure will be delivered by an exception. If the application client uses the sendCAS() asynchronous API, the failure will be delivered via the normal callback listener that the application registered with the UIMA AS client.

See $UIMA_HOME/examples/src/org/apache/uima/examples/as/RunRemoteAsyncAE.java

JMX (Java Management Extensions) is a standard Java mechanism that is used to monitor and control Java applications. A standard Java tool provided with most Javas, called jconsole, is a GUI based application that can connect to a JVM and display the information JMX is providing, and also control the application in application-defined specific ways.

JMX information is provided by a hierarchy of JMX Beans. More background and information on JMX and the jconsole tool is available on the web.

Apache Camelis an integration framework based on Enterprise Integration Patterns which uses routes for rule-based message routing and mediation. The camel project has a large number of components which provide access to a wide variety of technologies and are the building blocks of the routes. The Asynchronous Scaleout Camel Driver is a component to integrate UIMA AS into Camel.

The Asynchronous Scaleout Camel Driver sends the camel message body (without headers) to a specified UIMA AS processing pipeline. Accessing the analysis results which are written into the CAS is not possible from a camel route. There are basically two usage scenarios. The Camel Driver can be used to drive the processing of a UIMA AS cluster in which each server instance runs a cas multiplier to fetch the actual document from a database. In this scenario the camel route only sends an ID of the document to the cas multiplier which does the actual fetching of the document. In the second usage scenario the Camel Driver is used to send a document in a one way fashion to a UIMA AS processing pipeline which then takes care of processing it. In case an error occurs inside the processing pipeline the exception is forwarded to camel and set on the message as response. Error handling is described in the Error handling in Camel documentation.

The camel driver expects a string message body; if it is not of the type string it might be automatically converted by camel type converters. The string message body is set as the CAS's document text. An Analysis Engine calls CAS CAS.getDocumentText() to retrieve the string.

The Asynchronous Scaleout Camel Driver is configured with a configuration string. The configuration string must contain the broker location and name of the JMS queue used to communicate with UIMA AS. It has the following format

uimadriver:brokerURL?queue=nameofqueue&CasPoolSize=n&Timeout=t

which could for example be specified as

uimadriver:tcp://localhost:61616?queue=TextAnalysisQueue

. The CasPoolSize parameter is optional but if it is present n must be an integer which is larger than zero, otherwise the UIMA AS default will be used. The Timeout parameter is optional but if present t in milliseconds must zero or larger.

Camel enables a developer to create quickly all kinds of applications out of existing and custom components. The sample demonstrates how UIMA AS is integrated with other technologies. Readers who are new to camel should read the Getting Started chapter in the camel documentation.

First a simple sample: A user wants to test a UIMA AS processing pipeline, sending it a set of test documents to process. The plain text test documents are located in a folder "/test-data". A camel route for this defined with Java DSL could look like this:

     from("file://test-data?noop=true").
     to("uimadriver:tcp://localhost:61616?queue=TextAnalysisQueue");
    

In the route above the file component sends a message for every file to the uimadriver component. The message contains a reference to the file but not the content of the file itself. The uimadriver component expects a message with string body as input. An internal camel type converter will read in the bytes of the file, decode them into characters with the default platform encoding and then create a string object which is passed to the uimadriver component. The uimadriver then puts the string into a CAS and sends it via the UIMA AS Client API to a processing pipeline. Note that results from the returned CAS cannot be retrieved in a camel route.

A more complex sample. A web site has an area where people can upload pictures. The pictures must be checked for appropriate content. The pictures are pushed to the site via http, stored in a database and assigned to the human controllers to classify them either as appropriate or non-appropriate. That is achieved with an existing camel route and a servlet which receives the images and sends them to the camel route.

    from("direct:start").
    to("imagewriter").
    to("jms:queue:HumanPictureAnalysisQueue");
    

The message containing the image is received by the direct:start endpoint, the image is written to a database and replaced with a string identifier by the "imagewriter" component, in the last step the camel jms component posts the identifier on a JMS queue to notify the reception of a new image. The notification is received by a client tool which the human controllers use to classify an image.

To lessen the workload on the human reviewers, a system should automatically classify the pictures and only assign questionable cases to human reviewers. The automatic classification is done by an UIMA Analysis Engine. The AE can mark an image with one of three classes appropriate, non-appropriate and unknown. In the case of unknown the AE is not confident enough which of the first two classes is correct. To be scalable, the processing pipeline is hosted by UIMA AS and contains three AEs, one to fetch the image from the database, a classification AE and an AE to write the class of the image back to the database. The first AE is typical a cas multiplier and receives a CAS which only contains the string identifier but not the actual image. The cas muliplier uses the identifer to fetch the image from the database and outputs a new CAS with the actual image. The Camel route blocks until the CAS is processed by the following two AEs and depending on the class in the database the picture is assigned to a human controller or not.

    from("direct:start").
    to("imagewriter").
    to("uimadriver:tcp://localhost:61616?queue=UimaPictureAnalysisQueue").
    to("class-retriever").
    // filters messages with class appropriate and non-appropriate
    filter(header("picture-class").isEqualTo("unkown")).
    to("jms:queue:HumanPictureAnalysisQueue");
    

The first part is identical, after the imagewriter the string identifier is send to the UIMA AS processing pipeline which writes the image class back to the database. The class is retrieved with the custom class-retriever component and written to a message header field, only if the class is unknown the image is assigned for human classification.

Note: instead of using a CAS multiplier, a more straight-forward approach would use just one CAS, having 2 views: one view would contain the string identifier of the image, and the other view would have the image to be analyzed (or a reference to it in the DB).

The Asynchronous Scaleout Camel Driver is a typical camel component. The camel documentation Writing Components describes how camel components are written. The source code can be found in the uimaj-as-camel project. The implementation defines an asynchronous producer endpoint, which is implemented in the org.apache.uima.camel.UimaAsProducer class. The UimaAsProducer.process method gets the string body of the message, wraps it in a CAS object and sends it to UIMA AS. Since the producer is asynchronous the camel message is registered with the reference id of the sent CAS in an intermediate map, when the CAS comes back from UIMA AS, the camel message is looked up with the reference id of the CAS and the processing of the camel message is completed. For further details please read the UimaAsProducer implementation code.