apache flink-无法对flinkkinesisconsumer使用本地kinesis

hc8w905p 于 2021-06-24 发布在 Flink

关注(0)|答案(2)|浏览(387)

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

使用非aws运动端点进行测试

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:main方法导致错误：对于flinkkinesisconsumer，必须在配置中设置aws region（'aws.region'）或aws endpoint（'aws.endpoint'）。
这是连接器中的错误吗？我看到一个相关的公关：https://github.com/apache/flink/pull/6045 .
我在flink的邮件列表上找到了一个解决方法，但是他们把这个描述为生产者的问题，而不是消费者的问题，而我看到了相反的情况（我想），所以对此不确定。真让人困惑。

Java apache-flink amazon-kinesis

来源：https://stackoverflow.com/questions/56893685/apache-flink-unable-to-use-local-kinesis-for-flinkkinesisconsumer

2条答案

按热度按时间

1l5u6lss1#

自从提出这个问题以来，已经取得了一些进展。
询问者把这个问题推到第二次圣战中，被标记为第二次圣战的复制品。
这个问题现在应该已经解决了，并且1.10及更高版本的修复程序可用。

赞(0）回复(0）举报 2021-06-24

vhmi4jdf2#

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

要设置自定义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)

赞(0）回复(0）举报 2021-06-24

我来回答

apache flink-无法对flinkkinesisconsumer使用本地kinesis

使用非aws运动端点进行测试

2条答案

相关问题

热门标签

最新问答