到目前為止，我已經(jīng)按照為 Flink 的運(yùn)動(dòng)連接器記錄的說明使用本地運(yùn)動(dòng)。使用非 AWS Kinesis 終端節(jié)點(diǎn)進(jìn)行測(cè)試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");對(duì)于 Flink 生產(chǎn)者，這些指令適用于本地運(yùn)動(dòng)（我使用 Kinesalite）。但是，對(duì)于 Flink 消費(fèi)者，我得到一個(gè)例外，aws.region并且aws.endpoint不允許兩者都被允許。但是區(qū)域是必需的，這意味著它不可能覆蓋端點(diǎn)。org.apache.flink.client.program.ProgramInvocationException：主要方法導(dǎo)致錯(cuò)誤：對(duì)于 FlinkKinesisConsumer，必須在配置中設(shè)置 AWS 區(qū)域（'aws.region'）或 AWS 端點(diǎn)（'aws.endpoint'）。這是連接器中的錯(cuò)誤嗎？我看到一個(gè)相關(guān)的 PR： https: //github.com/apache/flink/pull/6045。我在 Flink 的郵件列表上找到了一個(gè)解決方法，但他們將此描述為生產(chǎn)者而非消費(fèi)者的問題，而我看到相反的情況（我認(rèn)為），所以對(duì)此不確定。這真的很混亂。
2 回答
拉丁的傳說
TA貢獻(xiàn)1789條經(jīng)驗(yàn) 獲得超8個(gè)贊
自提出這個(gè)問題以來，已經(jīng)取得了一些進(jìn)展。
提問者在這個(gè) jira中推送了這個(gè)問題，這個(gè) jira 被標(biāo)記為第二個(gè) jira的副本。
該問題現(xiàn)在應(yīng)該已解決，并且該修復(fù)程序適用于 1.10 及更高版本。
一只甜甜圈
TA貢獻(xiàn)1836條經(jīng)驗(yàn) 獲得超5個(gè)贊
XOR該問題與驗(yàn)證檢查中的條件有關(guān)。如您所見，該validateConsumerConfiguration方法在 if 語句中執(zhí)行 XOR 驗(yàn)證。因此，您只能指定選中的兩個(gè)參數(shù)之一。 
要設(shè)置自定義 URL，您需要?jiǎng)h除AWSConfigConstants.AWS_REGION屬性并僅使用鏈接。
// Set the given URL
consumerConfig.put(AWSConfigConstants.AWS_ENDPOINT, URL);
// Remove the region
consumerConfig.remove(AWSConfigConstants.AWS_REGION);
此解決方案，修復(fù)了與以下 StackTrace 相關(guān)的錯(cuò)誤：
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)
添加回答
舉報(bào)
0/150
提交
取消