Err/app 787/person service update nrt tables#948
Conversation
… service to conditionally write to RDB_MODERN
…xplicitly after writing to nrt tables
Documents the conditional flow for Patient/Provider/AuthUser CDC events, including the person-service-direct-write fork and how both paths converge on the shared postprocessing pipeline. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…ervice off processNrtMessage Producer-side cache mutations (idCache/cdCache/pbCache/obsCache/sumCache/dmCache) were not synchronized against the scheduled drain's snapshot-then-clear, so a concurrent add() could land in a queue right before it was orphaned by clear() -- silently and permanently dropping ids rather than just delaying them. Synchronize all producer-side mutations on the existing cacheLock, add a public enqueue(topic, uid) API for direct-write callers, and route PersonService's patient/provider/auth_user direct-write postprocessing through it instead of the in-process processNrtMessage() call. Adds a concurrency regression test proving no ids are lost under concurrent enqueue()/drain contention. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Both direct-write and legacy paths now converge on PostProcessingService.enqueue() rather than processNrtMessage() (which is now just the legacy listener's payload-parsing step that itself calls enqueue()). Documents the cacheLock guard added around producer-side cache writes to fix the producer/drain race. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Covers single-threaded correctness and contract behavior not exercised by the existing Kafka-message-path tests: basic add, multi-id accumulation, multi-topic isolation, drain integration for patient/provider/auth_user (parity with PersonService's direct-write call sites), serviceEnable bypass, null uid/topic rejection, and silent-drop-with-warning for an unrecognized topic. Concurrency safety is covered separately by PostProcessingServiceCacheConcurrencyTest. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…omMessage Restores the pre-fix, unsynchronized cdCache/idCache writes in extractIdFromMessage. The producer/drain race these guarded against is tracked separately for follow-up rather than fixed inline here.
…rvice-update-nrt-tables
Removes the synchronized(cacheLock) wrappers added around extractValFromMessage (pbCache/obsCache writes), extractSummaryCase (sumCache writes), and processDmMessage (dmCache writes) in 67befff. The producer/drain race these guarded against is now tracked in APP-815 for deliberate follow-up instead of being bundled into this branch. enqueue() is unaffected.
Adds a two-step functional test covering PersonService's direct-write path for providers: creating a new provider and updating its middle name/email, verifying both nrt_provider and D_PROVIDER pick up the changes. Both steps were validated live against the dev stack before being committed. Also switches the local nbs-mssql image tag to latest and removes the stale APP-787-Changes.md notes file.
|
|
||
| ## Key conditionals | ||
|
|
||
| - **`cd` field** routes `nbs_Person` events to patient vs. provider stored procs. |
There was a problem hiding this comment.
I thought the idea was to avoid the race between two things listening to the nrt_ topics (the sink connector, and the reporting-service) by ensuring that the nrt_ tables are always written first.
In this diagram, it looks like the suggestion is to avoid the race by not writing to the nrt_ topics at all, in the Person case.
But how does this compare to this alternative:
Before writing to the nrt_ topics, write to the nrt_ tables. Then send the message to the nrt_ topic, which is read by reporting-service and processed. (Turn off the sink connector for those topics, so there is no race as only one thing reads that topic.) The preserves some of the decoupling via messaging which may(?) help with scaling.
There was a problem hiding this comment.
If we publish to the nrt topics after writing to the nrt tables then Kafka-Connect will write to the nrt tables as well. We'll likely see errors in Kafka-Connect since the records would be there already (e.g., violation of KEY constraint) and who knows what issues that could cause.
Yes, we could modify the Kafka-Connect config to include the non-person nrt topics, but we would be forcing STLTs to perform additional manual processes so that they could switch between the 2 pathways.
- Update the feature flag for direct write
- Update the Kafka Connect config
- Restart/Redeploy reporting-service-pipeline
- Restart/Redeploy Kafka-Connect
There was a problem hiding this comment.
Yes, the kafka connect configuration would need to be modified to stop writing.
The feature flag is not for users, but for RTR developers so that the code can be merged safely to the main branch while this work is under development. When the feature is fully completed and tested, the feature flag and old code paths can be removed.
There was a problem hiding this comment.
Makes sense to me. We discussed during the DRAGON demo day and there were different opinions on this. Perhaps a team review will help determine what is best.
There was a problem hiding this comment.
I might be speaking in ignorance here but I believe this statement is incorrect:
We'll likely see errors in Kafka-Connect since the records would be there already
(e.g., violation of KEY constraint)
I'm pretty sure the Kafka Sink Connector performs an upsert and will avoid any errors if data is already present.
I do like the idea of continuing to write to the nrt_* topics as the signal to drive post processing. This allows us to manage retry mechanisms and scaling at that level instead of having to restart processing by querying ODSE if something goes wrong.
There was a problem hiding this comment.
Makes sense @mpeels . It doesn't hurt to continue writing to the topics, but wondering if we need to determine if a config update is needed to handle duplicates because I believe a standard INSERT will cause the task to halt in Kafka-Connect if a key violation error is raised. https://docs.confluent.io/kafka-connectors/jdbc/current/sink-connector/overview.html#jdbc-sink-idempotent-writes
| @Column(name = "race_white_2") | ||
| private String raceWhite2; | ||
|
|
||
| @Column(name = "race_white_3") | ||
| private String raceWhite3; | ||
|
|
||
| @Column(name = "race_white_gt3_ind") | ||
| private String raceWhiteGt3Ind; | ||
|
|
||
| @Column(name = "race_white_all") | ||
| private String raceWhiteAll; | ||
|
|
||
| private String patientNumber; | ||
| private String patientNumberAuth; | ||
| private String entryMethod; |
There was a problem hiding this comment.
(nit, non-blocking): Some fields are annotated with @Column definitions and others are not. I personally prefer that all or none of them be annotated.
I believe that the Spring Boot Hibernate implementation auto maps camel case to snake case (patientNumber becomes patient_number) by default. So we might not need any of the @Column annotations unless the field name does not match for some reason.
There was a problem hiding this comment.
makes sense to me. will update soon. Thanks!
…ency Addresses mpeels' PR review comment preferring all-or-none @column usage rather than a mix of annotated and unannotated fields.
3 participants
Description
Routes
PersonService's patient/provider/auth_user postprocessing through a direct write to theirnrt_*tables in RDB_MODERN instead of the Kafka round-trip (processNrtMessage), gated behind aPERSON_SERVICE_DIRECT_WRITEflag. AddsNrtAuthUser/NrtPatient/NrtProviderentities and repositories, a newPostProcessingService.enqueue()API so direct-write callers still go through the same priority-ordered batch drain, and functional tests covering the auth-user and provider direct-write paths (including a provider-update step).Related Issue
APP-787
Additional Notes
While working on
enqueue()and the producer paths intoPostProcessingService's in-memory caches, we found a pre-existing producer/drain race: producer writes toidCache/cdCache/pbCache/obsCache/sumCache/dmCachearen't synchronized against the scheduled drain's snapshot-then-clear, so a concurrentadd()can land in a queue right as it's being cleared and get silently, permanently dropped. A fix was prototyped but reverted out of this PR's scope so it can be tracked and applied deliberately — see APP-815.Checklist