自定义标量函数（UDSF）

配置参数

通过以下参数控制异步标量函数的运行行为。参数应在提交查询或生成执行计划前设置。

配置无序输出

默认情况下，异步函数虽然可以并发执行，但后续记录已完成的结果仍需等待前面的记录完成后再输出。当外部请求耗时差异较大，且业务允许结果乱序时，可以开启无序输出，减少慢请求造成的等待。

在执行查询前增加以下配置：

SET 'table.exec.async-scalar.output-mode' = 'ALLOW_UNORDERED';

自定义异步标量函数示例

以下示例通过商品服务的 HTTP 接口查询商品名称。函数在 open 中创建并复用一个 HttpClient，在 eval 中调用 sendAsync() 发起请求，通过完成回调返回结果。sendAsync() 返回 CompletableFuture，适合将 HTTP 调用与异步标量函数衔接，详情请参见 JDK HttpClient 文档。

示例使用 3 秒连接超时、5 秒请求超时，适用于响应体较小的商品名称查询。请根据实际接口的响应时间调整。

package com.example.udf;

import org.apache.flink.table.functions.AsyncScalarFunction;
import org.apache.flink.table.functions.FunctionContext;

import java.io.IOException;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;

/**
 * 通过异步 HTTP 接口查询商品名称。
 *
 * <p>接口约定：200 返回 UTF-8 商品名称，404 表示商品不存在。
 * 本示例适用于返回短文本的只读查询；其他状态码作为调用失败交给引擎处理。
 */
public class AsyncProductName extends AsyncScalarFunction {

    private transient HttpClient client;
    private transient URI serviceBaseUri;

    @Override
    public void open(FunctionContext context) {
        String baseUrl = context.getJobParameter("product.service.base-url", null);
        if (baseUrl == null || baseUrl.isBlank()) {
            throw new IllegalArgumentException("请配置 product.service.base-url");
        }
        serviceBaseUri = URI.create(baseUrl);
        if (!("https".equalsIgnoreCase(serviceBaseUri.getScheme())
                        || "http".equalsIgnoreCase(serviceBaseUri.getScheme()))
                || serviceBaseUri.getHost() == null
                || serviceBaseUri.getRawQuery() != null
                || serviceBaseUri.getRawFragment() != null
                || !baseUrl.endsWith("/")) {
            throw new IllegalArgumentException("服务地址须为以 / 结尾的 HTTP(S) 基础地址");
        }
        client = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(3)).build();
    }

    @Override
    public boolean isDeterministic() {
        // 商品信息可能变化，相同 ID 在不同时间查询可能得到不同结果。
        return false;
    }

    public void eval(CompletableFuture<String> result, Long productId) {
        if (productId == null) {
            result.complete(null);
            return;
        }

        try {
            HttpRequest request =
                    HttpRequest.newBuilder(
                                    serviceBaseUri.resolve("products/" + productId + "/name"))
                            .timeout(Duration.ofSeconds(5))
                            .header("Accept", "text/plain")
                            .GET()
                            .build();

            client.sendAsync(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8))
                    .thenApply(
                            response -> {
                                if (response.statusCode() == 200) {
                                    return response.body();
                                }
                                if (response.statusCode() == 404) {
                                    return null;
                                }
                                throw new CompletionException(
                                        new IOException(
                                                "商品查询失败，HTTP 状态码：" + response.statusCode()));
                            })
                    .whenComplete(
                            (name, error) -> {
                                if (error != null) {
                                    result.completeExceptionally(error);
                                } else {
                                    result.complete(name);
                                }
                            });
        } catch (RuntimeException e) {
            result.completeExceptionally(e);
        }
    }
}

开发时请注意：

• 调用 result.complete(value) 返回结果；需要返回 SQL NULL 时，调用 result.complete(null)。

• 调用失败时，通过 result.completeExceptionally(exception) 返回异常，交由引擎按重试策略处理。

• eval 在发起请求后直接返回。使用 sendAsync() 的完成回调传递结果，避免在 eval 或回调中调用阻塞式 send()、get()、join()，或执行耗时计算。

• 每个函数实例在 open 中初始化并复用客户端，避免每条记录创建客户端。JDK 11/17 的 HttpClient 没有公开的 close() 方法，其内部资源由 JDK 管理。使用提供关闭接口的其他异步客户端时，应在函数的 close 中释放客户端。

• 商品信息会随时间变化，因此示例将 isDeterministic() 设置为 false。涉及更新流时，还需结合查询语义处理非确定性结果。