Configure Custom Validators for Confluent Manager for Apache Flink

A custom validator is Java code that you write and that Confluent Manager for Apache Flink® (CMF) runs on every request that creates or updates a resource. The validator gets the resource as JSON. It accepts the request, or it rejects the request with a reason that CMF returns to the caller. For example, a validator can reject Flink SQL statements that contain DROP DATABASE, or applications that set a Flink configuration option that you don’t allow.

You package your validators in a JAR file and mount the JAR into the CMF pod. CMF loads the validators when it starts.

Important

Custom validators are a preview feature in CMF 2.5. The Java interface, the JSON that CMF passes to it, and the way CMF loads and runs validators might change in future releases.

How CMF runs validators

CMF runs the validators on the following requests. The kind column shows the value of the kind field in the JSON that a validator gets.

`kind`	Requests
`Environment`	Create or update an environment.
`EnvironmentSecretMapping`	Create or update the mapping of a secret to an environment.
`FlinkApplication`	Create or update an application.
`Statement`	Create or update a statement. Stopping, resuming, and rescaling a statement are updates, too.
`ComputePool`	Create or update a compute pool.
`KafkaCatalog`	Create or update a Kafka catalog.
`KafkaDatabase`	Create or update a Kafka database.
`Secret`	Create or update a secret.
`Artifact`	Upload a new artifact or a new version of an artifact. The validator gets the artifact metadata, not the uploaded file.
`KubernetesCluster`	Update a Kubernetes cluster.
`Savepoint`	Create a savepoint, create a detached savepoint, or update a savepoint.

These requests go through the validators no matter where they come from: the REST API, the Confluent CLI, the CMF UI, or the CMF MCP server. CMF first checks that the caller is authorized and that the request is valid. Only then does it run the validators. Requests that read or delete resources don’t go through the validators.

When a request arrives, CMF works as follows:

It runs the validators one after the other, in the alphabetical order of their JAR file names. Within a JAR, the order is the order of the entries in the service file.
If a validator rejects the request, CMF stops and doesn’t run the remaining validators. The caller gets HTTP 400 with an error message that names the validator, the reason, the kind, and the resource name. For example: Rejected by custom validator DestructiveDdlValidator: DROP DATABASE is not allowed here (Statement 'my-statement').
If a validator throws an exception or returns null, the request fails with HTTP 500. CMF doesn’t let a request through that it couldn’t check. The CMF log has the stack trace.
If all validators accept the request, CMF processes it as usual.

Note

Validators also run on resources that existed before you installed them. For example, if a validator rejects statements because of their SQL, users can no longer stop such a statement, because stopping is an update that includes the same SQL. Check the fields that matter for your rule, and accept the request otherwise.

Write a validator

A validator is a Java class that implements the CustomResourceValidator interface. CMF doesn’t publish a Maven artifact for this interface. Copy it into your project with the same package and name:

package io.confluent.cmf.validation.spi;

import tools.jackson.databind.JsonNode;
import java.util.Optional;

public interface CustomResourceValidator {
    Optional<String> validate(JsonNode resource);
}

You only compile against your copy. At runtime, CMF uses its own copy of the interface.

Your validate method must follow these rules:

Return Optional.empty() to accept the request, or Optional.of(reason) to reject it. CMF returns the reason to the caller.
Never return null.
Be thread-safe. CMF creates one instance of each validator and calls it from many requests at the same time.
Be fast. The validator runs while the caller waits for the response.

Data that the validator gets

The resource argument is the request body as a Jackson 3 JsonNode (package tools.jackson.databind).

kind always has the value from the table above. CMF sets it from the endpoint that received the request, so a request with a wrong kind can’t skip your checks.
apiVersion, metadata, and spec are as the caller sent them. Fields that the request doesn’t set are missing.
An Environment has no apiVersion. Depending on the client, its name is in metadata.name or in the older top-level name field.
The JSON isn’t masked. Secret values and credentials in other resources, for example in Kafka catalog options or in Flink configuration, reach your validator in clear text. Don’t log or forward them.

The JSON doesn’t contain the environment name from the request path, whether the request is a create or an update, or who sent the request.

Example

The following validator rejects statements whose SQL contains DROP DATABASE:

package com.example;

import io.confluent.cmf.validation.spi.CustomResourceValidator;
import tools.jackson.databind.JsonNode;
import java.util.Optional;

public class DestructiveDdlValidator implements CustomResourceValidator {
    @Override
    public Optional<String> validate(JsonNode resource) {
        if (!"Statement".equals(resource.path("kind").asString())) {
            return Optional.empty();
        }
        String sql = resource.path("spec").path("statement").asString();
        if (sql.toUpperCase().contains("DROP DATABASE")) {
            return Optional.of("DROP DATABASE is not allowed here");
        }
        return Optional.empty();
    }
}

Register and package the validator

Register the class for the Java ServiceLoader. Add the file src/main/resources/META-INF/services/io.confluent.cmf.validation.spi.CustomResourceValidator to your project. Put the fully qualified name of each validator class on its own line:
```
com.example.DestructiveDdlValidator
```
Add Jackson, and SLF4J if your validator logs, with the provided scope. CMF loads its own copies of these libraries first, so copies in your JAR are never used. Compile against the versions that your CMF release uses, so that the methods you call exist at runtime. CMF 2.5.0 uses the versions in the following example:
```
<dependency>
  <groupId>tools.jackson.core</groupId>
  <artifactId>jackson-databind</artifactId>
  <version>3.1.7</version>
  <scope>provided</scope>
</dependency>

<dependency>
  <groupId>org.slf4j</groupId>
  <artifactId>slf4j-api</artifactId>
  <version>2.0.20</version>
  <scope>provided</scope>
</dependency>
```
If your JAR calls a method that the CMF copy of a library doesn’t have, requests fail with HTTP 500 and the CMF log shows a NoSuchMethodError or NoClassDefFoundError.
Build a JAR that contains your classes, the service file, and any other library that your validator needs. Don’t add the CustomResourceValidator interface, Jackson, or SLF4J to the JAR.

Install validators

CMF loads validators from the JAR files in /opt/cmf/custom-validators when it starts. The directory exists and is empty in the CMF image, so without your JARs CMF runs no validators. To add or change a validator, replace the JAR and restart CMF.

The following steps mount the JAR from a ConfigMap. This works for JAR files up to 1 MiB, the size limit of a Kubernetes object.

Create a ConfigMap from your JAR in the namespace where CMF runs:

kubectl create configmap cmf-custom-validators \
    --from-file=my-validators.jar \
    --namespace <your-cmf-namespace>

Mount the ConfigMap at /opt/cmf/custom-validators with the mountedVolumes Helm value:

# custom-validators-values.yaml
mountedVolumes:
  volumes:
    - name: custom-validators
      configMap:
        name: cmf-custom-validators
  volumeMounts:
    - name: custom-validators
      mountPath: /opt/cmf/custom-validators
      readOnly: true

Upgrade your CMF release:

helm upgrade --install cmf confluentinc/confluent-manager-for-apache-flink \
    --version "~2.5.0" \
    -f custom-validators-values.yaml \
    --namespace <your-cmf-namespace>

Check the CMF log for a line like the following, which lists the validators that CMF loaded:

Loaded 1 custom validator(s) from '/opt/cmf/custom-validators': [com.example.DestructiveDdlValidator]

For a JAR larger than 1 MiB, use an init container that copies the JAR into an emptyDir volume, and mount that volume at /opt/cmf/custom-validators in the CMF container. You can also add the JAR to your own CMF image, as described in Build Confluent Platform for Apache Flink Container Images from Published Artifacts.

CMF doesn’t start if it can’t load a validator. This happens when a JAR in the directory can’t be read, when the service file names a class that is missing or has no public constructor without arguments, or when the configured path is a file and not a directory. The CMF log names the JAR or class that caused the error.

Configuration reference

Helm value	Default	Description
`cmf.customValidators.dir`	`/opt/cmf/custom-validators`	Directory from which CMF loads validator JAR files at startup. If you change it, mount your volume at the new path. If the directory doesn’t exist, CMF starts without validators. Set it to an empty string to turn off custom validators.