ksqlDB UDFs: Creating Custom Functions for Stream Processing

In the operational landscape of Stream Processing, platform and SRE engineers regularly hit business logic that ksqlDB’s built-in scalar, aggregate, and table functions cannot express: parsing a proprietary log format, applying a domain-specific scoring model, masking PII with an in-house library, or running a custom feature transformation inline. ksqlDB’s User-Defined Functions (UDFs) are the supported extension point for exactly this. This guide covers the full lifecycle — building, deploying, securing, and monitoring a custom scalar UDF in production Apache Kafka environments — with the gotchas that actually matter when a function runs inside every stream thread of a persistent query.

Everything below is grounded in the official ksqlDB UDF reference.

How ksqlDB Loads and Executes UDFs

A ksqlDB UDF is a plain Java class whose methods are annotated so the server can discover and register them. At startup, each ksqlDB server scans the jars in its extensions directory (configured via ksql.extension.dir, default ext/) for classes annotated with @UdfDescription (scalar UDF), @UdafDescription (aggregate), or @UdtfDescription (table function). The annotated methods become callable functions in SQL — usable in both persistent queries and pull queries.

Two operational facts follow directly from this model and should shape your design:

UDFs run on the stream thread. A persistent query compiles into a Kafka Streams topology, and your UDF is invoked synchronously inside the processing of each record, on the stream thread that owns that partition’s task. There is no separate worker pool. A slow or blocking UDF stalls that thread, which means consumer lag accumulates on its partitions and — if the call blocks long enough — the consumer can miss max.poll.interval.ms and trigger a rebalance. Design UDFs to be CPU-bound and fast. If you need per-event I/O such as an HTTP enrichment call, that belongs in a custom Kafka Streams processor with proper async handling (or an external lookup pattern), not in a synchronous UDF.

Jars are scanned only at startup; there is no hot reload. ksqlDB loads UDFs once when the server boots. To add or update a UDF you replace the jar in the extensions directory and restart every ksqlDB server in the cluster. Roll the restart so the cluster stays available, and keep the same UDF version on every node — a mixed deployment where some servers have a function and others do not will fail queries depending on where they run.

UDF jars are loaded with a child-first classloader isolated from ksqlDB’s own classpath, so your dependencies do not collide with the server’s. Build an uber/shaded jar that bundles everything your function needs. Watch for libraries with native code or static initializers, and always validate on a staging deployment that mirrors the production Java version and OS.

Building a Scalar UDF

We’ll build a UDF that normalizes a raw RFC 3164 syslog line into a structured JSON string for downstream parsing — a realistic, CPU-bound transformation.

Add the ksqlDB UDF API to your pom.xml. The dependency is provided: ksqlDB supplies these classes at runtime, so they must not be bundled into your uber jar. Match the version to your Confluent Platform / ksqlDB server version.

<dependency>
    <groupId>io.confluent.ksql</groupId>
    <artifactId>ksqldb-udf</artifactId>
    <version>7.5.0</version>
    <scope>provided</scope>
</dependency>

(The io.confluent.ksql artifacts resolve from Confluent’s Maven repository, https://packages.confluent.io/maven/.)

The class is annotated with @UdfDescription; each invocable method with @Udf; each parameter optionally with @UdfParameter. @UdfDescription supports name and description (both required) plus optional author, version, and category — author and version surface in DESCRIBE FUNCTION output and are genuinely useful for tracing which build is deployed.

import io.confluent.ksql.function.udf.Udf;
import io.confluent.ksql.function.udf.UdfDescription;
import io.confluent.ksql.function.udf.UdfParameter;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@UdfDescription(
    name = "parse_syslog",
    author = "platform-team",
    version = "1.0.0",
    description = "Parses a raw RFC 3164 syslog line into a structured JSON string"
)
public class SyslogParserUdf {

    // <PRI>MMM dd HH:MM:SS HOST TAG[PID]: message
    private static final Pattern SYSLOG_PATTERN = Pattern.compile(
        "^<(\\d+)>(\\w{3}\\s+\\d+\\s\\d{2}:\\d{2}:\\d{2})\\s+(\\S+)\\s+([^\\[:\\s]+)(?:\\[(\\d+)\\])?:\\s+(.*)$"
    );

    @Udf(description = "Parse a raw syslog message into a JSON string")
    public String parseSyslog(
        @UdfParameter(value = "rawMessage", description = "Raw syslog line")
        final String rawMessage
    ) {
        if (rawMessage == null || rawMessage.isEmpty()) {
            return null;
        }

        final Matcher m = SYSLOG_PATTERN.matcher(rawMessage);
        if (!m.matches()) {
            return String.format(
                "{\"error\":\"unparseable\",\"raw\":%s}",
                jsonString(rawMessage)
            );
        }

        final String pid = m.group(5);
        return String.format(
            "{\"priority\":%s,\"timestamp\":%s,\"hostname\":%s,"
                + "\"app_name\":%s,\"pid\":%s,\"message\":%s}",
            m.group(1),                       // numeric, no quotes
            jsonString(m.group(2)),
            jsonString(m.group(3)),
            jsonString(m.group(4)),
            pid == null ? "null" : pid,       // numeric or JSON null
            jsonString(m.group(6))
        );
    }

    // Minimal JSON string escaping for quotes and backslashes.
    private static String jsonString(final String s) {
        return "\"" + s.replace("\\", "\\\\").replace("\"", "\\\"") + "\"";
    }
}

A few design points that the prose above promised and the code actually does:

• The parameter type drives null handling. ksqlDB treats a boxed parameter (String, Integer, Long…) as nullable and a primitive (int, long…) as never-null. Because syslog input can be null, we accept String and return null (which becomes SQL NULL) for empty input rather than throwing.
• Returning a JSON string vs. a STRUCT. Returning String keeps this example simple, and downstream queries can apply EXTRACTJSONFIELD. If you would rather return a true STRUCT, ksqlDB requires you to declare the return schema — either statically via the schema field on @Udf, or dynamically with a method annotated @UdfSchemaProvider referenced by @Udf(schemaProvider = "..."). The same applies to DECIMAL returns.
• No hand-rolled string concatenation for the JSON payload. The jsonString helper escapes backslashes before quotes so the output is valid JSON; doing this inline (as naive examples often do) corrupts any value containing a quote or backslash.

Build the uber jar

Use the Maven Shade plugin to produce a single self-contained jar (the Gradle equivalent in Confluent’s tutorial is the Shadow plugin):

<plugin>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-shade-plugin</artifactId>
    <version>3.5.1</version>
    <executions>
        <execution>
            <phase>package</phase>
            <goals><goal>shade</goal></goals>
        </execution>
    </executions>
</plugin>

mvn package produces the shaded jar. Because the ksqldb-udf dependency is provided, it is correctly excluded from the output.

Deploying the UDF

1. Copy the uber jar into the extensions directory on every ksqlDB server (the directory pointed to by ksql.extension.dir; the default is ext/ under the install root, and the Docker image uses KSQL_KSQL_EXTENSION_DIR).
2. Restart each server — jars are scanned only at startup.
3. Confirm registration from the ksqlDB CLI:

SHOW FUNCTIONS;

SHOW FUNCTIONS is the only listing command — there is no LIST FUNCTIONS. Then inspect your function’s signatures, author, and version:

DESCRIBE FUNCTION PARSE_SYSLOG;

And use it in a persistent query:

CREATE STREAM parsed_logs AS
    SELECT PARSE_SYSLOG(raw_line) AS log_json
    FROM raw_syslog_stream
    EMIT CHANGES;

Securing UDF Execution

By default ksqlDB installs a Java security manager for UDF execution. It is intentionally narrow: it blocks UDFs from forking processes from the ksqlDB server and from calling System.exit(). It is not a full sandbox — it does not restrict file or network access, so it will not contain a UDF that reads arbitrary files or opens sockets.

The relevant server configs:

Two important caveats for production:

• The Java SecurityManager API is deprecated for removal (JEP 411, deprecated since Java 17). On JDKs where it is disabled or removed, ksqlDB’s security-manager protection cannot apply. Do not treat it as a security boundary going forward.
• Because UDFs run with the full privileges of the ksqlDB server process, the real isolation control is operational: review and pin UDF source, control who can drop jars into the extensions directory, and run untrusted or experimental functions on a separate ksqlDB cluster rather than alongside critical queries.

Monitoring UDFs in Production

ksqlDB does not emit a per-UDF latency metric out of the box, so you monitor the queries that use the UDF. The engine publishes an aggregated MBean (one per server, keyed by service id, not per query):

io.confluent.ksql.metrics:type=_confluent-ksql-engine-query-stats,ksql_service_id=<ksql.service.id>

Useful attributes on that MBean include:

• messages-consumed-per-sec, messages-consumed-total
• messages-produced-per-sec
• error-rate — messages consumed but not processed (e.g. a deserialization failure or an exception thrown inside a UDF), and
• num-active-queries, running-queries, error-queries, not-running-queries.

A climbing error-rate after a UDF deploy is the first signal that the function is throwing on real data. Per-query health (running vs. errored) is exposed under the io.confluent.ksql.metrics:type=ksql-queries MBean.

Because a persistent query is a Kafka Streams application, the more actionable latency and lag signals come from the underlying Streams and consumer metrics for that query’s application.id: records-lag-max (consumer) tells you whether a slow UDF is causing the query to fall behind, and the Streams process-latency-avg/-max give per-record processing time. Watch consumer lag on the query’s partitions as your primary indicator that a UDF has become a bottleneck.

A rough throughput ceiling

For a CPU-bound UDF on a single stream thread, an order-of-magnitude ceiling is:

max_records_per_sec_per_thread ≈ 1000 / udf_latency_ms

A 2 ms UDF caps one thread near 500 records/sec. This is a back-of-the-envelope figure only: real throughput is also bounded by deserialization, serialization, state-store access, network, and commit overhead, and you scale horizontally by increasing the source topic’s partition count and the query’s stream-thread count. Treat the formula as a sanity check on whether a heavy function will keep up, not as a capacity guarantee — measure against the query’s consumer lag under production load.

Conclusion

ksqlDB UDFs let you push genuinely custom logic into a persistent query, but they run on the critical path of every record on the stream thread, are loaded only at server startup, and execute with the server’s full privileges. Keep them fast and CPU-bound, push I/O into a Streams processor instead, deploy the same shaded jar to every node and restart, and monitor the query’s error-rate and consumer lag rather than expecting a per-function metric. Built this way, a UDF is a durable extension point rather than an operational liability.

Related Pages

• Windowed Aggregations and Joins in ksqlDB