Bulk Import Operation

Question

Hi,

I am trying to use a feature Bulk import/Export to transfer legacy data (~130millins records) to our database through Kafka. There is a also live data stream coming from cloud Kafka already persisting on that same database.

During bulk Import batch job, I get error "ResourceVersionConflictException" and batch job fails because data we have on the prod have different version_Id tham the data from bulk Import.

For example Observation/81920296/histroy/_1 and Observation/81920296/history/_2 cannot both exists for the same Observation Id.(ERROR : Trying to update Observation/81920296/_history/2 but this is not the current version)

I want to keep all the history data. I do not want to trim any information. If there are any data related to a specific resource Id(FHIR_ID) I want to keep the records of them in the database.

Could you please help me with the necessary configuration setting or changes I might to do with my files(it's Binary json urls from an Bulk Export job )?

From research I find out I should put properties like below but I am not sure that it is enough.
initialImportMode = false
update_with_history_rewrite_enabled = true

preserve_existing_ids = true

Any help is greatly appreciated.

Thank you,
Niloufar

Answer 1

Hello Niloufar,

To resolve the ResourceVersionConflictException and successfully maintain your history while importing, you need to configure your $import request to handle non-sequential versions.

The configuration properties you found during your research (initialImportMode = false, update_with_history_rewrite_enabled = true, preserve_existing_ids = true) are not valid properties for the Azure Health Data Services $import JSON payload.

Instead, you should rely on IncrementalLoad and the allowNegativeVersions flag directly in your HTTP request body.

According to the official Microsoft documentation on FHIR Import:

Incremental mode: "Enables ingestion of multiple versions of FHIR resources in single batch while maintaining resource history and original lastUpdated values... Allows you to load resources in a nonsequential order of versions."

allowNegativeVersions: "Allows FHIR server assigning negative versions for resource records with explicit lastUpdated value and no version specified when input doesn't fit in contiguous space of positive versions existing in the store."

Here is the correct configuration for your API request payload:

POST https://<<FHIR service base URL>>/$import
Prefer: respond-async
Content-Type: application/fhir+json

{
  "inputFormat": "application/fhir+ndjson",
  "mode": "IncrementalLoad",
  "allowNegativeVersions": true,
  "input": [
    {
      "type": "Observation",
      "url": "https://<your-storage-account>.blob.core.windows.net/<container>/Observation.ndjson"
    }
  ]
}

Handling Live Stream Conflicts: A ResourceVersionConflictException is a standard concurrency control exception. Because you have a live Kafka stream pushing updates to the same database simultaneously, if the live stream and the bulk import attempt to modify the exact same resource ID at the exact same time, the server will block one to prevent data corruption.

If this continues to cause batch failures even with IncrementalLoad, you may need to temporarily pause the Kafka stream for the specific legacy Observation resources being backfilled or segregate the legacy IDs from the live IDs until the import completes.

Note: This response is drafted with the help of AI systems.

Answer 2

Hi Rakesh,

Thank you so much for the detailed explanation and configuration snip code. I did apply the changes and did not have batch failures anymore and the task is in progressing. Though as you advised, for data quality I requested to hold on streaming live data until all legacy has been seeded.

Thank you very much again for your time and support.

Niloufar

Answer 3

Hello Niloufar !

Thank you for posting on MS Learn Q&A.

I think you have a version history conflict not only a missing configuration issue.

In Azure Health Data Services FHIR service, historical versions can be imported only when the resource history can be applied consistently. So the mode for preserving historical versions and original meta.lastUpdated values is incremental import and not initial import because initial mode is intended for an empty server and only retains the latest version, while incremental mode supports versioned resources and concurrent ingestion.

https://learn.microsoft.com/en-us/azure/healthcare-apis/fhir/import-data

For your case, the problem is likely that the same Observation/{id} is being written by the live Kafka stream while the bulk import is trying to import older versions for that same resource ID. FHIR history is linear, you cannot have two different Observation/81920296/_history/2 records for the same resource ID. If prod already has a different current version importing legacy _history/1, _history/2... can fail with a conflict.

For Azure Health Data Services, the config should be :

"importConfiguration": {
  "enabled": true,
  "initialImportMode": false,
  "integrationDataStore": "<storage-account-name>"
}

And the import request should use:

{
  "resourceType": "Parameters",
  "parameter": [
    {
      "name": "inputFormat",
      "valueString": "application/fhir+ndjson"
    },
    {
      "name": "mode",
      "valueString": "IncrementalLoad"
    },
    {
      "name": "allowNegativeVersions",
      "valueBoolean": true
    },
    {
      "name": "input",
      "part": [
        {
          "name": "type",
          "valueString": "Observation"
        },
        {
          "name": "url",
          "valueUri": "https://<storage>.blob.core.windows.net/<container>/Observation.ndjson"
        }
      ]
    }
  ]
}

allowNegativeVersions can help only in specific cases where records have explicit lastUpdated values but no usable version sequence and it will not solve a conflict where the same FHIR ID already has a different version history in the target database.