Fixes for zk ensemble operations

- Fix some concurrent event issues which were
  born with new tests
- Enhance tests for dist machine and ensemble
- Add new features to testing system
- Polish and add more logging

spring-statemachine-core/src/main/java/org/springframework/statemachine/config/AbstractStateMachineFactory.java

@@ -212,7 +212,7 @@ public abstract class AbstractStateMachineFactory<S, E> extends LifecycleObjectS
 		// we wrap previously build machine with a distributed
 		// state machine and set it to use given ensemble.
 		if (stateMachineConfigurationConfig.getStateMachineEnsemble() != null) {
-			DistributedStateMachine<S, E> distributedStateMachine = new DistributedStateMachine<>(
+			DistributedStateMachine<S, E> distributedStateMachine = new DistributedStateMachine<S, E>(
 					stateMachineConfigurationConfig.getStateMachineEnsemble(), machine);
 			distributedStateMachine.setAutoStartup(stateMachineConfigurationConfig.isAutoStart());
 			distributedStateMachine.afterPropertiesSet();

spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/DistributedStateMachine.java

@@ -75,13 +75,15 @@ public class DistributedStateMachine<S, E> extends LifecycleObjectSupport implem
 
 	@Override
 	protected void onInit() throws Exception {
-		delegate.getStateMachineAccessor().doWithAllRegions(new StateMachineFunction<StateMachineAccess<S, E>>() {
+		// TODO: should we register with all, not just top one?
+		delegate.getStateMachineAccessor().doWithRegion(new StateMachineFunction<StateMachineAccess<S, E>>() {
 
 			@Override
 			public void apply(StateMachineAccess<S, E> function) {
 				function.addStateMachineInterceptor(interceptor);
 			}
 		});
+
 	}
 
 	@Override
@@ -176,6 +178,9 @@ public class DistributedStateMachine<S, E> extends LifecycleObjectSupport implem
 		@Override
 		public void preStateChange(State<S, E> state, Message<E> message, Transition<S, E> transition,
 				StateMachine<S, E> stateMachine) {
+			if (log.isTraceEnabled()) {
+				log.trace("Received preStateChange from " + stateMachine + " for delegate " + delegate);
+			}
 			// only handle if state change originates from this dist machine
 			if (message != null
 					&& ObjectUtils.nullSafeEquals(delegate.getId(),

spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java

@@ -417,6 +417,8 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
 		if (currentState != null) {
 			buf.append(StringUtils.collectionToCommaDelimitedString(currentState.getIds()));
 		}
+		buf.append(" / id=");
+		buf.append(id);
 		return buf.toString();
 	}
 

spring-statemachine-test/src/main/java/org/springframework/statemachine/test/StateMachineTestPlan.java

@@ -22,13 +22,19 @@ import static org.junit.Assert.assertThat;
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.hamcrest.Matcher;
 import org.springframework.statemachine.StateMachine;
 import org.springframework.statemachine.test.StateMachineTestPlanBuilder.StateMachineTestPlanStep;
 import org.springframework.statemachine.test.support.LatchStateMachineListener;
+import org.springframework.util.StringUtils;
 
 /**
  * {@code StateMachineTestPlan} is fully constructed plan how
@@ -41,18 +47,25 @@ import org.springframework.statemachine.test.support.LatchStateMachineListener;
  */
 public class StateMachineTestPlan<S, E> {
 
-	private final List<StateMachine<S, E>> stateMachines;
+	private final static Log log = LogFactory.getLog(StateMachineTestPlan.class);
+	private final Map<Object, StateMachine<S, E>> stateMachines;
 	private final List<StateMachineTestPlanStep<S, E>> steps;
+	private Integer defaultAwaitTime = 10;
 
 	/**
 	 * Instantiates a new state machine test plan.
 	 *
 	 * @param stateMachines the state machines
 	 * @param steps the steps
+	 * @param defaultAwaitTime the default await time in seconds
 	 */
-	public StateMachineTestPlan(List<StateMachine<S, E>> stateMachines, List<StateMachineTestPlanStep<S, E>> steps) {
+	public StateMachineTestPlan(Map<Object, StateMachine<S, E>> stateMachines, List<StateMachineTestPlanStep<S, E>> steps,
+			Integer defaultAwaitTime) {
 		this.stateMachines = stateMachines;
 		this.steps = steps;
+		if (defaultAwaitTime != null) {
+			this.defaultAwaitTime = defaultAwaitTime;
+		}
 	}
 
 	/**
@@ -62,39 +75,144 @@ public class StateMachineTestPlan<S, E> {
 	 */
 	public void test() throws Exception {
 
-		List<LatchStateMachineListener<S, E>> listeners = new ArrayList<LatchStateMachineListener<S, E>>();
-		for (StateMachine<S, E> stateMachine : stateMachines) {
+		Map<StateMachine<S, E>, LatchStateMachineListener<S, E>> listeners =
+				new HashMap<StateMachine<S,E>, LatchStateMachineListener<S,E>>();
+		for (StateMachine<S, E> stateMachine : stateMachines.values()) {
 			LatchStateMachineListener<S, E> listener = new LatchStateMachineListener<S, E>();
-			listeners.add(listener);
+			listeners.put(stateMachine, listener);
 			stateMachine.addStateListener(listener);
 			stateMachine.start();
 		}
+		log.info("Running test plan for machines "
+				+ StringUtils.collectionToCommaDelimitedString(stateMachines.values()));
 
-
+		int stepCounter = 0;
 		for (StateMachineTestPlanStep<S, E> step : steps) {
-			for (LatchStateMachineListener<S, E> listener : listeners) {
-				listener.reset(step.expectStateChanged != null ? step.expectStateChanged : 0, 0, 0, 0, 0, 0, 0, 0, 0);
+			log.info("Running test plan step " + stepCounter++);
+			for (LatchStateMachineListener<S, E> listener : listeners.values()) {
+				listener.reset(
+						step.expectStateChanged != null ? step.expectStateChanged : 0,
+						step.expectStateEntered != null ? step.expectStateEntered : 0,
+						step.expectStateExited != null ? step.expectStateExited : 0,
+						step.expectEventNotAccepted != null ? step.expectEventNotAccepted : 0,
+						step.expectTransition != null ? step.expectTransition : 0,
+						step.expectTransitionStarted != null ? step.expectTransitionStarted : 0,
+						step.expectTransitionEnded != null ? step.expectTransitionEnded : 0,
+						step.expectStateMachineStarted != null ? step.expectStateMachineStarted : 0,
+						step.expectStateMachineStopped != null ? step.expectStateMachineStopped : 0);
 			}
 
-			if (step.sendEvent != null) {
-				stateMachines.get(0).sendEvent(step.sendEvent);
-			}
-
-			if (step.expectStateChanged != null) {
-				for (LatchStateMachineListener<S, E> listener : listeners) {
-					assertThat(listener.getStateChangedLatch().await(5, TimeUnit.SECONDS), is(true));
-					assertThat(listener.getStateChanged().size(), is(step.expectStateChanged));
+			if (step.expectStateMachineStarted != null) {
+				for (Entry<StateMachine<S, E>, LatchStateMachineListener<S, E>> entry : listeners.entrySet()) {
+					assertThat("StateMachineStarted Await not matched for machine " + entry.getKey(), entry.getValue()
+							.getStateMachineStartedLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true));
+					assertThat("StateMachineStarted count not matched for machine " + entry.getKey(), entry.getValue()
+							.getStateMachineStarted().size(), is(step.expectStateMachineStarted));
 				}
 			}
 
-			if (step.expectState != null) {
-				for (StateMachine<S, E> stateMachine : stateMachines) {
+			if (step.sendEvent != null) {
+				StateMachine<S, E> sendVia = null;
+				if (step.sendEventMachineId != null) {
+					sendVia = stateMachines.get(step.sendEventMachineId);
+				} else {
+					sendVia = stateMachines.values().iterator().next();
+				}
+				assertThat("Error finding machine to send via", sendVia, notNullValue());
+				log.info("Sending test event " + step.sendEvent + " via machine " + sendVia);
+				sendVia.sendEvent(step.sendEvent);
+			}
+
+			if (step.expectStateChanged != null) {
+				for (Entry<StateMachine<S, E>, LatchStateMachineListener<S, E>> entry : listeners.entrySet()) {
+					assertThat("StateChanged Await not matched for machine " + entry.getKey(), entry.getValue()
+							.getStateChangedLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true));
+					assertThat("StateChanged count not matched for machine " + entry.getKey(), entry.getValue()
+							.getStateChanged().size(), is(step.expectStateChanged));
+				}
+			}
+
+			if (step.expectStateEntered != null) {
+				for (LatchStateMachineListener<S, E> listener : listeners.values()) {
+					assertThat(listener.getStateEnteredLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true));
+					assertThat(listener.getStateEntered().size(), is(step.expectStateEntered));
+				}
+			}
+
+			if (step.expectStateExited != null) {
+				for (LatchStateMachineListener<S, E> listener : listeners.values()) {
+					assertThat(listener.getStateExitedLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true));
+					assertThat(listener.getStateExited().size(), is(step.expectStateExited));
+				}
+			}
+
+			if (step.expectEventNotAccepted != null) {
+				for (LatchStateMachineListener<S, E> listener : listeners.values()) {
+					assertThat(listener.getEventNotAcceptedLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true));
+					assertThat(listener.getEventNotAccepted().size(), is(step.expectEventNotAccepted));
+				}
+			}
+
+			if (step.expectTransition != null) {
+				for (LatchStateMachineListener<S, E> listener : listeners.values()) {
+					assertThat(listener.getTransitionLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true));
+					assertThat(listener.getTransition().size(), is(step.expectTransition));
+				}
+			}
+
+			if (step.expectTransitionStarted != null) {
+				for (LatchStateMachineListener<S, E> listener : listeners.values()) {
+					assertThat(listener.getTransitionStartedLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true));
+					assertThat(listener.getTransitionStarted().size(), is(step.expectTransitionStarted));
+				}
+			}
+
+			if (step.expectTransitionEnded != null) {
+				for (LatchStateMachineListener<S, E> listener : listeners.values()) {
+					assertThat(listener.getTransitionEndedLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true));
+					assertThat(listener.getTransitionEnded().size(), is(step.expectTransitionEnded));
+				}
+			}
+
+			if (step.expectStateMachineStopped != null) {
+				for (LatchStateMachineListener<S, E> listener : listeners.values()) {
+					assertThat(listener.getStateMachineStoppedLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true));
+					assertThat(listener.getStateMachineStopped().size(), is(step.expectStateMachineStopped));
+				}
+			}
+
+			if (!step.expectStates.isEmpty()) {
+				for (StateMachine<S, E> stateMachine : stateMachines.values()) {
 					assertThat(stateMachine.getState(), notNullValue());
 					Collection<Matcher<? super S>> itemMatchers = new ArrayList<Matcher<? super S>>();
-					itemMatchers.add(is(step.expectState));
+					for (S expectState : step.expectStates) {
+						itemMatchers.add(is(expectState));
+					}
 					assertThat(stateMachine.getState().getIds(), containsInAnyOrder(itemMatchers));
 				}
 			}
+
+			if (!step.expectVariableKeys.isEmpty()) {
+				for (StateMachine<S, E> stateMachine : stateMachines.values()) {
+					Map<Object, Object> variables = stateMachine.getExtendedState().getVariables();
+					for (Object key : step.expectVariableKeys) {
+						assertThat("Key " + key + " doesn't exist in extended state variables",
+								variables.containsKey(key), is(true));
+					}
+				}
+			}
+
+			if (!step.expectVariables.isEmpty()) {
+				for (StateMachine<S, E> stateMachine : stateMachines.values()) {
+					Map<Object, Object> variables = stateMachine.getExtendedState().getVariables();
+					for (Entry<Object, Object> entry : step.expectVariables.entrySet()) {
+						assertThat("Key " + entry.getKey() + " doesn exist in extended state variables",
+								variables.containsKey(entry.getKey()), is(true));
+						assertThat("Variable " + entry.getKey() + " doesn't match in extended state variables",
+								variables.get(entry.getKey()), is(entry.getValue()));
+					}
+				}
+			}
 		}
 	}
 

spring-statemachine-test/src/main/java/org/springframework/statemachine/test/StateMachineTestPlanBuilder.java

@@ -16,7 +16,11 @@
 package org.springframework.statemachine.test;
 
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import org.springframework.statemachine.StateMachine;
 
@@ -30,8 +34,9 @@ import org.springframework.statemachine.StateMachine;
  */
 public class StateMachineTestPlanBuilder<S, E> {
 
-	private List<StateMachine<S, E>> stateMachines = new ArrayList<StateMachine<S, E>>();
+	private Map<Object, StateMachine<S, E>> stateMachines = new HashMap<Object, StateMachine<S, E>>();
 	private final List<StateMachineTestPlanStep<S, E>> steps = new ArrayList<StateMachineTestPlanStep<S, E>>();
+	private Integer defaultAwaitTime;
 
 	/**
 	 * Gets a new instance of this builder.
@@ -49,7 +54,33 @@ public class StateMachineTestPlanBuilder<S, E> {
 	 * @return the state machine test plan builder
 	 */
 	public StateMachineTestPlanBuilder<S, E> stateMachine(StateMachine<S, E> stateMachine) {
-		this.stateMachines.add(stateMachine);
+		return stateMachine(stateMachine, stateMachine);
+	}
+
+	/**
+	 * Associate a state machine with this builder.
+	 *
+	 * @param stateMachine the state machine
+	 * @param machineId the machine id to use for sending
+	 * @return the state machine test plan builder
+	 */
+	public StateMachineTestPlanBuilder<S, E> stateMachine(StateMachine<S, E> stateMachine, Object machineId) {
+		this.stateMachines.put(machineId, stateMachine);
+		return this;
+	}
+
+	/**
+	 * Sets default await time. This is in seconds how long a latch
+	 * will be waited for listening various callbacks.
+	 *
+	 * @param seconds the default await time in seconds
+	 * @return the state machine test plan builder
+	 */
+	public StateMachineTestPlanBuilder<S, E> defaultAwaitTime(int seconds) {
+		if (seconds < 0) {
+			throw new IllegalArgumentException("Default await time cannot be negative, was " + seconds);
+		}
+		this.defaultAwaitTime = seconds;
 		return this;
 	}
 
@@ -68,7 +99,7 @@ public class StateMachineTestPlanBuilder<S, E> {
 	 * @return the state machine test plan
 	 */
 	public StateMachineTestPlan<S, E> build() {
-		return new StateMachineTestPlan<S, E>(stateMachines, steps);
+		return new StateMachineTestPlan<S, E>(stateMachines, steps, defaultAwaitTime);
 	}
 
 	/**
@@ -77,8 +108,20 @@ public class StateMachineTestPlanBuilder<S, E> {
 	public class StateMachineTestPlanStepBuilder {
 
 		E sendEvent;
-		S expectState;
+		Object sendEventMachineId;
+		final Collection<S> expectStates = new ArrayList<S>();
 		Integer expectStateChanged;
+		Integer expectStateEntered;
+		Integer expectStateExited;
+		Integer expectEventNotAccepted;
+		Integer expectTransition;
+		Integer expectTransitionStarted;
+		Integer expectTransitionEnded;
+		Integer expectStateMachineStarted;
+		Integer expectStateMachineStopped;
+		final Collection<Object> expectVariableKeys = new ArrayList<Object>();
+		final Map<Object, Object> expectVariables = new HashMap<Object, Object>();
+
 
 		/**
 		 * Expect a state {@code S}.
@@ -87,18 +130,68 @@ public class StateMachineTestPlanBuilder<S, E> {
 		 * @return the state machine test plan step builder
 		 */
 		public StateMachineTestPlanStepBuilder expectState(S state) {
-			this.expectState = state;
+			this.expectStates.add(state);
 			return this;
 		}
 
 		/**
-		 * Send an event {@code E}.
+		 * Expect a states {@code S}.
+		 *
+		 * @param states the states
+		 * @return the state machine test plan step builder
+		 */
+		@SuppressWarnings("unchecked")
+		public StateMachineTestPlanStepBuilder expectStates(S... states) {
+			this.expectStates.addAll(Arrays.asList(states));
+			return this;
+		}
+
+		/**
+		 * Send an event {@code E}. In case multiple state machines
+		 * exists, a random one will be chosen to send this event.
 		 *
 		 * @param event the event
 		 * @return the state machine test plan step builder
 		 */
 		public StateMachineTestPlanStepBuilder sendEvent(E event) {
+			return sendEvent(event, null);
+		}
+
+		/**
+		 * Send an event {@code E} into a state machine identified
+		 * by {@code machineId}.
+		 *
+		 * @param event the event
+		 * @param machineId the machine identifier for sending event
+		 * @return the state machine test plan step builder
+		 */
+		public StateMachineTestPlanStepBuilder sendEvent(E event, Object machineId) {
 			this.sendEvent = event;
+			this.sendEventMachineId = machineId;
+			return this;
+		}
+
+		/**
+		 * Expect variable to exist in extended state variables.
+		 *
+		 * @param key the key
+		 * @return the state machine test plan step builder
+		 */
+		public StateMachineTestPlanStepBuilder expectVariable(Object key) {
+			this.expectVariableKeys.add(key);
+			return this;
+		}
+
+		/**
+		 * Expect variable to exist in extended state variables and match
+		 * with the value.
+		 *
+		 * @param key the key
+		 * @param value the value
+		 * @return the state machine test plan step builder
+		 */
+		public StateMachineTestPlanStepBuilder expectVariable(Object key, Object value) {
+			this.expectVariables.put(key, value);
 			return this;
 		}
 
@@ -116,6 +209,118 @@ public class StateMachineTestPlanBuilder<S, E> {
 			return this;
 		}
 
+		/**
+		 * Expect state enter happening {@code count} times.
+		 *
+		 * @param count the count
+		 * @return the state machine test plan step builder
+		 */
+		public StateMachineTestPlanStepBuilder expectStateEntered(int count) {
+			if (count < 0) {
+				throw new IllegalArgumentException("Expected count cannot be negative, was " + count);
+			}
+			this.expectStateEntered = count;
+			return this;
+		}
+
+		/**
+		 * Expect state exit happening {@code count} times.
+		 *
+		 * @param count the count
+		 * @return the state machine test plan step builder
+		 */
+		public StateMachineTestPlanStepBuilder expectStateExited(int count) {
+			if (count < 0) {
+				throw new IllegalArgumentException("Expected count cannot be negative, was " + count);
+			}
+			this.expectStateExited = count;
+			return this;
+		}
+
+		/**
+		 * Expect event not accepter happening {@code count} times.
+		 *
+		 * @param count the count
+		 * @return the state machine test plan step builder
+		 */
+		public StateMachineTestPlanStepBuilder expectEventNotAccepted(int count) {
+			if (count < 0) {
+				throw new IllegalArgumentException("Expected count cannot be negative, was " + count);
+			}
+			this.expectEventNotAccepted = count;
+			return this;
+		}
+
+		/**
+		 * Expect transition happening {@code count} times.
+		 *
+		 * @param count the count
+		 * @return the state machine test plan step builder
+		 */
+		public StateMachineTestPlanStepBuilder expectTransition(int count) {
+			if (count < 0) {
+				throw new IllegalArgumentException("Expected count cannot be negative, was " + count);
+			}
+			this.expectTransition = count;
+			return this;
+		}
+
+		/**
+		 * Expect transition start happening {@code count} times.
+		 *
+		 * @param count the count
+		 * @return the state machine test plan step builder
+		 */
+		public StateMachineTestPlanStepBuilder expectTransitionStarted(int count) {
+			if (count < 0) {
+				throw new IllegalArgumentException("Expected count cannot be negative, was " + count);
+			}
+			this.expectTransitionStarted = count;
+			return this;
+		}
+
+		/**
+		 * Expect transition end happening {@code count} times.
+		 *
+		 * @param count the count
+		 * @return the state machine test plan step builder
+		 */
+		public StateMachineTestPlanStepBuilder expectTransitionEnded(int count) {
+			if (count < 0) {
+				throw new IllegalArgumentException("Expected count cannot be negative, was " + count);
+			}
+			this.expectTransitionEnded = count;
+			return this;
+		}
+
+		/**
+		 * Expect state machine start happening {@code count} times.
+		 *
+		 * @param count the count
+		 * @return the state machine test plan step builder
+		 */
+		public StateMachineTestPlanStepBuilder expectStateMachineStarted(int count) {
+			if (count < 0) {
+				throw new IllegalArgumentException("Expected count cannot be negative, was " + count);
+			}
+			this.expectStateMachineStarted = count;
+			return this;
+		}
+
+		/**
+		 * Expect state machine stop happening {@code count} times.
+		 *
+		 * @param count the count
+		 * @return the state machine test plan step builder
+		 */
+		public StateMachineTestPlanStepBuilder expectStateMachineStopped(int count) {
+			if (count < 0) {
+				throw new IllegalArgumentException("Expected count cannot be negative, was " + count);
+			}
+			this.expectStateMachineStopped = count;
+			return this;
+		}
+
 		/**
 		 * Add a new step and return {@link StateMachineTestPlanBuilder}
 		 * for chaining.
@@ -123,7 +328,10 @@ public class StateMachineTestPlanBuilder<S, E> {
 		 * @return the state machine test plan builder for chaining
 		 */
 		public StateMachineTestPlanBuilder<S, E> and() {
-			steps.add(new StateMachineTestPlanStep<S, E>(sendEvent, expectState, expectStateChanged));
+			steps.add(new StateMachineTestPlanStep<S, E>(sendEvent, sendEventMachineId, expectStates,
+					expectStateChanged, expectStateEntered, expectStateExited, expectEventNotAccepted,
+					expectTransition, expectTransitionStarted, expectTransitionEnded, expectStateMachineStarted,
+					expectStateMachineStopped, expectVariableKeys, expectVariables));
 			return StateMachineTestPlanBuilder.this;
 		}
 
@@ -131,12 +339,39 @@ public class StateMachineTestPlanBuilder<S, E> {
 
 	static class StateMachineTestPlanStep<S, E> {
 		E sendEvent;
-		S expectState;
+		Object sendEventMachineId;
+		final Collection<S> expectStates;
 		Integer expectStateChanged;
-		public StateMachineTestPlanStep(E sendEvent, S expectState, Integer expectStateChanged) {
+		Integer expectStateEntered;
+		Integer expectStateExited;
+		Integer expectEventNotAccepted;
+		Integer expectTransition;
+		Integer expectTransitionStarted;
+		Integer expectTransitionEnded;
+		Integer expectStateMachineStarted;
+		Integer expectStateMachineStopped;
+		final Collection<Object> expectVariableKeys;
+		final Map<Object, Object> expectVariables;
+
+		public StateMachineTestPlanStep(E sendEvent, Object sendEventMachineId, Collection<S> expectStates, Integer expectStateChanged,
+				Integer expectStateEntered, Integer expectStateExited, Integer expectEventNotAccepted,
+				Integer expectTransition, Integer expectTransitionStarted, Integer expectTransitionEnded,
+				Integer expectStateMachineStarted, Integer expectStateMachineStopped,
+				Collection<Object> expectVariableKeys, Map<Object, Object> expectVariables) {
 			this.sendEvent = sendEvent;
-			this.expectState = expectState;
+			this.sendEventMachineId = sendEventMachineId;
+			this.expectStates = expectStates;
 			this.expectStateChanged = expectStateChanged;
+			this.expectStateEntered = expectStateEntered;
+			this.expectStateExited = expectStateExited;
+			this.expectEventNotAccepted = expectEventNotAccepted;
+			this.expectTransition = expectTransition;
+			this.expectTransitionStarted = expectTransitionStarted;
+			this.expectTransitionEnded = expectTransitionEnded;
+			this.expectStateMachineStarted = expectStateMachineStarted;
+			this.expectStateMachineStopped = expectStateMachineStopped;
+			this.expectVariableKeys = expectVariableKeys;
+			this.expectVariables = expectVariables;
 		}
 
 	}

spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java

@@ -32,6 +32,7 @@ import org.apache.curator.framework.imps.CuratorFrameworkState;
 import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex;
 import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode;
 import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode.Mode;
+import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.data.Stat;
 import org.springframework.statemachine.StateMachine;
@@ -182,9 +183,18 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 			if (stateWrapper != null) {
 				stat.setVersion(stateWrapper.version);
 			}
+			if (log.isDebugEnabled()) {
+				log.debug("Requesting persist write " + context + " with version " + stat.getVersion() + " for ensemble " + uuid);
+			}
 			persist.write(context, stat);
 			stateRef.set(new StateWrapper(context, stat.getVersion()));
 		} catch (Exception e) {
+			if (e instanceof StateMachineException) {
+				if (((StateMachineException)e).contains(KeeperException.BadVersionException.class)) {
+					notifyError(new StateMachineEnsembleException("Cas error during write id=[" + uuid
+							+ "] for context=[" + context + "]", e));
+				}
+			}
 			throw new StateMachineException("Error persisting data", e);
 		}
 	}
@@ -275,9 +285,16 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 	private void mayNotifyStateChanged(StateWrapper wrapper) {
 		StateWrapper notifyWrapper = notifyRef.get();
 		if (notifyWrapper == null) {
+			if (log.isDebugEnabled()) {
+				log.debug("notifyWrapper null, wrapper=[" + wrapper + "] for " + this);
+			}
 			notifyRef.set(wrapper);
 			notifyStateChanged(wrapper.context);
 		} else if (wrapper.version > notifyWrapper.version) {
+			if (log.isDebugEnabled()) {
+				log.debug("Wrapper version higher that notifyWrapper version, notifyWrapper=[" + notifyWrapper
+						+ "], wrapper=[" + wrapper + "] for " + this);
+			}
 			notifyRef.set(wrapper);
 			notifyStateChanged(wrapper.context);
 		}
@@ -285,11 +302,16 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 
 	private void traceLogWrappers(StateWrapper currentWrapper, StateWrapper notifyWrapper, StateWrapper newWrapper) {
 		if (log.isTraceEnabled()) {
-			log.trace("Wrappers \ncurrentWrapper=[" + currentWrapper + "] \nnotifyWrapper=[" + notifyWrapper
-					+ "] \nnewWrapper=[" + newWrapper + "]");
+			log.trace("Wrappers id=" + uuid + "\ncurrentWrapper=[" + currentWrapper + "] \nnotifyWrapper=["
+					+ notifyWrapper + "] \nnewWrapper=[" + newWrapper + "]");
 		}
 	}
 
+	@Override
+	public String toString() {
+		return "ZookeeperStateMachineEnsemble [uuid=" + uuid + "]";
+	}
+
 	private class StateWatcher implements CuratorWatcher {
 
 		// zk is not really reliable for watching events because
@@ -304,7 +326,7 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 		@Override
 		public void process(WatchedEvent event) throws Exception {
 			if (log.isTraceEnabled()) {
-				log.trace("Process WatchedEvent: " + event);
+				log.trace("Process WatchedEvent: id=" + uuid + " " + event);
 			}
 			switch (event.getType()) {
 			case NodeDataChanged:
@@ -333,14 +355,25 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 								// check if we're behind more than a log size meaning we can't
 								// replay full history, notify and break out from a loop
 								if (i + logSize < ver) {
-									notifyError(new StateMachineEnsembleException("Current version behind more that log size"));
+									notifyError(new StateMachineEnsembleException("Current version behind more than log size"));
 									break;
 								}
 								if (log.isDebugEnabled()) {
 									log.debug("Replay position " + i + " with version " + ver);
 									log.debug("Context in position " + i + " " + context);
 								}
+
 								StateWrapper wrapper = new StateWrapper(context, ver);
+
+								// need to set stateRef when replaying if its
+								// context is not set or otherwise just set
+								// if stateRef still is currentWrapper
+								StateWrapper currentWrapperx = stateRef.get();
+								if (currentWrapperx.context == null) {
+									stateRef.set(wrapper);
+								} else if (wrapper.version == currentWrapperx.version + 1){
+									stateRef.set(wrapper);
+								}
 								mayNotifyStateChanged(wrapper);
 							} catch (Exception e) {
 								log.error("error reading log", e);

spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java

@@ -17,6 +17,7 @@ package org.springframework.statemachine.zookeeper;
 
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
 import static org.junit.Assert.assertThat;
@@ -29,12 +30,14 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.curator.framework.CuratorFramework;
+import org.apache.zookeeper.KeeperException;
 import org.junit.Test;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.messaging.Message;
 import org.springframework.statemachine.ExtendedState;
 import org.springframework.statemachine.StateMachine;
 import org.springframework.statemachine.StateMachineContext;
+import org.springframework.statemachine.StateMachineException;
 import org.springframework.statemachine.access.StateMachineAccessor;
 import org.springframework.statemachine.ensemble.EnsembleListeger;
 import org.springframework.statemachine.ensemble.StateMachineEnsembleException;
@@ -229,6 +232,146 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests {
 		}
 	}
 
+	@Test
+	public void testContextEventsNotMissedBurstNoOverflow2() throws Exception {
+		context.register(ZkServerConfig.class, BaseConfig.class);
+		context.refresh();
+
+		CuratorFramework curatorClient =
+				context.getBean("curatorClient", CuratorFramework.class);
+
+		ZookeeperStateMachineEnsemble<String, String> ensemble1 =
+				new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
+		ZookeeperStateMachineEnsemble<String, String> ensemble2 =
+				new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
+
+		TestEnsembleListener listener1 = new TestEnsembleListener();
+		ensemble1.addEnsembleListener(listener1);
+		TestEnsembleListener listener2 = new TestEnsembleListener();
+		ensemble2.addEnsembleListener(listener2);
+
+		ensemble1.afterPropertiesSet();
+		ensemble1.start();
+		ensemble2.afterPropertiesSet();
+		ensemble2.start();
+
+		listener1.reset(0, 10);
+		listener2.reset(0, 10);
+
+		for (int i = 0; i < 10; i++) {
+			ensemble1.setState(new DefaultStateMachineContext<String, String>("S" + i, "E" + i,
+					new HashMap<String, Object>(), new DefaultExtendedState()));
+		}
+
+		assertThat(listener1.eventLatch.await(3, TimeUnit.SECONDS), is(true));
+		assertThat(listener1.events.size(), is(10));
+		assertThat(listener2.eventLatch.await(3, TimeUnit.SECONDS), is(true));
+		assertThat(listener2.events.size(), is(10));
+
+		for (int i = 0; i < 10; i++) {
+			assertThat(listener1.events.get(i).getEvent(), is("E" + i));
+			assertThat(listener2.events.get(i).getEvent(), is("E" + i));
+		}
+	}
+
+	@Test
+	public void testContextEventsNotMissedBurstNoOverflow3() throws Exception {
+		context.register(ZkServerConfig.class, BaseConfig.class);
+		context.refresh();
+
+		CuratorFramework curatorClient =
+				context.getBean("curatorClient", CuratorFramework.class);
+
+		ZookeeperStateMachineEnsemble<String, String> ensemble1 =
+				new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
+		ZookeeperStateMachineEnsemble<String, String> ensemble2 =
+				new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
+
+		TestEnsembleListener listener1 = new TestEnsembleListener();
+		ensemble1.addEnsembleListener(listener1);
+		TestEnsembleListener listener2 = new TestEnsembleListener();
+		ensemble2.addEnsembleListener(listener2);
+
+		ensemble1.afterPropertiesSet();
+		ensemble1.start();
+		ensemble2.afterPropertiesSet();
+		ensemble2.start();
+
+		listener1.reset(0, 10);
+		listener2.reset(0, 10);
+
+		Exception e = null;
+		try {
+			for (int i = 0; i < 10; i++) {
+				if (((i % 2) == 0)) {
+					ensemble1.setState(new DefaultStateMachineContext<String, String>("S" + i, "E" + i,
+							new HashMap<String, Object>(), new DefaultExtendedState()));
+				} else {
+					ensemble2.setState(new DefaultStateMachineContext<String, String>("S" + i, "E" + i,
+							new HashMap<String, Object>(), new DefaultExtendedState()));
+				}
+			}
+		} catch (Exception ee) {
+			e = ee;
+		}
+
+		if (e != null) {
+			assertThat(e, instanceOf(StateMachineException.class));
+			assertThat(((StateMachineException)e).contains(KeeperException.BadVersionException.class), is(true));
+		} else {
+			// miracle happened and no cas error, well then check events
+			assertThat(listener1.eventLatch.await(3, TimeUnit.SECONDS), is(true));
+			assertThat(listener1.events.size(), is(10));
+			assertThat(listener2.eventLatch.await(3, TimeUnit.SECONDS), is(true));
+			assertThat(listener2.events.size(), is(10));
+
+			for (int i = 0; i < 10; i++) {
+				assertThat(listener1.events.get(i).getEvent(), is("E" + i));
+				assertThat(listener2.events.get(i).getEvent(), is("E" + i));
+			}
+		}
+	}
+
+	@Test
+	public void testContextEventsNotMissedBurstNoOverflow4() throws Exception {
+		context.register(ZkServerConfig.class, BaseConfig.class);
+		context.refresh();
+
+		CuratorFramework curatorClient =
+				context.getBean("curatorClient", CuratorFramework.class);
+
+		ZookeeperStateMachineEnsemble<String, String> ensemble1 =
+				new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
+		ZookeeperStateMachineEnsemble<String, String> ensemble2 =
+				new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
+
+		TestEnsembleListener listener1 = new TestEnsembleListener();
+		ensemble1.addEnsembleListener(listener1);
+		TestEnsembleListener listener2 = new TestEnsembleListener();
+		ensemble2.addEnsembleListener(listener2);
+
+		ensemble1.afterPropertiesSet();
+		ensemble1.start();
+		ensemble2.afterPropertiesSet();
+		ensemble2.start();
+
+		for (int i = 0; i < 10; i++) {
+			listener1.reset(0, 1);
+			listener2.reset(0, 1);
+			if (((i % 2) == 0)) {
+				ensemble1.setState(new DefaultStateMachineContext<String, String>("S" + i, "E" + i,
+						new HashMap<String, Object>(), new DefaultExtendedState()));
+			} else {
+				ensemble2.setState(new DefaultStateMachineContext<String, String>("S" + i, "E" + i,
+						new HashMap<String, Object>(), new DefaultExtendedState()));
+			}
+			assertThat(listener1.eventLatch.await(3, TimeUnit.SECONDS), is(true));
+			assertThat(listener1.events.size(), is(1));
+			assertThat(listener2.eventLatch.await(3, TimeUnit.SECONDS), is(true));
+			assertThat(listener2.events.size(), is(1));
+		}
+	}
+
 	@Test
 	public void testContextEventsNotMissedSlowNoOverflow() throws Exception {
 		context.register(ZkServerConfig.class, BaseConfig.class);
@@ -261,6 +404,108 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests {
 		}
 	}
 
+	@Test
+	public void testContextEventsNotMissedSlowNoOverflow2() throws Exception {
+		context.register(ZkServerConfig.class, BaseConfig.class);
+		context.refresh();
+
+		CuratorFramework curatorClient =
+				context.getBean("curatorClient", CuratorFramework.class);
+
+		ZookeeperStateMachineEnsemble<String, String> ensemble1 =
+				new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
+		ZookeeperStateMachineEnsemble<String, String> ensemble2 =
+				new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
+
+		TestEnsembleListener listener1 = new TestEnsembleListener();
+		TestEnsembleListener listener2 = new TestEnsembleListener();
+		ensemble1.addEnsembleListener(listener1);
+		ensemble2.addEnsembleListener(listener2);
+
+		ensemble1.afterPropertiesSet();
+		ensemble1.start();
+		ensemble2.afterPropertiesSet();
+		ensemble2.start();
+		listener1.reset(0, 10);
+		listener2.reset(0, 10);
+
+		for (int i = 0; i < 10; i++) {
+			ensemble1.setState(new DefaultStateMachineContext<String, String>("S" + i, "E" + i,
+					new HashMap<String, Object>(), new DefaultExtendedState()));
+			Thread.sleep(500);
+		}
+
+		assertThat(listener1.eventLatch.await(3, TimeUnit.SECONDS), is(true));
+		assertThat(listener1.events.size(), is(10));
+		assertThat(listener2.eventLatch.await(3, TimeUnit.SECONDS), is(true));
+		assertThat(listener2.events.size(), is(10));
+
+		for (int i = 0; i < 10; i++) {
+			assertThat(listener1.events.get(i).getEvent(), is("E" + i));
+			assertThat(listener2.events.get(i).getEvent(), is("E" + i));
+		}
+	}
+
+	@Test
+	public void testDoesNotThrowCasError() throws Exception {
+		context.register(ZkServerConfig.class, BaseConfig.class);
+		context.refresh();
+
+		CuratorFramework curatorClient =
+				context.getBean("curatorClient", CuratorFramework.class);
+
+		ZookeeperStateMachineEnsemble<String, String> ensemble1 =
+				new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
+		ZookeeperStateMachineEnsemble<String, String> ensemble2 =
+				new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
+
+		TestEnsembleListener listener1 = new TestEnsembleListener();
+		TestEnsembleListener listener2 = new TestEnsembleListener();
+		ensemble1.addEnsembleListener(listener1);
+		ensemble2.addEnsembleListener(listener2);
+
+		ensemble1.afterPropertiesSet();
+		ensemble1.start();
+		ensemble2.afterPropertiesSet();
+		ensemble2.start();
+		listener1.reset(0, 9);
+		listener2.reset(0, 9);
+
+		for (int i = 0; i < 9; i++) {
+			ensemble1.setState(new DefaultStateMachineContext<String, String>("S" + i, "E" + i,
+					new HashMap<String, Object>(), new DefaultExtendedState()));
+		}
+
+		assertThat(listener1.eventLatch.await(3, TimeUnit.SECONDS), is(true));
+		assertThat(listener1.events.size(), is(9));
+		assertThat(listener2.eventLatch.await(3, TimeUnit.SECONDS), is(true));
+		assertThat(listener2.events.size(), is(9));
+
+		for (int i = 0; i < 9; i++) {
+			assertThat(listener1.events.get(i).getEvent(), is("E" + i));
+			assertThat(listener2.events.get(i).getEvent(), is("E" + i));
+		}
+
+		listener1.reset(0, 1);
+		listener2.reset(0, 1);
+
+		// should not throw BadVersionException when we immediately
+		for (int i = 9; i < 10; i++) {
+			ensemble2.setState(new DefaultStateMachineContext<String, String>("S" + i, "E" + i,
+					new HashMap<String, Object>(), new DefaultExtendedState()));
+		}
+
+		assertThat(listener1.eventLatch.await(3, TimeUnit.SECONDS), is(true));
+		assertThat(listener1.events.size(), is(1));
+		assertThat(listener2.eventLatch.await(3, TimeUnit.SECONDS), is(true));
+		assertThat(listener2.events.size(), is(1));
+
+		for (int i = 0; i < 1; i++) {
+			assertThat(listener1.events.get(i).getEvent(), is("E" + (i+9)));
+			assertThat(listener2.events.get(i).getEvent(), is("E" + (i+9)));
+		}
+	}
+
 	@Test
 	public void testEventsOverflow() throws Exception {
 		context.register(ZkServerConfig.class, BaseConfig.class);

spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineTests.java

@@ -19,6 +19,7 @@ import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.assertThat;
 
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -29,14 +30,19 @@ import org.springframework.context.SmartLifecycle;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.task.SyncTaskExecutor;
+import org.springframework.statemachine.StateContext;
 import org.springframework.statemachine.StateMachine;
+import org.springframework.statemachine.action.Action;
 import org.springframework.statemachine.config.EnableStateMachine;
+import org.springframework.statemachine.config.StateMachineBuilder;
 import org.springframework.statemachine.config.StateMachineConfigurerAdapter;
 import org.springframework.statemachine.config.builders.StateMachineConfigurationConfigurer;
 import org.springframework.statemachine.config.builders.StateMachineStateConfigurer;
 import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer;
 import org.springframework.statemachine.ensemble.DistributedStateMachine;
 import org.springframework.statemachine.ensemble.StateMachineEnsemble;
+import org.springframework.statemachine.guard.Guard;
 import org.springframework.statemachine.listener.StateMachineListenerAdapter;
 import org.springframework.statemachine.state.State;
 import org.springframework.statemachine.test.StateMachineTestPlan;
@@ -111,6 +117,67 @@ public class ZookeeperStateMachineTests extends AbstractZookeeperTests {
 		assertThat(machine2.getState().getIds(), containsInAnyOrder("S2"));
 	}
 
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testStateChangesManualSetupSendDifferentMachines() throws Exception {
+		context.register(ZkServerConfig.class, BaseConfig.class, Config1.class, Config2.class);
+		context.refresh();
+
+		StateMachine<String, String> machine1 =
+				context.getBean("sm1", StateMachine.class);
+		StateMachine<String, String> machine2 =
+				context.getBean("sm2", StateMachine.class);
+
+		TestListener listener1 = new TestListener();
+		TestListener listener2 = new TestListener();
+		machine1.addStateListener(listener1);
+		machine2.addStateListener(listener2);
+
+		CuratorFramework curatorClient =
+				context.getBean("curatorClient", CuratorFramework.class);
+
+		ZookeeperStateMachineEnsemble<String, String> ensemble1 =
+				new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
+		ZookeeperStateMachineEnsemble<String, String> ensemble2 =
+				new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
+		ensemble1.afterPropertiesSet();
+		ensemble2.afterPropertiesSet();
+		ensemble1.start();
+		ensemble2.start();
+
+		DistributedStateMachine<String, String> machine1s =
+				new DistributedStateMachine<String, String>(ensemble1, machine1);
+
+		DistributedStateMachine<String, String> machine2s =
+				new DistributedStateMachine<String, String>(ensemble2, machine2);
+
+		machine1s.afterPropertiesSet();
+		machine2s.afterPropertiesSet();
+
+		machine1s.start();
+		machine2s.start();
+
+		listener1.reset(1);
+		listener2.reset(1);
+		machine1s.sendEvent("E1");
+		assertThat(listener1.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true));
+		assertThat(listener1.stateChangedCount, is(1));
+		assertThat(listener2.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true));
+		assertThat(listener2.stateChangedCount, is(1));
+		assertThat(machine1.getState().getIds(), containsInAnyOrder("S1"));
+		assertThat(machine2.getState().getIds(), containsInAnyOrder("S1"));
+
+		listener1.reset(1);
+		listener2.reset(1);
+		machine2s.sendEvent("E2");
+		assertThat(listener1.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true));
+		assertThat(listener1.stateChangedCount, is(1));
+		assertThat(listener2.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true));
+		assertThat(listener2.stateChangedCount, is(1));
+		assertThat(machine1.getState().getIds(), containsInAnyOrder("S2"));
+		assertThat(machine2.getState().getIds(), containsInAnyOrder("S2"));
+	}
+
 	@Test
 	@SuppressWarnings("unchecked")
 	public void testLifecycle() throws Exception {
@@ -151,6 +218,45 @@ public class ZookeeperStateMachineTests extends AbstractZookeeperTests {
 		plan.test();
 	}
 
+	@Test
+	public void testVariousChangesInShowcase() throws Exception {
+		context.register(ZkServerConfig.class, BaseConfig.class);
+		context.refresh();
+
+		CuratorFramework curatorClient =
+				context.getBean("curatorClient", CuratorFramework.class);
+
+		StateMachine<String, String> machine1 =
+				buildTestStateMachine(curatorClient);
+
+		StateMachine<String, String> machine2 =
+				buildTestStateMachine(curatorClient);
+
+		StateMachineTestPlan<String, String> plan =
+				StateMachineTestPlanBuilder.<String, String>builder()
+					.defaultAwaitTime(2)
+					.stateMachine(machine1)
+					.stateMachine(machine2)
+					.step()
+						.expectStates("S0", "S1", "S11")
+						.expectVariable("foo")
+						.expectVariable("foo", 0)
+						.and()
+					.step()
+						.sendEvent("C", machine1)
+						.expectStateChanged(3)
+						.expectStates("S0", "S2", "S21", "S211")
+						.and()
+					.step()
+						.sendEvent("C", machine2)
+						.expectStateChanged(2)
+						.expectStates("S0", "S1", "S11")
+						.and()
+					.build();
+
+		plan.test();
+	}
+
 	@Test
 	@SuppressWarnings("unchecked")
 	public void testJoinLaterShouldSyncState() throws Exception {
@@ -330,4 +436,161 @@ public class ZookeeperStateMachineTests extends AbstractZookeeperTests {
 
 	}
 
+	private static StateMachineEnsemble<String, String> stateMachineEnsemble(CuratorFramework curatorClient) {
+		ZookeeperStateMachineEnsemble<String, String> ensemble = new ZookeeperStateMachineEnsemble<String, String>(
+				curatorClient, "/foo");
+		ensemble.afterPropertiesSet();
+		ensemble.start();
+		return ensemble;
+	}
+
+	private StateMachine<String, String> buildTestStateMachine(CuratorFramework curatorClient)
+			throws Exception {
+		StateMachineBuilder.Builder<String, String> builder = StateMachineBuilder.builder();
+
+		builder.configureConfiguration()
+			.withConfiguration()
+				.taskExecutor(new SyncTaskExecutor())
+				.autoStartup(true)
+				.and()
+			.withDistributed()
+				.ensemble(stateMachineEnsemble(curatorClient));
+
+		builder.configureStates()
+				.withStates()
+					.initial("S0", fooAction())
+					.state("S0")
+					.and()
+					.withStates()
+						.parent("S0")
+						.initial("S1")
+						.state("S1")
+						.and()
+						.withStates()
+							.parent("S1")
+							.initial("S11")
+							.state("S11")
+							.state("S12")
+							.and()
+					.withStates()
+						.parent("S0")
+						.state("S2")
+						.and()
+						.withStates()
+							.parent("S2")
+							.initial("S21")
+							.state("S21")
+							.and()
+							.withStates()
+								.parent("S21")
+								.initial("S211")
+								.state("S211")
+								.state("S212");
+
+		builder.configureTransitions()
+				.withExternal()
+					.source("S1").target("S1").event("A")
+					.guard(foo1Guard())
+					.and()
+				.withExternal()
+					.source("S1").target("S11").event("B")
+					.and()
+				.withExternal()
+					.source("S21").target("S211").event("B")
+					.and()
+				.withExternal()
+					.source("S1").target("S2").event("C")
+					.and()
+				.withExternal()
+					.source("S2").target("S1").event("C")
+					.and()
+				.withExternal()
+					.source("S1").target("S0").event("D")
+					.and()
+				.withExternal()
+					.source("S211").target("S21").event("D")
+					.and()
+				.withExternal()
+					.source("S0").target("S211").event("E")
+					.and()
+				.withExternal()
+					.source("S1").target("S211").event("F")
+					.and()
+				.withExternal()
+					.source("S2").target("S11").event("F")
+					.and()
+				.withExternal()
+					.source("S11").target("S211").event("G")
+					.and()
+				.withExternal()
+					.source("S211").target("S0").event("G")
+					.and()
+				.withInternal()
+					.source("S0").event("H")
+					.guard(foo0Guard())
+					.action(fooAction())
+					.and()
+				.withInternal()
+					.source("S2").event("H")
+					.guard(foo1Guard())
+					.action(fooAction())
+					.and()
+				.withInternal()
+					.source("S1").event("H")
+					.and()
+				.withExternal()
+					.source("S11").target("S12").event("I")
+					.and()
+				.withExternal()
+					.source("S211").target("S212").event("I")
+					.and()
+				.withExternal()
+					.source("S12").target("S212").event("I");
+
+		return builder.build();
+	}
+
+	private static FooGuard foo0Guard() {
+		return new FooGuard(0);
+	}
+
+	private static FooGuard foo1Guard() {
+		return new FooGuard(1);
+	}
+
+	private static FooAction fooAction() {
+		return new FooAction();
+	}
+
+	private static class FooGuard implements Guard<String, String> {
+
+		private final int match;
+
+		public FooGuard(int match) {
+			this.match = match;
+		}
+
+		@Override
+		public boolean evaluate(StateContext<String, String> context) {
+			Object foo = context.getExtendedState().getVariables().get("foo");
+			return !(foo == null || !foo.equals(match));
+		}
+	}
+
+	private static class FooAction implements Action<String, String> {
+
+		@Override
+		public void execute(StateContext<String, String> context) {
+			Map<Object, Object> variables = context.getExtendedState().getVariables();
+			Integer foo = context.getExtendedState().get("foo", Integer.class);
+			if (foo == null) {
+				variables.put("foo", 0);
+			} else if (foo == 0) {
+				variables.put("foo", 1);
+			} else if (foo == 1) {
+				variables.put("foo", 0);
+			}
+		}
+	}
+
 }