Apache Flink - 无法为 FlinkKinesisConsumer 使用本地 Kinesis答案

【问题标题】：Apache Flink - Unable to use local Kinesis for FlinkKinesisConsumerApache Flink - 无法为 FlinkKinesisConsumer 使用本地 Kinesis
【发布时间】：2019-07-04 19:56:19
【问题描述】：

到目前为止，我已按照为 Flink 的 kinesis 连接器记录的说明使用本地 Kinesis。

Using Non-AWS Kinesis Endpoints for Testing

Properties producerConfig = new Properties();
producerConfig.put(AWSConfigConstants.AWS_REGION, "us-east-1");
producerConfig.put(AWSConfigConstants.AWS_ACCESS_KEY_ID, "aws_access_key_id");
producerConfig.put(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, "aws_secret_access_key");
producerConfig.put(AWSConfigConstants.AWS_ENDPOINT, "http://localhost:4567");

对于 Flink 生产者，这些指令适用于本地 kinesis（我使用 Kinesalite）。

但是，对于 Flink 消费者，我得到一个例外，即 aws.region 和 aws.endpoint 不允许两者。但是 region 是必需的，这意味着它不可能覆盖端点。

org.apache.flink.client.program.ProgramInvocationException：主要方法导致错误：对于 FlinkKinesisConsumer，必须在配置中设置 AWS 区域（'aws.region'）或 AWS 端点（'aws.endpoint'） .

这是连接器中的错误吗？我看到一个相关的 PR：https://github.com/apache/flink/pull/6045。

我找到了workaround on Flink's mailing list，但他们将此描述为生产者而非消费者的问题，而我看到相反的情况（我认为），所以对此不确定。真的很混乱。

【问题讨论】：

我也在 Flink 的 jira 上标记了它issues.apache.org/jira/browse/FLINK-13111

标签： java apache-flink amazon-kinesis

【解决方案1】：

自从提出这个问题以来，已经取得了一些进展。

提问者将问题推送到this jira，该问题被标记为与this second jira 重复。

该问题现在应该已解决，并且该修复适用于 1.10 及更高版本。

【讨论】：

【解决方案2】：

问题与验证检查中的XOR 条件有关。如您所见，validateConsumerConfiguration 方法在 if 语句中执行 XOR 验证。因此，您只能指定选中的两个参数之一。

要设置自定义 URL，您需要删除 AWSConfigConstants.AWS_REGION 属性并仅使用链接。

// Set the given URL
consumerConfig.put(AWSConfigConstants.AWS_ENDPOINT, URL);
// Remove the region
consumerConfig.remove(AWSConfigConstants.AWS_REGION);

这个解决方案，修复以下StackTrace相关的错误：

java.lang.IllegalArgumentException: For FlinkKinesisConsumer either AWS region ('aws.region') or AWS endpoint ('aws.endpoint') must be set in the config.

at org.apache.flink.streaming.connectors.kinesis.util.KinesisConfigUtil.validateConsumerConfiguration(KinesisConfigUtil.java:92)

【讨论】：