HealthLake FHIR Import Job Health Gate

A failed HealthLake FHIR import is not a pipeline inconvenience, it is patient data that never landed in the clinical record the rest of the system now assumes is complete. HealthLake reports exactly what happened to every batch, jobStatus, message, timing, but by default nothing reads that report after the import call returns. This blueprint reacts to the job itself finishing, fetches the exact outcome, and alerts Slack with the job's own error message, never a guess, the moment a batch fails or partially lands.

How it works

1. healthlake_job_terminal (io.kestra.plugin.aws.healthlake.Trigger) polls the datastore named by HEALTHLAKE_DATASTORE_ID every interval and starts this flow the moment the most recently submitted job of jobType reaches a terminal state (COMPLETED, COMPLETED_WITH_ERRORS, FAILED, CANCEL_COMPLETED, or CANCEL_FAILED). It exposes trigger.jobId and trigger.jobStatus to every task below, and persists the last job ID it fired on in the namespace KV store so a job already handled is never re-fired on the next poll.
2. probe_job_detail (io.kestra.plugin.aws.healthlake.DescribeImportJob) re-fetches the same job by trigger.jobId, with allowFailure: true so an API hiccup does not stop the flow before it can be classified as PROBE_FAILED below. The trigger already knows the terminal jobStatus, this step exists specifically to get the job's own message field, the actual error text, which the trigger's output does not carry.
3. classify_job_outcome (io.kestra.plugin.core.debug.Return) returns PROBE_FAILED when probe_job_detail produced no output at all (guarded with is not defined), HEALTHY on COMPLETED, CANCELLED on CANCEL_COMPLETED, DEGRADED on COMPLETED_WITH_ERRORS unless partial_success_is_critical is true, in which case it is FAILED, and FAILED for anything else (FAILED or CANCEL_FAILED).
4. route_by_job_outcome (io.kestra.plugin.core.flow.Switch) branches on that status:
   • HEALTHY, CANCELLED: log_healthy / log_cancelled log it.
   • DEGRADED, FAILED, PROBE_FAILED: a SlackIncomingWebhook task alerts with the job's own truncated message, never an invented explanation.
5. log_audit_summary always runs last and logs the final classification.
6. The errors block posts to Slack if the flow itself fails outside the gate logic above.

What you get

• A real-time reaction to a FHIR import failing or partially landing, instead of finding out from a downstream clinical system querying incomplete data.
• A Slack alert carrying the job's own error message, truncated to a readable length, never an invented explanation of what went wrong.
• A policy choice, partial_success_is_critical, for datastores where "mostly landed" is not an acceptable outcome, versus ones where a small error rate is tolerated and logged as DEGRADED instead of paging anyone.
• Automatic deduplication: the trigger's own KV-backed state means a terminal job fires this flow exactly once, not on every poll until a new job is submitted.

Who it's for

• Healthcare data engineering and integration teams operating AWS HealthLake as a FHIR datastore, who need to know the moment a bulk import fails or partially lands, not at the next manual check.
• Compliance and data governance teams who need an audit trail proving every import batch was checked, with the real outcome and error text attached to each execution.
• Teams evaluating the HealthLake plugin who want a working example of its dedicated polling Trigger, paired with DescribeImportJob for the detail the trigger itself does not carry.

Why orchestrate this with Kestra

HealthLake's own StartFHIRImportJob API returns once the job is submitted and has no concept of a callback, a policy on partial success, or an alert built from the real error text. Kestra adds the event-driven reaction via the plugin's own Trigger, the classification policy, the Slack alert carrying HealthLake's own message, and a full execution history of every import outcome, all as one declarative YAML file you can version and review like code.

Prerequisites

• An AWS HealthLake FHIR data store that receives bulk imports via StartImportJob (elsewhere in your pipeline) or the AWS console or CLI.
• An AWS credential with healthlake:DescribeFHIRImportJob and healthlake:ListFHIRImportJobs permission.
• A Slack incoming webhook for alerts.
• HealthLake is a managed AWS service with no self-hostable open-source emulator, there is no local container to start for HealthLake itself. This plugin calls the AWS SDK directly from the Kestra worker process, there is no sub-container taskRunner and no Docker network to join.

Secrets

• AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_DEFAULT_REGION: credentials used by the trigger and probe_job_detail.

• HEALTHLAKE_DATASTORE_ID: the FHIR data store this flow monitors.

• SLACK_WEBHOOK: Slack incoming webhook URL used by the alert tasks and the errors block.

  In Kestra OSS, set all five as base64-encoded environment variables: SECRET_AWS_ACCESS_KEY_ID, SECRET_AWS_SECRET_ACCESS_KEY, SECRET_AWS_DEFAULT_REGION, SECRET_HEALTHLAKE_DATASTORE_ID, SECRET_SLACK_WEBHOOK.

Inputs

Unlike a Schedule-triggered flow, the job identity and datastore here come from the trigger's own configuration and output (trigger.jobId, trigger.jobStatus), not from flow inputs, that is how every Kestra trigger works. The inputs below only tune the classification and alert text.

Input	Type	Default	Allowed values
job_type_label	SELECT	IMPORT	IMPORT, EXPORT
partial_success_is_critical	BOOL	false	true, false
error_message_truncate_length	INT	300	any non-negative integer

Outputs

• {{ trigger.jobId }}: ID of the HealthLake import or export job that fired the trigger.
• {{ trigger.jobStatus }}: Terminal job status detected by the polling trigger (COMPLETED, COMPLETED_WITH_ERRORS, FAILED, CANCEL_COMPLETED, or CANCEL_FAILED).
• {{ outputs.probe_job_detail.jobStatus }}: Confirmed job status returned by DescribeImportJob.
• {{ outputs.probe_job_detail.message }}: Service error or diagnostic message attached to the job.
• {{ outputs.classify_job_outcome.value }}: Outcome classification evaluated by the gate (HEALTHY, CANCELLED, DEGRADED, FAILED, or PROBE_FAILED).

Quick start

1. Add the AWS, HEALTHLAKE_DATASTORE_ID, and SLACK_WEBHOOK secrets.
2. Submit a test import with StartImportJob against a non-production datastore.
3. Enable the healthlake_job_terminal trigger and confirm the execution fires once the test job reaches a terminal state, with the right classification in log_audit_summary.
4. Set partial_success_is_critical to true on datastores where a partially-landed batch must always page someone.

How to extend

• Add io.kestra.plugin.aws.healthlake.DescribeDatastore as a second probe to also check the datastore's own datastoreStatus is ACTIVE, catching the rarer case where the datastore itself is degraded independently of any one job.
• Duplicate the trigger with jobType: EXPORT and job_type_label: "EXPORT" as a second flow to monitor export jobs the same way, since one trigger only watches one job type.
• Push trigger.jobId and the classification into a Kestra KV entry keyed by date, to build a rolling import success-rate report instead of only per-job alerts.

Pitfalls

• The trigger's own Javadoc is explicit that listFhirImportJobs with maxResults(1) returns the most recently submitted job but does not guarantee strict submit-time ordering. If two imports are submitted in quick succession, confirm which one actually fired by checking trigger.jobId against your own records, do not assume it is always the very last one you submitted.
• One Trigger instance watches one jobType (IMPORT or EXPORT) and one datastore. Monitoring both job types, or more than one datastore, needs a separate trigger and flow per combination, not one flow with an input switch.
• The trigger deduplicates by job ID in namespace KV, if you delete or rename the KV entry manually, the next poll can re-fire on a job this flow already alerted on.
• message can be empty on some terminal states, this blueprint falls back to "no message returned" rather than rendering a blank alert.

Links

• AWS plugin documentation
• AWS HealthLake FHIR bulk import and export