Redact a Column with a User-Defined Function

This example builds a scalar Flink user-defined function (UDF), mask, that keeps the last few characters of a string and replaces the rest with asterisks. You upload it as an artifact, register it in an environment catalog, and call it in a query to redact an email column. For the reference material behind each step, see User-Defined Functions in Confluent Manager for Apache Flink.

Prerequisites

Complete the examples setup first. It provides the test environment, the pool compute pool, artifact management, and the JDK and Maven you use to build the function JAR.

Build the function JAR

The function extends ScalarFunction and needs only flink-table-common, which is already on the CMF and Flink classpath, so declare that dependency as provided. Compile for Java 11 to match the Flink cluster runtime.

MaskFunction.java:

package io.confluent.examples.flink.udf;

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

public class MaskFunction extends ScalarFunction {
    public String eval(String input, Integer visible) {
        if (input == null) {
            return null;
        }
        int keep = (visible == null || visible < 0) ? 0 : Math.min(visible, input.length());
        int hide = input.length() - keep;
        return "*".repeat(hide) + input.substring(hide);
    }
}

Package it with this pom.xml. Because flink-table-common is provided, the JAR carries only your class and needs no shading:

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>io.confluent.examples.flink</groupId>
  <artifactId>udfs</artifactId>
  <version>1.0.0</version>
  <packaging>jar</packaging>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.release>11</maven.compiler.release>
    <flink.version>1.19.1</flink.version>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-common</artifactId>
      <version>${flink.version}</version>
      <scope>provided</scope>
    </dependency>
  </dependencies>

  <build>
    <finalName>udfs</finalName>
  </build>
</project>

Build it:

mvn package

The build writes target/udfs.jar.

Upload the JAR as an artifact

A UDF JAR is referenced by a cmf:// URI, so it does not need the connector or format labels. Put the metadata in artifact.json:

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

Upload it:

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

The response has status.version of 1.

Register the function

Register mask in the environment catalog with CREATE FUNCTION, pointing USING JAR at the artifact. This is metadata only and completes immediately.

confluent flink statement create udf-create-mask \
  --url http://cmf:8080 \
  --environment test \
  --compute-pool pool \
  --catalog _env_test \
  --database default \
  --sql "CREATE FUNCTION mask AS 'io.confluent.examples.flink.udf.MaskFunction' USING JAR 'cmf://test/udfs.jar';"

The statement reaches phase COMPLETED with sqlKind of CREATE_FUNCTION.

Call the function

Call mask like any built-in function. The unqualified name resolves only when _env_test and default are the current catalog and database, so start the SQL shell with the same --catalog and --database flags:

confluent flink shell \
  --url http://cmf:8080 \
  --environment test \
  --compute-pool pool \
  --catalog _env_test \
  --database default

This query redacts customer emails from the examples.marketplace.customers datagen table, keeping the last four characters:

SELECT customer_id, email, mask(email, 4) AS masked_email FROM examples.marketplace.customers;

The statement runs on the compute pool, and the shell shows results such as:

customer_id  email                        masked_email
       eufemia.jenkins@yahoo.com    *********************.com
       anabel.sipes@gmail.com       ******************.com
       billy.tillman@gmail.com      *******************.com

Clean up

The shell prints the statement name when it submits the query. For example, cli-2026-10-19-120056-c98ff916. Stop the query by deleting that statement, and drop the function if you no longer need it:

confluent flink statement delete <statement-name> \
  --url http://cmf:8080 --environment test

confluent flink statement create udf-drop-mask \
  --url http://cmf:8080 --environment test --compute-pool pool \
  --catalog _env_test --database default \
  --sql "DROP FUNCTION IF EXISTS mask;"