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, and create a table that uses it in an 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

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

{
  "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:

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.

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.

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 does not resolve or download the connector JAR yet. 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:

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