到目前為止，我已經按照為 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 的郵件列表上找到了一個解決方法，但他們將此描述為生產者而非消費者的問題，而我看到相反的情況（我認為），所以對此不確定。這真的很混亂。
2 回答
拉丁的傳說
TA貢獻1789條經驗 獲得超8個贊
一只甜甜圈
TA貢獻1836條經驗 獲得超5個贊
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)
添加回答
舉報
0/150
提交
取消