Realtime Compute for Apache Flink：自訂表格值函數（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));
    }
}