KinesisConfigUtil { public static void validateConsumerConfiguration(Properties config) { checkNotNull(config, "config can not be null"); validateAwsConfiguration(config); 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.STREAM_DESCRIBE_BACKOFF_BASE, "Invalid value given for describe stream operation base backoff milliseconds. Must be a valid non-negative long value."); validateOptionalPositiveLongProperty(config, ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_MAX, "Invalid value given for describe stream operation max backoff milliseconds. Must be a valid non-negative long value."); validateOptionalPositiveDoubleProperty(config, ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_EXPONENTIAL_CONSTANT, "Invalid value given for describe stream 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 KinesisProducerConfiguration getValidatedProducerConfiguration(Properties config); static void validateAwsConfiguration(Properties config);  }
@Test public void testUnrecognizableStreamInitPositionTypeInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid initial position in stream"); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "wrongInitPosition"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testStreamInitPositionTypeSetToAtTimestampButNoInitTimestampSetInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Please set value for initial timestamp ('" + ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP + "') when using AT_TIMESTAMP initial position."); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testUnparsableDateForInitialTimestampInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for initial timestamp for AT_TIMESTAMP initial position in stream."); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP"); testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, "unparsableDate"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testIllegalValueForInitialTimestampInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for initial timestamp for AT_TIMESTAMP initial position in stream."); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP"); testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, "-1.0"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testDateStringForValidateOptionDateProperty() { String timestamp = "2016-04-04T19:58:46.480-00:00"; Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP"); testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, timestamp); try { KinesisConfigUtil.validateConsumerConfiguration(testConfig); } catch (Exception e) { e.printStackTrace(); fail(); } }
@Test public void testUnixTimestampForValidateOptionDateProperty() { String unixTimestamp = "1459799926.480"; Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP"); testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, unixTimestamp); try { KinesisConfigUtil.validateConsumerConfiguration(testConfig); } catch (Exception e) { e.printStackTrace(); fail(); } }
@Test public void testInvalidPatternForInitialTimestampInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for initial timestamp for AT_TIMESTAMP initial position in stream."); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP"); testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, "2016-03-14"); testConfig.setProperty(ConsumerConfigConstants.STREAM_TIMESTAMP_DATE_FORMAT, "InvalidPattern"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@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 = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); 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); }
@Test public void testDateStringForUserDefinedDateFormatForValidateOptionDateProperty() { String unixTimestamp = "2016-04-04"; String pattern = "yyyy-MM-dd"; Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER, "BASIC"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "AT_TIMESTAMP"); testConfig.setProperty(ConsumerConfigConstants.STREAM_INITIAL_TIMESTAMP, unixTimestamp); testConfig.setProperty(ConsumerConfigConstants.STREAM_TIMESTAMP_DATE_FORMAT, pattern); try { KinesisConfigUtil.validateConsumerConfiguration(testConfig); } catch (Exception e) { e.printStackTrace(); fail(); } }
@Test public void testUnparsableLongForDescribeStreamBackoffBaseMillisInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for describe stream operation base backoff milliseconds"); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_BASE, "unparsableLong"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testUnparsableLongForDescribeStreamBackoffMaxMillisInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for describe stream operation max backoff milliseconds"); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_MAX, "unparsableLong"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testUnparsableDoubleForDescribeStreamBackoffExponentialConstantInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for describe stream operation backoff exponential constant"); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_EXPONENTIAL_CONSTANT, "unparsableDouble"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testUnparsableIntForGetRecordsRetriesInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for maximum retry attempts for getRecords shard operation"); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_RETRIES, "unparsableInt"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testUnparsableIntForGetRecordsMaxCountInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for maximum records per getRecords shard operation"); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_MAX, "unparsableInt"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testUnparsableLongForGetRecordsBackoffBaseMillisInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for get records operation base backoff milliseconds"); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_BASE, "unparsableLong"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testUnparsableLongForGetRecordsBackoffMaxMillisInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for get records operation max backoff milliseconds"); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_MAX, "unparsableLong"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testUnparsableDoubleForGetRecordsBackoffExponentialConstantInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for get records operation backoff exponential constant"); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_EXPONENTIAL_CONSTANT, "unparsableDouble"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testUnparsableLongForGetRecordsIntervalMillisInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for getRecords sleep interval in milliseconds"); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.SHARD_GETRECORDS_INTERVAL_MILLIS, "unparsableLong"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testUnparsableIntForGetShardIteratorRetriesInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for maximum retry attempts for getShardIterator shard operation"); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.SHARD_GETITERATOR_RETRIES, "unparsableInt"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testUnparsableLongForGetShardIteratorBackoffBaseMillisInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for get shard iterator operation base backoff milliseconds"); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_BASE, "unparsableLong"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testUnparsableLongForGetShardIteratorBackoffMaxMillisInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for get shard iterator operation max backoff milliseconds"); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_MAX, "unparsableLong"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testUnparsableDoubleForGetShardIteratorBackoffExponentialConstantInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for get shard iterator operation backoff exponential constant"); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.SHARD_GETITERATOR_BACKOFF_EXPONENTIAL_CONSTANT, "unparsableDouble"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }
@Test public void testUnparsableLongForShardDiscoveryIntervalMillisInConfig() { exception.expect(IllegalArgumentException.class); exception.expectMessage("Invalid value given for shard discovery sleep interval in milliseconds"); Properties testConfig = new Properties(); testConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); testConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "accessKeyId"); testConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey"); testConfig.setProperty(ConsumerConfigConstants.SHARD_DISCOVERY_INTERVAL_MILLIS, "unparsableLong"); KinesisConfigUtil.validateConsumerConfiguration(testConfig); }