docs: public static RestartStrategyFactory createRestartStrategyFactory(Configuration configuration) throws Exception {
  		String restartStrategyName = configuration.getString(ConfigConstants.RESTART_STRATEGY, null);
  
  		if (restartStrategyName == null) {
  			// support deprecated ConfigConstants values
  			final int numberExecutionRetries = configuration.getInteger(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS,
  				ConfigConstants.DEFAULT_EXECUTION_RETRIES);
  			String pauseString = configuration.getString(AkkaOptions.WATCH_HEARTBEAT_PAUSE);
  			String delayString = configuration.getString(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY,
  				pauseString);
  
  			long delay;
  
  			try {
  				delay = Duration.apply(delayString).toMillis();
  			} catch (NumberFormatException nfe) {
  				if (delayString.equals(pauseString)) {
  					throw new Exception("Invalid config value for " +
  						AkkaOptions.WATCH_HEARTBEAT_PAUSE.key() + ": " + pauseString +
  						". Value must be a valid duration (such as '10 s' or '1 min')");
  				} else {
  					throw new Exception("Invalid config value for " +
  						ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY + ": " + delayString +
  						". Value must be a valid duration (such as '100 milli' or '10 s')");
  				}
  			}
  
  			if (numberExecutionRetries > 0 && delay >= 0) {
  				return new FixedDelayRestartStrategy.FixedDelayRestartStrategyFactory(numberExecutionRetries, delay);
  			} else {
  				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();
  		}
  	}