ksqlDB java 客户端：无法从 client.executeStatement() 反序列化数据

Question

我正在尝试使用 Java ksqlDb-api-client:0.24.0 使用下面的代码查询汇合云上的 KSQL 流

        ClientOptions options = ClientOptions.create()
                .setHost(KSQLDB_SERVER_HOST)
                .setPort(KSQLDB_SERVER_HOST_PORT)
                .setUseTls(true)
                .setUseAlpn(true)
                .setBasicAuthCredentials(USER_NAME,PASSWORD);

        Client client = Client.create(options);
        Map<String, Object> properties = Collections.singletonMap("auto.offset.reset", "earliest");

        client.streamQuery("select * from process_payments EMIT CHANGES;", properties)
                .thenAccept(streamedQueryResult -> {
                    System.out.println("Query has started. Query ID: " + streamedQueryResult.queryID());
//                    RowSubscriber subscriber = new RowSubscriber();
//                    streamedQueryResult.subscribe(subscriber);

                }).exceptionally(e -> {
                    e.printStackTrace();
                    System.out.println("Request failed: " + e);
                    return null;
                });

即使 RowSubscriber 被注释掉，我仍然收到以下异常，仅供参考，我的消息就像单个字符串列，以便于调试：

Caused by: java.lang.NoClassDefFoundError: io/confluent/ksql/schema/utils/FormatOptions
    at java.base/java.lang.Class.getDeclaredMethods0(Native Method)
    at java.base/java.lang.Class.privateGetDeclaredMethods(Class.java:3166)
    at java.base/java.lang.Class.getDeclaredMethods(Class.java:2309)
    at com.fasterxml.jackson.databind.util.ClassUtil.getClassMethods(ClassUtil.java:1231)
    ... 88 more
Caused by: java.lang.ClassNotFoundException: io.confluent.ksql.schema.utils.FormatOptions
    at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:581)
    at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178)
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:522)
    ... 92 more

Request failed: java.util.concurrent.CompletionException: com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Failed on call to `getDeclaredMethods()` on class `io.confluent.ksql.schema.ksql.LogicalSchema`, problem: (java.lang.NoClassDefFoundError) io/confluent/ksql/schema/utils/FormatOptions
 at [Source: (byte[])"{"queryId":"transient_PROCESS_PAYMENTS_2977877146871246906","columnNames":["STATUS"],"columnTypes":["STRING"]}"; line: 1, column: 1]

我错过了什么吗？

Answer 1

看起来您需要将 ksqldb-udf 添加到您的类路径中。该 jar 包含 FormatOptions。

Answer 2

我认为这是由于版本冲突造成的。我在 7.2.1 版本中也遇到了完全相同的错误。它在 6.0.1 版本中运行良好，无需额外依赖。