Apache Flink - 无法为 FlinkKinesisConsumer 使用本地

Question

到目前为止，我已经按照为 Flink 的运动连接器记录的说明使用本地运动。使用非 AWS Kinesis 终端节点进行测试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 生产者，这些指令适用于本地运动（我使用 Kinesalite）。但是，对于 Flink 消费者，我得到一个例外，aws.region并且aws.endpoint不允许两者都被允许。但是区域是必需的，这意味着它不可能覆盖端点。org.apache.flink.client.program.ProgramInvocationException：主要方法导致错误：对于 FlinkKinesisConsumer，必须在配置中设置 AWS 区域（'aws.region'）或 AWS 端点（'aws.endpoint'）。这是连接器中的错误吗？我看到一个相关的 PR： https: //github.com/apache/flink/pull/6045。我在 Flink 的邮件列表上找到了一个解决方法，但他们将此描述为生产者而非消费者的问题，而我看到相反的情况（我认为），所以对此不确定。这真的很混乱。

Answer 1

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

提问者在这个 jira中推送了这个问题，这个 jira 被标记为第二个 jira的副本。

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

Answer 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)