[FLINK-10316] Add check to KinesisProducer that aws.region is set

• Type: Bug
• Status: Closed
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 1.5.0, 1.6.0, 1.7.0
• Fix Version/s: 1.5.5, 1.6.2, 1.7.0
• Component/s: Connectors / Kinesis
• Labels: pull-request-available
• Assignee: Unassigned
• Reporter: Joseph Sims
• Created: 10/Sep/18 21:26
• Updated: 17/Oct/18 08:27
• Resolved: 10/Oct/18 09:13

Description

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:

    validateAwsConfiguration(config)

Where the original fix was introduced ("aws.region" XOR "aws.endpoint")
But a couple of lines later, getValidatedProducerConfiguration() then calls:

    kpc.setRegion(config.getProperty(AWSConfigConstants.AWS_REGION));

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:

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));
}

But should be:

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));
}

Issue Links

• links to: GitHub Pull Request #6789