带debezium-json格式的Flink 1.11

Question

在Flink 1.11中，我正在尝试使用debezium格式，以下应该可以工作，对吗？我在试着遵循文档[1]

    TableResult products = bsTableEnv.executeSql(
            "CREATE TABLE products (\n" +
                    "  id BIGINT,\n" +
                    "  name STRING,\n" +
                    "  description STRING,\n" +
                    "  weight DECIMAL(10, 2)\n" +
                    ") WITH (\n" +
                    " 'connector' = 'kafka',\n" +
                    " 'topic' = 'dbserver1.inventory.products',\n" +
                    " 'properties.bootstrap.servers' = 'localhost:9092',\n" +
                    " 'properties.group.id' = 'testGroup',\n" +
                    "'scan.startup.mode'='earliest-offset',\n" +
                    " 'format' = 'debezium-json'" +
                    ")"
    );

    bsTableEnv.executeSql("SHOW TABLES").print(); // This seems to work; 
    bsTableEnv.executeSql("SELECT id FROM products").print();

输出代码段/异常：

+------------+
| table name |
+------------+
|   products |
+------------+
1 row in set
Exception in thread "main" org.apache.flink.table.api.TableException: AppendStreamTableSink doesn't support consuming update and delete changes which is produced by node TableSourceScan(table=[[default_catalog, default_database, products]], fields=[id, name, description, weight])

我已经验证了Debezium设置，dbserver1中有消息。库存产品主题。我可以使用其他方法阅读Flink中的Kafka主题，但如前所述，我希望使用debezium json格式。

此外，我知道Flink 1.12引入了新的Kafka Upett连接器，但我现在无法使用1.11。

我对Flink很陌生，所以我很可能错过了一些明显的东西。

提前感谢

[1]https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/formats/debezium.html

Answer 1

似乎问得太早了。如果它可能对其他人有帮助，我可以让它一起工作

Table results = bsTableEnv.sqlQuery("SELECT id, name FROM products");
bsTableEnv.toRetractStream(results, Row.class).print();