@Test public void testUnparsableDateForUserDefinedDateFormatForInitialTimestampInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for initial timestamp for AT_TIMESTAMP initial position in stream."); Properties testConfig = TestUtils.getStandardProperties(); testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC"); testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP"); testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, "stillUnparsable"); testConfig.setProperty(ConsumerConfigConstants.STREAM_TIMESTAMP_DATE_FORMAT, "yyyy-MM-dd"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
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))) { 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); 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()); } 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." ); } }
KinesisConfigUtil { 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))) { 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); 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()); } 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." ); } } }
KinesisConfigUtil { 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))) { 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); 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()); } 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." ); } }  }
KinesisConfigUtil { 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))) { 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); 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()); } 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." ); } }  static void validateConsumerConfiguration(Properties config); static Properties replaceDeprecatedProducerKeys(Properties configProps); static Properties backfillConsumerKeys(Properties configProps); static KinesisProducerConfiguration getValidatedProducerConfiguration(Properties config); static void validateAwsConfiguration(Properties config); }
KinesisConfigUtil { 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))) { 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); 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()); } 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." ); } }  static void validateConsumerConfiguration(Properties config); static Properties replaceDeprecatedProducerKeys(Properties configProps); static Properties backfillConsumerKeys(Properties configProps); static KinesisProducerConfiguration getValidatedProducerConfiguration(Properties config); static void validateAwsConfiguration(Properties config);  }