class RestartStrategyFactory implements Serializable {

	public static RestartStrategyFactory createRestartStrategyFactory(Configuration configuration) throws Exception {
		String restartStrategyName = configuration.getString(RestartStrategyOptions.RESTART_STRATEGY, null);

		if (restartStrategyName == null) {
			return new NoOrFixedIfCheckpointingEnabledRestartStrategyFactory();
		}

		switch (restartStrategyName.toLowerCase()) {
			case "none":
			case "off":
			case "disable":
				return NoRestartStrategy.createFactory(configuration);
			case "fixeddelay":
			case "fixed-delay":
				return FixedDelayRestartStrategy.createFactory(configuration);
			case "failurerate":
			case "failure-rate":
				return FailureRateRestartStrategy.createFactory(configuration);
			default:
				try {
					Class<?> clazz = Class.forName(restartStrategyName);

					if (clazz != null) {
						Method method = clazz.getMethod(CREATE_METHOD, Configuration.class);

						if (method != null) {
							Object result = method.invoke(null, configuration);

							if (result != null) {
								return (RestartStrategyFactory) result;
							}
						}
					}
				} catch (ClassNotFoundException cnfe) {
					LOG.warn("Could not find restart strategy class {}.", restartStrategyName);
				} catch (NoSuchMethodException nsme) {
					LOG.warn("Class {} does not has static method {}.", restartStrategyName, CREATE_METHOD);
				} catch (InvocationTargetException ite) {
					LOG.warn("Cannot call static method {} from class {}.", CREATE_METHOD, restartStrategyName);
				} catch (IllegalAccessException iae) {
					LOG.warn("Illegal access while calling method {} from class {}.", CREATE_METHOD, restartStrategyName);
				}

				// fallback in case of an error
				return new NoOrFixedIfCheckpointingEnabledRestartStrategyFactory();
		}
	}

}