User-Defined Functions in Confluent Manager for Apache Flink

Starting with version 2.4.0, Confluent Manager for Apache Flink® (CMF) lets you extend Flink SQL with your own user-defined functions (UDFs). You implement a UDF as a Java class, package it as a JAR, upload the JAR to artifact management, and register the function in an environment catalog with a CREATE FUNCTION statement.

After registration, the function is available to statements that run in that environment, the same way a built-in function is.

Prerequisites

Write and package a function

Implement the function as a Java class that extends the appropriate Flink base class, for example ScalarFunction for a scalar function:

package com.example;

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

public class ToUpperCase extends ScalarFunction {
    public String eval(String input) {
        return input == null ? null : input.toUpperCase();
    }
}

Compile the class against org.apache.flink:flink-table-common as a provided dependency, target Java 11, and package it as a JAR. If the function depends on libraries that are not on the CMF or Flink classpath, shade them into the JAR rather than widening the class allowlist (see Limitations).

Deliver the function JAR

Upload the JAR to artifact management, then reference it from CREATE FUNCTION with a cmf:// URI. CMF downloads the JAR when it compiles the statement, and the Flink cluster fetches it when it runs the job, so you do not need to rebuild any images and you can version the JAR independently.

Put the artifact metadata in a file, artifact.json:

{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "Artifact",
  "metadata": {
    "name": "udfs.jar"
  },
  "spec": {}
}

Upload the JAR, passing artifact.json in the artifact part:

curl -X POST http://cmf:8080/cmf/api/v1/environments/prod/artifacts \
  -F 'artifact=@artifact.json;type=application/json' \
  -F 'file=@target/udfs.jar'

Register a function

Register a function with CREATE FUNCTION, using the USING JAR clause to reference the uploaded artifact.

Syntax

CREATE FUNCTION [IF NOT EXISTS] function_name
  AS 'fully.qualified.ClassName'
  USING JAR 'cmf://{environment}/{artifact-name}[?version={N}]'
    [, JAR 'cmf://{environment}/{artifact-name}[?version={N}]']*;

  • IF NOT EXISTS: Prevents an error if a function with the same name already exists.

  • AS: The fully qualified name of the Java class that implements the function.

  • USING JAR: The cmf:// URI of the artifact that contains the class. If the function’s classes span more than one artifact, add additional , JAR '...' clauses. For the URI form and version behavior, see Reference artifacts in SQL statements.

Example

Register a function from a managed artifact and pin it to a version:

confluent flink statement create create-to-upper \
  --url http://cmf:8080 \
  --environment prod \
  --compute-pool pool \
  --catalog _env_prod \
  --database default \
  --sql "CREATE FUNCTION to_upper AS 'com.example.ToUpperCase' USING JAR 'cmf://prod/udfs.jar?version=1';"

Use a function

Call the function in a query the same way you call a built-in function, and list the registered functions with SHOW USER FUNCTIONS:

SHOW USER FUNCTIONS;

SELECT to_upper(name) AS name_upper FROM users;

Drop a function

DROP FUNCTION [IF EXISTS] function_name;

Dropping a function does not affect jobs that are already running; they keep using the version they were compiled with.

Limitations

  • User-defined functions must be implemented in Java. The LANGUAGE clause for other languages, such as Python or Scala, is not supported.

  • Process table functions (PTFs) are not supported.

  • ALTER FUNCTION is not supported. To change a function’s class or JAR, drop the function and create it again.

  • CREATE TEMPORARY FUNCTION and CREATE TEMPORARY SYSTEM FUNCTION are not supported.

  • When CMF compiles a statement, function code can load only classes whose fully qualified names match the cmf.sql.classAllowlist configuration. The default allowlist covers the JDK, Flink, Confluent Flink packages, and SLF4J. Shade any other dependencies into the function JAR instead of widening the allowlist.

Security considerations

Review the following guidance before you enable user-defined functions in production.

Important

CMF does not sandbox user-defined function code. To compile a statement, CMF loads and runs the function class in the CMF process, and the Flink cluster runs it during the job. Treat a UDF JAR as trusted code that runs with CMF privileges. Review and scan function code before you upload it, restrict who can upload artifacts and submit statements with RBAC, and run UDF workloads on dedicated compute pools rather than shared ones. The class allowlist is a best-effort control, not a security boundary.

Related content