Err/app 787/person service update nrt tables

STRESS_TEST_OBSERVABILITY.md

@@ -0,0 +1,160 @@
+# Stress Testing `reporting-pipeline-service`: What to Watch
+
+Context: load-testing the pipeline (Debezium → Kafka → `reporting-pipeline-service` →
+Kafka Connect JDBC sink → SQL Server) to check for data loss from insufficient
+concurrency guard rails. No standalone Prometheus is deployed; Rancher is the
+container platform. This documents what's already available in the existing
+toolset, grounded in the current code/config.
+
+## 1. The service already exports Prometheus metrics — no Prometheus server required to use them
+
+`reporting-pipeline-service/src/main/resources/application.yaml` sets:
+
+```yaml
+management:
+  endpoints:
+    web:
+      exposure:
+        include: health,metrics,prometheus,liquibase,lag
+  prometheus:
+    metrics:
+      export:
+        enabled: true
+```
+
+So the running service (port `8095`) already serves `GET /actuator/prometheus`
+in plain-text exposition format. You can scrape it by hand during the load run:
+
+```bash
+watch -n2 'curl -s localhost:8095/actuator/prometheus | grep -E "hikaricp_connections|jvm_memory_used|jvm_threads_live"'
+```
+
+Or stand up a throwaway Prometheus + Grafana pointed at it and reuse the
+dashboards **already checked into the repo**:
+`reporting-pipeline-service/src/main/resources/grafana-dashboard/*.json`
+(`person`, `organization`, `investigation`, `observation`, `ldfdata`,
+`postprocessing`). They're wired to an `aws-prometheus` datasource UID for
+prod — swap that UID for a local one and the panels render as-is.
+
+### Metrics that matter most for the concurrency hypothesis
+
+- **`hikaricp_connections_active` / `hikaricp_connections_idle` /
+  `hikaricp_connections_pending` / `hikaricp_connections_timeout_total`** —
+  the likely smoking gun. No `spring.datasource.hikari.maximum-pool-size` is
+  set anywhere, so it runs the Spring Boot default (10 connections). Every
+  domain service spins up its **own** `Executors.newFixedThreadPool` doing
+  synchronous JDBC/stored-proc calls against that one shared pool:
+
+  | Service | Executors |
+  |---|---|
+  | `PersonService` | `rtrExecutor` (`nproc*2`), `prsExecutor` (`featureFlag.thread-pool-size`) |
+  | `OrganizationService` | `rtrExecutor` (`nproc*2`), `org` pool (`thread-pool-size`) |
+  | `InvestigationService` | `phc` pool (`nProc*2`), `inv` pool (`thread-pool-size`) |
+  | `ObservationService` | `obs` pool (`thread-pool-size`) |
+  | `LdfDataService` | `ldf` pool (`thread-pool-size`) |
+  | `ProcessDatamartData` | `dynDmExecutor` (`nProc`) |
+
+  If `hikaricp_connections_pending > 0` and `hikaricp_connections_timeout_total`
+  climbs during the load test, that confirms the connection pool as the
+  concurrency bottleneck.
+
+- **`jvm_memory_used_bytes` / `jvm_gc_pause_seconds`** — `Executors.newFixedThreadPool`
+  backs onto an **unbounded** `LinkedBlockingQueue`. There's no bound tying
+  queued work to consumer throughput, so under sustained high volume the queue
+  can grow without limit. A memory climb followed by a crash, with Kafka lag
+  looking fine right up until the crash, is the signature of this.
+
+- **Per-domain counters/timers** — e.g. `person_msg_processed`,
+  `person_msg_success`, `person_msg_failure`, `inv_msg_failure_total`,
+  `post_dm_success`, `post_dm_failure`, `*_msg_processing_seconds`. Diff
+  processed vs. success vs. failure to spot silent drops.
+
+## 2. The custom `/actuator/lag` endpoint
+
+`gov.cdc.nbs.report.pipeline.lag.LagEndpoint` reports real-time backlog for
+both the pipeline consumer group and the Kafka Connect sink group, with a
+per-topic breakdown:
+
+```bash
+curl -s localhost:8095/actuator/lag | jq
+```
+
+Status is `READY` when both groups are caught up, `PROCESSING` otherwise.
+Poll this on an interval during the load test:
+
+- Sink group lag growing while the pipeline group stays caught up → the JDBC
+  sink connector (writing to SQL Server) is the bottleneck.
+- Pipeline group lag growing → the service's own thread pools/DB contention
+  are the bottleneck.
+
+## 3. Kafka / Kafka Connect layer (no extra tooling needed)
+
+- From inside the `kafka` container:
+  ```bash
+  kafka-consumer-groups.sh --bootstrap-server kafka:29092 --describe --group pipeline-consumer-app
+  kafka-consumer-groups.sh --bootstrap-server kafka:29092 --describe --group connect-Kafka-Connect-SqlServer-Sink
+  ```
+  Same data as `/actuator/lag` but per-partition, straight from the source.
+- Kafka Connect REST API for connector/task health:
+  ```bash
+  curl -s localhost:8083/connectors/<name>/status | jq   # JDBC sink
+  curl -s localhost:8085/connectors/<name>/status | jq   # Debezium source
+  ```
+  A task going `FAILED` (not just lagging) is a distinct data-loss mode from
+  consumer-side loss — worth ruling out separately.
+- Retry/DLT topics (from `spring.kafka.dlq.retry-suffix` / `dlq-suffix` in
+  `application.yaml`): consume `*_retry` / `*_dlt` topics directly with
+  `kafka-console-consumer` to see exactly which payloads are dying, correlated
+  with `nrt_dead_letter_log`.
+
+## 4. SQL Server side — the three tables plus DMVs
+
+- **`nrt_dead_letter_log`** — columns: `origin_topic`, `payload_key`,
+  `payload`, `original_consumer_group`, `exception_stack_trace`,
+  `exception_message`, `exception_fqcn`, `exception_cause_fqcn`,
+  `received_at`. Query the rate of growth during the load window
+  (`COUNT(*) GROUP BY DATEPART(minute, received_at)`), and group by
+  `exception_fqcn` — failures clustering around one exception type (timeout
+  vs. deadlock vs. constraint violation) points to very different root causes.
+- **`nrt_backfill`** — columns: `entity`, `record_uid_list`, `batch_id`,
+  `err_description`, `status_cd`, `retry_count`. Watch for rows stuck at high
+  `retry_count` with a non-terminal `status_cd` — that's the backfill-retry
+  path failing to keep up, a second concurrency-guardrail symptom distinct
+  from the DLQ.
+- **`job_flow_log`** — correlate stored-proc start/end/failure timestamps
+  against the HikariCP pending/timeout spikes above. If proc executions start
+  queuing or timing out at the same moment `hikaricp_connections_pending`
+  spikes, that directly implicates the shared connection pool.
+- **SQL Server DMVs** — none of the above show lock contention directly:
+  ```sql
+  SELECT * FROM sys.dm_exec_requests;
+  SELECT * FROM sys.dm_os_waiting_tasks;
+  SELECT * FROM sys.dm_exec_session_wait_stats
+    WHERE wait_type IN ('THREADPOOL','LCK_M_X','RESOURCE_SEMAPHORE');
+  ```
+  These distinguish connection-pool starvation, row/table lock contention
+  (plausible since multiple threads across multiple executors may hit the
+  same stored procs concurrently for related entities), and SQL Server
+  worker-thread starvation.
+
+## 5. Rancher (in place of Prometheus/Grafana-as-infra)
+
+Rancher's built-in cluster monitoring UI (project/workload metrics view) gives
+per-pod CPU, memory, and **restart count** without a dedicated Prometheus.
+Watch `reporting-pipeline-service`'s memory graph and restart count during the
+load test:
+
+- Restart correlated with a memory climb → corroborates the unbounded-queue
+  hypothesis above.
+- Restart with flat memory but Kafka rebalance activity → points elsewhere,
+  e.g. a health/liveness probe failing due to blocked DB calls.
+
+## Suggested sequence
+
+1. Watch `/actuator/lag` and Rancher pod memory/restarts live — cheapest signals.
+2. If memory climbs unbounded, pull `/actuator/prometheus` for
+   `hikaricp_connections_pending` / `hikaricp_connections_timeout_total` to
+   confirm pool exhaustion is the trigger.
+3. Cross-reference timestamps against `nrt_dead_letter_log` and
+   `job_flow_log` to identify which entity/stored-proc path is actually
+   dropping or failing data.

docker-compose.yaml

@@ -1,6 +1,6 @@
 services:
   nbs-mssql:
-    image: ghcr.io/cdcent/nedssdb:6.0.17.0-replacement-beta
+    image: ghcr.io/cdcent/nedssdb:latest
     platform: linux/amd64
     environment:
       - DATABASE_VERSION=6.0.18.1

reporting-pipeline-service/README.md

@@ -7,18 +7,33 @@ If you don't already have a local application config create one using the existi
 cp src/main/resources/application.yaml src/main/resources/application-local.yaml
 ```
 
-Update `src/main/resources/application-local.yaml` with the following values:
+Create a `src/main/resources/application-local.yaml` file. Sample below:
 
 ```yaml
 spring:
   datasource:
-    password: reporting_pipeline_service
-    username: reporting_pipeline_service_rdb
+    password: PizzaIsGood33!
+    username: sa
     url: jdbc:sqlserver://localhost:3433;databaseName=RDB_MODERN;encrypt=true;trustServerCertificate=true;
   kafka:
     bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVER:localhost:9092}
+  kafka-connect:
+    url: http://localhost:8083
+
+  liquibase:
+    enabled: true
+    change-log: classpath:db/changelog/db.changelog-master.yaml
+    user: sa
+    password: PizzaIsGood33!
+
+  featureFlag:
+    person-service-direct-write: true
 ```
 
+Stop the container if it is running:
+```shell
+docker stop nedss-datareporting-pipeline-service-1
+```
 Run `reporting-pipeline-service` using the following Gradle command:
 
 ```shell

reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/ARCHITECTURE.md

@@ -0,0 +1,102 @@
+# Person Event Processing — Conditional Data Flow
+
+Describes how `PersonService` routes Patient/Provider/Auth User CDC events, including the
+`person-service-direct-write` branch and how both paths converge on the shared postprocessing
+pipeline.
+
+```mermaid
+%%{init: {'theme': 'base', 'themeVariables': {
+  'primaryColor': '#16273d',
+  'primaryTextColor': '#d7e3f0',
+  'primaryBorderColor': '#3d5a7a',
+  'lineColor': '#7d93ad',
+  'secondaryColor': '#16273d',
+  'tertiaryColor': '#16273d',
+  'fontFamily': 'ui-monospace, SFMono-Regular, Consolas, monospace',
+  'fontSize': '15px'
+}}}%%
+flowchart TD
+    subgraph SRC["CDC Source"]
+        ODSEPerson[(ODSE.Person)]
+        ODSEAuthUser[(ODSE.Auth_user)]
+    end
+
+    ODSEPerson -->|Debezium CDC| TopicPerson[["nbs_Person"]]
+    ODSEAuthUser -->|Debezium CDC| TopicAuthUser[["nbs_Auth_user"]]
+
+    TopicPerson --> PS["PersonService.processMessage()"]
+    TopicAuthUser --> PS
+
+    PS --> TopicSwitch{"which topic?"}
+    TopicSwitch -->|nbs_Person| CdSwitch{"cd field"}
+    TopicSwitch -->|nbs_Auth_user| ComputeAuthUser["sp_auth_user_event"]
+
+    CdSwitch -->|PAT| ComputePatient["sp_Patient_Event"]
+    CdSwitch -->|PRV| ComputeProvider["sp_provider_event"]
+
+    subgraph DIRECTWRITE["Conditional: person-service-direct-write"]
+        DW{"directWrite flag"}
+        DW -->|true| JpaSave["JPA save<br/>→ nrt_patient / nrt_provider / nrt_auth_user"]
+        DW -->|false| KafkaPublish["Publish enriched JSON<br/>→ nrt.patient / nrt.provider / nrt.auth-user"]
+        KafkaPublish --> KafkaConnect["Kafka-Connect JDBC Sink<br/>upserts nrt_* table"]
+        KafkaPublish --> PPListener["PostProcessingService<br/>@KafkaListener on nrt.* topic"]
+    end
+
+    ComputePatient --> DW
+    ComputeProvider --> DW
+    ComputeAuthUser --> DW
+
+    JpaSave --> Enqueue["PostProcessingService.enqueue(topic, uid)<br/>(direct in-process call — no Kafka round-trip)"]
+    PPListener --> ProcessNrt["processNrtMessage()<br/>parses Kafka payload"]
+    ProcessNrt --> Enqueue
+
+    subgraph POSTPROC["Shared postprocessing pipeline (priority-ordered batch)"]
+        Enqueue --> IdCache[("idCache<br/>keyed by topic<br/>producer adds + drain snapshot/clear both under cacheLock")]
+        IdCache -->|"@Scheduled processCachedIds()<br/>sorted by Entity.priority"| Drain["Priority-ordered SP execution"]
+        Drain --> PatientSP["sp_nrt_patient_postprocessing<br/>→ D_PATIENT"]
+        Drain --> ProviderSP["sp_nrt_provider_postprocessing<br/>→ D_PROVIDER"]
+        Drain --> AuthUserSP["sp_user_profile_postprocessing<br/>→ USER_PROFILE"]
+        Drain -.->|"same batch,<br/>later priority"| InvSP["Investigation / case_management<br/>postprocessing (unrelated to Person, unchanged)"]
+    end
+
+    PatientSP -->|"returns Covid_*_Datamart rows<br/>(empty for Provider/AuthUser happy path)"| DmData["DatamartData"]
+
+    subgraph DATAMART["Datamart routing"]
+        DmData --> DatamartTopic[["nbs_Datamart"]]
+        DatamartTopic --> DmCache[("dmCache")]
+        DmCache -->|"@Scheduled processDatamartIds()"| DatamartSP["Datamart-specific stored procs<br/>e.g. sp_covid_case_datamart_postprocessing"]
+    end
+
+    InvSP -.->|"LEFT JOIN D_PATIENT<br/>(ordering-dependent)"| DatamartSP2["sp_std_hiv_datamart_postprocessing"]
+
+    classDef highlight fill:#3a2a12,stroke:#e7a53d,color:#f4dcae,stroke-width:2px;
+    classDef muted fill:#16273d,stroke:#48607e,color:#6c839c,stroke-dasharray: 3 3;
+    class DW,JpaSave highlight;
+    class KafkaPublish,KafkaConnect,PPListener,ProcessNrt muted;
+```
+
+🟧 amber = `person-service-direct-write` (active path) · ⬜ dashed = Kafka-Connect (legacy, feature-flagged off)
+
+## Key conditionals
+
+- **`cd` field** routes `nbs_Person` events to patient vs. provider stored procs.
+- **`person-service-direct-write` flag** is the main fork: JPA save (direct-write) vs. Kafka
+  publish → Kafka-Connect (legacy). Both paths converge on the same
+  `PostProcessingService.enqueue()` call into the shared `idCache` — direct-write calls it
+  in-process, the legacy path via its existing `@KafkaListener` → `processNrtMessage()`, which
+  itself now delegates to `enqueue()` for id-caching.
+- **Priority-ordered batch drain** is what keeps `D_PATIENT` hydration ahead of
+  investigation/case_management processing within the same cycle. Direct-write must go through
+  this same shared pipeline rather than calling postprocessing stored procedures itself —
+  bypassing it breaks that ordering guarantee (see APP-787).
+- **`cacheLock`** guards every producer-side cache write (`enqueue()` and the legacy path's
+  payload-enrichment writes) against the scheduled drain's snapshot-then-clear. Without it, an
+  add landing between the drain's snapshot and its `clear()` was silently and permanently
+  dropped rather than merely delayed — this affected every entity type processed by this shared
+  service, not just Patient/Provider/AuthUser (see APP-787 concurrency follow-up).
+- **Datamart routing** only actually carries rows in the happy path for Patient (Covid
+  datamarts); Provider/AuthUser postprocessing stored procedures return empty result sets there.
+
+Not shown (orthogonal, non-blocking flags that don't affect this core routing):
+`elasticSearchEnable` (publishes to `elastic_search_patient`/`elastic_search_provider` regardless
+of direct-write) and `phcDatamartEnable` (async PHC fact datamart update).

reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/model/entity/NrtAuthUser.java

@@ -0,0 +1,56 @@
+package gov.cdc.nbs.report.pipeline.person.model.entity;
+
+import static gov.cdc.nbs.report.pipeline.util.UtilHelper.parseDateTime;
+
+import gov.cdc.nbs.report.pipeline.person.model.dto.user.AuthUser;
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.Table;
+import java.time.LocalDateTime;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Entity
+@Table(name = "nrt_auth_user")
+@Data
+@NoArgsConstructor
+public class NrtAuthUser {
+  @Id
+  @Column(name = "auth_user_uid")
+  private Long authUserId;
+
+  private String userId;
+  private String firstNm;
+  private String lastNm;
+  private Long nedssEntryId;
+  private Long providerUid;
+  private LocalDateTime addTime;
+  private Long addUserId;
+  private LocalDateTime lastChgTime;
+  private Long lastChgUserId;
+  private String recordStatusCd;
+  private LocalDateTime recordStatusTime;
+
+  // GENERATED ALWAYS AS ROW START: SQL Server populates this column; it must never be
+  // included in the INSERT/UPDATE column list.
+  @Column(name = "refresh_datetime", insertable = false, updatable = false)
+  private LocalDateTime refreshDatetime;
+
+  public static NrtAuthUser from(AuthUser authUser) {
+    NrtAuthUser nrtAuthUser = new NrtAuthUser();
+    nrtAuthUser.setAuthUserId(authUser.getAuthUserUid());
+    nrtAuthUser.setUserId(authUser.getUserId());
+    nrtAuthUser.setFirstNm(authUser.getFirstNm());
+    nrtAuthUser.setLastNm(authUser.getLastNm());
+    nrtAuthUser.setNedssEntryId(authUser.getNedssEntryId());
+    nrtAuthUser.setProviderUid(authUser.getProviderUid());
+    nrtAuthUser.setAddTime(parseDateTime(authUser.getAddTime()));
+    nrtAuthUser.setAddUserId(authUser.getAddUserId());
+    nrtAuthUser.setLastChgTime(parseDateTime(authUser.getLastChgTime()));
+    nrtAuthUser.setLastChgUserId(authUser.getLastChgUserId());
+    nrtAuthUser.setRecordStatusCd(authUser.getRecordStatusCd());
+    nrtAuthUser.setRecordStatusTime(parseDateTime(authUser.getRecordStatusTime()));
+    return nrtAuthUser;
+  }
+}