Bugfix/task scheduler lifecycle

[RenanSFreitas]

ConcurrentTaskScheduler doesn't expose any way for it's inner executor to be shutdown, when it's instance is created with the default constructor. This can lead to a scenario where a library user ends up with many unused threads with no clear cues about where they came from. The unmanaged creation of those threads can result in an OutOfMemoryError for a client application.

This commit replaces the usage of ConcurrentTaskScheduler with ThreadPoolTaskScheduler, a type of similar semantics but better API, which includes methods for managing its internal resources lifecycle. Also it implements spring-beans interfaces which makes it more suitable to be managed by a Spring application context.

• Fixes https://github.com/spring-projects/spring-statemachine/issues/624

[RenanSFreitas]

This pull requests aims to fix the scenario exemplified in the code below, where usage of spring-statemachine 2.5.1 leads to a hanging JVM with many spawned threads. Increasing the number of iterations in the main loop results in more threads being created, which can result in an OOM scenario.

package com.example.springstatemachinethreadpoolbug;

import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.statemachine.StateMachine;
import org.springframework.statemachine.config.StateMachineBuilder;
import org.springframework.statemachine.config.configurers.StateConfigurer;
import org.springframework.statemachine.support.DefaultStateMachineContext;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;

@SpringBootApplication
public class Application implements CommandLineRunner {
	enum StateMachineEvent {
		E0, E1, E2
	}

	enum StateMachineState {
		S0, S1, S2
	}
	static final EnumSet<StateMachineState> ALL_STATES = EnumSet.allOf(StateMachineState.class);

	public static void main(String[] args) {
		final SpringApplication app = new SpringApplication(Application.class);
		app.run();
	}

	@Override
	public void run(final String... args) {
		runTestCase();
	}

	private class Iteration {
		StateMachineState iterationInitialState;
		StateMachineEvent iterationInitialEvent;
		StateMachineEvent iterationEventToPush;

		public Iteration(final StateMachineState iterationInitialState,
						 final StateMachineEvent iterationInitialEvent,
						 final StateMachineEvent iterationEventToPush) {
			this.iterationInitialState = iterationInitialState;
			this.iterationInitialEvent = iterationInitialEvent;
			this.iterationEventToPush = iterationEventToPush;
		}
	}
	private void runTestCase() {
		final var beforeThreadCount = Thread.activeCount();

		// Do the same thing a few times, to demonstrate the increase in the thread count
		for (int i = 0; i < 100; i++) {
			for (final var iteration : List.of(
					new Iteration(StateMachineState.S0, StateMachineEvent.E0, StateMachineEvent.E0),
					new Iteration(StateMachineState.S1, StateMachineEvent.E1, StateMachineEvent.E1)
			)) {
				final var iterationInitialState = iteration.iterationInitialState;
				final var iterationInitialEvent = iteration.iterationInitialEvent;
				final var stateMachine = buildStateMachine(iterationInitialState, iterationInitialEvent);
				stateMachine.start();
				stateMachine.sendEvent(MessageBuilder.withPayload(iteration.iterationEventToPush)
						.setHeader("HEADER_A", "HEADER_A_VALUE")
						.build());
				stateMachine.stop();
			}
		}

		System.out.println("Before thread count: " + beforeThreadCount);
		System.out.println("After thread count: " + Thread.activeCount());
	}

	public StateMachine<StateMachineState, StateMachineEvent> buildStateMachine(StateMachineState state, StateMachineEvent event) {
		try {
			final StateMachineBuilder.Builder<StateMachineState, StateMachineEvent> builder = buildStateMachine();

			final StateConfigurer<StateMachineState, StateMachineEvent> stateConfigurer = builder.configureStates()
					.withStates()
					.initial(StateMachineState.S0)
					.states(ALL_STATES)
					.end(StateMachineState.S2);

			builder.configureTransitions()
					.withExternal()
					.source(StateMachineState.S0)
					.target(StateMachineState.S1)
					.event(StateMachineEvent.E0)
					.and()
					.withExternal()
					.source(StateMachineState.S1)
					.target(StateMachineState.S2)
					.event(StateMachineEvent.E1);

			stateConfigurer.state(StateMachineState.S2, context -> System.out.println("State machine action at " + StateMachineState.S2));
			// it happens with stateDo too
//			stateConfigurer.stateDo(StateMachineState.STATE_2, context -> System.out.println("State machine action at " + StateMachineState.S2));

			final var stateMachine = builder.build();

			stateMachine.getStateMachineAccessor()
					.doWithAllRegions((function) -> function.resetStateMachine(new DefaultStateMachineContext<>(state,
							event,
							Map.of(),
							stateMachine.getExtendedState(),
							null,
							stateMachine.getId()))
					);

			return stateMachine;
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

	private StateMachineBuilder.Builder<StateMachineState, StateMachineEvent> buildStateMachine()
			throws Exception {
		final StateMachineBuilder.Builder<StateMachineState, StateMachineEvent> result = StateMachineBuilder.builder();

		result.configureConfiguration()
				.withConfiguration()
				.autoStartup(false)
				.machineId("STATE_MACHINE_ID");

		return result;
	}
}

[pdalfarr]

@RenanSFreitas Did you receive any answer about this PR ?