Java UDTFs

Configuration parameters

The following parameters control the runtime behavior of asynchronous table functions.

Development example

When you implement an AsyncTableFunction, declare a public, non-static eval method. The first parameter is a CompletableFuture<Collection<T>> used to submit results, and the remaining parameters are the business parameters passed in by the SQL call.

The function returns promptly after it sends the asynchronous request, and submits results in the completion callback by calling result.complete(rows), or submits an exception by calling result.completeExceptionally(error). Multiple rows of a single call are submitted as one collection. Submit an empty collection when there is no result.

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 performs network requests asynchronously, so the function does not need to create its own thread pool. The callback only parses short text and submits results. When you use another client, prefer its native asynchronous interface as well. For the behavior of the JDK client, see HttpClient API.

Product tags may change over time, so the example sets isDeterministic() to false. For a function that always returns the same result for the same input, keep the default behavior.

Custom timeout handling

By default, if an eval call does not complete the CompletableFuture before the timeout expires and the configured retries are exhausted, the framework throws a java.util.concurrent.TimeoutException and fails the job. You can declare a timeout(...) method that matches the eval method in an AsyncTableFunction subclass to return a fallback row or implement custom timeout handling.

Timeouts are relatively more likely when calling a remote LLM cluster, and a timeout usually should not fail the job directly. With a declared timeout method, a fallback result is returned on timeout and the job continues.

The timeout method must meet all of the following conditions:

• Method declaration: The method must be a public instance method, cannot be static, and must be named timeout.

• Signature aligned with eval: The parameter list must be strictly identical to that of the corresponding eval method. The first parameter is a CompletableFuture whose generic type must be exactly the same as that of eval, and the types and order of the remaining parameters must also be exactly the same. timeout supports overloading: you can declare a separate timeout method for each eval overload that requires fallback logic.

• Synchronous completion (mandatory): This callback runs on the mailbox thread of the operator and must complete the future before the method returns. Immediately after the method returns, the framework checks future.isDone(). If the future is not complete, the framework forcibly completes it with an IllegalStateException. Therefore:

  • Do not start another asynchronous call in the method body, such as another retry or a second query, and expect its callback to complete the future. By the time the callback is triggered, the framework has already given up on the record.

  • Do not start a new thread to complete the future asynchronously, for the same reason.

  • The method body does only one thing: provide a lightweight fallback result, such as a constant row, a NULL row, or an empty collection, or call completeExceptionally(...) with a custom business exception.

• Throwing exceptions is safe: Throwing an exception synchronously in the method body is acceptable. The framework passes the exception to the downstream ResultFuture, which is equivalent to explicitly calling future.completeExceptionally(thrown).

Beyond these constraints, the framework also guarantees stable behavior in the following scenarios:

• No timeout method declared: The default TimeoutException of the framework is used, and no failure occurs during code generation.

• Incompatible signature: If the timeout method has valid visibility but its parameter list cannot match the actual argument types at the call site, the framework fails during planning (code generation) with a ValidationException. The error message contains the fully qualified class name of the function and shows both the expected and the actual signatures.

• Empty collection as fallback: When future.complete(Collections.emptyList()) is called, the row is discarded for an INNER lookup join, and the right-table fields are filled with NULL for a LEFT OUTER lookup join.

The following example adds a timeout callback to an asynchronous table function and shows how to complete the future synchronously and return a fallback row:

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));
    }
}