[
https://issues.apache.org/jira/browse/FLINK-10316?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16644663#comment-16644663
]
ASF GitHub Bot commented on FLINK-10316:
----------------------------------------
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:
[email protected]
> Kinesalite bug fix (FLINK-9402) doesn't fully fix the problem.
> --------------------------------------------------------------
>
> Key: FLINK-10316
> URL: https://issues.apache.org/jira/browse/FLINK-10316
> Project: Flink
> Issue Type: Bug
> Components: Kinesis Connector
> Affects Versions: 1.5.0, 1.6.0, 1.7.0
> Reporter: Joseph Sims
> Priority: Minor
> Labels: pull-request-available
>
> Related to: https://issues.apache.org/jira/browse/FLINK-9402
> A fix (Flink-9402) was introduced to force the configurations to have
> *exactly* one of "aws.region" or "aws.endpoint", but the underlying problem
> wasn't entirely fixed.
> If "aws.endpoint" is set and "aws.region" is unset, a different exception is
> thrown, instead.
> In *KinesisConfigUtil.java*, *getValidatedProducerConfiguration()* calls:
> {code:java}
> validateAwsConfiguration(config)
> {code}
> Where the original fix was introduced ("aws.region" XOR "aws.endpoint")
> But a couple of lines later, *getValidatedProducerConfiguration()* then
> calls:
> {code:java}
> kpc.setRegion(config.getProperty(AWSConfigConstants.AWS_REGION));
> {code}
> Which explicitly checks for the existence of "aws.region", and throws an
> exception if it is not set.
> Thus, this class needs to be fixed such that "aws.region" is only required
> if "aws.endpoint" is unset, as the original ticket indicated.
> Also, *validateAwsConfiguration()* has a minor bug where is labels the
> "aws.endpoint" incorrectly.
> Current:
> {code:java}
> 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));
> }
> {code}
> But should be:
> {code:java}
> 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, ConsumerConfigConstants.AWS_ENDPOINT));
> }
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
