class CheckpointSchedulerAction implements ISchedulerAction {

    @Override
    public void init(final ISchedulerEntry entry, final IObject args)
            throws SchedulerActionInitializationException {
        try {
            IObject recoverConfig = (IObject) args.getValue(recoverFieldName);
            String recoverStrategyId = (String) recoverConfig.getValue(strategyFieldName);
            IObject message = (IObject) args.getValue(messageFieldName);

            if (null == message) {
                throw new SchedulerActionInitializationException(
                        "Checkpoint scheduler action arguments should contain message object.", null);
            }

            entry.getState().setValue(messageFieldName, message);

            IRecoverStrategy strategy = IOC.resolve(Keys.getOrAdd(recoverStrategyId));

            strategy.init(entry.getState(), recoverConfig, (IMessageProcessor) args.getValue(processorFieldName));

            entry.getState().setValue(recoverStrategyFieldName, recoverStrategyId);

            // Store id's of currently and previously responsible for the message checkpoints and identifier of the message in storage of
            // previous checkpoint in the entry state.
            entry.getState().setValue(responsibleCheckpointIdFieldName, args.getValue(responsibleCheckpointIdFieldName));
            entry.getState().setValue(prevCheckpointIdFieldName, args.getValue(prevCheckpointIdFieldName));
            entry.getState().setValue(prevCheckpointEntryIdFieldName, args.getValue(prevCheckpointEntryIdFieldName));
        } catch (ReadValueException | InvalidArgumentException | ResolutionException | RecoverStrategyInitializationException
                | ChangeValueException e) {
            throw new SchedulerActionInitializationException("Error occurred initializing checkpoint action.", e);
        }
    }

    public  CheckpointSchedulerAction();

}