--- name: kafka-connect description: Generates a new Kafka Connect connector instance as Terraform in the terraform-infra//kafka-infra/ tree, plus the corresponding Kafka topic resources in kafka-cluster*/topics/. Use this skill when the user asks to "create a snowflake sink connector", "write to snowflake from kafka", "add a snowflake sink", "create a snowflake source", "read from snowflake into kafka", "snowflake jdbc source with watermark", "create a neo4j source", "neo4j cdc connector", "stream neo4j changes to kafka", "kafka connect terraform for snowflake/neo4j", or similar phrases. Supports three connector types — `snowflake_sink` (Kafka topics → Snowflake tables via Snowpipe Streaming), `jdbc_source_snowflake` (Snowflake table → Kafka topic via JDBC, with query.sql + timestamp watermark), and `neo4j_cdc_source` (Neo4j graph CDC → Kafka topics, with node/relationship pattern mappings). Validates topic naming, MSK cluster choice, required fields, and warns when the underlying connector type may need new env vars in the kafka-connect repo. Does not generate AVRO schemas, does not run terraform, does not open PRs. --- # Kafka Connector — terraform-infra generator This skill is a **deterministic code generator** for Kafka Connect connector instances. It is scoped to three connector types: Snowflake sink, Snowflake source (JDBC), and Neo4j CDC source. Other connector types (JDBC MySQL/Postgres, S3, Debezium) are intentionally out of scope. It produces: 1. A new directory under `terraform-infra//kafka-infra//` containing `main.tf`, `variables.tf`, `versions.tf`, and (for `jdbc_source_snowflake`) `query.sql`. The `` comes from the connector's `instance_path_pattern` in the registry: - `snowflake_sink` → nested: `snowflake_sink//` - `jdbc_source_snowflake` → nested: `jdbc_source/snowflake//` - `neo4j_cdc_source` → nested: `neo4j_cdc_source//` 2. New `kafka_topic` resource block(s) appended to (or created as) `terraform-infra//kafka-infra/kafka-cluster*/topics/_topics.tf`. For `neo4j_cdc_source`, **one topic per `topic_mappings` entry** is generated. 3. A validation report (errors block, warnings don't) and a checklist of related changes the user must still make in the `kafka-connect` repo. It is **not** a deployment tool. It does not run terraform, does not commit, does not open PRs. ## Prerequisites - The user has the `terraform-infra` repo checked out locally. **Resolve its root at runtime** (from the working directory or by asking) — never assume an absolute path or username; paths in this skill are repo-relative. - The user has the `kafka-connect` repo checked out (used for the detect-and-warn step). - A registry of supported connector types lives at `references/connector_registry.json`. - MSK clusters and naming rules live at `references/msk_clusters.json` and `references/topic_naming_rules.md`. ## Workflow ### Step 1 — Parse user intent and gather inputs Extract from the user's message any of: - **Connector type** — `snowflake_sink`, `jdbc_source_snowflake`, or `neo4j_cdc_source`. Disambiguate as below. - **Purpose** — short snake_case identifier (e.g. `payments`, `label_participant`, `nr_graph`, `neo4j_v5`, `mg_fingerprint`). - **Environment** — `dev`, `qa`, or `prod`. - **MSK cluster** — pick a key from `msk_clusters.json` (may be inferred from environment). The skill uses its `msk_cluster_name_short` for `var.msk_cluster_name`; **every connector resolves brokers via `data "aws_msk_cluster" "kafka_infra" { cluster_name = "${var.environment}-${var.msk_cluster_name}" }` → `bootstrap_brokers_tls`** — never a hardcoded broker list. Do not ask the user for bootstrap servers. - **`application_family`** — user-provided (passed to `terraform-default-tags`). At ask-time, **list** `terraform-infra/prod/iam/application_family/` (the directory contains one entry per known family) and surface the entries as `AskUserQuestion` options so the user can pick one. **Do NOT validate** the user's choice against the list — they may be adding a new family that doesn't yet exist in that directory. Accept whatever the user provides. Required for all connector types. (Note: `terraform-default-tags` itself doesn't validate this string at plan time either — the directory list is org convention, not module-enforced.) - **`team_name`** — user-provided, **required for ALL connector types** (all three now use `terraform-default-tags` 2.0.0). Ask the user for the value. **No validation in the plugin** — accept any input. `terraform-default-tags` 2.0.0 itself validates the value at plan time against the live Datadog team-handle list (`data "datadog_teams" "pde"`); if invalid, `terraform plan` fails with the full list of valid handles. - **`notification_endpoints`** and **`escalation_notification_endpoints`** — user-provided. Both passed to the `terraform-datadog/kafka_connector` module. These are Datadog monitor mention strings (e.g. `@slack-`, `@pagerduty-`; space-separated for multiple). Ask the user for BOTH at creation time — required for all three connector types. **No validation, no hardcoded defaults.** Common starting points to suggest (do not enforce): `@slack-kafka-data-highway-alerts`, `@slack-data-alerts-`, `@slack-data-alarm-`, `@pagerduty-`. Make it clear when asking that `escalation_notification_endpoints` is for higher-severity / paged alerts (typically includes a `@pagerduty-...` mention) while `notification_endpoints` is for routine alerts (typically Slack-only). - **`kafka_connect_repo_path`** (optional) — ask the user whether they have the `kafka-connect` repo (separate from `terraform-infra`) checked out locally. If yes, **ask them for its absolute path** — do NOT propose or pre-fill a path, and never assume a username or `/Users/...` prefix (the plugin runs on many developers' machines with different layouts). You MAY offer to locate it: it's commonly a sibling of the terraform-infra repo (`/kafka-connect`); if you find a candidate, confirm it with the user before using it. Used to register the new connector in the repo's `Jenkinsfile` SERVICES map — see Step 5c. If the user doesn't provide/confirm a path, Step 5c emits the change as a manual follow-up instead. - **`deploy_to_prod`** (Jenkinsfile flag) — **derived from `environment`**: `prod` → `true`, `qa`/`dev` → `false`. Don't ask the user; set it from the chosen environment. The user may still override if they explicitly request it. Only used by Step 5c. - **Topics** — for sinks: list of input topics; for sources: the single produced topic (Snowflake source) or one topic per `topic_mappings` entry (Neo4j source). - **Topic middle segment — ALWAYS confirm with the user.** Topics follow the convention `..` (e.g. `cdc.fingerprinting.vendorLabel`). The middle segment (`` / dataset namespace — e.g. `fingerprinting`, `musicGraphV5`, `facts`) is NOT necessarily the same as the connector `purpose`/`connector_name`. Before building any topic name, use `AskUserQuestion` to confirm the middle segment to use. Propose a default (often the purpose) but let the user override. Apply the confirmed middle segment consistently across every topic for this connector. - **Snowflake fields** (`snowflake_sink` / `jdbc_source_snowflake`) — database, schema, warehouse (source only), user, table (source only), and **role**: - `jdbc_source_snowflake`: role required. - `snowflake_sink`: role is **conditionally required, keyed on the ingestion method**. Default `SNOWFLAKE_INGESTION_METHOD` is `SNOWPIPE_STREAMING`, which **requires** a role — so by default **ask the user for `snowflake_role`**. If the user instead chooses plain `SNOWPIPE`, **skip role entirely** (don't ask, don't emit — it isn't consumed). Confirm the ingestion method (default `SNOWPIPE_STREAMING`) so you know whether to ask. See `connector_registry.json[snowflake_sink].snowflake_role_note`. - **Topic→table map (`snowflake_sink` only) — DERIVED by default; do NOT ask for it.** For each input topic, default the Snowflake table name to: split the topic on `.`, convert each segment camelCase→snake_case, UPPERCASE, and join segments with `__` (double underscore). Example: `cdc.musicGraphV5.acrid` → `CDC__MUSIC_GRAPH_V5__ACRID`. Only ask the user for **overrides** (`topic_table_overrides`) when a topic must land in a pre-existing table whose name differs from the derived form — e.g. some older prod tables used a single `_` for the dot (`event.musicEvent.socialMedia` → `EVENT_MUSIC_EVENT_SOCIAL_MEDIA`), which would need an explicit override. The skill resolves the final map (derived ∪ overrides) and renders it into `SNOWFLAKE_TOPIC_TABLE_MAP`. Full spec: `connector_registry.json[snowflake_sink].topic_to_table_naming`. - **Watermark column** — for `jdbc_source_snowflake`: column name used as `TIMESTAMP_COLUMN_NAME` (e.g. `CREATED_AT`). - **Neo4j fields** (`neo4j_cdc_source`) — `neo4j_instance_type` (`on_prem` or `aura`), `neo4j_server_uri`, `neo4j_database_name`, `connector_name`, and `topic_mappings` (list of `{mapping_key, topic, pattern, key_strategy?}`). Each mapping declares one Kafka topic plus a Cypher node/relationship pattern. Optional: `neo4j_connection_timeout`, `neo4j_connection_acquisition_timeout` (recommend `60s` each for Aura). See `connector_registry.json[neo4j_cdc_source].pattern_syntax_reference` for full pattern syntax. Authoritative docs: - CDC mode: https://neo4j.com/docs/kafka/current/source/cdc/ - QUERY mode (not currently supported by the current kafka-connect image): https://neo4j.com/docs/kafka/current/source/query/ Quick examples: - Node, all changes: `(:LabelSoundRecording)` - Node, include props: `(:User{name, surname})` - Node, exclude props: `(:User{-address, -dob})` - Relationship, any-to-any: `()-[:HAS_PROFILE]->()` - Relationship, typed ends: `(:Vendor)-[:HAS_LABEL_PARTICIPANT]->(:LabelParticipant)` **Relationship direction — ALWAYS confirm with the user (do not infer).** When the user provides a relationship together with node labels, the start → end direction and which labels sit on each end are NOT obvious from the relationship name. Before generating any relationship pattern, use `AskUserQuestion` to confirm, for each relationship, the exact `(:StartLabel)-[:REL]->(:EndLabel)` shape — including direction (which side is the arrow tail vs head) and the endpoint labels (or `()` for any-to-any). Never guess the direction or endpoints. Present the candidate combinations and let the user pick/correct them. Only after the user confirms do you build the `topic_mappings` entries. **Connector-type disambiguation:** - "write to snowflake", "snowflake sink", "kafka into snowflake", "load kafka topic into a snowflake table" → `snowflake_sink` - "read from snowflake", "snowflake source", "snowflake jdbc source", "ingest snowflake table into kafka", references to a watermark column on a Snowflake table → `jdbc_source_snowflake` - "neo4j source", "neo4j cdc", "stream neo4j to kafka", "graph cdc", node/relationship pattern mentions → `neo4j_cdc_source` - **on-prem vs Aura** — set `neo4j_instance_type`: - URI matches `^neo4j\+s://[a-z0-9]+\.databases\.neo4j\.io$` → `aura` - User mentions "aura", "cloud neo4j", "managed neo4j", "neo4j cloud" → `aura` - Anything else (internal hostnames like `*.theorchard.io`, explicit port `:7687`) → `on_prem` - Always confirm with the user when auto-detecting; defaults differ (e.g. database name `neo4j` for Aura vs `graph.db` on-prem). - If the user describes something unrelated to Snowflake or Neo4j (S3, Postgres, MySQL/Debezium, JDBC against MySQL, etc.) → **stop** and tell the user this skill is scoped to Snowflake + Neo4j only; other connector types are out of scope. Use `AskUserQuestion` for any required field still missing after parsing. Do **not** invent defaults for sensitive Snowflake fields (`snowflake_warehouse`, `snowflake_role`, `snowflake_user`) or Neo4j fields (`neo4j_server_uri`, `neo4j_database_name`). Required-field reference per connector type is in `connector_registry.json` → `.required_inputs`. ### Step 2 — Validate inputs Hard errors (block generation; report all then stop): 1. `connector_type` must be `snowflake_sink`, `jdbc_source_snowflake`, or `neo4j_cdc_source`. Reject anything else with a message that this skill is scoped to Snowflake + Neo4j only. 2. `environment` must be one of `dev`, `qa`, `prod`. 3. **Naming**: `` must be snake_case (no hyphens, no dots). The Fargate `service_name` is set by the template, and the terraform-fargate module prepends `-` to produce the deployed AWS name: - `snowflake_sink` → service_name composed once in variables.tf as `local.service_name = "${var.cluster_name}-${var.connector_type}-${var.connector_name}"` = `kc-sfsink-` (defaults: `cluster_name=kc`, `connector_type=sfsink`). `connector_name` is the hyphenated instance id — **derive it from `` by replacing `_` with `-`** unless the user gives one explicitly (e.g. purpose `mg_gsr` → connector_name `mg-gsr`). The directory still uses the snake_case `` (`snowflake_sink/mg_gsr/`). Deployed name `-kc-sfsink-` (e.g. `prod-kc-sfsink-mg-gsr`). Every name reference uses `local.service_name` — never repeat the composed string inline. - `jdbc_source_snowflake` → service_name `${var.connector_type}-` where `var.connector_type` defaults to `kc-jdbc-src` (e.g. `kc-jdbc-src-label-participant`). - `neo4j_cdc_source` → service_name `kc-neo-src-`; deployed name `-kc-neo-src-` (e.g. `prod-kc-neo-src-fingerprinting`). `connector_name` is the user-provided short identifier (no env / kc / neo / src prefix). 3a. **Name length (32-char ALB limit) — applies to ALL connectors.** The terraform-fargate module names the ALB `-` and AWS caps ALB names at 32 chars. Compute the deployed-name length per connector and HARD ERROR if it exceeds 32. Each connector's full `name_length_limit` (deployed pattern, formula, per-env max) is in `connector_registry.json[].name_length_limit`. Quick reference (using each connector's defaults): | Connector | Deployed-name pattern | prod max | qa max | dev max | |---|---|---|---|---| | `snowflake_sink` | `-kc-sfsink-` | name ≤ 17 | ≤ 19 | ≤ 18 | | `jdbc_source_snowflake` | `-kc-jdbc-src-` | purpose ≤ 15 | ≤ 17 | ≤ 16 | | `neo4j_cdc_source` | `-kc-neo-src-` | name ≤ 16 | ≤ 18 | ≤ 17 | If any name overflows, the skill proposes shortening the `connector_name`/`purpose` before regenerating. 4. **Topic naming**: every topic must match a pattern from `references/topic_naming_rules.md`. Reject `cdc.*`, `stream.*`, `event.*`, `etl.*`, `dlq.*` violations as a hard error unless the user explicitly says to ignore — then keep it but emit a warning. 5. **MSK cluster** must be a key in `msk_clusters.json` and its `environment` field must equal the requested environment. 6. **No plaintext secrets**: no input value may look like a password, AWS key, private key, or JWT. Reject if found. 7. **Required type-specific fields** present per the registry. 7a. **(`snowflake_sink`) Snowflake role vs ingestion method**: if `SNOWFLAKE_INGESTION_METHOD` is `SNOWPIPE_STREAMING` (the default), `snowflake_role` is **required** — if missing, ask for it before generating. If the method is plain `SNOWPIPE`, `snowflake_role` must be **omitted** (not emitted). See `connector_registry.json[snowflake_sink].conditionally_required_inputs`. 8. **Target directory must not already exist** at `terraform-infra//kafka-infra//`. If it does, refuse and tell the user to delete or rename — this skill never overwrites existing infra. 9. **Neo4j Aura validation** (when `neo4j_instance_type = "aura"`): - `neo4j_server_uri` must match `^neo4j\+s://[a-z0-9]+\.databases\.neo4j\.io$` (TLS scheme, no port). - `NEO4J_ENCRYPTION_ENABLED` must be `"true"` — reject if the user tries to set `false`. - If `neo4j_database_name` is unset, default to `"neo4j"` (not `"graph.db"`). 10. **Neo4j on-prem validation** (when `neo4j_instance_type = "on_prem"`): - `neo4j_server_uri` must include a port (typically `:7687`) and an internal hostname. - If `neo4j_database_name` is unset, default to `"graph.db"`. Warnings (don't block): - For sinks: a topic in `topics` is not yet terraformed in `kafka-cluster*/topics/`. Surface and ask whether to also generate it. - For sources: remind the user the produced topic must exist before the connector starts. - Default `task_cpu` / `task_memory` from the registry differ from values the user explicitly provided. - For `jdbc_source_snowflake`: warehouse / role / user values may be placeholders the user must verify before applying. - For `neo4j_cdc_source`: **CDC must be enabled on the Neo4j database** (see Step 2.5 below). Always emit a top-line warning even when the user confirms — the connector will fail to start otherwise. ### Step 2.5 — Neo4j CDC enablement confirmation (neo4j_cdc_source only) Before rendering files, **ask the user to confirm CDC is enabled** on the target Neo4j database. CDC is OFF by default on both Neo4j Enterprise and Aura. The connector requires `txLogEnrichment` to be `DIFF` or `FULL` and will fail to start otherwise. Tell the user how to verify and enable: **Verification (any instance type)** — run against the target database via Cypher Shell or Neo4j Browser: ```cypher SHOW DATABASES YIELD name, options WHERE name = "" RETURN name, options.txLogEnrichment AS cdcMode ``` Expected: `cdcMode` is `DIFF` or `FULL`. If it's `OFF`, CDC must be enabled before the connector will work. **Enable on on-prem Neo4j Enterprise 5** — admin runs against the `system` database: ```cypher ALTER DATABASE SET OPTION txLogEnrichment 'FULL' ``` (`FULL` captures old + new property values; `DIFF` only captures changes. Database restarts briefly.) Docs: https://neo4j.com/docs/cdc/current/getting-started/ **Enable on Aura** — Aura console → Database → CDC → Enable. The instance restarts briefly. Docs: https://neo4j.com/docs/aura/auradb/managing-databases/cdc/ This step does **not block generation** — the user can terraform the connector before CDC is enabled (the connector just won't start). But the validation report must include a top-line warning + the enablement steps as follow-up action #1. ### Step 3 — Confirm output location Default base path: - `snowflake_sink` → `terraform-infra//kafka-infra/snowflake_sink//` - `jdbc_source_snowflake` → `terraform-infra//kafka-infra/jdbc_source/snowflake//` - `neo4j_cdc_source` → `terraform-infra//kafka-infra/neo4j_cdc_source//` Use `AskUserQuestion` to confirm the absolute base path. Show the full path (resolve relative to the user's home if needed). The user may override (e.g. for a fork or worktree). For topics, the path is determined by the chosen MSK cluster's `topics_dir` field in `msk_clusters.json`. ### Step 4 — Detect-and-warn for kafka-connect repo This check only runs **if the user provided `kafka_connect_repo_path`** (Step 1). It uses that path — never a hardcoded/assumed location. - **If `kafka_connect_repo_path` was provided**: look up `connector_registry.json[connector_type].kafka_connect_dir` (`snowflake_sink`, `jdbc_source`, or `neo4j_cdc_source`) and optional `kafka_connect_subdir` (`snowflake` for `jdbc_source_snowflake`), and check whether `//` exists. - If it **does not exist**: stop with a hard error — the connector image must exist in the kafka-connect repo before generating Terraform for it. - If it exists: continue. (Do NOT compare `.env.shadow` env vars — that file is local-dev only; the prod path sets env vars via Fargate.) - **If `kafka_connect_repo_path` was NOT provided**: skip this check. Emit a warning in the report that the connector-image existence couldn't be verified, and that the connector type is assumed to already exist in the kafka-connect repo. Do **not** modify the kafka-connect repo at this step (the Jenkinsfile edit happens in Step 5c). ### Step 5 — Render Terraform files Render templates from `references/templates/` substituting all collected inputs. Use the template named in `connector_registry.json[connector_type].template`. Each connector has its own variables template (`variables_template` field). For `jdbc_source_snowflake`, also render `query.sql.j2`. Module versions come from the registry: `_meta.module_versions_default`, overridden per-connector by `module_version_overrides` if present. - `snowflake_sink` pinned to verified-prod: terraform-fargate 6.1.1, default-tags 2.0.0, datadog 6.13.4, terraform 1.11.4 (application_family is user-provided). - `jdbc_source_snowflake` uses latest verified: terraform-fargate 6.5.0, default-tags 2.0.0, datadog 6.18.2, terraform 1.14.3 (application_family is user-provided). - `neo4j_cdc_source` pinned to qa neo4j_v5: terraform-fargate 6.3.0, default-tags 2.0.0, datadog 6.15.3, terraform 1.14.0 (application_family is user-provided). `versions.tf` is rendered with `omit_datadog_provider=true`. All three connectors use **terraform-default-tags 2.0.0**, so `team_name` is a required input for every connector type. Files to write: ``` snowflake_sink: terraform-infra//kafka-infra/snowflake_sink// ├── main.tf (from snowflake_sink.tf.j2) ├── variables.tf (from snowflake_sink_variables.tf.j2) └── versions.tf (from versions.tf.j2) jdbc_source_snowflake: terraform-infra//kafka-infra/jdbc_source/snowflake// ├── main.tf (from jdbc_source_snowflake.tf.j2) ├── variables.tf (from jdbc_source_snowflake_variables.tf.j2) ├── versions.tf (from versions.tf.j2 with omit_datadog_provider=true) └── query.sql (from query.sql.j2) neo4j_cdc_source: terraform-infra//kafka-infra/neo4j_cdc_source// ├── main.tf (from neo4j_cdc_source.tf.j2) ├── variables.tf (from neo4j_cdc_source_variables.tf.j2) └── versions.tf (from versions.tf.j2 with omit_datadog_provider=true) AKHQ registration (all connector types — see Step 5b): terraform-infra//kafka-infra/akhq/main.tf └── append { name, url } entry under akhq.connections..connect: Jenkinsfile registration (separate repo — see Step 5c): /Jenkinsfile └── append SERVICES entry under the matching section comment (e.g. // Neo4j CDC Source) (Applied locally only if the user provided kafka_connect_repo_path; otherwise a manual follow-up.) ``` For each topic that needs to be created, render `topic_block.tf.j2` and append to: ``` terraform-infra//kafka-infra//_topics.tf ``` (`` from `msk_clusters.json` for the chosen cluster.) If the file already exists, append to it; if not, create it. After writing, run `terraform fmt` on the generated directory if the `terraform` CLI is available (best effort — don't fail if missing). ### Step 5b — Register the connector in AKHQ After generating the connector + topics, **also append the new connector to the AKHQ terraform config** so it appears in the AKHQ UI. AKHQ won't auto-discover Fargate-deployed Kafka Connect instances — they have to be listed explicitly inside the YAML heredoc in `akhq/main.tf`. 1. **Locate the file**: `terraform-infra//kafka-infra/akhq/main.tf` (the exact path is in `msk_clusters.json` → `akhq_terraform_path`). 2. **Find the right cluster block**: inside the `AKHQ_CONFIGURATION` heredoc, find `akhq.connections.:` — the key comes from `msk_clusters.json[].akhq_connection_key` (e.g. `prod-managed-kafka-cdc-destination` → `prod-cdc`). 3. **Compute the new entry**: - `name`: the human-friendly AKHQ label — render the connector's `akhq_display_name_pattern` from the registry (this is NOT the service name). Examples: - `neo4j_cdc_source` with `connector_name = fingerprinting` → `neo-src-fingerprinting` - `snowflake_sink` with `connector_name = mg-gsr` → `snowflake-sink-mg-gsr` - `jdbc_source_snowflake` with `purpose = label_participant` → `jdbc-snowflake-src-label-participant` - `url`: always `https://-.theorchard.io` — the terraform-fargate module names the ALB `-`, and AKHQ talks to the Kafka Connect REST over that ALB on 443/HTTPS. (For `snowflake_sink` the service name is `kc-sfsink-`, so the URL host is `-kc-sfsink-`.) 4. **Append** the new two-line item at the end of the existing `connect:` list for that cluster, immediately before the next sibling (`ksqldb:` or the next cluster block). Preserve the 16-space indentation used inside the heredoc: ```yaml - name: url: "https://-.theorchard.io" ``` Use the `Edit` tool. Do **not** reorder existing entries or modify other cluster blocks. 5. **If the cluster's connect block doesn't exist yet** (some clusters in `akhq/main.tf` don't have a `connect:` list at all): emit a warning, list the connector entry in the validation report under follow-ups so the user can add it manually, and continue. This step does NOT live in a separate template — it's a targeted text edit inside `akhq/main.tf`. The atlantis apply for the AKHQ config is a separate PR/apply cycle from the connector itself; mention this in the follow-up report. ### Step 5c — Register the connector in the kafka-connect Jenkinsfile For the connector to actually be **built and deployed** by Jenkins, it must be listed in the kafka-connect repo's `Jenkinsfile` — specifically in the `SERVICES = [...]` Groovy map. This file lives in a **separate repo** (`theorchard/kafka-connect`), not in `terraform-infra`. **Conditional behavior:** - **If the user provided `kafka_connect_repo_path` in Step 1**: edit `/Jenkinsfile` directly. - **If they did not**: skip the edit and emit a precise manual follow-up in the validation report. **What to insert:** ```groovy '' : [project: '', deployToProd: ], ``` - ``: the same value passed to terraform-fargate's `service_name` (no env prefix). E.g. `kc-neo-src-fingerprinting` for `neo4j_cdc_source`, `kc-jdbc-src-label-participant` for `jdbc_source_snowflake`, `kc-sfsink-mg-gsr` for `snowflake_sink`. - ``: from the registry — `snowflake_sink` / `jdbc_source` / `neo4j_cdc_source`. - ``: derived from `environment` — `true` for `prod`, `false` for `qa`/`dev`. (User may override on explicit request.) **Where to insert:** Find the section comment matching the connector's `jenkinsfile_section_comment` (e.g. `// Neo4j CDC Source`). Append the new entry at the **end of that section's existing entries**, before the next section comment. Preserve the existing alignment/quoting style of nearby entries. **Already-present service (qa→prod promotion):** the SERVICES key is the bare `service_name` (env-independent — the env prefix only appears on the deployed ALB name, not here). So the SAME connector created first in qa then in prod maps to ONE SERVICES entry. Before inserting, check whether `''` already exists in the map: - If it exists and this run is for `prod` (so `deploy_to_prod = true`) while the existing entry has `deployToProd: false`, **update that entry's flag to `true`** instead of adding a duplicate. - If it exists and already matches, do nothing (idempotent) — note it in the report. - Only append a brand-new entry when the service_name isn't present. **Skill behavior:** 1. If `kafka_connect_repo_path` was provided: verify `/Jenkinsfile` exists; locate the existing `''` entry (update its `deployToProd` if needed per the promotion rule above) or, if absent, locate the section comment and append the new entry via the `Edit` tool. If the section comment isn't found (uncommon but possible), emit a warning and treat as "path not provided" — fall back to the follow-up. 2. If not provided: include in the validation report follow-ups the exact line and where to put it, with file path, e.g.: *"In `/kafka-connect/Jenkinsfile`, under the `// Neo4j CDC Source` section in the SERVICES map, add (or, if it already exists, set deployToProd accordingly): `'kc-neo-src-fingerprinting' : [project: 'neo4j_cdc_source', deployToProd: true],`"*. Note: the Jenkinsfile change goes to a **separate repo** with its own PR/merge cycle. The connector image won't build until that PR merges. This is independent of the terraform-infra apply. ### Step 6 — Emit validation report Print a structured summary to chat: ``` ✓ Generated / in - main.tf (~XX lines) - variables.tf (~XX lines) - versions.tf (XX lines) - query.sql (jdbc_source_snowflake only) - /_topics.tf (N topic resources) Validation: Errors: none Warnings: Follow-up actions (the skill did NOT do these): 1. (snowflake_sink / jdbc_source_snowflake) Provision the Snowflake **service user + role** with key-pair auth. Define them in Terraform under `terraform-infra/prod/snowflake/delphi/` — the service user goes in `service_users/` (and its RSA **public** key under `service_users/keys/`), the role in `roles/`. Generate the RSA key pair following the Snowflake key-pair guide: https://app.notion.com/p/Snowflake-7c88cc17b0034e7db669a88fd2962bab#debbb574070e4059afd93520d392bca3 (The `snowflake.user.name` / `snowflake.role.name` the connector uses must match the user/role created here.) 2. Populate the SNOWFLAKE_PRIVATE_KEY secret value in AWS Secrets Manager at `//SNOWFLAKE_PRIVATE_KEY` (and SNOWFLAKE_PRIVATE_KEY_PASSPHRASE for snowflake_sink) — this is the **private** half of the key pair from step 1. **Note**: the secret *resources* themselves are created automatically by atlantis when the terraform PR applies (via the `terraform-secrets-manager` module the plugin emits). This step is only about populating the actual secret *values* in the AWS console / CLI after the apply completes — terraform creates an empty placeholder. Open THREE separate PRs in terraform-infra against master, in this order: 3. **PR 1 — Topics** (apply FIRST so topics exist before any source connector starts): - Changes: `/kafka-infra/kafka-cluster*/topics/_topics.tf` - Get approval → `atlantis apply` → merge. 4. **PR 2 — Connector**: - Changes: `/kafka-infra//` (main.tf, variables.tf, versions.tf, query.sql if applicable) - Get approval → `atlantis apply` → merge. 5. **PR 3 — AKHQ registration**: - Changes: `/kafka-infra/akhq/main.tf` - Different terraform state — separate atlantis apply cycle. - Get approval → `atlantis apply` → merge. 6. **Jenkinsfile PR** (separate `kafka-connect` repo): - Status: APPLIED LOCALLY (if user provided `kafka_connect_repo_path`) — user still needs to commit + open the PR. - Status: MANUAL FOLLOW-UP (if path not provided) — show the user the exact line and section to add. - Without this PR merged, Jenkins won't build/deploy the new connector image. 7. (Sources) Confirm the produced topic(s) exist before merging the connector PR (PR 1 above must apply first). 8. (neo4j_cdc_source) **Confirm CDC is enabled on the target Neo4j database** — verify with: `SHOW DATABASES YIELD name, options WHERE name = "" RETURN options.txLogEnrichment` Expected: `DIFF` or `FULL`. If `OFF`: - on-prem: `ALTER DATABASE SET OPTION txLogEnrichment 'FULL'` (run against `system` db as admin) - Aura: enable via Aura console → Database → CDC → Enable Also populate the NEO4J_CREDENTIALS secret value (`{"username": "...", "password": "..."}`) in AWS Secrets Manager **after the terraform PR applies** (the secret resource is created by atlantis via the `terraform-secrets-manager` module the plugin emits; only the value needs to be filled in by the user). The Neo4j user + password must exist on the Neo4j side before the connector can connect. ``` ### Step 7 — Stop Do not run terraform, do not stage git changes, do not open a PR. The user is responsible for those steps. ## Inputs reference For the canonical list of required and default inputs per connector type, see `references/connector_registry.json`. The following are **always** required: - `connector_type`, `purpose`, `environment`, **MSK cluster** (pick a key from `msk_clusters.json`; the skill passes its `msk_cluster_name_short` to `var.msk_cluster_name` and the connector resolves brokers via `data.aws_msk_cluster` — **no bootstrap-server list is ever an input**), `application_family` (user-provided, no validation; surface options from `/prod/iam/application_family/`), `team_name` (user-provided, no validation; Datadog-validated at plan time by terraform-default-tags 2.0.0 — required for ALL connectors), `notification_endpoints`, `escalation_notification_endpoints` (both user-provided, no validation; Datadog monitor mentions) - For `snowflake_sink`: `topics`, `dlq_topic_name`, `snowflake_database`, `snowflake_schema`, `snowflake_user`. (`snowflake_role` is **conditionally required** — required under the default `SNOWPIPE_STREAMING` ingestion method, skipped for plain `SNOWPIPE`. Target table per topic is **derived by default** — see `topic_to_table_naming`; `topic_table_overrides` is optional.) - For `jdbc_source_snowflake`: `snowflake_database`, `snowflake_schema`, `snowflake_table`, `snowflake_warehouse`, `snowflake_role`, `snowflake_user`, `mode`, `timestamp_column_name`, `topic_prefix`. - For `neo4j_cdc_source`: `neo4j_instance_type` (`on_prem` | `aura`), `neo4j_server_uri`, `neo4j_database_name`, `topic_mappings` (list of `{mapping_key, topic, pattern, key_strategy?}`), `connector_name`. ## Naming rules (strict) - **Instance path** comes from `connector_registry.json[connector_type].instance_path_pattern`: - `snowflake_sink` → `snowflake_sink/` (nested under the shared snowflake_sink/ catalog) - `jdbc_source_snowflake` → `jdbc_source/snowflake/` (nested) - `neo4j_cdc_source` → `neo4j_cdc_source/` (nested) - **Fargate service_name** — composed once in each connector's variables.tf as `local.service_name` and referenced everywhere (terraform-fargate prepends `-` for the deployed AWS name): - `snowflake_sink` → `${var.cluster_name}-${var.connector_type}-${var.connector_name}` = `kc-sfsink-` (defaults `cluster_name=kc`, `connector_type=sfsink`). - `jdbc_source_snowflake` → `${var.connector_type}-` (with `var.connector_type = "kc-jdbc-src"`). - `neo4j_cdc_source` → `kc-neo-src-${var.connector_name}` → deployed `-kc-neo-src-` (e.g. `prod-kc-neo-src-fingerprinting`). - **32-char ALB cap** — enforced for **all three connectors**. See Step 2 hard error #3a for the per-connector formula and limits. HARD ERROR if exceeded; the skill proposes shortening the `connector_name`/`purpose` before regenerating. - **Backend state key** — from the connector's `backend_state_key_pattern`: - `snowflake_sink` → `/kafka-infra/snowflake_sink//terraform.tfstate` - `jdbc_source_snowflake` → `/kafka-infra/jdbc-source/snowflake//terraform.tfstate` - `neo4j_cdc_source` → `/kafka-infra/neo4j_cdc_source//terraform.tfstate` - These differ per connector (snowflake_sink uses underscores throughout; jdbc_source_snowflake uses a hyphenated `jdbc-source` prefix; neo4j hyphenates the purpose) — use each connector's `backend_state_key_pattern` from the registry verbatim rather than assuming one convention. - **Topic resource Terraform name** = `topic_name.replace('.', '_')`. ## What this skill does NOT do - Does not support connectors other than Snowflake source/sink and Neo4j CDC source (no JDBC MySQL/Postgres, no S3, no Debezium, no Neo4j sink). - Does not run `terraform init/plan/apply`. - Does not commit, push, or open PRs. - Does not generate AVRO schemas. - Does not modify the `kafka-connect` repo (only warns about needed changes). - Does not edit existing connector instances (refuses if target directory already exists). - Does not validate Kafka cluster connectivity (not a deployment tool). - Does not provision Snowflake users, roles, or grants. - Does not populate AWS Secrets Manager secret values. ## References - `references/connector_registry.json` — `snowflake_sink` + `jdbc_source_snowflake` + `neo4j_cdc_source` registry - `references/msk_clusters.json` — MSK cluster catalog with bootstrap servers and topics dir - `references/topic_naming_rules.md` — Notion-aligned topic naming rules + validation regex - `references/templates/` — Jinja2-style templates for both connector types - Notion guidebook: [Kafka Data Highway](https://www.notion.so/f556c9cb814b461bb93f23082ec7dc6e) - Naming conventions: [Kafka Naming Conventions](https://www.notion.so/49d48bb6a549442cb65c6d36120c7069) - Neo4j Kafka source — CDC mode: https://neo4j.com/docs/kafka/current/source/cdc/ - Neo4j Kafka source — QUERY mode: https://neo4j.com/docs/kafka/current/source/query/