docs: public static void validateConsumerConfiguration(Properties config) {
  		checkNotNull(config, "config can not be null");
  
  		validateAwsConfiguration(config);
  
  		if (!(config.containsKey(AWSConfigConstants.AWS_REGION) ^ config.containsKey(ConsumerConfigConstants.AWS_ENDPOINT))) {
  			// per validation in AwsClientBuilder
  			throw new IllegalArgumentException(String.format("For FlinkKinesisConsumer either AWS region ('%s') or AWS endpoint ('%s') must be set in the config.",
  					AWSConfigConstants.AWS_REGION, AWSConfigConstants.AWS_ENDPOINT));
  		}
  
  		if (config.containsKey(ConsumerConfigConstants.STREAM_INITIAL_POSITION)) {
  			String initPosType = config.getProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION);
  
  			// specified initial position in stream must be either LATEST, TRIM_HORIZON or AT_TIMESTAMP
  			try {
  				InitialPosition.valueOf(initPosType);
  			} catch (IllegalArgumentException e) {
  				StringBuilder sb = new StringBuilder();
  				for (InitialPosition pos : InitialPosition.values()) {
  					sb.append(pos.toString()).append(", ");
  				}
  				throw new IllegalArgumentException("Invalid initial position in stream set in config. Valid values are: " + sb.toString());
  			}
  
  			// specified initial timestamp in stream when using AT_TIMESTAMP
  			if (InitialPosition.valueOf(initPosType) == InitialPosition.AT_TIMESTAMP) {
  				if (!config.containsKey(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP)) {
  					throw new IllegalArgumentException("Please set value for initial timestamp ('"
  						+ ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP + "') when using AT_TIMESTAMP initial position.");
  				}
  				validateOptionalDateProperty(config,
  					ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP,
  					config.getProperty(ConsumerConfigConstants.STREAM_TIMESTAMP_DATE_FORMAT, ConsumerConfigConstants.DEFAULT_STREAM_TIMESTAMP_DATE_FORMAT),
  					"Invalid value given for initial timestamp for AT_TIMESTAMP initial position in stream. "
  						+ "Must be a valid format: yyyy-MM-dd'T'HH:mm:ss.SSSXXX or non-negative double value. For example, 2016-04-04T19:58:46.480-00:00 or 1459799926.480 .");
  			}
  		}
  
  		validateOptionalPositiveIntProperty(config, ConsumerConfigConstants.SHARD_GETRECORDS_MAX,
  			"Invalid value given for maximum records per getRecords shard operation. Must be a valid non-negative integer value.");
  
  		validateOptionalPositiveIntProperty(config, ConsumerConfigConstants.SHARD_GETRECORDS_RETRIES,
  			"Invalid value given for maximum retry attempts for getRecords shard operation. Must be a valid non-negative integer value.");
  
  		validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_BASE,
  			"Invalid value given for get records operation base backoff milliseconds. Must be a valid non-negative long value.");
  
  		validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_MAX,
  			"Invalid value given for get records operation max backoff milliseconds. Must be a valid non-negative long value.");
  
  		validateOptionalPositiveDoubleProperty(config, ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_EXPONENTIAL_CONSTANT,
  			"Invalid value given for get records operation backoff exponential constant. Must be a valid non-negative double value.");
  
  		validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.SHARD_GETRECORDS_INTERVAL_MILLIS,
  			"Invalid value given for getRecords sleep interval in milliseconds. Must be a valid non-negative long value.");
  
  		validateOptionalPositiveIntProperty(config, ConsumerConfigConstants.SHARD_GETITERATOR_RETRIES,
  			"Invalid value given for maximum retry attempts for getShardIterator shard operation. Must be a valid non-negative integer value.");
  
  		validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_BASE,
  			"Invalid value given for get shard iterator operation base backoff milliseconds. Must be a valid non-negative long value.");
  
  		validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_MAX,
  			"Invalid value given for get shard iterator operation max backoff milliseconds. Must be a valid non-negative long value.");
  
  		validateOptionalPositiveDoubleProperty(config, ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_EXPONENTIAL_CONSTANT,
  			"Invalid value given for get shard iterator operation backoff exponential constant. Must be a valid non-negative double value.");
  
  		validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.SHARD_DISCOVERY_INTERVAL_MILLIS,
  			"Invalid value given for shard discovery sleep interval in milliseconds. Must be a valid non-negative long value.");
  
  		validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.LIST_SHARDS_BACKOFF_BASE,
  			"Invalid value given for list shards operation base backoff milliseconds. Must be a valid non-negative long value.");
  
  		validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.LIST_SHARDS_BACKOFF_MAX,
  			"Invalid value given for list shards operation max backoff milliseconds. Must be a valid non-negative long value.");
  
  		validateOptionalPositiveDoubleProperty(config, ConsumerConfigConstants.LIST_SHARDS_BACKOFF_EXPONENTIAL_CONSTANT,
  			"Invalid value given for list shards operation backoff exponential constant. Must be a valid non-negative double value.");
  
  		if (config.containsKey(ConsumerConfigConstants.SHARD_GETRECORDS_INTERVAL_MILLIS)) {
  			checkArgument(
  				Long.parseLong(config.getProperty(ConsumerConfigConstants.SHARD_GETRECORDS_INTERVAL_MILLIS))
  					< ConsumerConfigConstants.MAX_SHARD_GETRECORDS_INTERVAL_MILLIS,
  				"Invalid value given for getRecords sleep interval in milliseconds. Must be lower than " +
  					ConsumerConfigConstants.MAX_SHARD_GETRECORDS_INTERVAL_MILLIS + " milliseconds."
  			);
  		}
  	}