public Optional<CheckpointManager> getCheckpointManager(MetricsRegistry metricsRegistry) {
    return Optional.ofNullable(get(CHECKPOINT_MANAGER_FACTORY))
        .filter(StringUtils::isNotBlank)
        .map(checkpointManagerFactoryName -> ReflectionUtil.getObj(checkpointManagerFactoryName,
            CheckpointManagerFactory.class).getCheckpointManager(this, metricsRegistry));
  }
Create the checkpoint manager

@param metricsRegistry Registry of metrics to use. Can be null if not using metrics.
@return CheckpointManager object if checkpoint manager factory is configured, otherwise empty.
samza-core/src/main/java/org/apache/samza/config/TaskConfig.java