luciano/amazon-kinesis-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

removed region validation as this was not a requirement for stringArn support in multilangdaemon

Browse source
This commit is contained in:
Ryan Pelaez 2023-06-19 15:50:19 -07:00
parent f7c30c5e54
commit e11cc9a398
3 changed files with 17 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
amazon-kinesis-client-multilang/src/main/java/software/amazon/kinesis/multilang/config/KinesisClientLibConfigurator.java
View file

@ -84,11 +84,6 @@ public class KinesisClientLibConfigurator {
} }
Validate.isTrue(configuration.getKinesisCredentialsProvider().isDirty(), "A basic set of AWS credentials must be provided"); Validate.isTrue(configuration.getKinesisCredentialsProvider().isDirty(), "A basic set of AWS credentials must be provided");
//Verify Region is real
final String regionCode = String.valueOf(configuration.getKinesisClient().get("region"));
if (regionCode == null || Region.regions().stream().filter(x -> x.id().equalsIgnoreCase(regionCode)).count() == 0) {
throw new IllegalArgumentException(String.format("Unsupported region: %s", regionCode));
}
return configuration; return configuration;
} }

5
amazon-kinesis-client-multilang/src/test/java/software/amazon/kinesis/multilang/MultiLangDaemonConfigTest.java
View file

@ -101,11 +101,6 @@ public class MultiLangDaemonConfigTest {
setup("", "arn:aws:kinesis:us-east-2:ACCOUNT_ID:BadFormatting:stream/" + TEST_STREAM_NAME_IN_ARN, TEST_REGION); setup("", "arn:aws:kinesis:us-east-2:ACCOUNT_ID:BadFormatting:stream/" + TEST_STREAM_NAME_IN_ARN, TEST_REGION);
} }
@Test(expected = IllegalArgumentException.class)
public void testConstructorFailsBecauseInvalidRegion() throws Exception {
setup("", TEST_STREAM_ARN, "us-east-1000");
}
@Test(expected = IllegalArgumentException.class) @Test(expected = IllegalArgumentException.class)
public void testConstructorFailsBecauseStreamNameAndArnAreEmpty() throws Exception { public void testConstructorFailsBecauseStreamNameAndArnAreEmpty() throws Exception {
setup("", "", TEST_REGION); setup("", "", TEST_REGION);

34
amazon-kinesis-client-multilang/src/test/java/software/amazon/kinesis/multilang/config/KinesisClientLibConfiguratorTest.java
View file

@ -69,7 +69,7 @@ public class KinesisClientLibConfiguratorTest {
@Test @Test
public void testWithBasicSetup() { public void testWithBasicSetup() {
MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "streamName = a", MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "streamName = a",
"applicationName = b", "AWSCredentialsProvider = " + credentialName1, "workerId = 123", "regionName = us-east-1" }, '\n')); "applicationName = b", "AWSCredentialsProvider = " + credentialName1, "workerId = 123" }, '\n'));
assertEquals(config.getApplicationName(), "b"); assertEquals(config.getApplicationName(), "b");
assertEquals(config.getStreamName(), "a"); assertEquals(config.getStreamName(), "a");
assertEquals(config.getWorkerIdentifier(), "123"); assertEquals(config.getWorkerIdentifier(), "123");
@ -81,7 +81,7 @@ public class KinesisClientLibConfiguratorTest {
public void testWithLongVariables() { public void testWithLongVariables() {
MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "applicationName = app", MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "applicationName = app",
"streamName = 123", "AWSCredentialsProvider = " + credentialName1 + ", " + credentialName2, "streamName = 123", "AWSCredentialsProvider = " + credentialName1 + ", " + credentialName2,
"workerId = 123", "failoverTimeMillis = 100", "shardSyncIntervalMillis = 500", "regionName = us-east-1" }, '\n')); "workerId = 123", "failoverTimeMillis = 100", "shardSyncIntervalMillis = 500" }, '\n'));
assertEquals(config.getApplicationName(), "app"); assertEquals(config.getApplicationName(), "app");
assertEquals(config.getStreamName(), "123"); assertEquals(config.getStreamName(), "123");
@ -95,7 +95,7 @@ public class KinesisClientLibConfiguratorTest {
long epochTimeInSeconds = 1617406032; long epochTimeInSeconds = 1617406032;
MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "applicationName = app", MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "applicationName = app",
"streamName = 123", "AWSCredentialsProvider = " + credentialName1 + ", " + credentialName2, "streamName = 123", "AWSCredentialsProvider = " + credentialName1 + ", " + credentialName2,
"initialPositionInStreamExtended = " + epochTimeInSeconds, "regionName = us-east-1"}, '\n')); "initialPositionInStreamExtended = " + epochTimeInSeconds}, '\n'));
assertEquals(config.getInitialPositionInStreamExtended().getTimestamp(), new Date(epochTimeInSeconds * 1000L)); assertEquals(config.getInitialPositionInStreamExtended().getTimestamp(), new Date(epochTimeInSeconds * 1000L));
assertEquals(config.getInitialPositionInStream(), InitialPositionInStream.AT_TIMESTAMP); assertEquals(config.getInitialPositionInStream(), InitialPositionInStream.AT_TIMESTAMP);
@ -135,7 +135,7 @@ public class KinesisClientLibConfiguratorTest {
public void testWithUnsupportedClientConfigurationVariables() { public void testWithUnsupportedClientConfigurationVariables() {
MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join( MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(
new String[] { "AWSCredentialsProvider = " + credentialName1 + ", " + credentialName2, "workerId = id", new String[] { "AWSCredentialsProvider = " + credentialName1 + ", " + credentialName2, "workerId = id",
"kinesisClientConfig = {}", "streamName = stream", "applicationName = b", "regionName = us-east-1" }, "kinesisClientConfig = {}", "streamName = stream", "applicationName = b" },
'\n')); '\n'));
assertEquals(config.getApplicationName(), "b"); assertEquals(config.getApplicationName(), "b");
@ -149,7 +149,7 @@ public class KinesisClientLibConfiguratorTest {
MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "streamName = kinesis", MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "streamName = kinesis",
"AWSCredentialsProvider = " + credentialName2 + ", " + credentialName1, "workerId = w123", "AWSCredentialsProvider = " + credentialName2 + ", " + credentialName1, "workerId = w123",
"maxRecords = 10", "metricsMaxQueueSize = 20", "applicationName = kinesis", "maxRecords = 10", "metricsMaxQueueSize = 20", "applicationName = kinesis",
"retryGetRecordsInSeconds = 2", "maxGetRecordsThreadPool = 1", "regionName = us-east-1" }, '\n')); "retryGetRecordsInSeconds = 2", "maxGetRecordsThreadPool = 1" }, '\n'));
assertEquals(config.getApplicationName(), "kinesis"); assertEquals(config.getApplicationName(), "kinesis");
assertEquals(config.getStreamName(), "kinesis"); assertEquals(config.getStreamName(), "kinesis");
@ -179,7 +179,7 @@ public class KinesisClientLibConfiguratorTest {
public void testWithStringVariables() { public void testWithStringVariables() {
MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "streamName = a", MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "streamName = a",
"applicationName = b", "AWSCredentialsProvider = ABCD," + credentialName1, "workerId = 1", "applicationName = b", "AWSCredentialsProvider = ABCD," + credentialName1, "workerId = 1",
"kinesisEndpoint = https://kinesis", "metricsLevel = SUMMARY", "regionName = us-east-1" }, '\n')); "kinesisEndpoint = https://kinesis", "metricsLevel = SUMMARY" }, '\n'));
assertEquals(config.getWorkerIdentifier(), "1"); assertEquals(config.getWorkerIdentifier(), "1");
assertEquals(config.getKinesisClient().get("endpointOverride"), URI.create("https://kinesis")); assertEquals(config.getKinesisClient().get("endpointOverride"), URI.create("https://kinesis"));
@ -190,7 +190,7 @@ public class KinesisClientLibConfiguratorTest {
public void testWithSetVariables() { public void testWithSetVariables() {
MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "streamName = a", MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "streamName = a",
"applicationName = b", "AWSCredentialsProvider = ABCD," + credentialName1, "workerId = 1", "applicationName = b", "AWSCredentialsProvider = ABCD," + credentialName1, "workerId = 1",
"metricsEnabledDimensions = ShardId, WorkerIdentifier", "regionName = us-east-1" }, '\n')); "metricsEnabledDimensions = ShardId, WorkerIdentifier" }, '\n'));
Set<String> expectedMetricsEnabledDimensions = ImmutableSet.<String> builder() Set<String> expectedMetricsEnabledDimensions = ImmutableSet.<String> builder()
.add("ShardId", "WorkerIdentifier").build(); .add("ShardId", "WorkerIdentifier").build();
@ -201,7 +201,7 @@ public class KinesisClientLibConfiguratorTest {
public void testWithInitialPositionInStreamTrimHorizon() { public void testWithInitialPositionInStreamTrimHorizon() {
MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "streamName = a", MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "streamName = a",
"applicationName = b", "AWSCredentialsProvider = ABCD," + credentialName1, "workerId = 123", "applicationName = b", "AWSCredentialsProvider = ABCD," + credentialName1, "workerId = 123",
"initialPositionInStream = TriM_Horizon", "regionName = us-east-1" }, '\n')); "initialPositionInStream = TriM_Horizon" }, '\n'));
assertEquals(config.getInitialPositionInStream(), InitialPositionInStream.TRIM_HORIZON); assertEquals(config.getInitialPositionInStream(), InitialPositionInStream.TRIM_HORIZON);
} }
@ -210,7 +210,7 @@ public class KinesisClientLibConfiguratorTest {
public void testWithInitialPositionInStreamLatest() { public void testWithInitialPositionInStreamLatest() {
MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "streamName = a", MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "streamName = a",
"applicationName = b", "AWSCredentialsProvider = ABCD," + credentialName1, "workerId = 123", "applicationName = b", "AWSCredentialsProvider = ABCD," + credentialName1, "workerId = 123",
"initialPositionInStream = LateSt", "regionName = us-east-1" }, '\n')); "initialPositionInStream = LateSt" }, '\n'));
assertEquals(config.getInitialPositionInStream(), InitialPositionInStream.LATEST); assertEquals(config.getInitialPositionInStream(), InitialPositionInStream.LATEST);
} }
@ -219,7 +219,7 @@ public class KinesisClientLibConfiguratorTest {
public void testSkippingNonKCLVariables() { public void testSkippingNonKCLVariables() {
MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "streamName = a", MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "streamName = a",
"applicationName = b", "AWSCredentialsProvider = ABCD," + credentialName1, "workerId = 123", "applicationName = b", "AWSCredentialsProvider = ABCD," + credentialName1, "workerId = 123",
"initialPositionInStream = TriM_Horizon", "abc = 1", "regionName = us-east-1" }, '\n')); "initialPositionInStream = TriM_Horizon", "abc = 1" }, '\n'));
assertEquals(config.getApplicationName(), "b"); assertEquals(config.getApplicationName(), "b");
assertEquals(config.getStreamName(), "a"); assertEquals(config.getStreamName(), "a");
@ -231,7 +231,7 @@ public class KinesisClientLibConfiguratorTest {
public void testEmptyOptionalVariables() { public void testEmptyOptionalVariables() {
MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "streamName = a", MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(new String[] { "streamName = a",
"applicationName = b", "AWSCredentialsProvider = ABCD," + credentialName1, "workerId = 123", "applicationName = b", "AWSCredentialsProvider = ABCD," + credentialName1, "workerId = 123",
"initialPositionInStream = TriM_Horizon", "maxGetRecordsThreadPool = 1", "regionName = us-east-1" }, '\n')); "initialPositionInStream = TriM_Horizon", "maxGetRecordsThreadPool = 1" }, '\n'));
assertThat(config.getMaxGetRecordsThreadPool(), equalTo(1)); assertThat(config.getMaxGetRecordsThreadPool(), equalTo(1));
assertThat(config.getRetryGetRecordsInSeconds(), nullValue()); assertThat(config.getRetryGetRecordsInSeconds(), nullValue());
} }
@ -241,7 +241,7 @@ public class KinesisClientLibConfiguratorTest {
String test = StringUtils.join(new String[] { "streamName = a", "applicationName = b", String test = StringUtils.join(new String[] { "streamName = a", "applicationName = b",
"AWSCredentialsProvider = ABCD," + credentialName1, "workerId = 123", "AWSCredentialsProvider = ABCD," + credentialName1, "workerId = 123",
"initialPositionInStream = TriM_Horizon", "maxGetRecordsThreadPool = 0", "initialPositionInStream = TriM_Horizon", "maxGetRecordsThreadPool = 0",
"retryGetRecordsInSeconds = 0", "regionName = us-east-1" }, '\n'); "retryGetRecordsInSeconds = 0" }, '\n');
InputStream input = new ByteArrayInputStream(test.getBytes()); InputStream input = new ByteArrayInputStream(test.getBytes());
try { try {
@ -255,7 +255,7 @@ public class KinesisClientLibConfiguratorTest {
@Test @Test
public void testWithInvalidIntValue() { public void testWithInvalidIntValue() {
String test = StringUtils.join(new String[] { "streamName = a", "applicationName = b", String test = StringUtils.join(new String[] { "streamName = a", "applicationName = b",
"AWSCredentialsProvider = " + credentialName1, "workerId = 123", "failoverTimeMillis = 100nf", "regionName = us-east-1" }, '\n'); "AWSCredentialsProvider = " + credentialName1, "workerId = 123", "failoverTimeMillis = 100nf" }, '\n');
InputStream input = new ByteArrayInputStream(test.getBytes()); InputStream input = new ByteArrayInputStream(test.getBytes());
try { try {
@ -268,7 +268,7 @@ public class KinesisClientLibConfiguratorTest {
@Test @Test
public void testWithNegativeIntValue() { public void testWithNegativeIntValue() {
String test = StringUtils.join(new String[] { "streamName = a", "applicationName = b", String test = StringUtils.join(new String[] { "streamName = a", "applicationName = b",
"AWSCredentialsProvider = " + credentialName1, "workerId = 123", "failoverTimeMillis = -12", "regionName = us-east-1" }, '\n'); "AWSCredentialsProvider = " + credentialName1, "workerId = 123", "failoverTimeMillis = -12" }, '\n');
InputStream input = new ByteArrayInputStream(test.getBytes()); InputStream input = new ByteArrayInputStream(test.getBytes());
// separate input stream with getConfiguration to explicitly catch exception from the getConfiguration statement // separate input stream with getConfiguration to explicitly catch exception from the getConfiguration statement
@ -296,7 +296,7 @@ public class KinesisClientLibConfiguratorTest {
public void testWithMissingWorkerId() { public void testWithMissingWorkerId() {
String test = StringUtils.join( String test = StringUtils.join(
new String[] { "streamName = a", "applicationName = b", "AWSCredentialsProvider = " + credentialName1, new String[] { "streamName = a", "applicationName = b", "AWSCredentialsProvider = " + credentialName1,
"failoverTimeMillis = 100", "shardSyncIntervalMillis = 500", "regionName = us-east-1" }, "failoverTimeMillis = 100", "shardSyncIntervalMillis = 500" },
'\n'); '\n');
InputStream input = new ByteArrayInputStream(test.getBytes()); InputStream input = new ByteArrayInputStream(test.getBytes());
MultiLangDaemonConfiguration config = configurator.getConfiguration(input); MultiLangDaemonConfiguration config = configurator.getConfiguration(input);
@ -375,7 +375,7 @@ public class KinesisClientLibConfiguratorTest {
"AWSCredentialsProvider = " + credentialNameKinesis, "AWSCredentialsProvider = " + credentialNameKinesis,
"AWSCredentialsProviderDynamoDB = " + credentialNameDynamoDB, "AWSCredentialsProviderDynamoDB = " + credentialNameDynamoDB,
"AWSCredentialsProviderCloudWatch = " + credentialNameCloudWatch, "failoverTimeMillis = 100", "AWSCredentialsProviderCloudWatch = " + credentialNameCloudWatch, "failoverTimeMillis = 100",
"shardSyncIntervalMillis = 500", "regionName = us-east-1" }, '\n'); "shardSyncIntervalMillis = 500" }, '\n');
InputStream input = new ByteArrayInputStream(test.getBytes()); InputStream input = new ByteArrayInputStream(test.getBytes());
// separate input stream with getConfiguration to explicitly catch exception from the getConfiguration statement // separate input stream with getConfiguration to explicitly catch exception from the getConfiguration statement
@ -404,7 +404,7 @@ public class KinesisClientLibConfiguratorTest {
"AWSCredentialsProvider = " + credentialNameKinesis, "AWSCredentialsProvider = " + credentialNameKinesis,
"AWSCredentialsProviderDynamoDB = " + credentialName2, "AWSCredentialsProviderDynamoDB = " + credentialName2,
"AWSCredentialsProviderCloudWatch = " + credentialName2, "failoverTimeMillis = 100", "AWSCredentialsProviderCloudWatch = " + credentialName2, "failoverTimeMillis = 100",
"shardSyncIntervalMillis = 500", "regionName = us-east-1" }, '\n'); "shardSyncIntervalMillis = 500" }, '\n');
InputStream input = new ByteArrayInputStream(test.getBytes()); InputStream input = new ByteArrayInputStream(test.getBytes());
// separate input stream with getConfiguration to explicitly catch exception from the getConfiguration statement // separate input stream with getConfiguration to explicitly catch exception from the getConfiguration statement