Apache Kafka Connect

marmotdata/kafkaconnect

Discover connectors, tasks and topics from Kafka Connect clusters with lineage to the systems they move data between

Pipelines Assets Lineage Experimental

The Kafka Connect plugin discovers connectors, their tasks and the topics they move data through from a Kafka Connect cluster, and links each connector to the tables, collections and buckets it reads or writes.

It talks to the Connect REST API (self-hosted workers and Confluent Cloud) with optional basic auth.

Assets

Type	Name	Provider
Pipeline	connector name	Kafka Connect
Task	`<connector>.task-<id>`	Kafka Connect
Topic	topic name	Kafka

Topics use the identity the Kafka plugin gives them, so a topic both plugins see is one asset with two sources.

Lineage

Pipeline CONTAINS Task
Source connectors: Table FEEDS Pipeline, Pipeline PRODUCES Topic
Sink connectors: Topic FEEDS Pipeline, Pipeline PRODUCES Table, Collection, Bucket or Container

Table, collection and bucket edges are built from the connector config and point at the asset the owning Marmot plugin creates; this plugin never creates those assets itself. Resolvers exist for Debezium (PostgreSQL, MySQL, SQL Server, MongoDB, Oracle) and the matching Confluent Cloud CDC connectors, the Confluent JDBC source and sink, the S3, GCS and Azure Blob sinks, the Snowflake, BigQuery and Elasticsearch sinks, and the MongoDB source and sink. Other connector classes get topic edges only.

Topics come from the worker's active topics endpoint (KIP-558, Kafka 2.5+) when it has records for the connector, otherwise from the config: topics, topic, kafka.topic and each family's naming rule, with RegexRouter transforms applied.

In the UI

Point-and-click, no config file needed.

1 Open Runs Create pipeline
2 Pick Apache Kafka Connect from the plugin list.
3 Fill in the wizard, set a schedule, save.

Read the UI guide

With the CLI

Save a YAML config, then run marmot ingest.

ingest.yaml yaml

name: my-kafkaconnect-pipeline
runs:
  - kafkaconnect:
      host: "<host>"

$ marmot ingest -c ingest.yaml

Read the CLI guide

Not using plugins? Other ways to populate Marmot

Terraform Declare assets as infrastructure-as-code. Pulumi Native TypeScript, Python, or Go resources. Kubernetes Operator Sync assets from Kubernetes custom resources.

Configuration

11 top-level fields. * marks required fields.

tags multiselect

Tags to apply to discovered assets

external_links object[]

External links to show on all assets

name string

Display name for the link

icon string

Icon identifier for the link

url string

URL to the external resource

filter object

Filter discovered assets by name (regex)

include multiselect

Include patterns for resource names (regex)

exclude multiselect

Exclude patterns for resource names (regex)

host string

Kafka Connect REST URL (e.g. http://connect:8083)

username string

Username for basic authentication

password password

Password for basic authentication

verify_ssl bool

Whether to verify the TLS certificate of the Connect REST endpoint

default: true

include_tasks bool

Whether to discover connector tasks as Task assets

default: true

include_topics bool

Whether to discover the topics connectors read and write as Topic assets

default: true

include_config bool

Whether to store each connector's config in its metadata, with credentials masked

default: true

discover_lineage bool

Whether to link connectors to the topics and datasets they move data between

default: true

Assets emitted

Metadata this plugin attaches to each discovered asset.

Pipeline

KafkaConnectPipelineFields

14 fields

The metadata fields the plugin emits for a connector's Pipeline asset.

connector_class string

Java class of the connector

connector_type string

Connector direction (source, sink)

state string

Connector state (RUNNING, PAUSED, FAILED, UNASSIGNED)

worker_id string

Worker the connector instance runs on

task_count int

Number of tasks the connector is split into

task_states object

State of each task, keyed by task id

topics string[]

Topics the connector reads or writes

plugin_version string

Version of the installed connector plugin

connect_version string

Version of the Connect worker

kafka_cluster_id string

Id of the Kafka cluster the worker is attached to

config object

Connector config with credential values masked

description string

Description from the connector config, when set

error string

First 500 characters of the failure trace, when the connector or a task has failed

url string

REST URL of the connector

Task

KafkaConnectTaskFields

5 fields

The metadata fields the plugin emits for a Task asset.

task_id int

Task id within the connector

state string

Task state (RUNNING, PAUSED, FAILED, UNASSIGNED)

worker_id string

Worker the task runs on

connector string

Name of the connector the task belongs to

error string

First 500 characters of the failure trace, when the task has failed

Topic

KafkaConnectTopicFields

3 fields

The metadata fields the plugin emits for a Topic asset.

topic_name string

Topic name

producers string[]

Source connectors writing to the topic

consumers string[]

Sink connectors reading from the topic