<a id="cmf-sql-custom-connectors"></a>

# Custom Connectors and Formats in Confluent Manager for Apache Flink

Starting with version 2.4.0, Confluent Manager for Apache Flink® (CMF) lets you use custom Flink
connectors and formats in Flink SQL. They follow the same `CREATE TABLE`
syntax as open-source Flink SQL. This makes it possible to read from and
write to systems other than Kafka, for example a change-data-capture (CDC) source
or a filesystem sink, and to use data formats that are not derived from Schema Registry.

You package the connector or format as a JAR, upload it to
[artifact management](../../configure/artifacts.md#cmf-artifacts), and create a table that uses it in an
[environment catalog](../../configure/environment-catalog.md#cmf-environment-catalog) with a `CREATE TABLE`
statement. This is different from the built-in Kafka catalogs, where tables are
backed by Kafka topics and use the `confluent` connector.

## Prerequisites

- The [environment catalog](../../configure/environment-catalog.md#cmf-environment-catalog-enable) and
  [artifact management](../../configure/artifacts.md#cmf-artifacts-enable) must both be enabled.
- Submit statements with the current catalog set to the environment catalog
  (`_env_<environment>`) and the current database set to `default`. For
  details, see [Manage Environment Catalogs in Confluent Manager for Apache Flink](../../configure/environment-catalog.md#cmf-environment-catalog).
- The compute pool’s Flink clusters must have filesystem access to the artifact
  storage, so they can fetch the connector JAR (see
  [Storage access for the Flink cluster](../../configure/artifacts.md#cmf-artifacts-cluster-access)).

## Package a connector or format

Build the connector or format as a JAR that includes its Flink factory. The JAR
must contain a service file at
`META-INF/services/org.apache.flink.table.factories.Factory` that lists the
fully qualified factory class or classes. Target Java 11, and shade any
dependencies that are not on the Flink cluster classpath into the JAR.

Off-the-shelf Flink SQL connector JARs, such as
`flink-sql-connector-mongodb`, already include the factory service file and
can be uploaded as is.

## Upload the connector as a labeled artifact

Upload the JAR as an [artifact](../../configure/artifacts.md#cmf-artifacts-upload) with two labels that
tell CMF how to match it to a table:

- `cmf.confluent.io/sql-artifact-type` — `connector` or `format`.
- `cmf.confluent.io/sql-artifact-identifier` — the Flink factory identifier,
  which is the value you put in the `connector` or `format` table option.

The artifact name must end with a `.jar` extension. When a statement first
uses the table, CMF resolves the connector or format and rejects an artifact
whose name is not a `.jar`, because the Flink cluster loads it with
`ADD JAR`.

Put the artifact metadata and labels in a file, `artifact.json`:

```json
{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "Artifact",
  "metadata": {
    "name": "mongodb-connector.jar",
    "labels": {
      "cmf.confluent.io/sql-artifact-type": "connector",
      "cmf.confluent.io/sql-artifact-identifier": "mongodb"
    }
  },
  "spec": {}
}
```

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

```bash
curl -X POST http://cmf:8080/cmf/api/v1/environments/prod/artifacts \
  -F 'artifact=@artifact.json;type=application/json' \
  -F 'file=@flink-sql-connector-mongodb.jar'
```

For the matching rules and how CMF ships the JAR to the Flink cluster, see
[Reference artifacts in SQL statements](../../configure/artifacts.md#cmf-artifacts-reference-sql).

## Create a table

Create a table in the environment catalog and set the `connector` option (and,
if needed, the `format` option) to the factory identifiers you labeled the
artifacts with. The remaining options in the `WITH` clause are the connector’s
own options.

```sql
CREATE TABLE orders (
  `_id` STRING,
  amount DOUBLE,
  PRIMARY KEY (`_id`) NOT ENFORCED
) WITH (
  'connector' = 'mongodb',
  'uri' = 'mongodb://user:password@mongo:27017/?authSource=admin',
  'database' = 'shop',
  'collection' = 'orders'
);
```

`CREATE TABLE` stores the table definition. It checks that a connector with
the identifier in the `connector` option is available, either built-in or
provided by a labeled artifact in the environment, and rejects the statement if
it is not. It does not check the `format` option and does not download the
connector JAR. Resolution happens the first time a statement uses the table.

## Query and write to the table

Reading from or writing to the table triggers CMF to resolve the connector
(and any format), ship the matching JARs to the Flink cluster, and run the job:

```sql
SELECT * FROM orders;

INSERT INTO orders SELECT * FROM staged_orders;
```

## Limitations

- Connector and format artifacts always resolve to their latest version. There
  is no syntax to pin a table to an older connector or format version.
- `CREATE TABLE AS` (CTAS) and `CREATE OR REPLACE TABLE AS` (RTAS) statements
  are not supported in the environment catalog.
- If more than one artifact in the environment matches the same
  `connector` or `format` identifier, statements that use the table fail with
  an ambiguous-identifier error. Keep one artifact per identifier per environment.
- When CMF compiles a statement, connector and format factory code can load
  only classes whose fully qualified names match the `cmf.sql.classAllowlist`
  configuration. Shade other dependencies into the JAR instead of widening the
  allowlist.
- A single JAR that bundles both a connector factory and a format factory, and is
  labeled as a `connector`, makes the bundled format available only to tables
  that also use that connector. To reuse a format independently, upload it as a
  separate `format` artifact.

## Security considerations

Review the following guidance before you enable custom connectors in production.

#### IMPORTANT
CMF does not sandbox connector or format code. To compile a statement, CMF
loads and runs the connector’s factory in the CMF process, and the Flink
cluster runs the connector during the job. Treat a connector or format JAR as
trusted code that runs with CMF privileges. Review and scan the code before
you upload it, restrict who can upload artifacts and submit statements with
RBAC, and run these workloads on dedicated compute pools rather than shared
ones.

## Related content

- [Manage Environment Catalogs in Confluent Manager for Apache Flink](../../configure/environment-catalog.md#cmf-environment-catalog)
- [Manage Artifacts in Confluent Manager for Apache Flink](../../configure/artifacts.md#cmf-artifacts)
- [User-Defined Functions in Confluent Manager for Apache Flink](user-defined-functions.md#cmf-sql-udf)
- [Flink SQL Examples in Confluent Manager for Apache Flink](examples/index.md#cmf-sql-examples)
- [Table Operations in Confluent Manager for Apache Flink](table-operations.md#flink-sql-table-operations)
- [Features & Support for Statements in Confluent Manager for Apache Flink](features-support.md#flink-sql-features-support)