ksqlDB Java Client无法捕获连接断开异常问题排查

使用io.confluent.ksql:ksqldb-api-client:7.3.1时，网络断开后Flux的doOnError未触发，核心原因是原代码未正确捕获异步操作异常、未传递订阅阶段的错误，且缺少客户端层面的连接存活检测配置。以下是具体修复方案：

1. 捕获streamQuery异步操作的异常 #

原代码仅处理了client.streamQuery()的成功场景，未捕获异步调用阶段的异常。需要通过exceptionally将异步错误传递到FluxSink：

return Flux.create((FluxSink<io.confluent.ksql.api.client.Row> sink) ->
        client.streamQuery("select * from " + tableName + " " + emitChangesString(entityClass) + ";")
                .thenAccept(streamedQueryResult -> {
                    // 自定义订阅逻辑，确保错误传递
                    streamedQueryResult.subscribe(new RowSubscriber() {
                        @Override
                        public void onNext(Row row) {
                            sink.next(row);
                        }

                        @Override
                        public void onError(Throwable throwable) {
                            sink.error(throwable);
                        }

                        @Override
                        public void onComplete() {
                            sink.complete();
                        }
                    });
                })
                .exceptionally(throwable -> {
                    // 捕获streamQuery调用阶段的异常
                    sink.error(throwable);
                    return null;
                })
)

2. 自定义RowSubscriber传递订阅错误 #

RowSubscriber.fromSink()默认实现可能未正确处理onError事件，导致连接断开时的错误无法注入Flux流。手动实现RowSubscriber，确保onError触发时通过sink.error()将错误传递给Flux。

3. 配置客户端连接与心跳参数 #

构建ksqlDB客户端时添加超时和WebSocket心跳配置，主动检测连接存活状态：

KsqlClient client = KsqlClient.create(KsqlClientOptions.builder()
        .setHost("your-ksqldb-host")
        .setPort(8088)
        .setConnectTimeout(Duration.ofSeconds(10))
        .setReadTimeout(Duration.ofSeconds(30))
        // WebSocket心跳配置，定期检测连接
        .setWebSocketPingInterval(Duration.ofSeconds(10))
        .setWebSocketPongTimeout(Duration.ofSeconds(5))
        .build());

4. 完善Flux流的错误处理 #

将对象映射阶段的错误也注入流中，确保doOnError能捕获所有异常：

.flatMap(row -> {
    try {
        return Flux.just(objectMapper.readValue(row.asObject().toJsonString(), entityClass));
    } catch (JsonProcessingException e) {
        logger.error("ksqlDB对象映射异常：{}", e.getMessage(), e);
        return Flux.error(e); // 将映射错误传递到流中
    }
})
.doOnError(throwable -> logger.error("ksqlDB Push Query处理异常：{}", throwable.getMessage(), throwable))

内容的提问来源于stack exchange，提问作者Farhan Islam