Realtime Compute for Apache Flink:Java UDTFs

Parameter konfigurasi

Parameter berikut mengontrol perilaku runtime asynchronous table functions.

Contoh pengembangan

Saat mengimplementasikan AsyncTableFunction, deklarasikan metode eval publik yang non-static. Parameter pertama adalah CompletableFuture<Collection<T>> yang digunakan untuk mengirimkan hasil, dan parameter sisanya adalah parameter bisnis yang diteruskan oleh pemanggilan SQL.

Fungsi segera mengembalikan kontrol setelah mengirim permintaan asinkron, dan mengirimkan hasil dalam callback penyelesaian dengan memanggil result.complete(rows), atau mengirimkan exception dengan memanggil result.completeExceptionally(error). Beberapa baris dari satu pemanggilan dikirimkan sebagai satu koleksi. Kirim koleksi kosong jika tidak ada hasil.

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;

/**
 * Queries product tags through an asynchronous HTTP interface.
 *
 * <p>Interface contract: 200 returns UTF-8 text with one tag per line; 204 and 404 indicate no tags.
 * This example suits read-only queries that return short text. Other status codes are treated as invocation failures and handed over to the engine.
 */
@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("Configure 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("The service URL must be an HTTP(S) base URL that ends with a forward slash (/).");
        }
        client = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(3)).build();
    }

    @Override
    public boolean isDeterministic() {
        // Product information may change. The same ID may return different results at different times.
        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(
                                                "Failed to query the product. HTTP status code: " + response.statusCode()));
                            })
                    .whenComplete(
                            (rows, error) -> {
                                if (error != null) {
                                    result.completeExceptionally(error);
                                } else {
                                    result.complete(rows);
                                }
                            });
        } catch (RuntimeException e) {
            result.completeExceptionally(e);
        }
    }
}

HttpClient.sendAsync melakukan permintaan jaringan secara asinkron, sehingga fungsi tidak perlu membuat kolam thread sendiri. Callback hanya mengurai teks pendek dan mengirimkan hasil. Saat menggunakan klien lain, lebih baik gunakan juga antarmuka asinkron aslinya. Untuk perilaku klien JDK, lihat HttpClient API.

Tag produk dapat berubah seiring waktu, sehingga contoh ini menetapkan isDeterministic() ke false. Untuk fungsi yang selalu mengembalikan hasil yang sama untuk input yang sama, pertahankan perilaku default.

Penanganan timeout kustom

Secara default, jika pemanggilan eval tidak menyelesaikan CompletableFuture sebelum timeout berakhir dan retry yang dikonfigurasi telah habis, framework akan melemparkan java.util.concurrent.TimeoutException dan gagal menjalankan pekerjaan. Anda dapat mendeklarasikan metode timeout(...) yang sesuai dengan metode eval dalam subclass AsyncTableFunction untuk mengembalikan baris fallback atau mengimplementasikan penanganan timeout kustom.

Timeout cenderung lebih sering terjadi saat memanggil kluster LLM remote, dan timeout biasanya tidak boleh langsung menyebabkan pekerjaan gagal. Dengan metode timeout yang dideklarasikan, hasil fallback dikembalikan saat timeout dan pekerjaan dilanjutkan.

Metode timeout harus memenuhi semua kondisi berikut:

• Deklarasi metode: Metode harus berupa metode instans public, tidak boleh static, dan harus bernama timeout.

• Signature selaras dengan eval: Daftar parameter harus benar-benar identik dengan metode eval yang sesuai. Parameter pertama adalah CompletableFuture yang tipe generiknya harus persis sama dengan eval, dan tipe serta urutan parameter sisanya juga harus persis sama. timeout mendukung overloading: Anda dapat mendeklarasikan metode timeout terpisah untuk setiap overload eval yang memerlukan logika fallback.

• Penyelesaian sinkron (wajib): Callback ini berjalan pada thread mailbox operator dan harus menyelesaikan future sebelum metode mengembalikan kontrol. Segera setelah metode mengembalikan kontrol, framework memeriksa future.isDone(). Jika future belum selesai, framework akan memaksanya diselesaikan dengan IllegalStateException. Oleh karena itu:

  • Jangan memulai pemanggilan asinkron lain dalam isi metode, seperti retry lain atau kueri kedua, dan mengharapkan callback-nya menyelesaikan future. Pada saat callback dipicu, framework sudah menyerah pada catatan tersebut.

  • Jangan memulai thread baru untuk menyelesaikan future secara asinkron, karena alasan yang sama.

  • Isi metode hanya melakukan satu hal: memberikan hasil fallback ringan, seperti baris konstan, baris NULL, atau koleksi kosong, atau memanggil completeExceptionally(...) dengan exception bisnis kustom.

• Melemparkan exception aman: Melemparkan exception secara sinkron dalam isi metode diperbolehkan. Framework meneruskan exception tersebut ke ResultFuture downstream, yang setara dengan memanggil secara eksplisit future.completeExceptionally(thrown).

Di luar batasan ini, framework juga menjamin perilaku stabil dalam skenario berikut:

• Tidak ada metode timeout yang dideklarasikan: TimeoutException default dari framework digunakan, dan tidak terjadi kegagalan selama pembuatan kode.

• Signature tidak kompatibel: Jika metode timeout memiliki visibilitas yang valid tetapi daftar parameternya tidak dapat sesuai dengan tipe argumen aktual di lokasi pemanggilan, framework akan gagal selama perencanaan (pembuatan kode) dengan ValidationException. Pesan error berisi nama kelas lengkap fungsi dan menunjukkan signature yang diharapkan dan yang aktual.

• Koleksi kosong sebagai fallback: Saat future.complete(Collections.emptyList()) dipanggil, baris tersebut dibuang untuk INNER lookup join, dan bidang tabel kanan diisi dengan NULL untuk LEFT OUTER lookup join.

Contoh berikut menambahkan callback timeout ke asynchronous table function dan menunjukkan cara menyelesaikan future secara sinkron serta mengembalikan baris fallback:

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) {
        // ...same logic as the preceding development example: dispatch the asynchronous task,
        // and then complete `future` in the callback thread.
    }

    // The parameter list must align with eval(): the same CompletableFuture generic type,
    // and the same types and order for the remaining parameters (Integer waitMax).
    // The method body must complete `future` synchronously. Do not start another asynchronous call
    // or a new thread to complete it asynchronously. When this callback is triggered,
    // the operator has already given up on this record.
    public void timeout(CompletableFuture<Collection<Long>> future, Integer waitMax) {
        future.complete(Collections.singletonList(-1L));
        // You can also use a custom business exception instead of the default TimeoutException:
        // future.completeExceptionally(new RuntimeException(waitMax));
    }
}