<a id="cmf-custom-validators"></a>

# 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.

<a id="cmf-custom-validators-run"></a>

## 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<br/>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<br/>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.

<a id="cmf-custom-validators-write"></a>

## 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:

```java
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-custom-validators-run). 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`:

```java
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

1. 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:
   ```text
   com.example.DestructiveDdlValidator
   ```
2. 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:
   ```xml
   <dependency>
     <groupId>tools.jackson.core</groupId>
     <artifactId>jackson-databind</artifactId>
     <version>3.1.7</version>
     <scope>provided</scope>
   </dependency>
   <!-- Only if the validator logs -->
   <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`.
3. 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.

<a id="cmf-custom-validators-install"></a>

## 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.

1. Create a ConfigMap from your JAR in the namespace where CMF runs:
   ```bash
   kubectl create configmap cmf-custom-validators \
       --from-file=my-validators.jar \
       --namespace <your-cmf-namespace>
   ```
2. Mount the ConfigMap at `/opt/cmf/custom-validators` with the
   `mountedVolumes` Helm value:
   ```yaml
   # 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
   ```
3. Upgrade your CMF release:
   ```bash
   helm upgrade --install cmf confluentinc/confluent-manager-for-apache-flink \
       --version "~2.5.0" \
       -f custom-validators-values.yaml \
       --namespace <your-cmf-namespace>
   ```
4. Check the CMF log for a line like the following, which lists the
   validators that CMF loaded:
   ```text
   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](build-images.md#cpf-build-images).

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<br/>you change it, mount your volume at the new path. If the directory<br/>doesn’t exist, CMF starts without validators. Set it to an empty<br/>string to turn off custom validators. |

## Related content

- [Install Confluent Manager for Apache Flink with Helm](helm.md#install-cmf-helm)
- [Build Confluent Platform for Apache Flink Container Images from Published Artifacts](build-images.md#cpf-build-images)
- [Configure Authorization for Confluent Manager for Apache Flink](authorization.md#cmf-authorize)
- [Create Statements with Confluent Manager for Apache Flink](../jobs/sql-statements/create-statements.md#flink-sql-statements-create)