Java客户端从ksqlDB拉取数据失败，streamQuery.get()持续阻塞

1. 修正客户端连接核心配置 #

你的代码存在两个连接配置错误，这是导致get()阻塞的主要原因：

• 使用0.0.0.0作为连接地址：这是服务端的监听地址，客户端应使用localhost连接本地服务
• 开启ALPN：本地开发环境下ksqlDB默认未启用ALPN协议，强制开启会导致连接异常，进而阻塞调用

修改后的连接配置：

ClientOptions options = ClientOptions.create()
        .setHost("localhost") // 替换0.0.0.0为localhost
        .setPort(KSQLDB_SERVER_HOST_PORT)
        .setUseTls(false)
        .setUseAlpn(false); // 关闭不必要的ALPN

2. 正确传递消费者配置 #

你定义的auto.offset.reset配置未传递给客户端，导致客户端默认从最新偏移量开始消费，若没有新数据则会一直等待。需通过setProperties方法传入配置：

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

ClientOptions options = ClientOptions.create()
        .setHost("localhost")
        .setPort(KSQLDB_SERVER_HOST_PORT)
        .setUseTls(false)
        .setUseAlpn(false)
        .setProperties(properties); // 新增此行传递配置

3. 优化流查询的消费逻辑 #

streamQuery返回的是持续查询结果，poll()方法默认会无限阻塞等待新数据。建议添加超时时间，或循环消费已有数据：

public static void main(String[] args) throws ExecutionException, InterruptedException {
    Map<String, Object> properties = Collections.singletonMap(
            "auto.offset.reset", "earliest"
    );

    ClientOptions options = ClientOptions.create()
            .setHost("localhost")
            .setPort(KSQLDB_SERVER_HOST_PORT)
            .setUseTls(false)
            .setUseAlpn(false)
            .setProperties(properties);
    Client client = Client.create(options);

    String streamQuery = "select name, countrycode from USERS_STREAM emit changes;";

    StreamedQueryResult streamedQueryResult = client.streamQuery(streamQuery).get();
    
    // 设置5秒超时，避免无限阻塞
    Row row;
    while ((row = streamedQueryResult.poll(Duration.ofSeconds(5))) != null) {
        System.out.println("## " + row.values());
    }
    
    client.close();
}

4. 验证ksqlDB服务可用性 #

通过curl确认ksqlDB服务正常监听8088端口：

curl http://localhost:8088/info

返回包含版本信息的JSON则说明服务正常。

内容的提问来源于stack exchange，提问作者Jeff Cook