static String getAutoOffsetResetValue(final String autoOffsetReset, final String samzaOffsetDefault) {
    // valid kafka consumer values
    final String kafkaOffsetLatest = "latest";
    final String kafkaOffsetEarliest = "earliest";
    final String kafkaOffsetNone = "none";

    // if the value for KafkaConsumer is set - use it.
    if (!StringUtils.isBlank(autoOffsetReset)) {
      if (autoOffsetReset.equals(kafkaOffsetEarliest) || autoOffsetReset.equals(kafkaOffsetLatest)
          || autoOffsetReset.equals(kafkaOffsetNone)) {
        return autoOffsetReset;
      }
      // translate old kafka consumer values into new ones (SAMZA-1987 top remove it)
      String newAutoOffsetReset;
      switch (autoOffsetReset) {
        case "largest":
          newAutoOffsetReset = kafkaOffsetLatest;
          LOG.warn("Using old (deprecated) value for kafka consumer config auto.offset.reset = {}. The right value should be {}", autoOffsetReset, kafkaOffsetLatest);
          break;
        case "smallest":
          newAutoOffsetReset = kafkaOffsetEarliest;
          LOG.warn("Using old (deprecated) value for kafka consumer config auto.offset.reset = {}. The right value should be {}", autoOffsetReset, kafkaOffsetEarliest);
          break;
        default:
          throw new SamzaException("Using invalid value for kafka consumer config auto.offset.reset " + autoOffsetReset + ". See KafkaConsumer config for the correct values.");
      }

      LOG.info("Auto offset reset value converted from {} to {}", autoOffsetReset, newAutoOffsetReset);
      return newAutoOffsetReset;
    }

    // in case kafka consumer configs are not provided we should match them to Samza's ones.
    String newAutoOffsetReset = kafkaOffsetLatest;
    if (!StringUtils.isBlank(samzaOffsetDefault)) {
      switch (samzaOffsetDefault) {
        case SystemConfig.SAMZA_SYSTEM_OFFSET_UPCOMING:
          newAutoOffsetReset = kafkaOffsetLatest;
          break;
        case SystemConfig.SAMZA_SYSTEM_OFFSET_OLDEST:
          newAutoOffsetReset = kafkaOffsetEarliest;
          break;
        default:
          throw new SamzaException("Using invalid value for samza default offset config " + autoOffsetReset + ". See samza config for the correct values");
      }
      LOG.info("Auto offset reset value for KafkaConsumer for system {} converted from {}(samza) to {}", samzaOffsetDefault, newAutoOffsetReset);
    }

    return newAutoOffsetReset;
  }
If settings for Kafka Consumer auto.offset.reset is set - use it.
If this setting is using the old (deprecated values) - translate them:
"largest" -> "latest"
"smallest" -> "earliest"

If no setting specified - match it to the Samza default offset setting.
If none defined - return "latest"
@param autoOffsetReset consumer.auto.offset.reset config
@param samzaOffsetDefault  samza system default
@return String representing the config value for "auto.offset.reset" property
samza-kafka/src/main/java/org/apache/samza/config/KafkaConsumerConfig.java