You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2018/10/10 08:58:13 UTC
[GitHub] asfgit closed pull request #6789: [FLINK-10316][kinesis] bug was
preventing FlinkKinesisProducer to connect to Kinesalite
asfgit closed pull request #6789: [FLINK-10316][kinesis] bug was preventing FlinkKinesisProducer to connect to Kinesalite
URL: https://github.com/apache/flink/pull/6789
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtil.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtil.java
index a4d60ed496d..75c84cdca5a 100644
--- a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtil.java
+++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtil.java
@@ -87,6 +87,12 @@ public static void validateConsumerConfiguration(Properties config) {
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);
@@ -213,6 +219,11 @@ public static KinesisProducerConfiguration getValidatedProducerConfiguration(Pro
validateAwsConfiguration(config);
+ if (!config.containsKey(AWSConfigConstants.AWS_REGION)) {
+ // per requirement in Amazon Kinesis Producer Library
+ throw new IllegalArgumentException(String.format("For FlinkKinesisProducer AWS region ('%s') must be set in the config.", AWSConfigConstants.AWS_REGION));
+ }
+
KinesisProducerConfiguration kpc = KinesisProducerConfiguration.fromProperties(config);
kpc.setRegion(config.getProperty(AWSConfigConstants.AWS_REGION));
@@ -266,12 +277,6 @@ public static void validateAwsConfiguration(Properties config) {
}
}
- if (!(config.containsKey(AWSConfigConstants.AWS_REGION) ^ config.containsKey(ConsumerConfigConstants.AWS_ENDPOINT))) {
- // per validation in AwsClientBuilder
- throw new IllegalArgumentException(String.format("Either AWS region ('%s') or AWS endpoint ('%s') must be set in the config.",
- AWSConfigConstants.AWS_REGION, AWSConfigConstants.AWS_REGION));
- }
-
if (config.containsKey(AWSConfigConstants.AWS_REGION)) {
// specified AWS Region name must be recognizable
if (!AWSUtil.isValidRegion(config.getProperty(AWSConfigConstants.AWS_REGION))) {
diff --git a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtilTest.java b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtilTest.java
index c4bfa17675c..ae2d21e9493 100644
--- a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtilTest.java
+++ b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtilTest.java
@@ -125,14 +125,10 @@ public void testCorrectlySetRegionInProducerConfiguration() {
assertEquals("incorrect region", region, kpc.getRegion());
}
- // ----------------------------------------------------------------------
- // validateAwsConfiguration() tests
- // ----------------------------------------------------------------------
-
@Test
- public void testMissingAwsRegionInConfig() {
- String expectedMessage = String.format("Either AWS region ('%s') or AWS endpoint ('%s') must be set in the config.",
- AWSConfigConstants.AWS_REGION, AWSConfigConstants.AWS_REGION);
+ public void testMissingAwsRegionInProducerConfig() {
+ String expectedMessage = String.format("For FlinkKinesisProducer AWS region ('%s') must be set in the config.",
+ AWSConfigConstants.AWS_REGION);
exception.expect(IllegalArgumentException.class);
exception.expectMessage(expectedMessage);
@@ -140,9 +136,13 @@ public void testMissingAwsRegionInConfig() {
testConfig.setProperty(AWSConfigConstants.AWS_ACCESS_KEY_ID, "accessKey");
testConfig.setProperty(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
- KinesisConfigUtil.validateAwsConfiguration(testConfig);
+ KinesisConfigUtil.getValidatedProducerConfiguration(testConfig);
}
+ // ----------------------------------------------------------------------
+ // validateAwsConfiguration() tests
+ // ----------------------------------------------------------------------
+
@Test
public void testUnrecognizableAwsRegionInConfig() {
exception.expect(IllegalArgumentException.class);
@@ -156,22 +156,6 @@ public void testUnrecognizableAwsRegionInConfig() {
KinesisConfigUtil.validateAwsConfiguration(testConfig);
}
- @Test
- public void testAwsRegionOrEndpointInConfig() {
- String expectedMessage = String.format("Either AWS region ('%s') or AWS endpoint ('%s') must be set in the config.",
- AWSConfigConstants.AWS_REGION, AWSConfigConstants.AWS_REGION);
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage(expectedMessage);
-
- Properties testConfig = new Properties();
- testConfig.setProperty(AWSConfigConstants.AWS_REGION, "us-east");
- testConfig.setProperty(AWSConfigConstants.AWS_ENDPOINT, "fake");
- testConfig.setProperty(AWSConfigConstants.AWS_ACCESS_KEY_ID, "accessKey");
- testConfig.setProperty(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
-
- KinesisConfigUtil.validateAwsConfiguration(testConfig);
- }
-
@Test
public void testCredentialProviderTypeSetToBasicButNoCredentialSetInConfig() {
exception.expect(IllegalArgumentException.class);
@@ -200,6 +184,22 @@ public void testUnrecognizableCredentialProviderTypeInConfig() {
// validateConsumerConfiguration() tests
// ----------------------------------------------------------------------
+ @Test
+ public void testAwsRegionOrEndpointInConsumerConfig() {
+ String expectedMessage = String.format("For FlinkKinesisConsumer either AWS region ('%s') or AWS endpoint ('%s') must be set in the config.",
+ AWSConfigConstants.AWS_REGION, AWSConfigConstants.AWS_ENDPOINT);
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage(expectedMessage);
+
+ Properties testConfig = new Properties();
+ testConfig.setProperty(AWSConfigConstants.AWS_REGION, "us-east-1");
+ testConfig.setProperty(AWSConfigConstants.AWS_ENDPOINT, "fake");
+ testConfig.setProperty(AWSConfigConstants.AWS_ACCESS_KEY_ID, "accessKey");
+ testConfig.setProperty(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, "secretKey");
+
+ KinesisConfigUtil.validateConsumerConfiguration(testConfig);
+ }
+
@Test
public void testUnrecognizableStreamInitPositionTypeInConfig() {
exception.expect(IllegalArgumentException.class);
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services