自定义表值函数（UDTF）

配置参数

通过以下参数控制异步表值函数的运行行为。

开发示例

实现 AsyncTableFunction 时，需要声明公开、非静态的 eval 方法。第一个参数是用于提交结果的 CompletableFuture<Collection<T>>，其余参数是 SQL 调用时传入的业务参数。

函数发起异步请求后及时返回，在请求完成的回调中调用 result.complete(rows) 提交结果，或调用 result.completeExceptionally(error) 提交异常。一次调用的多行结果通过一个集合提交，无结果时提交空集合。

package com.example.udf;

import org.apache.flink.table.functions.AsyncTableFunction;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.types.Row;
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.Collection;
import java.util.Collections;
import java.util.stream.Collectors;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;

/**
 * 通过异步 HTTP 接口查询商品标签。
 *
 * <p>接口约定：200 返回 UTF-8 文本，每行一个标签；204、404 表示无标签。
 * 本示例适用于返回短文本的只读查询；其他状态码作为调用失败交给引擎处理。
 */
@FunctionHint(output = @DataTypeHint("ROW<tag STRING>"))
public class AsyncProductTags extends AsyncTableFunction<Row> {

    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<Collection<Row>> result, Long productId) {
        if (productId == null) {
            result.complete(Collections.emptyList());
            return;
        }

        try {
            HttpRequest request =
                    HttpRequest.newBuilder(
                                    serviceBaseUri.resolve("products/" + productId + "/tags"))
                            .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().lines()
                                            .map(String::strip)
                                            .filter(tag -> !tag.isEmpty())
                                            .map(tag -> Row.of(tag))
                                            .collect(Collectors.toList());
                                }
                                if (response.statusCode() == 204 || response.statusCode() == 404) {
                                    return Collections.<Row>emptyList();
                                }
                                throw new CompletionException(
                                        new IOException(
                                                "商品查询失败，HTTP 状态码：" + response.statusCode()));
                            })
                    .whenComplete(
                            (rows, error) -> {
                                if (error != null) {
                                    result.completeExceptionally(error);
                                } else {
                                    result.complete(rows);
                                }
                            });
        } catch (RuntimeException e) {
            result.completeExceptionally(e);
        }
    }
}

HttpClient.sendAsync 异步执行网络请求，函数无需自行创建线程池。回调只完成短文本解析和结果提交。使用其他客户端时，也应优先选择其原生异步接口。JDK 客户端的行为请参见 HttpClient API。

商品标签可能随时间变化，因此示例将 isDeterministic() 设置为 false。对于同样的输入始终返回相同结果的函数，可保留默认行为。

自定义超时处理

默认情况下，如果某次 eval 调用在超时时间之后仍未完成 CompletableFuture，且配置的重试次数也已用尽，框架会抛出 java.util.concurrent.TimeoutException 并让作业失败。支持在 AsyncTableFunction 子类中声明一个与 eval 方法相匹配的 timeout(...) 方法，由该方法返回一行兜底数据，或实现自定义的超时处理逻辑。

调用远程 LLM 集群等场景的超时概率相对较高，通常不希望超时直接导致作业失败。声明 timeout 方法后，可在超时时返回兜底结果，避免作业中断。

timeout 方法需要同时满足以下条件：

• 方法声明：必须是 public 实例方法，不能是 static，方法名固定为 timeout。

• 签名与 eval 对齐：参数列表需要与对应的 eval 严格一致，第一个参数是 CompletableFuture，其泛型类型必须与 eval 完全相同，其余参数的类型和顺序也必须与 eval 完全相同。timeout 支持重载，每一个需要兜底逻辑的 eval 重载都可以单独声明一个对应的 timeout。

• 同步完成（强制要求）：该回调在算子的 mailbox 线程上运行，必须在方法返回前完成 future。方法返回后，框架会立即检查 future.isDone()；如果尚未完成，框架会强制以 IllegalStateException 完成 future。因此：

  • 不要在方法体内再发起一次异步调用（例如又一次重试或二次查询），并指望其回调最终完成 future。回调触发时，框架已放弃该条记录。

  • 不要启动新线程异步完成 future，原因同上。

  • 方法体只做一件事：给出轻量的兜底结果，例如返回一个常量行、一个 NULL 行、空集合，或使用业务自定义异常调用 completeExceptionally(...)。

• 抛异常是安全的：直接在方法体内同步抛出异常没有问题，框架会把异常转交给下游的 ResultFuture，效果等价于显式调用 future.completeExceptionally(thrown)。

在以上约束之外，框架还会在以下场景保证稳定的行为：

• 未声明 timeout 方法：沿用框架默认的 TimeoutException，不会引发代码生成阶段的失败。

• 签名不兼容：如果 timeout 方法可见性合法，但参数列表无法和当前调用点的实参类型对应，框架会在规划期（代码生成阶段）以 ValidationException 直接失败。错误信息中包含函数的全限定类名，并同时给出期望签名与实际签名。

• 空集合兜底：调用 future.complete(Collections.emptyList()) 时，对 INNER lookup join 而言该行会被丢弃；对 LEFT OUTER lookup join 而言，右表字段会被填充为 NULL。

以下示例在异步表值函数的基础上增加 timeout 回调，展示如何同步完成 future 并返回一行兜底数据：

import org.apache.flink.table.functions.AsyncTableFunction;

import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;

public static class BackgroundFunctionWithTimeout extends AsyncTableFunction<Long> {

    public void eval(CompletableFuture<Collection<Long>> future, Integer waitMax) {
        // ...逻辑与前述开发示例一致：把异步任务派发出去，
        // 然后在回调线程中完成 `future`。
    }

    // 参数列表必须与 eval() 对齐：CompletableFuture 的泛型相同，
    // 后续参数（Integer waitMax）的类型和顺序也相同。
    // 方法体必须同步完成 `future`——不要再发起异步调用，
    // 也不要启动新线程异步去完成它；这个回调被触发时，
    // 算子已经放弃了这条记录。
    public void timeout(CompletableFuture<Collection<Long>> future, Integer waitMax) {
        future.complete(Collections.singletonList(-1L));
        // 也可以选择用业务自定义异常代替默认的 TimeoutException：
        // future.completeExceptionally(new RuntimeException(waitMax));
    }
}