Replace ProcessingJobBytesToRead with more generic ProcessingUnitBytesToRead (#5127)

SergeyGaluzo · web-flow · commit d717b6e5af20 · 2025-09-05T01:13:22.000Z
* Add processing job bytes to read parameter to the doc

* removed new

* job -&gt; unit

* doc back
diff --git a/src/Microsoft.Health.Fhir.Core/Extensions/ImportMediatorExtensions.cs b/src/Microsoft.Health.Fhir.Core/Extensions/ImportMediatorExtensions.cs
@@ -28,13 +28,13 @@ public static async Task<CreateImportResponse> ImportAsync(
             bool allowNegativeVersions,
             string errorContainerName,
             bool eventualConsistency,
-            int processingJobBytesToRead,
+            int processingUnitBytesToRead,
             CancellationToken cancellationToken)
         {
             EnsureArg.IsNotNull(mediator, nameof(mediator));
             EnsureArg.IsNotNull(requestUri, nameof(requestUri));
 
-            var request = new CreateImportRequest(requestUri, inputFormat, inputSource, input, storageDetail, importMode, allowNegativeVersions, errorContainerName, eventualConsistency, processingJobBytesToRead);
+            var request = new CreateImportRequest(requestUri, inputFormat, inputSource, input, storageDetail, importMode, allowNegativeVersions, errorContainerName, eventualConsistency, processingUnitBytesToRead);
 
             CreateImportResponse response = await mediator.Send(request, cancellationToken);
             return response;
diff --git a/src/Microsoft.Health.Fhir.Core/Features/Operations/Import/CreateImportRequestHandler.cs b/src/Microsoft.Health.Fhir.Core/Features/Operations/Import/CreateImportRequestHandler.cs
@@ -62,7 +62,7 @@ public async Task<CreateImportResponse> Handle(CreateImportRequest request, Canc
                 AllowNegativeVersions = request.AllowNegativeVersions,
                 EventualConsistency = request.EventualConsistency,
                 ErrorContainerName = request.ErrorContainerName,
-                ProcessingJobBytesToRead = request.ProcessingJobBytesToRead,
+                ProcessingUnitBytesToRead = request.ProcessingUnitBytesToRead,
             };
 
             var jobInfo = (await _queueClient.EnqueueAsync(QueueType.Import, cancellationToken, definitions: definitionObj))[0];
diff --git a/src/Microsoft.Health.Fhir.Core/Features/Operations/Import/ImportOrchestratorJobDefinition.cs b/src/Microsoft.Health.Fhir.Core/Features/Operations/Import/ImportOrchestratorJobDefinition.cs
@@ -75,6 +75,6 @@ public class ImportOrchestratorJobDefinition : IJobData
         /// If not speficied it is 10 million bytes. In case of very large resources (binary data),
         /// this should be increased to the resource size to avoid unnecessary input file scans.
         /// </summary>
-        public int ProcessingJobBytesToRead { get; set; }
+        public int ProcessingUnitBytesToRead { get; set; }
     }
 }
diff --git a/src/Microsoft.Health.Fhir.Core/Features/Operations/Import/Models/ImportRequest.cs b/src/Microsoft.Health.Fhir.Core/Features/Operations/Import/Models/ImportRequest.cs
@@ -63,6 +63,6 @@ public class ImportRequest
         /// <summary>
         /// Number of bytes to be read by processing job.
         /// </summary>
-        public int ProcessingJobBytesToRead { get; set; }
+        public int ProcessingUnitBytesToRead { get; set; }
     }
 }
diff --git a/src/Microsoft.Health.Fhir.Core/Messages/BulkImport/CreateImportRequest.cs b/src/Microsoft.Health.Fhir.Core/Messages/BulkImport/CreateImportRequest.cs
@@ -24,7 +24,7 @@ public CreateImportRequest(
             bool allowNegativeVersions,
             string errorContainerName,
             bool eventualConsistency,
-            int processingJobBytesToRead)
+            int processingUnitBytesToRead)
         {
             EnsureArg.IsNotNull(requestUri, nameof(requestUri));
 
@@ -37,7 +37,7 @@ public CreateImportRequest(
             AllowNegativeVersions = allowNegativeVersions;
             ErrorContainerName = errorContainerName;
             EventualConsistency = eventualConsistency;
-            ProcessingJobBytesToRead = processingJobBytesToRead;
+            ProcessingUnitBytesToRead = processingUnitBytesToRead;
         }
 
         /// <summary>
@@ -92,6 +92,6 @@ public CreateImportRequest(
         /// <summary>
         /// Number of bytes to be read by processing job.
         /// </summary>
-        public int ProcessingJobBytesToRead { get; set; }
+        public int ProcessingUnitBytesToRead { get; set; }
     }
 }
diff --git a/src/Microsoft.Health.Fhir.Shared.Api/Controllers/ImportController.cs b/src/Microsoft.Health.Fhir.Shared.Api/Controllers/ImportController.cs
@@ -118,7 +118,7 @@ public async Task<IActionResult> Import([FromBody] Parameters importTaskParamete
                  importRequest.AllowNegativeVersions,
                  importRequest.ErrorContainerName,
                  importRequest.EventualConsistency,
-                 importRequest.ProcessingJobBytesToRead,
+                 importRequest.ProcessingUnitBytesToRead,
                  HttpContext.RequestAborted);
 
             var bulkImportResult = ImportResult.Accepted();
diff --git a/src/Microsoft.Health.Fhir.Shared.Api/Features/Operations/Import/ImportRequestExtensions.cs b/src/Microsoft.Health.Fhir.Shared.Api/Features/Operations/Import/ImportRequestExtensions.cs
@@ -26,7 +26,7 @@ public static class ImportRequestExtensions
         public const string ForceParameterName = "force";
         public const string AllowNegativeVersionsParameterName = "allowNegativeVersions";
         public const string EventualConsistencyParameterName = "eventualConsistency";
-        public const string ProcessingJobBytesToReadParameterName = "processingJobBytesToRead";
+        public const string ProcessingUnitBytesToReadParameterName = "processingUnitBytesToRead";
         public const string ErrorContainerNameParameterName = "errorContainerName";
         public const string DefaultStorageDetailType = "azure-blob";
 
@@ -105,9 +105,9 @@ public static Parameters ToParameters(this ImportRequest importRequest)
                 parameters.Add(ErrorContainerNameParameterName, new FhirString(importRequest.ErrorContainerName));
             }
 
-            if (importRequest.ProcessingJobBytesToRead > 0)
+            if (importRequest.ProcessingUnitBytesToRead > 0)
             {
-                parameters.Add(ProcessingJobBytesToReadParameterName, new FhirString(importRequest.ProcessingJobBytesToRead.ToString()));
+                parameters.Add(ProcessingUnitBytesToReadParameterName, new Integer(importRequest.ProcessingUnitBytesToRead));
             }
 
             return parameters;
@@ -193,9 +193,9 @@ public static ImportRequest ExtractImportRequest(this Parameters parameters)
                 importRequest.ErrorContainerName = errorContainerName;
             }
 
-            if (parameters.TryGetIntValue(ProcessingJobBytesToReadParameterName, out int bytesToRead))
+            if (parameters.TryGetIntValue(ProcessingUnitBytesToReadParameterName, out int bytesToRead))
             {
-                importRequest.ProcessingJobBytesToRead = bytesToRead;
+                importRequest.ProcessingUnitBytesToRead = bytesToRead;
             }
 
             return importRequest;
diff --git a/src/Microsoft.Health.Fhir.SqlServer/Features/Operations/Import/ImportOrchestratorJob.cs b/src/Microsoft.Health.Fhir.SqlServer/Features/Operations/Import/ImportOrchestratorJob.cs
@@ -236,9 +236,9 @@ private async Task EnqueueProcessingJobsAsync(JobInfo coord, ImportOrchestratorJ
             {
                 var blobLength = (long)(await _integrationDataStoreClient.GetPropertiesAsync(input.Url, cancellationToken))[IntegrationDataStoreClientConstants.BlobPropertyLength];
                 result.TotalBytes += blobLength;
-                var bytesToRead = coordDefinition.ProcessingJobBytesToRead == 0
+                var bytesToRead = coordDefinition.ProcessingUnitBytesToRead == 0
                                 ? BytesToReadDefault
-                                : coordDefinition.ProcessingJobBytesToRead;
+                                : coordDefinition.ProcessingUnitBytesToRead;
                 foreach (var offset in GetOffsets(blobLength, bytesToRead))
                 {
                     var newInput = input.Clone();
diff --git a/test/Microsoft.Health.Fhir.Shared.Tests.E2E/Rest/Import/ImportTests.cs b/test/Microsoft.Health.Fhir.Shared.Tests.E2E/Rest/Import/ImportTests.cs
@@ -336,17 +336,17 @@ private async Task<Uri> CreateNDJson(int resources)
         }
 
         [Fact]
-        public async Task ProcessingJobBytesToReadHonored()
+        public async Task ProcessingUnitBytesToReadHonored()
         {
             var ndJson = CreateTestPatient(Guid.NewGuid().ToString("N"));
 
             //// set small bytes to read, so there are multiple processing jobs
-            var request = CreateImportRequest((await ImportTestHelper.UploadFileAsync(ndJson, _fixture.StorageAccount)).location, ImportMode.IncrementalLoad, processingJobBytesToRead: 10);
+            var request = CreateImportRequest((await ImportTestHelper.UploadFileAsync(ndJson, _fixture.StorageAccount)).location, ImportMode.IncrementalLoad, processingUnitBytesToRead: 10);
             var result = await ImportCheckAsync(request, null, 0, true);
             Assert.Equal(7, result.Output.Count); // 7 processing jobs
 
             //// no details
-            request = CreateImportRequest((await ImportTestHelper.UploadFileAsync(ndJson, _fixture.StorageAccount)).location, ImportMode.IncrementalLoad, processingJobBytesToRead: 10);
+            request = CreateImportRequest((await ImportTestHelper.UploadFileAsync(ndJson, _fixture.StorageAccount)).location, ImportMode.IncrementalLoad, processingUnitBytesToRead: 10);
             result = await ImportCheckAsync(request, null, 0, false);
             Assert.Single(result.Output);
 
@@ -356,7 +356,7 @@ public async Task ProcessingJobBytesToReadHonored()
             Assert.Single(result.Output);
 
             //// 0 should be ovewritten by default in orchestrator job
-            request = CreateImportRequest((await ImportTestHelper.UploadFileAsync(ndJson, _fixture.StorageAccount)).location, ImportMode.IncrementalLoad, processingJobBytesToRead: 0);
+            request = CreateImportRequest((await ImportTestHelper.UploadFileAsync(ndJson, _fixture.StorageAccount)).location, ImportMode.IncrementalLoad, processingUnitBytesToRead: 0);
             result = await ImportCheckAsync(request, null, 0, true);
             Assert.Single(result.Output);
         }
@@ -1184,7 +1184,7 @@ private static DateTimeOffset GetLastUpdated(string lastUpdatedYear)
             return DateTimeOffset.Parse(lastUpdatedYear + "-01-01T00:00:00.000+00:00");
         }
 
-        private static ImportRequest CreateImportRequest(IList<Uri> locations, ImportMode importMode, bool setResourceType = true, bool allowNegativeVersions = false, string errorContainerName = null, bool eventualConsistency = false, int? processingJobBytesToRead = null)
+        private static ImportRequest CreateImportRequest(IList<Uri> locations, ImportMode importMode, bool setResourceType = true, bool allowNegativeVersions = false, string errorContainerName = null, bool eventualConsistency = false, int? processingUnitBytesToRead = null)
         {
             var input = locations.Select(location =>
             {
@@ -1209,17 +1209,17 @@ private static ImportRequest CreateImportRequest(IList<Uri> locations, ImportMod
                 ErrorContainerName = errorContainerName,
             };
 
-            if (processingJobBytesToRead.HasValue)
+            if (processingUnitBytesToRead.HasValue)
             {
-                request.ProcessingJobBytesToRead = processingJobBytesToRead.Value;
+                request.ProcessingUnitBytesToRead = processingUnitBytesToRead.Value;
             }
 
             return request;
         }
 
-        private static ImportRequest CreateImportRequest(Uri location, ImportMode importMode, bool setResourceType = true, bool allowNegativeVersions = false, string errorContainerName = null, bool eventualConsistency = false, int? processingJobBytesToRead = null)
+        private static ImportRequest CreateImportRequest(Uri location, ImportMode importMode, bool setResourceType = true, bool allowNegativeVersions = false, string errorContainerName = null, bool eventualConsistency = false, int? processingUnitBytesToRead = null)
         {
-            return CreateImportRequest([location], importMode, setResourceType, allowNegativeVersions, errorContainerName, eventualConsistency, processingJobBytesToRead);
+            return CreateImportRequest([location], importMode, setResourceType, allowNegativeVersions, errorContainerName, eventualConsistency, processingUnitBytesToRead);
         }
 
         private static string PrepareResource(string id, string version, string lastUpdatedYear)

Original file line number	Diff line number	Diff line change
`@@ -75,6 +75,6 @@ public class ImportOrchestratorJobDefinition : IJobData`
`75`	`75`	`/// If not speficied it is 10 million bytes. In case of very large resources (binary data),`
`76`	`76`	`/// this should be increased to the resource size to avoid unnecessary input file scans.`
`77`	`77`	`/// </summary>`
`78`		`- public int ProcessingJobBytesToRead { get; set; }`
	`78`	`+ public int ProcessingUnitBytesToRead { get; set; }`
`79`	`79`	`}`
`80`	`80`	`}`
Original file line number	Diff line number	Diff line change
`@@ -63,6 +63,6 @@ public class ImportRequest`
`63`	`63`	`/// <summary>`
`64`	`64`	`/// Number of bytes to be read by processing job.`
`65`	`65`	`/// </summary>`
`66`		`- public int ProcessingJobBytesToRead { get; set; }`
	`66`	`+ public int ProcessingUnitBytesToRead { get; set; }`
`67`	`67`	`}`
`68`	`68`	`}`
Original file line number	Diff line number	Diff line change
`@@ -24,7 +24,7 @@ public CreateImportRequest(`
`24`	`24`	`bool allowNegativeVersions,`
`25`	`25`	`string errorContainerName,`
`26`	`26`	`bool eventualConsistency,`
`27`		`- int processingJobBytesToRead)`
	`27`	`+ int processingUnitBytesToRead)`
`28`	`28`	`{`
`29`	`29`	`EnsureArg.IsNotNull(requestUri, nameof(requestUri));`
`30`	`30`
`@@ -37,7 +37,7 @@ public CreateImportRequest(`
`37`	`37`	`AllowNegativeVersions = allowNegativeVersions;`
`38`	`38`	`ErrorContainerName = errorContainerName;`
`39`	`39`	`EventualConsistency = eventualConsistency;`
`40`		`- ProcessingJobBytesToRead = processingJobBytesToRead;`
	`40`	`+ ProcessingUnitBytesToRead = processingUnitBytesToRead;`
`41`	`41`	`}`
`42`	`42`
`43`	`43`	`/// <summary>`
`@@ -92,6 +92,6 @@ public CreateImportRequest(`
`92`	`92`	`/// <summary>`
`93`	`93`	`/// Number of bytes to be read by processing job.`
`94`	`94`	`/// </summary>`
`95`		`- public int ProcessingJobBytesToRead { get; set; }`
	`95`	`+ public int ProcessingUnitBytesToRead { get; set; }`
`96`	`96`	`}`
`97`	`97`	`}`
Original file line number	Diff line number	Diff line change
`@@ -26,7 +26,7 @@ public static class ImportRequestExtensions`
`26`	`26`	`public const string ForceParameterName = "force";`
`27`	`27`	`public const string AllowNegativeVersionsParameterName = "allowNegativeVersions";`
`28`	`28`	`public const string EventualConsistencyParameterName = "eventualConsistency";`
`29`		`- public const string ProcessingJobBytesToReadParameterName = "processingJobBytesToRead";`
	`29`	`+ public const string ProcessingUnitBytesToReadParameterName = "processingUnitBytesToRead";`
`30`	`30`	`public const string ErrorContainerNameParameterName = "errorContainerName";`
`31`	`31`	`public const string DefaultStorageDetailType = "azure-blob";`
`32`	`32`
`@@ -105,9 +105,9 @@ public static Parameters ToParameters(this ImportRequest importRequest)`
`105`	`105`	`parameters.Add(ErrorContainerNameParameterName, new FhirString(importRequest.ErrorContainerName));`
`106`	`106`	`}`
`107`	`107`
`108`		`- if (importRequest.ProcessingJobBytesToRead > 0)`
	`108`	`+ if (importRequest.ProcessingUnitBytesToRead > 0)`
`109`	`109`	`{`
`110`		`- parameters.Add(ProcessingJobBytesToReadParameterName, new FhirString(importRequest.ProcessingJobBytesToRead.ToString()));`
	`110`	`+ parameters.Add(ProcessingUnitBytesToReadParameterName, new Integer(importRequest.ProcessingUnitBytesToRead));`
`111`	`111`	`}`
`112`	`112`
`113`	`113`	`return parameters;`
`@@ -193,9 +193,9 @@ public static ImportRequest ExtractImportRequest(this Parameters parameters)`
`193`	`193`	`importRequest.ErrorContainerName = errorContainerName;`
`194`	`194`	`}`
`195`	`195`
`196`		`- if (parameters.TryGetIntValue(ProcessingJobBytesToReadParameterName, out int bytesToRead))`
	`196`	`+ if (parameters.TryGetIntValue(ProcessingUnitBytesToReadParameterName, out int bytesToRead))`
`197`	`197`	`{`
`198`		`- importRequest.ProcessingJobBytesToRead = bytesToRead;`
	`198`	`+ importRequest.ProcessingUnitBytesToRead = bytesToRead;`
`199`	`199`	`}`
`200`	`200`
`201`	`201`	`return importRequest;`
Original file line number	Diff line number	Diff line change
`@@ -236,9 +236,9 @@ private async Task EnqueueProcessingJobsAsync(JobInfo coord, ImportOrchestratorJ`
`236`	`236`	`{`
`237`	`237`	`var blobLength = (long)(await _integrationDataStoreClient.GetPropertiesAsync(input.Url, cancellationToken))[IntegrationDataStoreClientConstants.BlobPropertyLength];`
`238`	`238`	`result.TotalBytes += blobLength;`
`239`		`- var bytesToRead = coordDefinition.ProcessingJobBytesToRead == 0`
	`239`	`+ var bytesToRead = coordDefinition.ProcessingUnitBytesToRead == 0`
`240`	`240`	`? BytesToReadDefault`
`241`		`- : coordDefinition.ProcessingJobBytesToRead;`
	`241`	`+ : coordDefinition.ProcessingUnitBytesToRead;`
`242`	`242`	`foreach (var offset in GetOffsets(blobLength, bytesToRead))`
`243`	`243`	`{`
`244`	`244`	`var newInput = input.Clone();`
Original file line number	Diff line number	Diff line change
`@@ -336,17 +336,17 @@ private async Task<Uri> CreateNDJson(int resources)`
`336`	`336`	`}`
`337`	`337`
`338`	`338`	`[Fact]`
`339`		`- public async Task ProcessingJobBytesToReadHonored()`
	`339`	`+ public async Task ProcessingUnitBytesToReadHonored()`
`340`	`340`	`{`
`341`	`341`	`var ndJson = CreateTestPatient(Guid.NewGuid().ToString("N"));`
`342`	`342`
`343`	`343`	`//// set small bytes to read, so there are multiple processing jobs`
`344`		`- var request = CreateImportRequest((await ImportTestHelper.UploadFileAsync(ndJson, _fixture.StorageAccount)).location, ImportMode.IncrementalLoad, processingJobBytesToRead: 10);`
	`344`	`+ var request = CreateImportRequest((await ImportTestHelper.UploadFileAsync(ndJson, _fixture.StorageAccount)).location, ImportMode.IncrementalLoad, processingUnitBytesToRead: 10);`
`345`	`345`	`var result = await ImportCheckAsync(request, null, 0, true);`
`346`	`346`	`Assert.Equal(7, result.Output.Count); // 7 processing jobs`
`347`	`347`
`348`	`348`	`//// no details`
`349`		`- request = CreateImportRequest((await ImportTestHelper.UploadFileAsync(ndJson, _fixture.StorageAccount)).location, ImportMode.IncrementalLoad, processingJobBytesToRead: 10);`
	`349`	`+ request = CreateImportRequest((await ImportTestHelper.UploadFileAsync(ndJson, _fixture.StorageAccount)).location, ImportMode.IncrementalLoad, processingUnitBytesToRead: 10);`
`350`	`350`	`result = await ImportCheckAsync(request, null, 0, false);`
`351`	`351`	`Assert.Single(result.Output);`
`352`	`352`
`@@ -356,7 +356,7 @@ public async Task ProcessingJobBytesToReadHonored()`
`356`	`356`	`Assert.Single(result.Output);`
`357`	`357`
`358`	`358`	`//// 0 should be ovewritten by default in orchestrator job`
`359`		`- request = CreateImportRequest((await ImportTestHelper.UploadFileAsync(ndJson, _fixture.StorageAccount)).location, ImportMode.IncrementalLoad, processingJobBytesToRead: 0);`
	`359`	`+ request = CreateImportRequest((await ImportTestHelper.UploadFileAsync(ndJson, _fixture.StorageAccount)).location, ImportMode.IncrementalLoad, processingUnitBytesToRead: 0);`
`360`	`360`	`result = await ImportCheckAsync(request, null, 0, true);`
`361`	`361`	`Assert.Single(result.Output);`
`362`	`362`	`}`
`@@ -1184,7 +1184,7 @@ private static DateTimeOffset GetLastUpdated(string lastUpdatedYear)`
`1184`	`1184`	`return DateTimeOffset.Parse(lastUpdatedYear + "-01-01T00:00:00.000+00:00");`
`1185`	`1185`	`}`
`1186`	`1186`
`1187`		`- private static ImportRequest CreateImportRequest(IList<Uri> locations, ImportMode importMode, bool setResourceType = true, bool allowNegativeVersions = false, string errorContainerName = null, bool eventualConsistency = false, int? processingJobBytesToRead = null)`
	`1187`	`+ private static ImportRequest CreateImportRequest(IList<Uri> locations, ImportMode importMode, bool setResourceType = true, bool allowNegativeVersions = false, string errorContainerName = null, bool eventualConsistency = false, int? processingUnitBytesToRead = null)`
`1188`	`1188`	`{`
`1189`	`1189`	`var input = locations.Select(location =>`
`1190`	`1190`	`{`
`@@ -1209,17 +1209,17 @@ private static ImportRequest CreateImportRequest(IList<Uri> locations, ImportMod`
`1209`	`1209`	`ErrorContainerName = errorContainerName,`
`1210`	`1210`	`};`
`1211`	`1211`
`1212`		`- if (processingJobBytesToRead.HasValue)`
	`1212`	`+ if (processingUnitBytesToRead.HasValue)`
`1213`	`1213`	`{`
`1214`		`- request.ProcessingJobBytesToRead = processingJobBytesToRead.Value;`
	`1214`	`+ request.ProcessingUnitBytesToRead = processingUnitBytesToRead.Value;`
`1215`	`1215`	`}`
`1216`	`1216`
`1217`	`1217`	`return request;`
`1218`	`1218`	`}`
`1219`	`1219`
`1220`		`- private static ImportRequest CreateImportRequest(Uri location, ImportMode importMode, bool setResourceType = true, bool allowNegativeVersions = false, string errorContainerName = null, bool eventualConsistency = false, int? processingJobBytesToRead = null)`
	`1220`	`+ private static ImportRequest CreateImportRequest(Uri location, ImportMode importMode, bool setResourceType = true, bool allowNegativeVersions = false, string errorContainerName = null, bool eventualConsistency = false, int? processingUnitBytesToRead = null)`
`1221`	`1221`	`{`
`1222`		`- return CreateImportRequest([location], importMode, setResourceType, allowNegativeVersions, errorContainerName, eventualConsistency, processingJobBytesToRead);`
	`1222`	`+ return CreateImportRequest([location], importMode, setResourceType, allowNegativeVersions, errorContainerName, eventualConsistency, processingUnitBytesToRead);`
`1223`	`1223`	`}`
`1224`	`1224`
`1225`	`1225`	`private static string PrepareResource(string id, string version, string lastUpdatedYear)`