airajena opened a new pull request, #102: URL: https://github.com/apache/fineract-business-intelligence/pull/102
## Summary This PR adds a second, real-time way of getting data into the warehouse, next to the existing Python extractor, and the pieces needed to run both without surprises. - **CDC engine.** Debezium reads each Fineract PostgreSQL database's change log, Kafka carries the changes, and a small Java sink writes them to the warehouse. A change shows up in the warehouse in about a second, and deletes are captured. - **One write path.** Both engines write through the same warehouse function, `meta.apply_staged_batch()`, so the rows end up identical whichever engine produced them. - **Feature flag.** A table, `meta.ingestion_ownership`, says which engine owns each (source, table). Moving a source from one engine to the other goes through a shadow copy, a row-by-row comparison and a second person's approval. Rolling back is the same procedure in reverse. - **Control plane and Airflow.** A source registry, a connector reconciler, an audit trail, and a set of Airflow DAGs that run the routines. Airflow is never in the per-event data path. - **Python-only setups are unchanged.** Kafka is only needed if you start the `cdc` profile. ## Motivation - The Python extractor runs on a schedule, so data can be up to an hour old. It can't see deleted rows without a separate comparison job, and it queries every source database on every run. - The target I had in mind is a large bank with hundreds to thousands of Fineract databases and somewhere between 100M and 1B+ row changes a day, with raw data fresh to within about a minute on the CDC engine. - Fineract is database-per-tenant, so an id is only unique inside one database. Two banks both have loan 42. Every key had to become `(source_id, tenant_id, id)`. - Switching a live source between engines had to be safe, reversible and auditable. A config flag can't give you that, so ownership is stored as data and enforced by the database itself. ## What is included | Area | What it adds | |---|---| | **Table contract** | `contracts/fineract_tables.json`, one list of tables and columns that every component reads | | **Warehouse write path** | `warehouse/schema/platform.sql`: the single write function (fencing, in-batch dedup, ordering guard, soft deletes, TOAST handling, change log), plus registry, ownership, audit, offsets, watermarks and reconciliation tables | | **Python engine** | `extractor/`: keyset-paginated extraction, backfill and incremental modes, one bookmark per table | | **CDC engine** | Debezium connector builder, Kafka topic provisioning, the Java sink in `ingestion/`, a dead-letter topic | | **Control plane** | `controlplane/`: registry, ownership state machine, connector reconciler, cutover gate, source reconciliation, `retire-source`, CLI | | **Feature flag** | `meta.ingestion_ownership` plus the `start_shadow`, `cutover_gate`, `cutover`, rollback, pause, resume and abort procedures | | **Airflow** | DAGs in `airflow/dags/` for connector reconciliation, Python ingestion, dbt, engine switching, onboarding, freshness and maintenance | | **dbt** | Change-driven incremental models, certified metric macros and catalog, new consistency tests | | **Superset** | Row-level security re-keyed to `(source_id, office_id)` | | **Load and chaos harness** | `loadtest/` and the load generator in `ingestion/` | | **Deployment** | `compose.yaml` profiles, `compose.kafka-cluster.yaml`, reference manifests in `deploy/k8s/` | | **CI** | Java build and license gate, dbt seed step, integration tests (write path, cutover, fencing, multi-bank RLS) | | **Docs** | `docs/getting-started.md`, `docs/walkthrough-powershell.md` | ## Architecture <img width="1672" height="941" alt="image" src="https://github.com/user-attachments/assets/90b0ab55-a079-4ef3-b6a6-d3b9dc2741d2"; /> Key decisions: 1. **One write path for both engines**, so rows are identical regardless of which engine wrote them. 2. **Ownership as data, with a fencing epoch.** The database rejects a writer that is out of date. 3. **One Kafka topic per table, shared by all sources.** A thousand sources do not mean thirteen thousand topics. 4. **Offsets are stored in the warehouse, in the same transaction as the data.** Kafka can redeliver and the result is still exact. 5. **Soft deletes plus an ordering guard** on `(capture_generation, LSN)`, so a late update cannot bring a deleted row back. 6. **Change-log-driven incremental dbt.** Only loans that changed are recomputed. 7. **Airflow orchestrates and never carries data.** ## Reviewer guide Suggested reading order: 1. `contracts/fineract_tables.json`: the shared table contract. 2. `warehouse/schema/platform.sql`: the write function, ownership and audit tables. **This is the core of the PR.** 3. `controlplane/ownership.py` and the rest of `controlplane/`: state machine, reconciler, CLI. 4. `ingestion/` (sink, parser, routing): the Kafka-to-warehouse writer. 5. `extractor/`: the Python engine. 6. `airflow/dags/`: orchestration. 7. `dbt/`: incremental models and metrics. 8. `superset/` and `docker/`: row-level security and runtime config. This is one big PR because the pieces depend on each other (the write path, the flag, both engines and the dbt keys all change together). If you'd rather review it in slices, the order above is the natural split, and I'm happy to split it. ## Breaking changes and migration - **Keys changed.** Raw and mart keys are now `(source_id, tenant_id, id)`. - **Old files removed.** `warehouse/schema/{raw_tables,staging_tables,mart_tables,pipeline_state}.sql` are replaced by `platform.sql`, and the old extractor unit tests (written for the previous single-source extractor) are replaced by `tests/unit/extractor/test_python_engine.py`. `meta.watermarks` is replaced by `meta.extract_watermarks`. - **Row-level security fix.** The old policy filtered on `office_id` alone, which would have shown office 101 of every bank to a branch manager. It now uses `(source_id, office_id)`, and there are tests for it. - **The warehouse is PostgreSQL 18** on a new volume. For an existing install: apply the schema (it is idempotent), run a Python backfill, then `dbt build`. - **Python-only deployments are unaffected.** Don't start the `cdc` profile and Kafka is never needed. ## How to run it ```powershell docker compose up -d warehouse fineract-db docker compose --profile cdc up -d kafka kafka-init connect sink ``` Then follow `docs/walkthrough-powershell.md` (every command explained) or `docs/getting-started.md`. The evidence comment shows the same path, in this order: 1. Start the stack and check health (E1). 2. Python engine: backfill and incremental (E2). 3. CDC: create a source, send traffic, follow each hop (E3). 4. Feature flag: shadow, gate, cutover, rollback (E4). 5. dbt, dashboards, Airflow (E5 to E7). 6. Health checks, load and chaos tests, automated tests, cleanup (E8 to E11). ## Testing | Suite | Count | Result | |---|---|---| | Python unit (state machine, connectors, contract, Python engine, Superset specs) | 123 | pass | | Warehouse integration (write path, fencing, cutover, reconciliation, multi-bank RLS) | 74 | pass | | Airflow DAG integrity (run inside the Airflow image) | 6 | pass | | Java unit tests and license gate (`mvn verify`) | 25 | pass | | dbt tests (schema and singular) | 102 | pass | | dbt build (models, seeds and tests together) | 125 | pass, 0 errors | | Load and chaos harness | 5 stages plus real Debezium | See E9 | | Incremental dbt equals full refresh | 1 | See E5 | All measurements were taken on a Windows 11 laptop with 12 vCPUs and 8 GB given to Docker, a single Kafka broker, and a single-node warehouse. They show headroom on small hardware. They are **not** production capacity numbers. Screenshots and command output for each step are in the evidence comment below. ## Known limitations and follow-ups Not covered by this PR: - No 24-hour soak test, and no run on server-class hardware. - Multi-broker Kafka (RF=3) is configured but not fault-tested. Source-database failover is described in the docs but I haven't drilled it. - Secret references support `env:` and file placeholders only. A Vault-style resolver is a follow-up. - No metrics server or alerting is included. The sink exposes a plain-text `/metrics` endpoint, and health is checked with the SQL views and commands shown in E8. Alert delivery to people is a follow-up. - No replay tool for dead-letter events yet. ## Security and licensing - Secrets are stored **by reference only** (`env:NAME`). A database constraint rejects anything that isn't a reference. - Least-privilege roles: a read-only user for the Python extractor, a `REPLICATION` + `SELECT` user for the connector, a loader user for the warehouse, and dashboard readers that can't read `raw` or `raw_shadow`. - Only the columns listed in the contract leave the source database (`column.include.list`). PII columns are flagged in the contract, and only `date_of_birth` is captured. - New Java dependencies: kafka-clients 4.1.1, HikariCP 6.3.0, jackson-databind 2.20.0, micrometer-core 1.15.4, PostgreSQL JDBC 42.7.13, Log4j 2 and SLF4J, with TestNG and AssertJ for tests. New images: `apache/kafka`, `quay.io/debezium/connect`, `apache/airflow`. All are Apache-2.0, MIT or BSD (ASF Category A), and `mvn verify` fails the build on Category B or X licenses. --- ## Verification evidence **Environment:** Windows 11, 12 vCPUs, 8 GB for Docker, PostgreSQL 18.3, Kafka 4.1.1 (KRaft, single broker), Debezium 3.4, Airflow 3.1.5 **Setup for every capture** (PowerShell, from the project folder): ```powershell cd <project folder> function wh { docker exec fineract-bi-warehouse psql -U analytics_admin -d analytics @args } function src { docker exec fineract-bi-fineract-db psql -U root -d fineract_default @args } ``` --- ### E1. Stack starts and is healthy **Command** ```powershell $env:FINERACT_CDC_PASSWORD = "<dev password>" $env:KAFKA_HEAP = "384m"; $env:CONNECT_HEAP = "384m" docker compose up -d warehouse fineract-db docker compose --profile cdc up -d kafka kafka-init connect sink docker compose --profile cdc ps -a --format "table {{.Service}}\t{{.State}}\t{{.Status}}" docker compose --profile cdc logs --tail 15 sink ``` <img width="1915" height="785" alt="image" src="https://github.com/user-attachments/assets/a16dbab5-875c-469b-a162-e44927880cf2"; /> **Proves:** the whole stack starts from the compose file, and the sink loads the ownership flag. --- ### E2. Python engine: backfill, then one changed loan **Setup** (point the extractor at the demo source and the warehouse): ```powershell $env:WAREHOUSE_DB_HOST="localhost"; $env:WAREHOUSE_DB_PORT="5434" $env:WAREHOUSE_DB_USER="analytics_loader"; $env:WAREHOUSE_DB_PASSWORD="<dev password>" $env:SOURCE_DB_HOST="localhost"; $env:SOURCE_DB_PORT="5433"; $env:SOURCE_DB_NAME="fineract_default" $env:SOURCE_DB_USER="fineract_reader"; $env:SOURCE_DB_PASSWORD="<dev password>" $env:SOURCE_DB_SCHEMA="bi_connector_source" $env:COB_LOOKBACK_HOURS="99999"; $env:LOG_LEVEL="WARNING" ``` **Command** ```powershell $r = (python -m extractor.cli backfill | Out-String) | ConvertFrom-Json $r | Format-Table source_id, status, rows_read, rows_applied, rows_deleted $r.tables | Format-Table table, target, rows_read, rows_applied, status ``` <img width="1373" height="491" alt="image" src="https://github.com/user-attachments/assets/9e1cc277-1d42-427e-833b-c82db557e1de"; /> **Command** ```powershell wh -c "SELECT count(*) AS raw_rows FROM raw.raw_m_loan WHERE source_id='default'" wh -c "SELECT table_name, target, cursor_value, last_pk FROM meta.extract_watermarks WHERE source_id='default' ORDER BY 1 LIMIT 6" wh -x -c "SELECT source_id, id, loan_id, amount, _source_lsn, _is_deleted, _ingest_engine, _ingested_at FROM raw.raw_m_loan_transaction WHERE source_id='default' AND NOT _is_deleted ORDER BY id LIMIT 1" ``` <img width="1897" height="626" alt="image" src="https://github.com/user-attachments/assets/51240bee-bf03-4d4e-a3f3-be3fb1a6a43f"; /> <img width="1898" height="378" alt="image" src="https://github.com/user-attachments/assets/bd5aa16d-871b-460d-905e-eb7a529ef05d"; /> **Command** ```powershell src -q -t -A -c "UPDATE m_loan SET last_modified_on_utc = clock_timestamp() WHERE id = (SELECT min(id) FROM m_loan) RETURNING id, last_modified_on_utc" $i = (python -m extractor.cli incremental | Out-String) | ConvertFrom-Json $i.tables | Where-Object { $_.table -eq "m_loan" } | Format-Table table, rows_read, rows_applied wh -c "SELECT id, last_modified_on_utc, _ingest_engine FROM raw.raw_m_loan WHERE source_id='default' ORDER BY last_modified_on_utc DESC LIMIT 1" ``` <img width="1898" height="561" alt="image" src="https://github.com/user-attachments/assets/92c744bc-d57c-484d-9acb-900d9fffb9c7"; /> **Proves:** the Python engine copies incrementally using bookmarks, and tags every row with its source and engine. --- ### E3. CDC engine, end to end #### E3.0 Create a test source database with CDC permissions **Command** ```powershell $tl = ((Get-Content contracts\fineract_tables.json -Raw | ConvertFrom-Json).tables.table_name | ForEach-Object { "public.$_" }) -join ", " docker exec fineract-bi-fineract-db psql -U root -d postgres -c "CREATE ROLE fbi_cdc LOGIN REPLICATION PASSWORD '<dev password>'" docker exec fineract-bi-fineract-db psql -U root -d postgres -c "CREATE DATABASE fineract_src_004" docker cp warehouse\seed\schema_fineract_source.sql fineract-bi-fineract-db:/tmp/schema.sql docker exec fineract-bi-fineract-db psql -U root -d fineract_src_004 -q -f /tmp/schema.sql docker exec fineract-bi-fineract-db psql -U root -d fineract_src_004 -c "GRANT CONNECT ON DATABASE fineract_src_004 TO fbi_cdc; GRANT USAGE ON SCHEMA public TO fbi_cdc; GRANT SELECT ON ALL TABLES IN SCHEMA public TO fbi_cdc; CREATE PUBLICATION fbi_cdc FOR TABLE $tl" ``` <img width="1899" height="647" alt="image" src="https://github.com/user-attachments/assets/157b8689-6d83-4035-a47c-844ef57a50f2"; /> **Proves:** a source needs only a limited replication login and a publication of the contract's tables. #### E3.1 Topics exist **Command** ```powershell docker compose --profile cdc exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list docker compose --profile cdc exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic fineract.cdc.m_loan ``` <img width="1912" height="714" alt="image" src="https://github.com/user-attachments/assets/1070c285-86c5-47af-bef7-da27cb033cc9"; /> **Proves:** topics are created from the contract, with auto-creation switched off. #### E3.2 Database side of CDC **Command** ```powershell docker exec fineract-bi-fineract-db psql -U root -d fineract_src_004 -c "SHOW wal_level" docker exec fineract-bi-fineract-db psql -U root -d fineract_src_004 -c "SELECT pubname, count(*) AS tables FROM pg_publication_tables GROUP BY 1" docker exec fineract-bi-fineract-db psql -U root -d postgres -c "SELECT slot_name, plugin, active, database FROM pg_replication_slots" ``` <img width="1467" height="558" alt="image" src="https://github.com/user-attachments/assets/13bff530-e531-45de-bbe8-2cfae7acd94b"; /> **Proves:** the source uses a DBA-owned publication and a least-privilege replication role. #### E3.3 Register the source and create its connector **Command** ```powershell $env:WAREHOUSE_DB_HOST="localhost"; $env:WAREHOUSE_DB_PORT="5434"; $env:WAREHOUSE_DB_USER="analytics_loader"; $env:WAREHOUSE_DB_PASSWORD="<dev password>"; $env:LOG_LEVEL="WARNING" python -m controlplane.cli --actor you register-source --source-id cdc-src-004 --tenant-id t-004 --region west --host fineract-bi-fineract-db --db fineract_src_004 --user fbi_cdc --secret-ref env:FINERACT_CDC_PASSWORD --engine KAFKA Start-Sleep 8 $env:FBI_CONNECT_CLUSTERS='{"default":"http://localhost:8083"}'; $env:FBI_DEBEZIUM_PUBLICATION_MODE="disabled"; $env:FBI_RECONCILER_DROP_STALE_SLOTS="false" $rep = (python -m controlplane.cli reconcile-connectors | Out-String) | ConvertFrom-Json $rep.actions | Format-Table kind, name ``` <img width="1900" height="755" alt="image" src="https://github.com/user-attachments/assets/3c5018d5-c425-49bd-ad05-c87df0abb009"; /> ```powershell Start-Sleep 10 Invoke-RestMethod http://localhost:8083/connectors/fbi-src-cdc-src-004-g0/status | ConvertTo-Json -Depth 4 docker exec fineract-bi-fineract-db psql -U root -d postgres -c "SELECT slot_name, active FROM pg_replication_slots WHERE slot_name LIKE '%004%'" ``` <img width="1703" height="648" alt="image" src="https://github.com/user-attachments/assets/778c7f40-4935-4804-8ea6-bc8dfb38f644"; /> **Proves:** the control plane creates connectors from the registry, and the password is supplied only as an environment reference. #### E3.4 Traffic, and each hop **Command** ```powershell $rootpw = (Select-String -Path .env -Pattern '^SOURCE_BOOTSTRAP_PASSWORD=').Line.Split('=')[1] docker compose --profile loadtest run --rm loadgen source-workload --sources=1 --jdbc-template=jdbc:postgresql://fineract-bi-fineract-db:5432/fineract_src_004 --db-user=root --db-password=$rootpw --seed-loans=200 --tps=100 --duration-sec=30 ``` <img width="1916" height="212" alt="image" src="https://github.com/user-attachments/assets/c377a295-981b-4447-9d8c-cff886962bac"; /> **Command (one per hop)** ```powershell # Hop 1: messages in Kafka docker compose --profile cdc exec kafka /opt/kafka/bin/kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic fineract.cdc.m_loan_transaction 2>$null # Hop 2: sink lag docker compose --profile cdc exec kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group fbi-warehouse-sink 2>$null | Select-String "m_loan_transaction" # Hop 3: sink counters docker compose --profile cdc exec sink bash -c "exec 3<>/dev/tcp/127.0.0.1/9404; printf 'GET /metrics HTTP/1.0\r\n\r\n' >&3; cat <&3" 2>$null | Select-String "^fbi_sink_(batches_total|db_errors_total|rows_written_total.*m_loan_transaction)" # Hop 4: warehouse wh -c "SELECT count(*) FILTER (WHERE NOT _is_deleted) AS live, count(*) FILTER (WHERE _is_deleted) AS tombstones FROM raw.raw_m_loan_transaction WHERE source_id='cdc-src-004'" ``` <img width="1917" height="748" alt="image" src="https://github.com/user-attachments/assets/e0766eff-2f21-4423-b8a4-d63f6f3b0db8"; /> <img width="1908" height="517" alt="image" src="https://github.com/user-attachments/assets/4c77f3a4-2363-4b99-bf29-4107ae629e88"; /> **Proves:** data flows through every hop and the sink keeps up. #### E3.5 One raw Debezium message **Command** ```powershell $m = docker compose --profile cdc exec kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic fineract.cdc.m_loan --partition 0 --offset earliest --max-messages 1 --timeout-ms 8000 2>$null $j = ($m | Select-Object -First 1) | ConvertFrom-Json "op=$($j.op) bank=$($j.source.name) table=$($j.source.table) log position=$($j.source.lsn)" $j.after | Select-Object id, account_no, principal_amount | Format-List ``` <img width="1333" height="409" alt="image" src="https://github.com/user-attachments/assets/b4f4f111-1e32-4f9c-a4a2-fc8c09023425"; /> **Proves:** messages carry the source id and log position the sink needs for routing and ordering. #### E3.6 Delete handling **Command** ```powershell $id = (docker exec fineract-bi-fineract-db psql -U root -d fineract_src_004 -t -A -c "SELECT max(id) FROM m_loan_transaction").Trim() docker exec fineract-bi-fineract-db psql -U root -d fineract_src_004 -c "DELETE FROM m_loan_transaction WHERE id = $id" Start-Sleep 4 wh -c "SELECT id, _is_deleted, _ingest_engine, _source_lsn FROM raw.raw_m_loan_transaction WHERE source_id='cdc-src-004' AND id = $id" ``` <img width="1749" height="459" alt="image" src="https://github.com/user-attachments/assets/b21a6b61-5e32-4348-9f55-d71c60f76d83"; /> **Proves:** deletes are captured, which the Python extractor can't do. The marker plus the ordering guard stops a late update from bringing the row back. #### E3.7 Source versus warehouse checksum **Command** ```powershell $r = (docker compose --profile python run --rm -T --no-deps -e FINERACT_CDC_PASSWORD=<dev password> extractor python -m controlplane.cli reconcile-source --source-id cdc-src-004 --bucket-size 1000 | Out-String) | ConvertFrom-Json $r.PSObject.Properties | ForEach-Object { "{0,-34} {1} rows={2}" -f $_.Name, $_.Value.status, $_.Value.compared_rows } ``` <img width="1911" height="598" alt="image" src="https://github.com/user-attachments/assets/c5d958a3-e432-4f28-aaac-7c1d4294ee85"; /> **Proves:** the warehouse matches the source database exactly. --- ### E4. Feature flag: switch engines and roll back This section moves the `default` source from the Python engine to Kafka CDC and back, using the real safety steps: a shadow copy, a row-by-row comparison, and a second person's approval. Run it in one PowerShell window (the commands share functions and variables defined in E4.0). Each step ends with what to expect and what to screenshot. The control-plane tools run inside Docker, so the source address they register (`fineract-bi-fineract-db`) is the same one Debezium uses. #### E4.0 Setup (once per window) ```powershell cd <project folder> function wh { docker exec fineract-bi-warehouse psql -U analytics_admin -d analytics @args } function src { docker exec fineract-bi-fineract-db psql -U root -d fineract_default @args } $env:FINERACT_CDC_PASSWORD = "fbi_cdc_dev_only" $tl = ((Get-Content contracts\fineract_tables.json -Raw | ConvertFrom-Json).tables.table_name | ForEach-Object { "public.$_" }) -join ", " src -c "GRANT CONNECT ON DATABASE fineract_default TO fbi_cdc; GRANT USAGE ON SCHEMA public, bi_connector_source TO fbi_cdc; GRANT SELECT ON ALL TABLES IN SCHEMA public, bi_connector_source TO fbi_cdc" src -c "CREATE PUBLICATION fbi_cdc FOR TABLE $tl" $E = @("-e","SOURCE_DB_HOST=fineract-bi-fineract-db","-e","SOURCE_DB_PORT=5432","-e","SOURCE_DB_NAME=fineract_default", "-e","SOURCE_DB_USER=fbi_cdc","-e","SOURCE_DB_PASSWORD=fbi_cdc_dev_only","-e","SOURCE_DB_SECRET_REF=env:FINERACT_CDC_PASSWORD", "-e","FINERACT_CDC_PASSWORD=fbi_cdc_dev_only","-e","SOURCE_DB_SCHEMA=bi_connector_source","-e","COB_LOOKBACK_HOURS=999999", "-e","FBI_DEBEZIUM_PUBLICATION_MODE=disabled","-e","LOG_LEVEL=WARNING") function ext { docker compose --profile python run --rm -T --no-deps @E extractor python -m extractor.cli @args } function cpc { docker compose --profile python run --rm -T --no-deps @E extractor python -m controlplane.cli @args } $q = "SELECT engine, coalesce(shadow_engine,'-') AS shadow, cutover_state, count(*) AS tables FROM meta.ingestion_ownership WHERE source_id='default' GROUP BY 1,2,3" ``` The `ext` and `cpc` calls each start a container, so a step can take 5 to 15 seconds. #### E4.1 Start from a known state ```powershell wh -c $q ``` <img width="1184" height="196" alt="image" src="https://github.com/user-attachments/assets/88672a3e-40b7-4ee2-b1f5-020c56a05750"; /> If it shows anything else, reset first: - `PYTHON / KAFKA / SHADOWING` (a switch left half-done): run ```powershell cpc --actor alice ownership-apply --source-id default --tables all --action abort_shadow --reason "reset" | Out-Null $rep = (cpc --actor alice reconcile-connectors | Out-String) | ConvertFrom-Json; $rep.actions | Format-Table kind, name wh -c $q ``` - `KAFKA / - / STEADY`: run E4.9 (rollback), then start again here. Then load the data with the Python engine: ```powershell (ext backfill --source-id default | Out-String) | ConvertFrom-Json | Format-Table source_id, status, rows_applied wh -c $q ``` <img width="1309" height="420" alt="image" src="https://github.com/user-attachments/assets/176e94dc-e3d7-4b18-b49d-0dea58c98141"; /> #### E4.2 Start the shadow with Kafka ```powershell cpc --actor alice ownership-apply --source-id default --tables all --action start_shadow --engine KAFKA --reason "migrate to CDC" | Out-Null wh -c $q ``` `--actor` is who is doing it and `--reason` is why. Both are recorded in the audit trail. **Expect:** `PYTHON / KAFKA / SHADOWING`. <img width="1585" height="275" alt="image" src="https://github.com/user-attachments/assets/563c77e2-a955-4e24-b1dd-73088e924683"; /> #### E4.3 Create the Debezium connector ```powershell $rep = (cpc --actor alice reconcile-connectors | Out-String) | ConvertFrom-Json $rep.actions | Format-Table kind, name Start-Sleep 10 (Invoke-RestMethod "http://localhost:8083/connectors?expand=status";).PSObject.Properties | ForEach-Object { "{0}: {1} tasks={2}" -f $_.Name, $_.Value.status.connector.state, (($_.Value.status.tasks | ForEach-Object { $_.state }) -join ",") } ``` <img width="1654" height="519" alt="image" src="https://github.com/user-attachments/assets/480228bf-26d3-4d2f-b9b0-605d42533da6"; /> #### E4.4 Wait for the trial copy to catch up ```powershell for ($i = 0; $i -lt 40; $i++) { $c = wh -t -A -c "SELECT (SELECT count(*) FROM raw_shadow.raw_m_loan_transaction WHERE source_id='default' AND NOT _is_deleted) || ' shadow vs ' || (SELECT count(*) FROM raw.raw_m_loan_transaction WHERE source_id='default' AND NOT _is_deleted) || ' production'"; $c; if ($c -match '^(\d+) shadow vs \1 production' -and $Matches[1] -ne '0') { break }; Start-Sleep 3 } ``` This counts the trial copy against production every 3 seconds and stops when they are equal and non-zero. <img width="1903" height="161" alt="image" src="https://github.com/user-attachments/assets/d8fd1b1e-87b4-449e-8218-2017701f7057"; /> #### E4.5 The cutover gate ```powershell $g = (cpc --actor alice cutover-gate --source-id default --settle-seconds 0 | Out-String) | ConvertFrom-Json $g.PSObject.Properties | ForEach-Object { "{0,-34} passed={1} rows={2} discrepancies={3}" -f $_.Name, $_.Value.passed, $_.Value.compared_rows, $_.Value.discrepancies } wh -c $q ``` The gate compares the trial copy with production, table by table, row by row. `--settle-seconds 0` skips the allowance for rows changed in the last N seconds (a real run would use something like 900). <img width="1864" height="579" alt="image" src="https://github.com/user-attachments/assets/bb77daca-028d-4454-aa0f-29fbce961d12"; /> #### E4.6 Self-approval must fail (dual control) ```powershell (cpc --actor alice ownership-apply --source-id default --tables all --action cutover --approved-by alice --reason x 2>&1 | Out-String) -split "`n" | Select-String "TransitionError:" | Select-Object -First 1 ``` <img width="1910" height="197" alt="image" src="https://github.com/user-attachments/assets/d690d98d-b9fe-4cbb-9d52-f723692f4133"; /> #### E4.7 Cutover, approved by a second person ```powershell cpc --actor alice ownership-apply --source-id default --tables all --action cutover --approved-by bob --reason "go live" | Out-Null wh -c $q ``` <img width="1485" height="290" alt="image" src="https://github.com/user-attachments/assets/27378363-be77-48f9-a2eb-271b48fdcb75"; /> #### E4.8 Prove CDC now feeds production ```powershell src -q -t -A -c "UPDATE m_loan SET last_modified_on_utc = clock_timestamp() WHERE id = (SELECT min(id) FROM m_loan) RETURNING last_modified_on_utc" Start-Sleep 5 wh -c "SELECT id, last_modified_on_utc, _ingest_engine FROM raw.raw_m_loan WHERE source_id='default' ORDER BY last_modified_on_utc DESC LIMIT 1" ``` <img width="1663" height="402" alt="image" src="https://github.com/user-attachments/assets/3ac8a828-7976-4a76-a89a-fe27561b68cf"; /> #### E4.9 Roll back to Python (the same steps, engines swapped) ```powershell cpc --actor carol ownership-apply --source-id default --tables all --action start_shadow --engine PYTHON --reason "rollback drill" | Out-Null (ext backfill --source-id default | Out-String) | ConvertFrom-Json | Out-Null $g = (cpc --actor carol cutover-gate --source-id default --settle-seconds 0 | Out-String) | ConvertFrom-Json "tables passed: {0} of {1}" -f @($g.PSObject.Properties | Where-Object { $_.Value.passed }).Count, @($g.PSObject.Properties).Count cpc --actor carol ownership-apply --source-id default --tables all --action cutover --approved-by dave --reason "rollback done" | Out-Null $rep = (cpc --actor carol reconcile-connectors | Out-String) | ConvertFrom-Json; $rep.actions | Format-Table kind, name wh -c $q ``` Python now fills the trial copy (the extractor writes to the shadow because that is its role), the gate compares, a different person approves, and the connector is deleted. <img width="1557" height="762" alt="image" src="https://github.com/user-attachments/assets/f239b9c4-5e54-49e8-90a0-666c80e826ce"; /> #### E4.10 The audit trail ```powershell wh -c "SELECT to_char(created_at,'HH24:MI:SS') AS at, action, actor, approved_by, reason FROM meta.ingestion_ownership_audit WHERE source_id='default' AND table_name='m_loan' ORDER BY audit_id DESC LIMIT 8" ``` **Expect:** every action in order, with who did it and who approved: `start_shadow` (alice), `begin_reconciliation`, `record_reconciliation`, `cutover` (alice, approved by bob), `start_shadow` (carol), `cutover` (carol, approved by dave). <img width="1918" height="371" alt="image" src="https://github.com/user-attachments/assets/672c69c1-8ee8-47bb-9d73-edefa0039332"; /> **Notes** - A line like `Batch fenced ... expected epoch N ... epoch N+1` in the sink log (`docker compose --profile cdc logs --tail 20 sink`) around E4.7 and E4.9 is the fencing protection working, not an error. It is worth one extra screenshot. - If the E4.4 loop never reaches equal counts, check that the connector task is `RUNNING` (E4.3) and look at the dead-letter topic. **Proves:** switching engines needs a passing comparison and a second approver, a stale writer is fenced out, rollback is the same procedure, and every step is audited. --- ### E5. dbt: build, incremental behavior and equivalence **Command** ```powershell docker compose up -d warehouse fineract-db dbt docker compose exec -T dbt dbt deps docker compose exec -T dbt dbt build 2>&1 | Select-String "OK|PASS|FAIL|ERROR|Done\." ``` <img width="1482" height="502" alt="image" src="https://github.com/user-attachments/assets/5051ad0e-029b-474b-99c3-d498e0770826"; /> <img width="1260" height="841" alt="image" src="https://github.com/user-attachments/assets/b7077b5d-5d06-486e-a97f-91a428c19c56"; /> **Command** (only changed loans are recomputed) ```powershell wh -c "SELECT last_change_id FROM meta.dbt_change_cursor" wh -c "SELECT max(change_id) AS newest_change FROM meta.entity_change_log" src -q -t -A -c "UPDATE m_loan SET last_modified_on_utc = clock_timestamp() WHERE id = (SELECT min(id) FROM m_loan) RETURNING id" python -m extractor.cli incremental | Out-Null wh -c "SELECT change_id, source_id, entity, entity_id, min_affected_date FROM meta.entity_change_log ORDER BY change_id DESC LIMIT 3" docker compose exec -T dbt dbt build 2>&1 | Select-String "int_snapshot_scope|fact_loan_snapshot|Done\." wh -c "SELECT count(*) AS loans_in_scope FROM intermediate.int_snapshot_scope" wh -c "SELECT last_change_id FROM meta.dbt_change_cursor" ``` (The `default` source must be owned by Python for the incremental step to ingest. After E4's rollback it is. Use the E2 environment variables in this window.) <img width="1211" height="649" alt="image" src="https://github.com/user-attachments/assets/e9b36cdb-4dde-43b4-9dae-ff58be00241c"; /> <img width="1140" height="769" alt="image" src="https://github.com/user-attachments/assets/7eae9f57-ea7c-451a-9bd4-5606269de7c2"; /> **Command** (incremental equals full refresh) ```powershell $fp = "SELECT count(*) AS rows, sum(principal_outstanding) AS principal, sum(total_outstanding) AS outstanding, count(*) FILTER (WHERE is_par_30) AS par30_rows FROM analytics.fact_loan_snapshot" wh -c $fp docker compose exec -T dbt dbt build --full-refresh 2>&1 | Select-String "Done\." wh -c $fp ``` <img width="908" height="446" alt="image" src="https://github.com/user-attachments/assets/0fcd3132-6179-4679-990a-24a320c708df"; /> **Command** (tests) ```powershell docker compose exec -T dbt dbt test 2>&1 | Select-String "FAIL|ERROR|Done\." ``` <img width="1675" height="222" alt="image" src="https://github.com/user-attachments/assets/a320c499-5b88-4872-8c04-d0e7d6c69902"; /> **Proves:** dbt recomputes only the loans that changed, gives the same result as a full rebuild, and the tests (including metric consistency and no orphan sources) pass. --- ### E6. Superset dashboards and row-level security **Command** ```powershell docker compose up -d warehouse superset docker compose ps superset --format "table {{.Service}}\t{{.State}}\t{{.Status}}" wh -c "SELECT username, role_name, region, source_id, office_id FROM meta.user_office_mapping ORDER BY 1" ``` <img width="1590" height="642" alt="image" src="https://github.com/user-attachments/assets/8ae21b15-f474-43eb-aeab-1846024daff2"; /> Wait for `(healthy)`, then open http://localhost:8088 and log in as `admin`, then `north_manager`, then `south_manager`. <img width="1770" height="822" alt="image" src="https://github.com/user-attachments/assets/c04239b6-320e-4754-8270-809fd5fc08c4"; /> **Proves:** each branch manager sees only their own `(source, office)`, and an unknown user sees nothing. --- ### E7. Airflow orchestration **Command** ```powershell $env:SOURCE_DB_HOST="fineract-bi-fineract-db"; $env:SOURCE_DB_HOST_PORT="5432"; $env:SOURCE_REPLICA_USER="fbi_cdc"; $env:SOURCE_REPLICA_PASSWORD="<dev password>" $env:SOURCE_DB_SECRET_REF="env:FINERACT_CDC_PASSWORD"; $env:FINERACT_CDC_PASSWORD="<dev password>"; $env:FBI_DEBEZIUM_PUBLICATION_MODE="disabled" $env:AIRFLOW_PARALLELISM="2"; $env:AIRFLOW_MAX_ACTIVE_TASKS_PER_DAG="2"; $env:FBI_POOL_SOURCE_EXTRACT="1" docker compose --profile airflow up -d airflow-apiserver airflow-scheduler airflow-dag-processor ``` <img width="1908" height="525" alt="image" src="https://github.com/user-attachments/assets/aecffcd1-2e0d-456b-9816-5120fc5d44a3"; /> **Command** (trigger the Python ingestion DAG and watch it) ```powershell $tok = (Invoke-RestMethod -Method Post -Uri http://localhost:8080/auth/token -ContentType "application/json" -Body '{"username":"admin","password":"<cropped>"}').access_token $h = @{ Authorization = "Bearer $tok" } (Invoke-RestMethod -Uri "http://localhost:8080/api/v2/dags?limit=50"; -Headers $h).dags | Format-Table dag_id, is_paused $run = Invoke-RestMethod -Method Post -Uri http://localhost:8080/api/v2/dags/fbi_python_ingestion/dagRuns -Headers $h -ContentType "application/json" -Body '{"logical_date": null, "conf": {}}' (Invoke-RestMethod -Uri "http://localhost:8080/api/v2/dags/fbi_python_ingestion/dagRuns/$($run.dag_run_id)" -Headers $h).state ``` <img width="1906" height="771" alt="image" src="https://github.com/user-attachments/assets/cd376a43-770e-4df8-b409-4e3459c9a2d3"; /> **Proves:** Airflow runs the routines (reconciler, ingestion, dbt), and a successful Python ingestion triggers the dbt build on its own. The data flow itself doesn't depend on Airflow. --- ### E8. Health, freshness and correctness checks **Command** ```powershell docker compose --profile cdc ps --format "table {{.Service}}\t{{.State}}\t{{.Status}}" (Invoke-RestMethod "http://localhost:8083/connectors?expand=status";).PSObject.Properties | ForEach-Object { "{0}: {1} tasks={2}" -f $_.Name, $_.Value.status.connector.state, (($_.Value.status.tasks | ForEach-Object { $_.state }) -join ",") } docker exec fineract-bi-fineract-db psql -U root -d postgres -c "SELECT slot_name, active FROM pg_replication_slots" wh -c "SELECT source_id, connector_name, observed_state, task_states FROM meta.connector_state" ``` <img width="1904" height="796" alt="image" src="https://github.com/user-attachments/assets/419c82a4-4ce4-4750-956d-f094129810b1"; /> **Command** (freshness) ```powershell wh -c "SELECT source_id, engine, SUM(rows_applied) AS rows, MAX(seconds_since_last_apply) AS secs_since_last_write FROM meta.v_table_freshness WHERE target='prod' GROUP BY 1,2 ORDER BY 4 DESC" wh -c "SELECT source_id, count(*) AS rows, round(percentile_cont(0.5) WITHIN GROUP (ORDER BY extract(epoch FROM (_ingested_at - _source_ts)))::numeric, 2) AS p50_sec, round(percentile_cont(0.95) WITHIN GROUP (ORDER BY extract(epoch FROM (_ingested_at - _source_ts)))::numeric, 2) AS p95_sec FROM raw.raw_m_loan_transaction WHERE _ingest_engine='KAFKA' AND _ingested_at > now() - interval '30 minutes' GROUP BY 1" ``` <img width="1902" height="805" alt="image" src="https://github.com/user-attachments/assets/b3828123-59fe-463c-8119-f003d427cb27"; /> <img width="1913" height="582" alt="image" src="https://github.com/user-attachments/assets/d107a1b3-58c8-47bd-9536-ff3c2f049750"; /> **Command** (correctness) ```powershell $dlt = docker compose --profile cdc exec kafka /opt/kafka/bin/kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic fineract.cdc.dlt 2>$null "dead-lettered events: " + (($dlt | ForEach-Object { [int]($_ -split ':')[2] }) | Measure-Object -Sum).Sum wh -c "SELECT table_name, SUM(events_consumed) AS consumed, SUM(rows_applied) AS applied, SUM(rows_skipped) AS skipped, SUM(events_dead_lettered) AS dead FROM meta.ingestion_audit WHERE minute_bucket > now() - interval '60 minutes' GROUP BY 1 ORDER BY 2 DESC" ``` <img width="1908" height="592" alt="image" src="https://github.com/user-attachments/assets/0f923348-5a34-4640-a05e-b223998f2cde"; /> **Proves:** health is measurable at every layer, and nothing disappears silently (consumed = applied + skipped + dead-lettered). --- ### E9. Load and chaos results This section checks that the CDC path stays **exactly right** (not just fast) under load, duplicates, reordering, deletes, a crashed sink and a warehouse outage. A load generator (`loadgen`) writes Debezium-style messages for 100 fake banks straight into Kafka, and remembers what the warehouse should look like at the end. The `verify` command then waits until the warehouse matches that expectation exactly: live rows, deleted rows, a sum of log positions and a sum of amounts. A stage passes only if nothing was lost, duplicated or wrong. Run this as one session on a quiet machine, in two PowerShell windows. **Window A** runs the stages. **Window B** is only used for `docker stats` in stage 2 and for the chaos step. Each window needs the setup block below. #### E9.0 Prepare (window A, once) Stop Airflow and Superset first. Airflow's reconciler and dbt DAGs would try to act on the fake banks, and Docker needs the memory. ```powershell cd <project folder> docker compose --profile airflow stop airflow-apiserver airflow-scheduler airflow-dag-processor docker compose stop superset function wh { docker exec fineract-bi-warehouse psql -U analytics_admin -d analytics @args } function Show-Report($n) { $r = Get-Content "loadtest\results\stage$n-report.json" | ConvertFrom-Json "stage $n : passed=$($r.passed) mismatched=$($r.mismatched_pairs) produced=$($r.generated_events) events at $($r.produce_eps)/s end-to-end=$($r.effective_end_to_end_eps)/s" $l = $r.end_to_end_latency_seconds.m_loan_transaction " delay source-to-warehouse (m_loan_transaction), seconds: p50=$($l.p50) p95=$($l.p95) p99=$($l.p99) max=$($l.max)" } $env:WAREHOUSE_DB_HOST="localhost"; $env:WAREHOUSE_DB_PORT="5434"; $env:WAREHOUSE_DB_USER="analytics_loader"; $env:WAREHOUSE_DB_PASSWORD="analytics_loader_dev_only"; $env:LOG_LEVEL="WARNING" docker compose --profile cdc ps --format "table {{.Service}}\t{{.State}}\t{{.Status}}" python -m controlplane.cli --actor loadtest register-synthetic --count 100 --prefix loadsrc --engine KAFKA Start-Sleep 8 ``` `kafka`, `connect`, `sink` and `warehouse` must be `healthy`. `register-synthetic` creates 100 fake banks owned by Kafka, with no real database or connector (the generator plays Debezium's part). The 8-second wait lets the sink learn about them. <img width="1384" height="474" alt="image" src="https://github.com/user-attachments/assets/56866b79-a753-4329-9b42-dd6012851e2c"; /> #### E9.1 Stage 1: correctness (about 1 minute) 10,000 events across 100 banks, with duplicates, out-of-order versions and deletes. ```powershell docker compose --profile loadtest run --rm loadgen produce --events=10000 --sources=100 --keys=5000 --rate=2000 --source-prefix=loadsrc --ground-truth=/results/stage1-truth.json docker compose --profile loadtest run --rm loadgen verify --ground-truth=/results/stage1-truth.json --report=/results/stage1-report.json --timeout-sec=300 Show-Report 1 ``` <img width="1912" height="571" alt="image" src="https://github.com/user-attachments/assets/16a7c61c-5eb0-4154-a514-4713f1fd18b6"; /> #### E9.2 Stage 2: volume (about 4 minutes) One million events at 8,000 per second, 3% deletes. In **window B** (`cd <project folder>`), have this ready but don't run it yet: ```powershell docker stats --no-stream --format "table {{.Name}}\t{{.CPUPerc}}\t{{.MemUsage}}" ``` In **window A**: ```powershell docker compose --profile loadtest run --rm loadgen produce --events=1000000 --sources=100 --keys=5000 --rate=8000 --delete-ratio=0.03 --source-prefix=loadsrc --ground-truth=/results/stage2-truth.json ``` About 60 seconds after `produce` starts, run the `docker stats` line in **window B**. <img width="1274" height="626" alt="image" src="https://github.com/user-attachments/assets/89aea6d8-ce96-4d26-9287-8f28da3097c5"; /> When `produce` finishes, in **window A**: ```powershell docker compose --profile loadtest run --rm loadgen verify --ground-truth=/results/stage2-truth.json --report=/results/stage2-report.json --timeout-sec=300 Show-Report 2 ``` <img width="1632" height="398" alt="image" src="https://github.com/user-attachments/assets/046d6a19-262e-4a30-bb14-923a07ab82ad"; /> #### E9.3 Stage 3: burst (about 6 minutes) 3,500 per second, a one-minute spike to 10,000 per second, then back down. ```powershell docker compose --profile loadtest run --rm loadgen produce --events=1020000 --sources=100 --keys=5000 --rate=3500 --burst-rate=10000 --burst-start-sec=60 --burst-duration-sec=60 --source-prefix=loadsrc --ground-truth=/results/stage3-truth.json ``` Optional but useful: in **window B**, about 70 seconds after it starts (during the spike), run this once and again about 60 seconds later: ```powershell docker compose --profile cdc exec kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group fbi-warehouse-sink 2>$null | Select-Object -Skip 2 | ForEach-Object { $f = ($_ -split '\s+'); if ($f.Count -ge 6 -and $f[5] -match '^\d+$') { [pscustomobject]@{Topic=$f[1]; Lag=[int]$f[5]} } } | Group-Object Topic | ForEach-Object { [pscustomobject]@{Topic=$_.Name; TotalLag=($_.Group | Measure-Object Lag -Sum).Sum} } | Sort-Object TotalLag -Descending | Select-Object -First 5 | Format-Table -AutoSize ``` <img width="1622" height="430" alt="image" src="https://github.com/user-attachments/assets/99e38475-a0b1-49e8-aaf2-a64baa6e92e1"; /> Then, in **window A**, after `produce` finishes: ```powershell docker compose --profile loadtest run --rm loadgen verify --ground-truth=/results/stage3-truth.json --report=/results/stage3-report.json --timeout-sec=300 Show-Report 3 ``` <img width="1626" height="415" alt="image" src="https://github.com/user-attachments/assets/fbf08454-cd6a-4ef6-ab9d-0897d4a15e94"; /> #### E9.4 Stage 4: backlog drain, your capacity ceiling (about 3 minutes) Stop the sink, flood Kafka at full speed, restart the sink, and time how long the warehouse takes to absorb the backlog. ```powershell docker compose --profile cdc stop sink docker compose --profile loadtest run --rm loadgen produce --events=300000 --sources=100 --keys=5000 --rate=0 --source-prefix=loadsrc --ground-truth=/results/stage4-truth.json docker compose --profile cdc start sink $t = Measure-Command { docker compose --profile loadtest run --rm loadgen verify --ground-truth=/results/stage4-truth.json --report=/results/stage4-report.json --timeout-sec=300 --poll-ms=1000 | Out-Null } "drain speed ~ " + [int](300000 / $t.TotalSeconds) + " events/s (lower bound, includes JVM start)" Show-Report 4 ``` <img width="1629" height="664" alt="image" src="https://github.com/user-attachments/assets/ca991f32-d8cb-474a-a6fe-50aa1b5df7dd"; /> #### E9.5 Chaos: kill the sink, then take the warehouse down (about 4 minutes) In **window A**, start a long run (about 2 minutes): ```powershell docker compose --profile loadtest run --rm loadgen produce --events=600000 --sources=100 --keys=5000 --rate=5000 --source-prefix=loadsrc --ground-truth=/results/stage5-truth.json ``` In **window B**, about 30 seconds after it starts: ```powershell docker kill --signal=KILL fineract-business-intelligence-sink-1 Start-Sleep 5 docker compose --profile cdc start sink Start-Sleep 20 docker stop fineract-bi-warehouse Start-Sleep 30 docker start fineract-bi-warehouse ``` This kills the sink with no warning, restarts it, takes the warehouse down for 30 seconds, and brings it back. <img width="1538" height="831" alt="image" src="https://github.com/user-attachments/assets/f15f7f1d-9908-4d63-9278-6267586f96e9"; /> Wait about 30 seconds for the warehouse to be healthy again. Check with `docker compose ps warehouse`. Then, in **window A**: ```powershell docker compose --profile loadtest run --rm loadgen verify --ground-truth=/results/stage5-truth.json --report=/results/stage5-report.json --timeout-sec=300 Show-Report 5 ``` <img width="1659" height="362" alt="image" src="https://github.com/user-attachments/assets/af31ce9b-7ab2-49d0-b186-07ff636c1378"; /> #### E9.6 Results table | Stage | Load | Result | |---|---|---| | 1 functional | 10k events, 100 sources, duplicates and reordering | exact, p95 `<s>` | | 2 volume | 1M events at 8k/s, 3% deletes | exact, p50 `<s>`, p95 `<s>`, p99 `<s>`, sink CPU `<%>` | | 3 burst | 3.5k/s, 10k/s for 60 s, 3.5k/s | exact, lag back to near 0 | | 4 backlog drain | 300k events with the sink stopped | Kafka intake `<n>/s`, sink drain `<n>/s`, exact | | 5 chaos | sink killed, warehouse down 30 s | exact | On my laptop (12 vCPUs, 8 GB for Docker, single broker) I measured: stage 1 p95 about 0.9 to 2.7 s, stage 2 p50 0.5 s and p99 1.5 s, stage 3 no backlog, Kafka intake 144k to 210k events/s, sink drain 28k to 33k events/s. Your numbers will differ with your hardware. These are headroom figures on small hardware, not production capacity. "Exact" means live rows, tombstones, the sum of winning log positions and the sum of amounts all match the generator's ground truth. The generator deliberately injects duplicates, out-of-order versions and deletes. #### E9.7 Clean up (do not skip) Fake banks left in the warehouse make `dbt build` very slow. dbt has not run since they were added, so deleting directly is safe. If dbt or Airflow ran in between, use `retire-source --source-id loadsrc-000NN --forget` for each id instead. ```powershell $tables = (Get-Content contracts\fineract_tables.json -Raw | ConvertFrom-Json).tables.raw_table foreach ($t in $tables) { wh -q -c "DELETE FROM raw.$t WHERE source_id LIKE 'loadsrc-%'"; wh -q -c "DELETE FROM raw_shadow.$t WHERE source_id LIKE 'loadsrc-%'" } wh -q -c "DELETE FROM meta.entity_change_log WHERE source_id LIKE 'loadsrc-%'" wh -q -c "DELETE FROM meta.ingestion_progress WHERE source_id LIKE 'loadsrc-%'" wh -c "DELETE FROM meta.source_registry WHERE source_id LIKE 'loadsrc-%'" wh -c "SELECT count(*) AS loadsrc_left FROM meta.source_registry WHERE source_id LIKE 'loadsrc-%'" docker compose --profile airflow start airflow-apiserver airflow-scheduler airflow-dag-processor docker compose start superset ``` `loadsrc_left` should be 0. The last two lines restart Airflow and Superset. <img width="1652" height="672" alt="image" src="https://github.com/user-attachments/assets/06b1f00e-4bac-4107-a6d2-123db782ef71"; /> **If something looks wrong** - A `verify` says `FAILED` or times out right after a stage: run the same `verify` once more. The sink may still be draining. - Docker runs out of memory: close other applications and confirm Airflow and Superset are stopped (E9.0). - `mismatched` above 0: stop there and keep the report file (`loadtest\results\stageN-report.json`) and the sink log (`docker compose --profile cdc logs --tail 100 sink`). Do not publish a failing stage as a pass. **Proves:** correctness under duplicates, reordering, crashes and outages, plus measured headroom on the test machine. --- ### E10. Automated tests **Command** ```powershell python -m pytest tests\unit python -m pytest tests\integration mvn -f ingestion\pom.xml verify docker compose exec -T dbt dbt test python -m ruff check . docker compose --profile airflow run --rm --no-deps --entrypoint bash -v "${PWD}:/repo" airflow-scheduler -c "pip install -q pytest && cd /repo && python -m pytest tests/airflow -q -p no:cacheprovider" ``` <img width="1646" height="430" alt="image" src="https://github.com/user-attachments/assets/93ac26bb-4117-47a0-a001-1f343fe78ce4"; /> <img width="1652" height="310" alt="image" src="https://github.com/user-attachments/assets/8ff0b13e-6e54-437d-914e-2932d941d9cb"; /> <img width="1154" height="344" alt="image" src="https://github.com/user-attachments/assets/eced73c3-de79-4442-96f6-4e885e8fa358"; /> <img width="1590" height="741" alt="image" src="https://github.com/user-attachments/assets/922bb06c-4de9-479a-a101-8f2549f0c14e"; /> <img width="1633" height="385" alt="image" src="https://github.com/user-attachments/assets/9ed12fa0-3feb-4586-ab9a-54654a5f87ba"; /> The last line runs the 6 Airflow DAG integrity tests inside the Airflow image (the package isn't installed on the host). Set the Airflow variables from E7 first if compose complains about missing ones. **Proves:** the automated suites pass on this branch. --- ### E11. Safe cleanup **Command** ```powershell python -m controlplane.cli retire-source --source-id cdc-src-004 --forget Invoke-RestMethod -Method Delete -Uri http://localhost:8083/connectors/fbi-src-cdc-src-004-g0 docker exec fineract-bi-fineract-db psql -U root -d postgres -c "SELECT pg_drop_replication_slot('fbi_cdc_src_004_g0')" docker exec fineract-bi-fineract-db psql -U root -d postgres -c "DROP DATABASE fineract_src_004" ``` <img width="1250" height="796" alt="image" src="https://github.com/user-attachments/assets/f9a6702e-6884-48f1-acab-24da438c67ea"; /> **Proves:** a source can be removed from every layer, including derived tables, so no orphan rows are left for dbt. Fixes: #101 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
