开发者社区 问答 正文

请问下,用上面的sql创建表时,出现下面这个错是为什么?#Flink

CREATE TABLE test ( myField2 DECIMAL, myField3 STRING, myField1 DECIMAL ) WITH ( 'connector' = 'kafka-0.11', 'topic' = 'test', 'properties.group.id' = 'test', 'scan.startup-mode' = 'group-offsets', 'properties.bootstrap.servers' = 'xxxx', 'properties.security.protocol' = 'SASL_PLAINTEXT', 'properties.sasl.mechanism' = 'SCRAM-SHA-256', 'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.scram.ScramLoginModule required username=aa password=aa', 'format' = 'json' ); 请问下,用上面的sql创建表时,出现下面这个错是为什么? Caused by: org.apache.flink.table.api.ValidationException: Unsupported options found for connector 'kafka-0.11'.

Unsupported options:

scan.startup-mode

Supported options:

connector format json.fail-on-missing-field json.ignore-parse-errors json.timestamp-format.standard properties.bootstrap.servers properties.group.id properties.sasl.jaas.config properties.sasl.mechanism properties.security.protocol property-version scan.startup.mode scan.startup.specific-offsets scan.startup.timestamp-millis sink.partitioner topic at org.apache.flink.table.factories.FactoryUtil$TableFactoryHelper.validate(FactoryUtil.java:487) at org.apache.flink.table.factories.FactoryUtil$TableFactoryHelper.validateExcept(FactoryUtil.java:519) at org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicTableFactoryBase.createDynamicTableSource(KafkaDynamicTableFactoryBase.java:77) at org.apache.flink.table.factories.FactoryUtil.createTableSource(FactoryUtil.java:122) ... 18 more flink版本: 1.11.0 #Flink

展开
收起
黄一刀 2020-08-06 16:54:16 2978 分享
分享
版权
举报
1 条回答
写回答
取消 提交回答
  • scan.startup.mode

    2020-08-06 16:54:43 举报
    赞同 评论

    评论

    全部评论 (0)

    登录后可评论
AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等