In This Section |
|
State Replication is the simpler of Talon's 2 HA models. With State Replication Talon replicates changes to state and the outbound messages emitted by your application's message handlers. In the event of failover to a backup your application's state is available at the same point where processing left off, and the engine will retransmit any outbound messages that were left in doubt as a result of the failure.
A basic state replication application is straightforward. A quick way to get started is to use the nvx-talon-sr-processor-archetype described in Maven Archetype Quick Starts. The general flow for creating a state replication application involves:
The steps outlined below assume that you have already modeled some messages and state for your application to use. See the Modeling Message and State sections to get started.
When modeling application state, Xbuf encoding is not yet recommended, Protobuf encoding should be used instead (Xbuf currently has a higher memory footprint than Protobuf generated entities). |
@AppHAPolicy(value=AepEngine.HAPolicy.StateReplication)
public class Application {
...
} |
An AEPEngine is application state agnostic: it deals with your application an plain old java object graph with a single root. Given the root object for your application's state the underlying store will track changes made to fields on the root (and its descendants) and replicate or persist those changes. As such, an AepEngine needs to be able to create a new application state during when your application is initialized. This is done by finding a method on your main application class anotated with @AppStateFactoryAccessor. The state factory accessor returns a newly initialized set of state for the engine to manage. As messages are processed the engine will pass the relevant state root back into the application to be operated upon.
@AppStateFactoryAccessor
final public IAepApplicationStateFactory getStateFactory() {
return new IAepApplicationStateFactory() {
@Override
final public Repository createState(MessageView view) {
return Repository.create();
}
};
} |
As your application makes changes to this root object (setting fields etc), the engine will monitor the root and replicate deltas to backup members or disk.
If your application will send messages, it can add an injection point for the underlying AEPEngine to inject a message sender.
private AepMessageSender messageSender;
@AppInjectionPoint
final public void setMessageSender(AepMessageSender messageSender) {
this.messageSender = messageSender;
} |
When working with an application using StateReplication, the underlying AepEngine will pass in the root object for your application state along with the message. Outbound message sends and state changes are managed by Talon as an atomic unit of work.
@EventHandler
final public void onMessage(Message message, Repository repository) {
// update state
repository.setCounter(repository.getCounter() + 1);
// send event
Event event = Event.create();
event.setVal(message.getVal());
event.setCount(repository.getCounter());
messageSender.sendMessage("events", event);
} |
When working with state replication both the ADM and message and state object factories need to be registered with the runtime. Registering the state factory allows the underlying state replication machinery to deserialize replicated state objects based on the ids encoded in the replication stream. The message factories allow are used for deserializing replicated outbound messages as well as messages received from message buses. The state factories can be declared in your config.xml or programmatically:
<app name="processor" mainClass="com.sample.Application">
<messaging>
<factories>
<factory name="com.sample.messages.MessageFactory"/>
</factories>
</messaging>
<storage enabled="true">
<factories>
<factory name="com.sample.state.StateFactory" />
</factories>
</storage>
</app> |
@AppInjectionPoint
public void initialize(AepEngine engine) {
// for messaging
engine.registerFactory(new com.sample.messages.MessageFactory());
// for state replication:
engine.registerFactory(new com.sample.state.StateFactory());
} |
To actually achieve high availability storage must be configured for the application. The primary means of storage is for Talon apps is through clustered replication to a backup instance. Talon also logs state changes to a disk based transaction log as a fallback mechanism. Storage and persistence can be enabled in the application's configuration xml.
<app name="processor" mainClass="com.sample.Application">
...
<storage enabled="true">
....
<clustering enabled="true"/>
<persistence enabled="true">
<!--
When using Xbuf encoded entities,
detached persist is not supported.
-->
<detachedPersist enabled="false"/>
</persistence>
</storage>
</app> |
Enabling clustering allows 2 applications of the same name to discover one another and form an HA cluster. When one or more instances of an application connect to one another one instance is elected as the primary via a leadership election algorithm. The primary member will establish messaging connections and begin invoking message handlers in your application.
Enabling persistence causes the replication stream that is sent to backup instances to also be logged locally on disk to a transaction log file. The transaction log file can be used to recover application state from a cold start, and is also used to initialize new cluster members that connect to it so when clustering is enabled, persistence must be enabled as well.
There are many configuration knobs that can be used to customize the store's behavior and performance. See DDL Config Reference for a listing of configuration knobs for storage.
For applications that use state replication with an unsolicited sender (i.e. those not done from a message handler) such as the Feeder app, the application should additionally configure the engine to replicate unsolicited sends so that state and messages are replicated for the sender. The Feeder is such an application, so it additionally sets this value to true:
descriptor.setReplicateUnsolicitedSends(true); |
Application state is always created by the AEP engine and is created in one of three different ways:
Two constraints need to be met when the state is created explicitly by the application.
The first constraint implies that the state must be created before the engine has started messaging. The second constraint implies that the state needs to be created during or after start (since the engine opens the store during start()) i.e. the only time the state can be explicitly created by the application is between start() and the receipt of the messaging started event. The messaging prestart event is such a point in the flow.
The below snippet comes from the Enricher message handler. With a message handler, application state should be retrieved using the Engine's getApplicationState(MessageView) accessor, which retrieves the state associated with the message's flow. In the Ticker sample, the flow is defined by the hashCode of the symbol and is set by the Feeder. Flows define the order in which messages are processed in the system. Flows essentially partition traffic into ordered parallel processed streams and, therefore, enable parallelized, concurrent message processing.
Under the covers, the Engine ensures that both messages and state are replicated to the backup to prevent loss or duplication in the event of a failure.
//to the enriched channel with a trend.
ApplicationState state = _engine.getApplicationState(tick);
state.setSno(state.getSno() + 1);
state.getTickHistory().add(tick);
..
_engine.sendMessage({_}enrichChannel, enriched);
//Clear out older entries:
if(state.getTickHistory().size() > HISTORY_SIZE) {
state.getTickHistory().remove();
} |
The Feeder operates slightly differently. Because it is the origin point of the traffic, it can't retrieve its state from the Engine using a MessageView, but as mentioned previously, Creation of State can only be done from an engine thread. It must therefore create its state from within an AepMessagingPrestartEvent handler. If the Feeder were to operate with multiple flows (and consequently multiple application state instances, it would need to create them all here.
// The messaging prestart event
@EventHandler
public void onMessagingPrestart( AepMessagingPrestartEvent evt) {
//In a more complex application one might futher initialize
//The state prior to returning it:
_state = _engine.getApplicationState(symbol.hashCode());
} |
The sample application is simple and operates with a single flow based on the symbol. For simplicity, we use the hashcode of the symbol name to map this to an integer id representing the flow. The sender creates the state and stores it in a private variable _state for use during runtime. The created state is not threadsafe, so in this context, it can only be used by the feeder's thread sending its messages.
See State Graph Limitations for a discussion of current state graph limitations.