# kafka plugin — `kafka:kafka-connect`


## Goal

Creating a new Kafka Connect connector is a mechanical but
error-prone chore: it touches several repos in a specific order, with naming and
tagging conventions that must stay consistent, and output that has to survive
`terraform fmt`, Checkov, and Atlantis. The `kafka-connect` skill turns that into
a guided, **deterministic generator**: you describe the connector you want, and it
produces the Terraform, the Kafka topic resources, the AKHQ registration, and the
Jenkins build/deploy entry — all matching the patterns of the connectors already
running in `terraform-infra`.

It is a **code generator, not a deployment tool**: it never runs `terraform`,
opens PRs, or populates secrets. A human reviews the generated files and drives
the apply.

## Scope

Currently supports three connector types:

| Type | Direction | One-liner |
|---|---|---|
| `snowflake_sink` | Kafka → Snowflake | Sinks Kafka topics into Snowflake tables (Snowpipe Streaming) |
| `jdbc_source_snowflake` | JDBC DB → Kafka | Polls a database table into a Kafka topic with a timestamp watermark — currently templated for **Snowflake** |
| `neo4j_cdc_source` | Neo4j → Kafka | Streams Neo4j graph CDC into Kafka topics via node/relationship patterns |

The JDBC source connector is **general-purpose** — the underlying `jdbc_source`
image (Confluent JDBC + drivers) works with any JDBC-capable database (MySQL,
Postgres, Sybase, Snowflake, …). The plugin currently only ships the **Snowflake
configuration** of it (`jdbc_source_snowflake`), because that matches the verified
instance it was modeled on; a generic non-Snowflake JDBC variant can be added as a
sibling template later.

Other connector types that exist in the `kafka-connect` repo (the JDBC source
against non-Snowflake databases, S3, Debezium, OpenSearch/Elasticsearch/Redis/DynamoDB
sinks, Salesforce, etc.) are **intentionally out of scope for now** — each is mostly a
registry entry plus a template that mirrors an existing instance, and can be added later.

## How it works

The skill is registry-driven. `references/connector_registry.json` is the source
of truth (per-connector inputs, defaults, naming, module versions); Jinja-style
templates under `references/templates/` render the actual Terraform; supporting
references describe the MSK clusters (`msk_clusters.json`) and topic-naming rules
(`topic_naming_rules.md`).

For a single connector it produces / edits:
1. a per-connector Terraform directory (`main.tf`, `variables.tf`, `versions.tf`, and `query.sql` for the Snowflake source) under `terraform-infra/<env>/kafka-infra/…`,
2. the `kafka_topic` resources for the connector's topics,
3. an entry in the AKHQ config so the connector shows up in the AKHQ UI,
4. (optionally) the SERVICES entry in the `kafka-connect` repo's `Jenkinsfile` so Jenkins builds/deploys it.

## Shared features (all connectors)

- **Registry-driven & deterministic** — same inputs produce the same output;
  templates mirror verified production/QA instances rather than inventing structure.
- **Broker resolution via data source** — every connector reads brokers from
  `data "aws_msk_cluster"` (`<env>-<cluster>` → `bootstrap_brokers_tls`); a hardcoded
  broker list is never an input.
- **Naming conventions, composed once** — each connector composes its service name
  into a single `local.service_name` (e.g. `kc-sfsink-<name>`, `kc-jdbc-src-<name>`,
  `kc-neo-src-<name>`) and references it everywhere, so there are no repeated literals.
- **32-char ALB name cap** — the deployed name (`<env>-<service_name>`) is validated
  against AWS's 32-char limit as a hard error, per connector.
- **Topic-name validation** — topic names are checked against the documented
  topic-naming conventions (`cdc.*` / `stream.*` / `event.*` / `etl.*` / `dlq.*`).
- **Tagging** — `application_family` and `team_name` are user-provided and passed to
  `terraform-default-tags` (v2.0.0); the plugin does not validate them — `team_name`
  is validated by the module at `terraform plan` against the live Datadog team list.
- **Monitoring** — Datadog notification + escalation endpoints are gathered per
  connector and wired into the Datadog dashboard module.
- **AKHQ + Jenkins registration** — connectors are registered in AKHQ (UI visibility)
  and, when the user provides the `kafka-connect` repo path, in the Jenkinsfile
  (build/deploy); `deployToProd` is derived from the target environment.
- **Secrets are referenced, not created** — the plugin terraforms the empty Secrets
  Manager resources; the secret *values* (Snowflake key, Neo4j creds) are populated
  by a human after apply.
- **Machine-agnostic** — no hardcoded local paths or usernames; repo roots are
  resolved at runtime or asked for.
- **Guard rails** — refuses to overwrite an existing connector directory; never runs
  Terraform, opens PRs, generates AVRO schemas, or provisions Snowflake/Neo4j users.
- **Suggested rollout** — generation is followed by a checklist: three terraform-infra
  PRs in order (topics → connector → AKHQ) plus the separate Jenkinsfile PR, with
  approvals/Atlantis apply left to the user.

## Per-connector highlights

### `neo4j_cdc_source` (Neo4j → Kafka)

- Streams Neo4j graph change-data-capture into Kafka topics; each topic maps to a
  Cypher **node or relationship pattern**.
- Works against both **on-prem Neo4j and Neo4j Aura** (auto-detected from the URI).
- **Relationship direction is always confirmed with the user** — a relationship name
  alone (e.g. `OWNS`, `BELONGS_TO`) doesn't imply which label is start vs end.
- Requires **CDC to be enabled** on the target database — surfaced as a prerequisite
  with the verification/enable commands.
- Nested under `neo4j_cdc_source/<purpose>/` by default; service name `kc-neo-src-<name>`.
- Auth via `NEO4J_CREDENTIALS` secret (read at startup by the Lenses AWS secret provider).


### `jdbc_source` (JDBC DB → Kafka)

- Polls a database table into a Kafka topic using the general-purpose JDBC source
  connector, with a timestamp **watermark** column for incremental reads. The JDBC
  connector works with any JDBC-capable database; this variant is configured for
  **Snowflake** (`SNOWFLAKE_CONFIGURATION=true` + Snowflake-native settings).
- Nested under `jdbc_source/snowflake/<purpose>/` by default; service name `kc-jdbc-src-<purpose>`.
- Auth via Snowflake key-pair.

### `snowflake_sink` (Kafka → Snowflake)

- Writes Kafka topics into Snowflake tables via Snowpipe Streaming.
- **Target table names are auto-derived** from each topic (`dot → __`,
  camelCase → snake, UPPERCASE; e.g. `cdc.musicGraphV5.acrid` → `CDC__MUSIC_GRAPH_V5__ACRID`);
  the user only supplies per-topic overrides for pre-existing tables.
- **Snowflake role is conditionally required** — needed under Snowpipe Streaming
  (the default ingestion method), skipped for plain Snowpipe.
- Lands in the shared `snowflake_sink/<purpose>/` catalog; service name `kc-sfsink-<name>`.
- Auth via Snowflake key-pair (private key in Secrets Manager).

## Where to go deeper

- `skills/kafka-connect/SKILL.md` — the full workflow the skill follows (gather →
  validate → render → register → report).
- `skills/kafka-connect/references/connector_registry.json` — per-connector source of truth.