feat: remove bloat from large files during upload (WIP) (#21757)

## Description

Currently, we try to upload large files by converting their binaries
into strings which leads to bloat in size. This is because converting to
bytes in a multi-byte encoding usually takes a larger space and white
characters are also included. We were also doing multiple modifications
which were just adding to the bloat.

Hence, we are now converting the binary data into an array buffer to
prevent this. This buffer is added to the multi-part form data request
as a new part and we add a pointer in the pace of the data which used to
be present earlier. This allows us to have minimal bloat on the payload
while sending the request.

TLDR: fix for uploading large files by changing the data type used for
upload.

*TODO:*
- [x] Client side payload changes
- [x] Server side double escape logic fixes
- [x] Server side tests
- [x] Server side refactor
- [ ] Cypress tests

Fixes #20642 

Media

## Type of change

- New feature (non-breaking change which adds functionality)


## How Has This Been Tested?

- Manual

### Test Plan
> Add Testsmith test cases links that relate to this PR

### Issues raised during DP testing
> Link issues raised during DP testing for better visiblity and tracking
(copy link from comments dropped on this PR)

## Checklist:
### Dev activity
- [x] My code follows the style guidelines of this project
- [x] I have performed a self-review of my own code
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [ ] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [ ] PR is being merged under a feature flag


### QA activity:
- [ ] Test plan has been approved by relevant developers
- [ ] Test plan has been peer reviewed by QA
- [ ] Cypress test cases have been added and approved by either SDET or
manual QA
- [ ] Organized project review call with relevant stakeholders after
Round 1/2 of QA
- [ ] Added Test Plan Approved label after reveiwing all Cypress test

---------

Co-authored-by: Nidhi Nair <nidhi@appsmith.com>

app/client/src/api/ActionAPI.tsx

@@ -49,7 +49,13 @@ export interface ExecuteActionRequest extends APIRequest {
   params?: Property[];
   paginationField?: PaginationField;
   viewMode: boolean;
-  paramProperties: Record<string, string | Record<string, string[]>>;
+  paramProperties: Record<
+    string,
+    | string
+    | Record<string, Array<string>>
+    | Record<string, string>
+    | Record<string, Record<string, Array<string>>>
+  >;
 }
 
 export type ExecuteActionResponse = ApiResponse & {

app/client/src/constants/WidgetConstants.tsx

@@ -216,3 +216,5 @@ export const WIDGET_PROPS_TO_SKIP_FROM_EVAL = {
  * It is also used to calculate widget positions and highlight placements.
  */
 export const FLEXBOX_PADDING = 4;
+
+export const FILE_SIZE_LIMIT_FOR_BLOBS = 5000 * 1024; // 5MB

app/client/src/sagas/ActionExecution/PluginActionSaga.ts

@@ -37,7 +37,18 @@ import {
   getAppMode,
   getCurrentApplication,
 } from "@appsmith/selectors/applicationSelectors";
-import { get, isArray, isString, set, find, isNil, flatten } from "lodash";
+import {
+  get,
+  isArray,
+  isString,
+  set,
+  find,
+  isNil,
+  flatten,
+  isArrayBuffer,
+  isEmpty,
+  unset,
+} from "lodash";
 import AppsmithConsole from "utils/AppsmithConsole";
 import { ENTITY_TYPE, PLATFORM_ERROR } from "entities/AppsmithConsole";
 import {
@@ -122,6 +133,7 @@ import { setDefaultActionDisplayFormat } from "./PluginActionSagaUtils";
 import { checkAndLogErrorsIfCyclicDependency } from "sagas/helper";
 import type { TRunDescription } from "workers/Evaluation/fns/actionFns";
 import { DEBUGGER_TAB_KEYS } from "components/editorComponents/Debugger/helpers";
+import { FILE_SIZE_LIMIT_FOR_BLOBS } from "constants/WidgetConstants";
 
 enum ActionResponseDataTypes {
   BINARY = "BINARY",
@@ -195,7 +207,16 @@ function* readBlob(blobUrl: string): any {
     if (fileType === FileDataTypes.Base64) {
       reader.readAsDataURL(file);
     } else if (fileType === FileDataTypes.Binary) {
-      reader.readAsBinaryString(file);
+      if (file.size < FILE_SIZE_LIMIT_FOR_BLOBS) {
+        //check size of the file, if less than 5mb, go with binary string method
+        // TODO: this method is deprecated, use readAsText instead
+        reader.readAsBinaryString(file);
+      } else {
+        // For files greater than 5 mb, use array buffer method
+        // This is to remove the bloat from the file which is added
+        // when using read as binary string method
+        reader.readAsArrayBuffer(file);
+      }
     } else {
       reader.readAsText(file);
     }
@@ -228,7 +249,9 @@ function* resolvingBlobUrls(
   //If array elements then dont push datatypes to payload.
   isArray
     ? arrDatatype?.push(dataType)
-    : (executeActionRequest.paramProperties[`k${index}`] = dataType);
+    : (executeActionRequest.paramProperties[`k${index}`] = {
+        datatype: dataType,
+      });
 
   if (isTrueObject(value)) {
     const blobUrlPaths: string[] = [];
@@ -242,6 +265,17 @@ function* resolvingBlobUrls(
       const blobUrl = value[blobUrlPath] as string;
       const resolvedBlobValue: unknown = yield call(readBlob, blobUrl);
       set(value, blobUrlPath, resolvedBlobValue);
+
+      // We need to store the url path map to be able to update the blob data
+      // and send the info to server
+
+      // Here we fetch the blobUrlPathMap from the action payload and update it
+      const blobUrlPathMap = get(value, "blobUrlPaths", {}) as Record<
+        string,
+        string
+      >;
+      set(blobUrlPathMap, blobUrlPath, blobUrl);
+      set(value, "blobUrlPaths", blobUrlPathMap);
     }
   } else if (isBlobUrl(value)) {
     // @ts-expect-error: Values can take many types
@@ -251,6 +285,28 @@ function* resolvingBlobUrls(
   return value;
 }
 
+// Function that updates the blob data in the action payload for large file
+// uploads
+function updateBlobDataFromUrls(
+  blobUrlPaths: Record<string, string>,
+  newVal: any,
+  blobMap: string[],
+  blobDataMap: Record<string, Blob>,
+) {
+  Object.entries(blobUrlPaths as Record<string, string>).forEach(
+    // blobUrl: string eg: blob:1234-1234-1234?type=binary
+    ([path, blobUrl]) => {
+      if (isArrayBuffer(newVal[path])) {
+        // remove the ?type=binary from the blob url if present
+        const sanitisedBlobURL = blobUrl.split("?")[0];
+        blobMap.push(sanitisedBlobURL);
+        set(blobDataMap, sanitisedBlobURL, new Blob([newVal[path]]));
+        set(newVal, path, sanitisedBlobURL);
+      }
+    },
+  );
+}
+
 /**
  * Api1
  * URL: https://example.com/{{Text1.text}}
@@ -298,6 +354,9 @@ function* evaluateActionParams(
   const bindingsMap: Record<string, string> = {};
   const bindingBlob = [];
 
+  // Maintain a blob data map to resolve blob urls of large files as array buffer
+  const blobDataMap: Record<string, Blob> = {};
+
   let recordFilePickerInstrumentation = false;
 
   // if json bindings have filepicker reference, we need to init the instrumentation object
@@ -311,12 +370,16 @@ function* evaluateActionParams(
     const key = bindings[i];
     let value = values[i];
 
+    let useBlobMaps = false;
+    // Maintain a blob map to resolve blob urls of large files
+    const blobMap: Array<string> = [];
+
     if (isArray(value)) {
       const tempArr = [];
-      const arrDatatype: string[] = [];
+      const arrDatatype: Array<string> = [];
       // array of objects containing blob urls that is loops and individual object is checked for resolution of blob urls.
       for (const val of value) {
-        const newVal: unknown = yield call(
+        const newVal: Record<string, any> = yield call(
           resolvingBlobUrls,
           val,
           executeActionRequest,
@@ -324,11 +387,22 @@ function* evaluateActionParams(
           true,
           arrDatatype,
         );
+
+        if (newVal.hasOwnProperty("blobUrlPaths")) {
+          updateBlobDataFromUrls(
+            newVal.blobUrlPaths,
+            newVal,
+            blobMap,
+            blobDataMap,
+          );
+          useBlobMaps = true;
+          unset(newVal, "blobUrlPaths");
+        }
+
         tempArr.push(newVal);
 
         if (key.includes(".files") && recordFilePickerInstrumentation) {
           filePickerInstrumentation["numberOfFiles"] += 1;
-          // @ts-expect-error: Values can take many types
           const { size, type } = newVal;
           filePickerInstrumentation["totalSize"] += size;
           filePickerInstrumentation["fileSizes"].push(size);
@@ -337,7 +411,7 @@ function* evaluateActionParams(
       }
       //Adding array datatype along with the datatype of first element of the array
       executeActionRequest.paramProperties[`k${i}`] = {
-        array: [arrDatatype[0]],
+        datatype: { array: [arrDatatype[0]] },
       };
       value = tempArr;
     } else {
@@ -352,17 +426,44 @@ function* evaluateActionParams(
     }
 
     if (typeof value === "object") {
+      // This is used in cases of large files, we store the bloburls with the path they were set in
+      // This helps in creating a unique map of blob urls to blob data when passing to the server
+      if (!!value && value.hasOwnProperty("blobUrlPaths")) {
+        updateBlobDataFromUrls(value.blobUrlPaths, value, blobMap, blobDataMap);
+        unset(value, "blobUrlPaths");
+      }
+
       value = JSON.stringify(value);
     }
 
-    value = new Blob([value], { type: "text/plain" });
+    // If there are no blob urls in the value, we can directly add it to the formData
+    // If there are blob urls, we need to add them to the blobDataMap
+    if (!useBlobMaps) {
+      value = new Blob([value], { type: "text/plain" });
+    }
     bindingsMap[key] = `k${i}`;
     bindingBlob.push({ name: `k${i}`, value: value });
+
+    // We need to add the blob map to the param properties
+    // This will allow the server to handle the scenaio of large files upload using blob data
+    const paramProperties = executeActionRequest.paramProperties[`k${i}`];
+    if (!!paramProperties && typeof paramProperties === "object") {
+      paramProperties["blobIdentifiers"] = blobMap;
+    }
   }
 
   formData.append("executeActionDTO", JSON.stringify(executeActionRequest));
   formData.append("parameterMap", JSON.stringify(bindingsMap));
   bindingBlob?.forEach((item) => formData.append(item.name, item.value));
+
+  // Append blob data map to formData if not empty
+  if (!isEmpty(blobDataMap)) {
+    // blobDataMap is used to resolve blob urls of large files as array buffer
+    // we need to add each blob data to formData as a separate entry
+    Object.entries(blobDataMap).forEach(([path, blobData]) =>
+      formData.append(path, blobData),
+    );
+  }
 }
 
 export default function* executePluginActionTriggerSaga(

app/client/src/widgets/FilePickerWidgetV2/widget/index.tsx

@@ -10,6 +10,7 @@ import UpIcon from "assets/icons/ads/up-arrow.svg";
 import { EventType } from "constants/AppsmithActionConstants/ActionConstants";
 import { Colors } from "constants/Colors";
 import type { WidgetType } from "constants/WidgetConstants";
+import { FILE_SIZE_LIMIT_FOR_BLOBS } from "constants/WidgetConstants";
 import { ValidationTypes } from "constants/WidgetValidation";
 import type { Stylesheet } from "entities/AppTheming";
 import { EvaluationSubstitutionType } from "entities/DataTree/dataTreeFactory";
@@ -689,7 +690,7 @@ class FilePickerWidget extends BaseWidget<
       const fileCount = this.props.selectedFiles?.length || 0;
       const fileReaderPromises = files.map((file, index) => {
         return new Promise((resolve) => {
-          if (file.size < 5000 * 1000) {
+          if (file.size < FILE_SIZE_LIMIT_FOR_BLOBS) {
             const reader = new FileReader();
             if (this.props.fileDataType === FileDataTypes.Base64) {
               reader.readAsDataURL(file.data);

app/server/appsmith-interfaces/src/main/java/com/appsmith/external/dtos/ExecuteActionDTO.java

@@ -26,31 +26,28 @@ public class ExecuteActionDTO {
 
     /* Sample value of paramProperties
         "paramProperties": {
-          "k1": "string",
-          "k2": "object",
-          "k3": "number",
-          "k4": {
-            "array": [
-              "string",
-              "number",
-              "string",
-              "boolean"
-            ]
-          },
-          "k5": "boolean"
+          "k0": {
+            "datatype": "string",
+            "blobIdentifiers": ["blobUrl1", "blobUrl2"]
+          }
         }
      */
-    Map<String, Object> paramProperties;
+    Map<String, ParamProperty> paramProperties;
 
-    Map<String, String> parameterMap; //e.g. {"Text1.text": "k1","Table1.data": "k2", "Api1.data": "k3"}
-    Map<String, String> invertParameterMap; //e.g. {"k1":"Text1.text","k2":"Table1.data", "k3": "Api1.data"}
+    Map<String, String> parameterMap; // e.g. {"Text1.text": "k1","Table1.data": "k2", "Api1.data": "k3"}
+
+    // This map is where we store the string values of the blob parts for replacement into evaluated value params
+    Map<String, String> blobValuesMap; // e.g. {"blobId": "stringified-blob-data"}
+
+    Map<String, String> invertParameterMap; // e.g. {"k1":"Text1.text","k2":"Table1.data", "k3": "Api1.data"}
 
     @JsonIgnore
     long totalReadableByteCount;
 
     public void setParameterMap(Map<String, String> parameterMap) {
         this.parameterMap = parameterMap;
-        invertParameterMap = parameterMap.entrySet().stream()
+        invertParameterMap = parameterMap.entrySet()
+                .stream()
                 .collect(Collectors.toMap(
                         Map.Entry::getValue,
                         Map.Entry::getKey

app/server/appsmith-interfaces/src/main/java/com/appsmith/external/dtos/ParamProperty.java

@@ -0,0 +1,19 @@
+package com.appsmith.external.dtos;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import java.util.List;
+
+@Getter
+@Setter
+@NoArgsConstructor
+@AllArgsConstructor
+public class ParamProperty {
+
+    Object datatype;
+
+    List<String> blobIdentifiers;
+}

app/server/appsmith-interfaces/src/main/java/com/appsmith/external/models/Datasource.java

@@ -3,7 +3,6 @@ package com.appsmith.external.models;
 import com.appsmith.external.views.Views;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonView;
-
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.Setter;
@@ -38,7 +37,7 @@ public class Datasource extends BranchAwareDomain {
     @Transient
     @JsonView(Views.Public.class)
     String pluginName;
-    
+
     //Organizations migrated to workspaces, kept the field as deprecated to support the old migration
     @Deprecated
     @JsonView(Views.Public.class)
@@ -122,6 +121,7 @@ public class Datasource extends BranchAwareDomain {
      * Intended to function like `.equals`, but only semantically significant fields, except for the ID. Semantically
      * significant just means that if two datasource have same values for these fields, actions against them will behave
      * exactly the same.
+     *
      * @return true if equal, false otherwise.
      */
     public boolean softEquals(Datasource other) {

app/server/appsmith-plugins/restApiPlugin/src/test/java/com/external/plugins/RestApiPluginTest.java

@@ -2,6 +2,7 @@ package com.external.plugins;
 
 import com.appsmith.external.datatypes.ClientDataType;
 import com.appsmith.external.dtos.ExecuteActionDTO;
+import com.appsmith.external.dtos.ParamProperty;
 import com.appsmith.external.helpers.PluginUtils;
 import com.appsmith.external.helpers.restApiUtils.connections.APIConnection;
 import com.appsmith.external.helpers.restApiUtils.helpers.HintMessageUtils;
@@ -600,7 +601,9 @@ public class RestApiPluginTest {
         param.setPseudoBindingName("k0");
 
         executeActionDTO.setParams(Collections.singletonList(param));
-        executeActionDTO.setParamProperties(Collections.singletonMap("k0", "string"));
+        ParamProperty paramProperty = new ParamProperty();
+        paramProperty.setDatatype("string");
+        executeActionDTO.setParamProperties(Collections.singletonMap("k0", paramProperty));
         executeActionDTO.setParameterMap(Collections.singletonMap("Input1.text", "k0"));
         executeActionDTO.setInvertParameterMap(Collections.singletonMap("k0", "Input1.text"));
 

app/server/appsmith-server/src/main/java/com/appsmith/server/services/ce/NewActionServiceCE.java

@@ -1,12 +1,12 @@
 package com.appsmith.server.services.ce;
 
 import com.appsmith.external.dtos.ExecuteActionDTO;
+import com.appsmith.external.models.ActionDTO;
 import com.appsmith.external.models.ActionExecutionResult;
 import com.appsmith.external.models.MustacheBindingToken;
 import com.appsmith.server.acl.AclPermission;
 import com.appsmith.server.domains.NewAction;
 import com.appsmith.server.domains.NewPage;
-import com.appsmith.external.models.ActionDTO;
 import com.appsmith.server.dtos.ActionViewDTO;
 import com.appsmith.server.dtos.LayoutActionUpdateDTO;
 import com.appsmith.server.services.CrudService;

app/server/appsmith-server/src/main/java/com/appsmith/server/services/ce/NewActionServiceCEImpl.java

@@ -5,6 +5,7 @@ import com.appsmith.external.datatypes.ClientDataType;
 import com.appsmith.external.dtos.DatasourceDTO;
 import com.appsmith.external.dtos.ExecuteActionDTO;
 import com.appsmith.external.dtos.ExecutePluginDTO;
+import com.appsmith.external.dtos.ParamProperty;
 import com.appsmith.external.exceptions.pluginExceptions.AppsmithPluginError;
 import com.appsmith.external.exceptions.pluginExceptions.AppsmithPluginException;
 import com.appsmith.external.exceptions.pluginExceptions.StaleConnectionException;
@@ -66,12 +67,14 @@ import com.appsmith.server.solutions.ApplicationPermission;
 import com.appsmith.server.solutions.DatasourcePermission;
 import com.appsmith.server.solutions.PagePermission;
 import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.micrometer.observation.ObservationRegistry;
 import jakarta.validation.Validator;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang.ArrayUtils;
+import org.apache.commons.lang.StringEscapeUtils;
 import org.apache.commons.lang3.ObjectUtils;
 import org.bson.types.ObjectId;
 import org.springframework.core.io.buffer.DataBufferUtils;
@@ -110,10 +113,20 @@ import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
 import static com.appsmith.external.constants.CommonFieldName.REDACTED_DATA;
-import static com.appsmith.external.constants.spans.ActionSpans.*;
+import static com.appsmith.external.constants.spans.ActionSpans.ACTION_EXECUTION_CACHED_ACTION;
+import static com.appsmith.external.constants.spans.ActionSpans.ACTION_EXECUTION_CACHED_DATASOURCE;
+import static com.appsmith.external.constants.spans.ActionSpans.ACTION_EXECUTION_CACHED_PLUGIN;
+import static com.appsmith.external.constants.spans.ActionSpans.ACTION_EXECUTION_DATASOURCE_CONTEXT;
+import static com.appsmith.external.constants.spans.ActionSpans.ACTION_EXECUTION_DATASOURCE_CONTEXT_REMOTE;
+import static com.appsmith.external.constants.spans.ActionSpans.ACTION_EXECUTION_EDITOR_CONFIG;
+import static com.appsmith.external.constants.spans.ActionSpans.ACTION_EXECUTION_REQUEST_PARSING;
+import static com.appsmith.external.constants.spans.ActionSpans.ACTION_EXECUTION_SERVER_EXECUTION;
+import static com.appsmith.external.constants.spans.ActionSpans.ACTION_EXECUTION_VALIDATE_AUTHENTICATION;
 import static com.appsmith.external.helpers.AppsmithBeanUtils.copyNewFieldValuesIntoOldObject;
 import static com.appsmith.external.helpers.DataTypeStringUtils.getDisplayDataTypes;
 import static com.appsmith.external.helpers.PluginUtils.setValueSafelyInFormData;
@@ -135,6 +148,13 @@ public class NewActionServiceCEImpl extends BaseService<NewActionRepository, New
     public static final PluginType JS_PLUGIN_TYPE = PluginType.JS;
     public static final String JS_PLUGIN_PACKAGE_NAME = "js-plugin";
 
+    static final String PARAM_KEY_REGEX = "^k\\d+$";
+    static final String BLOB_KEY_REGEX = "^blob:[0-9a-fA-F]{8}\\b-[0-9a-fA-F]{4}\\b-[0-9a-fA-F]{4}\\b-[0-9a-fA-F]{4}\\b-[0-9a-fA-F]{12}$";
+    static final String EXECUTE_ACTION_DTO = "executeActionDTO";
+    static final String PARAMETER_MAP = "parameterMap";
+
+    List<Pattern> patternList = new ArrayList<>();
+
     private final NewActionRepository repository;
     private final DatasourceService datasourceService;
     private final PluginService pluginService;
@@ -207,6 +227,11 @@ public class NewActionServiceCEImpl extends BaseService<NewActionRepository, New
         this.applicationPermission = applicationPermission;
         this.pagePermission = pagePermission;
         this.actionPermission = actionPermission;
+
+        this.patternList.add(Pattern.compile(PARAM_KEY_REGEX));
+        this.patternList.add(Pattern.compile(BLOB_KEY_REGEX));
+        this.patternList.add(Pattern.compile(EXECUTE_ACTION_DTO));
+        this.patternList.add(Pattern.compile(PARAMETER_MAP));
     }
 
     @Override
@@ -1072,131 +1097,9 @@ public class NewActionServiceCEImpl extends BaseService<NewActionRepository, New
     protected Mono<ExecuteActionDTO> createExecuteActionDTO(Flux<Part> partFlux) {
         final AtomicLong totalReadableByteCount = new AtomicLong(0);
         final ExecuteActionDTO dto = new ExecuteActionDTO();
-        return partFlux
-                .flatMap(part -> {
-                    final String key = part.name();
-                    if ("executeActionDTO".equals(key)) {
-                        return DataBufferUtils
-                                .join(part.content())
-                                .flatMap(executeActionDTOBuffer -> {
-                                    byte[] byteData = new byte[executeActionDTOBuffer.readableByteCount()];
-                                    executeActionDTOBuffer.read(byteData);
-                                    DataBufferUtils.release(executeActionDTOBuffer);
-                                    try {
-                                        return Mono.just(objectMapper.readValue(byteData, ExecuteActionDTO.class));
-                                    } catch (IOException e) {
-                                        log.error("Error in deserializing ExecuteActionDTO", e);
-                                        return Mono.error(new AppsmithException(AppsmithError.INVALID_PARAMETER, "executeActionDTO"));
-                                    }
-                                })
-                                .flatMap(executeActionDTO -> {
-                                    dto.setActionId(executeActionDTO.getActionId());
-                                    dto.setViewMode(executeActionDTO.getViewMode());
-                                    dto.setParamProperties(executeActionDTO.getParamProperties());
-                                    dto.setPaginationField(executeActionDTO.getPaginationField());
-                                    return Mono.empty();
-                                });
-                    } else if ("parameterMap".equals(key)) {
-                        return DataBufferUtils
-                                .join(part.content())
-                                .flatMap(executeActionDTOBuffer -> {
-                                    byte[] byteData = new byte[executeActionDTOBuffer.readableByteCount()];
-                                    executeActionDTOBuffer.read(byteData);
-                                    DataBufferUtils.release(executeActionDTOBuffer);
-                                    try {
-                                        return Mono.just(objectMapper.readValue(byteData, HashMap.class));
-                                    } catch (IOException e) {
-                                        return Mono.error(new AppsmithException(AppsmithError.INVALID_PARAMETER, "parameterMap"));
-                                    }
-                                })
-                                .flatMap(paramMap -> {
-                                    dto.setParameterMap(paramMap);
-                                    return Mono.empty();
-                                });
-                    }
-                    return Mono.just(part);
-                })
-                .flatMap(part -> {
-                    final Param param = new Param();
-                    param.setPseudoBindingName(part.name());
-                    return DataBufferUtils
-                            .join(part.content())
-                            .map(dataBuffer -> {
-                                byte[] bytes = new byte[dataBuffer.readableByteCount()];
-                                totalReadableByteCount.addAndGet(dataBuffer.readableByteCount());
-                                dataBuffer.read(bytes);
-                                DataBufferUtils.release(dataBuffer);
-                                param.setValue(new String(bytes, StandardCharsets.UTF_8));
-                                return param;
-                            });
-                })
+        return this.parsePartsAndGetParamsFlux(partFlux, totalReadableByteCount, dto)
                 .collectList()
-                .flatMap(params -> {
-                    if (dto.getActionId() == null) {
-                        return Mono.error(new AppsmithException(AppsmithError.INVALID_PARAMETER, FieldName.ACTION_ID));
-                    }
-
-                    dto.setTotalReadableByteCount(totalReadableByteCount.longValue());
-
-                    final Set<String> visitedBindings = new HashSet<>();
-                    /*
-                        Parts in multipart request can appear in any order. In order to avoid NPE original name of the parameters
-                        along with the client-side data type are set here as it's guaranteed at this point that the part having the parameterMap is already collected.
-                        Ref: https://github.com/appsmithorg/appsmith/issues/16722
-                     */
-                    params.forEach(
-                            param -> {
-                                String pseudoBindingName = param.getPseudoBindingName();
-                                String bindingValue = dto.getInvertParameterMap().get(pseudoBindingName);
-                                param.setKey(bindingValue);
-                                visitedBindings.add(bindingValue);
-                                //if the type is not an array e.g. "k1": "string" or "k1": "boolean"
-                                if (dto.getParamProperties()
-                                        .get(pseudoBindingName) instanceof String) {
-                                    param.setClientDataType(ClientDataType.valueOf(String.valueOf(dto.getParamProperties()
-                                                    .get(pseudoBindingName))
-                                            .toUpperCase()));
-                                } else if (dto.getParamProperties()
-                                        .get(pseudoBindingName) instanceof LinkedHashMap) {
-                                    //if the type is an array e.g. "k1": { "array": [ "string", "number", "string", "boolean"]
-                                    LinkedHashMap<String, ArrayList> stringArrayListLinkedHashMap =
-                                            (LinkedHashMap<String, ArrayList>) dto.getParamProperties()
-                                                    .get(pseudoBindingName);
-                                    Optional<String> firstKeyOpt = stringArrayListLinkedHashMap.keySet()
-                                            .stream()
-                                            .findFirst();
-                                    if (firstKeyOpt.isPresent()) {
-                                        String firstKey = firstKeyOpt.get();
-                                        param.setClientDataType(ClientDataType.valueOf(firstKey.toUpperCase()));
-                                        List<String> individualTypes = stringArrayListLinkedHashMap.get(firstKey);
-                                        List<ClientDataType> dataTypesOfArrayElements =
-                                                individualTypes.stream()
-                                                        .map(it -> ClientDataType.valueOf(String.valueOf(it)
-                                                                .toUpperCase()))
-                                                        .collect(Collectors.toList());
-                                        param.setDataTypesOfArrayElements(dataTypesOfArrayElements);
-                                    }
-                                }
-
-                            }
-                    );
-
-                    // In case there are parameters that did not receive a value in the multipart request,
-                    // initialize these bindings with empty strings
-                    if (dto.getParameterMap() != null) {
-                        dto.getParameterMap()
-                                .keySet()
-                                .stream()
-                                .forEach(parameter -> {
-                                    if (!visitedBindings.contains(parameter)) {
-                                        Param newParam = new Param(parameter, "");
-                                        params.add(newParam);
-                                    }
-                                });
-                    }
-                    dto.setParams(params);
-                    return Mono.just(dto);
-                })
+                .flatMap(params -> this.enrichExecutionParam(totalReadableByteCount, dto, params))
                 .name(ACTION_EXECUTION_REQUEST_PARSING)
                 .tap(Micrometer.observation(observationRegistry));
     }
@@ -2329,4 +2232,222 @@ public class NewActionServiceCEImpl extends BaseService<NewActionRepository, New
         return analyticsProperties;
     }
 
+    /**
+     * This method attempts to parse all incoming parts by type, in parallel
+     * The expectation is that each part gets processed by the time this flux ends,
+     * and the DTO is updated accordingly
+     *
+     * @param partFlux               Raw flux of parts as received in the execution request
+     * @param totalReadableByteCount An atomic type to store the total execution request size as and when we parse them
+     * @param dto                    The ExecuteActionDTO object to store all results in
+     * @return
+     */
+    protected Flux<Param> parsePartsAndGetParamsFlux(Flux<Part> partFlux, AtomicLong totalReadableByteCount, ExecuteActionDTO dto) {
+        return partFlux
+                .groupBy(part -> {
+                    // We're grouping parts by the type of processing required
+                    // Expected types: meta, value, blob
+
+                    for (Pattern pattern : patternList) {
+                        Matcher matcher = pattern.matcher(part.name());
+                        if (matcher.find()) {
+                            return pattern.pattern();
+                        }
+                    }
+                    return part.name();
+                })
+                .flatMap(groupedPartsFlux -> {
+                    String key = groupedPartsFlux.key();
+                    return switch (key) {
+                        case PARAM_KEY_REGEX ->
+                                groupedPartsFlux.flatMap(part -> this.parseExecuteParameter(part, totalReadableByteCount));
+                        case BLOB_KEY_REGEX ->
+                                this.parseExecuteBlobs(groupedPartsFlux, dto, totalReadableByteCount).then(Mono.empty());
+                        case EXECUTE_ACTION_DTO ->
+                                groupedPartsFlux.next().flatMap(part -> this.parseExecuteActionPart(part, dto)).then(Mono.empty());
+                        case PARAMETER_MAP ->
+                                groupedPartsFlux.next().flatMap(part -> this.parseExecuteParameterMapPart(part, dto)).then(Mono.empty());
+                        default ->
+                                Mono.error(new AppsmithException(AppsmithError.GENERIC_BAD_REQUEST, "Unexpected part found: " + key));
+                    };
+                });
+    }
+
+    protected Mono<ExecuteActionDTO> enrichExecutionParam(AtomicLong totalReadableByteCount, ExecuteActionDTO dto, List<Param> params) {
+        if (dto.getActionId() == null) {
+            return Mono.error(new AppsmithException(AppsmithError.INVALID_PARAMETER, FieldName.ACTION_ID));
+        }
+
+        dto.setTotalReadableByteCount(totalReadableByteCount.longValue());
+
+        final Set<String> visitedBindings = new HashSet<>();
+        /*
+            Parts in multipart request can appear in any order. In order to avoid NPE original name of the parameters
+            along with the client-side data type are set here as it's guaranteed at this point that the part having the parameterMap is already collected.
+            Ref: https://github.com/appsmithorg/appsmith/issues/16722
+         */
+        params.forEach(
+                param -> {
+                    String pseudoBindingName = param.getPseudoBindingName();
+                    String bindingValue = dto.getInvertParameterMap().get(pseudoBindingName);
+                    param.setKey(bindingValue);
+                    visitedBindings.add(bindingValue);
+                    // if the type is not an array e.g. "k1": "string" or "k1": "boolean"
+                    ParamProperty paramProperty = dto.getParamProperties().get(pseudoBindingName);
+                    if (paramProperty != null) {
+                        this.identifyExecutionParamDatatype(param, paramProperty);
+
+                        this.substituteBlobValuesInParam(dto, param, paramProperty);
+                    }
+
+                }
+        );
+
+        // In case there are parameters that did not receive a value in the multipart request,
+        // initialize these bindings with empty strings
+        if (dto.getParameterMap() != null) {
+            dto.getParameterMap()
+                    .keySet()
+                    .stream()
+                    .forEach(parameter -> {
+                        if (!visitedBindings.contains(parameter)) {
+                            Param newParam = new Param(parameter, "");
+                            params.add(newParam);
+                        }
+                    });
+        }
+        dto.setParams(params);
+        return Mono.just(dto);
+    }
+
+    private void substituteBlobValuesInParam(ExecuteActionDTO dto, Param param, ParamProperty paramProperty) {
+        // Check if this param has blobUrlPaths
+        if (paramProperty.getBlobIdentifiers() != null && !paramProperty.getBlobIdentifiers().isEmpty()) {
+            // If it does, trigger the replacement logic for each of these urlPaths
+            String replacedValue = this.replaceBlobValuesInParam(
+                    param.getValue(),
+                    paramProperty.getBlobIdentifiers(),
+                    dto.getBlobValuesMap());
+            // And then update the value for this param
+            param.setValue(replacedValue);
+        }
+    }
+
+    private void identifyExecutionParamDatatype(Param param, ParamProperty paramProperty) {
+        Object datatype = paramProperty.getDatatype();
+        if (datatype instanceof String) {
+            param.setClientDataType(ClientDataType.valueOf(String.valueOf(datatype).toUpperCase()));
+        } else if (datatype instanceof LinkedHashMap) {
+            // if the type is an array e.g. "k1": { "array": [ "string", "number", "string", "boolean"]
+            LinkedHashMap<String, ArrayList> stringArrayListLinkedHashMap =
+                    (LinkedHashMap<String, ArrayList>) datatype;
+            Optional<String> firstKeyOpt = stringArrayListLinkedHashMap.keySet()
+                    .stream()
+                    .findFirst();
+            if (firstKeyOpt.isPresent()) {
+                String firstKey = firstKeyOpt.get();
+                param.setClientDataType(ClientDataType.valueOf(firstKey.toUpperCase()));
+                List<String> individualTypes = stringArrayListLinkedHashMap.get(firstKey);
+                List<ClientDataType> dataTypesOfArrayElements =
+                        individualTypes.stream()
+                                .map(it -> ClientDataType.valueOf(String.valueOf(it)
+                                        .toUpperCase()))
+                                .collect(Collectors.toList());
+                param.setDataTypesOfArrayElements(dataTypesOfArrayElements);
+            }
+        }
+    }
+
+    protected Mono<Void> parseExecuteActionPart(Part part, ExecuteActionDTO dto) {
+        return DataBufferUtils
+                .join(part.content())
+                .flatMap(executeActionDTOBuffer -> {
+                    byte[] byteData = new byte[executeActionDTOBuffer.readableByteCount()];
+                    executeActionDTOBuffer.read(byteData);
+                    DataBufferUtils.release(executeActionDTOBuffer);
+                    try {
+                        return Mono.just(objectMapper.readValue(byteData, ExecuteActionDTO.class));
+                    } catch (IOException e) {
+                        log.error("Error in deserializing ExecuteActionDTO", e);
+                        return Mono.error(new AppsmithException(AppsmithError.INVALID_PARAMETER, EXECUTE_ACTION_DTO));
+                    }
+                })
+                .flatMap(executeActionDTO -> {
+                    dto.setActionId(executeActionDTO.getActionId());
+                    dto.setViewMode(executeActionDTO.getViewMode());
+                    dto.setParamProperties(executeActionDTO.getParamProperties());
+                    dto.setPaginationField(executeActionDTO.getPaginationField());
+                    return Mono.empty();
+                });
+    }
+
+    protected Mono<Void> parseExecuteParameterMapPart(Part part, ExecuteActionDTO dto) {
+        return DataBufferUtils
+                .join(part.content())
+                .flatMap(parameterMapBuffer -> {
+                    byte[] byteData = new byte[parameterMapBuffer.readableByteCount()];
+                    parameterMapBuffer.read(byteData);
+                    DataBufferUtils.release(parameterMapBuffer);
+                    try {
+                        return Mono.just(objectMapper.readValue(byteData, new TypeReference<Map<String, String>>() {
+                        }));
+                    } catch (IOException e) {
+                        return Mono.error(new AppsmithException(AppsmithError.INVALID_PARAMETER, PARAMETER_MAP));
+                    }
+                })
+                .flatMap(paramMap -> {
+                    dto.setParameterMap(paramMap);
+                    return Mono.empty();
+                });
+    }
+
+    protected Mono<Param> parseExecuteParameter(Part part, AtomicLong totalReadableByteCount) {
+        final Param param = new Param();
+        param.setPseudoBindingName(part.name());
+        return DataBufferUtils
+                .join(part.content())
+                .map(dataBuffer -> {
+                    byte[] bytes = new byte[dataBuffer.readableByteCount()];
+                    totalReadableByteCount.addAndGet(dataBuffer.readableByteCount());
+                    dataBuffer.read(bytes);
+                    DataBufferUtils.release(dataBuffer);
+                    param.setValue(new String(bytes, StandardCharsets.UTF_8));
+                    return param;
+                });
+    }
+
+    protected Mono<Void> parseExecuteBlobs(Flux<Part> partsFlux, ExecuteActionDTO dto, AtomicLong totalReadableByteCount) {
+        Map<String, String> blobMap = new HashMap<>();
+        dto.setBlobValuesMap(blobMap);
+
+        return partsFlux
+                .flatMap(part -> {
+                    return DataBufferUtils
+                            .join(part.content())
+                            .map(dataBuffer -> {
+                                byte[] bytes = new byte[dataBuffer.readableByteCount()];
+                                totalReadableByteCount.addAndGet(dataBuffer.readableByteCount());
+                                dataBuffer.read(bytes);
+                                DataBufferUtils.release(dataBuffer);
+                                blobMap.put(part.name(), new String(bytes, StandardCharsets.ISO_8859_1));
+                                return Mono.empty();
+                            });
+                })
+                .then();
+    }
+
+    protected String replaceBlobValuesInParam(String value, List<String> blobIdentifiers, Map<String, String> blobValuesMap) {
+        // If there is no blobId reference against this param, return as is
+        if (blobIdentifiers == null || blobIdentifiers.isEmpty()) {
+            return value;
+        }
+
+        // Otherwise, for each such blobId reference, replace the reference with the actual value from the blobMap
+        for (String blobId : blobIdentifiers) {
+            value = value.replace(blobId, StringEscapeUtils.escapeJava(blobValuesMap.get(blobId)));
+        }
+
+        return value;
+    }
+
 }

app/server/appsmith-server/src/test/java/com/appsmith/server/services/ce/NewActionServiceCEImplTest.java

@@ -1,8 +1,12 @@
 package com.appsmith.server.services.ce;
 
+import com.appsmith.external.datatypes.ClientDataType;
+import com.appsmith.external.dtos.ExecuteActionDTO;
+import com.appsmith.external.dtos.ParamProperty;
 import com.appsmith.external.models.ActionDTO;
 import com.appsmith.external.models.ActionExecutionResult;
 import com.appsmith.external.models.Datasource;
+import com.appsmith.external.models.Param;
 import com.appsmith.external.models.PluginType;
 import com.appsmith.server.acl.PolicyGenerator;
 import com.appsmith.server.constants.FieldName;
@@ -67,8 +71,10 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.atomic.AtomicLong;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
@@ -80,7 +86,7 @@ import static org.mockito.Mockito.spy;
 @Slf4j
 public class NewActionServiceCEImplTest {
 
-    NewActionServiceCE newActionService;
+    NewActionServiceCEImpl newActionService;
 
     @MockBean
     Scheduler scheduler;
@@ -219,9 +225,13 @@ public class NewActionServiceCEImplTest {
         MockServerHttpRequest mock = MockServerHttpRequest
                 .method(HttpMethod.POST, URI.create("https://example.com"))
                 .contentType(new MediaType("multipart", "form-data", Map.of("boundary", "boundary")))
-                .body("--boundary\r\n" +
-                        "Content-Disposition: form-data; name=\"executeActionDTO\"\r\n" + "\r\n" + "irrelevant content\r\n" +
-                        "--boundary--\r\n");
+                .body("""
+                        --boundary\r
+                        Content-Disposition: form-data; name="executeActionDTO"\r
+                        \r
+                        irrelevant content\r
+                        --boundary--\r
+                        """);
 
         final Flux<Part> partsFlux = BodyExtractors.toParts()
                 .extract(mock, this.context);
@@ -240,9 +250,13 @@ public class NewActionServiceCEImplTest {
         MockServerHttpRequest mock = MockServerHttpRequest
                 .method(HttpMethod.POST, URI.create("https://example.com"))
                 .contentType(new MediaType("multipart", "form-data", Map.of("boundary", "boundary")))
-                .body("--boundary\r\n" +
-                        "Content-Disposition: form-data; name=\"executeActionDTO\"\r\n" + "\r\n" + "{\"viewMode\":false}\r\n" +
-                        "--boundary--\r\n");
+                .body("""
+                        --boundary\r
+                        Content-Disposition: form-data; name="executeActionDTO"\r
+                        \r
+                        {"viewMode":false}\r
+                        --boundary--\r
+                        """);
 
         final Flux<Part> partsFlux = BodyExtractors.toParts()
                 .extract(mock, this.context);
@@ -312,20 +326,21 @@ public class NewActionServiceCEImplTest {
 
     @Test
     public void testExecuteAPIWithUsualOrderingOfTheParts() {
-        String usualOrderOfParts = "--boundary\r\n" +
-                "Content-Disposition: form-data; name=\"executeActionDTO\"\r\n" +
-                "\r\n" +
-                "{\"actionId\":\"63285a3388e48972c7519b18\",\"viewMode\":false,\"paramProperties\":{\"k0\":\"string\"}}\r\n" +
-                "--boundary\r\n" +
-                "Content-Disposition: form-data; name=\"parameterMap\"\r\n" +
-                "\r\n" +
-                "{\"Input1.text\":\"k0\"}\r\n" +
-                "--boundary\r\n" +
-                "Content-Disposition: form-data; name=\"k0\"; filename=\"blob\"\r\n" +
-                "Content-Type: text/plain\r\n" +
-                "\r\n" +
-                "xyz\r\n" +
-                "--boundary--";
+        String usualOrderOfParts = """
+                --boundary\r
+                Content-Disposition: form-data; name="executeActionDTO"\r
+                \r
+                {"actionId":"63285a3388e48972c7519b18","viewMode":false,"paramProperties":{"k0":{"datatype": "string"}}}\r
+                --boundary\r
+                Content-Disposition: form-data; name="parameterMap"\r
+                \r
+                {"Input1.text":"k0"}\r
+                --boundary\r
+                Content-Disposition: form-data; name="k0"; filename="blob"\r
+                Content-Type: text/plain\r
+                \r
+                xyz\r
+                --boundary--""";
 
         MockServerHttpRequest mock = MockServerHttpRequest
                 .method(HttpMethod.POST, URI.create("https://example.com"))
@@ -353,7 +368,7 @@ public class NewActionServiceCEImplTest {
         StepVerifier
                 .create(actionExecutionResultMono)
                 .assertNext(response -> {
-                    assertTrue(response instanceof ActionExecutionResult);
+                    assertNotNull(response);
                     assertTrue(response.getIsExecutionSuccess());
                     assertEquals(mockResult.getBody().toString(), response.getBody().toString());
                 })
@@ -362,20 +377,21 @@ public class NewActionServiceCEImplTest {
 
     @Test
     public void testExecuteAPIWithParameterMapAsLastPart() {
-        String parameterMapAtLast = "--boundary\r\n" +
-                "Content-Disposition: form-data; name=\"executeActionDTO\"\r\n" +
-                "\r\n" +
-                "{\"actionId\":\"63285a3388e48972c7519b18\",\"viewMode\":false,\"paramProperties\":{\"k0\":\"string\"}}\r\n" +
-                "--boundary\r\n" +
-                "Content-Disposition: form-data; name=\"k0\"; filename=\"blob\"\r\n" +
-                "Content-Type: text/plain\r\n" +
-                "\r\n" +
-                "xyz\r\n" +
-                "--boundary\r\n" +
-                "Content-Disposition: form-data; name=\"parameterMap\"\r\n" +
-                "\r\n" +
-                "{\"Input1.text\":\"k0\"}\r\n" +
-                "--boundary--";
+        String parameterMapAtLast = """
+                --boundary\r
+                Content-Disposition: form-data; name="executeActionDTO"\r
+                \r
+                {"actionId":"63285a3388e48972c7519b18","viewMode":false,"paramProperties":{"k0":{"datatype": "string"}}}\r
+                --boundary\r
+                Content-Disposition: form-data; name="k0"; filename="blob"\r
+                Content-Type: text/plain\r
+                \r
+                xyz\r
+                --boundary\r
+                Content-Disposition: form-data; name="parameterMap"\r
+                \r
+                {"Input1.text":"k0"}\r
+                --boundary--""";
 
         MockServerHttpRequest mock = MockServerHttpRequest
                 .method(HttpMethod.POST, URI.create("https://example.com"))
@@ -403,10 +419,81 @@ public class NewActionServiceCEImplTest {
         StepVerifier
                 .create(actionExecutionResultMono)
                 .assertNext(response -> {
-                    assertTrue(response instanceof ActionExecutionResult);
+                    assertNotNull(response);
                     assertTrue(response.getIsExecutionSuccess());
                     assertEquals(mockResult.getBody().toString(), response.getBody().toString());
                 })
                 .verifyComplete();
     }
+
+    @Test
+    public void testParsePartsAndGetParamsFlux_withBlobIdentifiers_replacesValueInParam() {
+        String partsWithBlobRefs = """
+                --boundary\r
+                Content-Disposition: form-data; name="executeActionDTO"\r
+                \r
+                {"actionId":"63285a3388e48972c7519b18","viewMode":false,"paramProperties":{"k0":{"datatype": "string", "blobIdentifiers": ["blob:12345678-1234-1234-1234-123456781234"]}}}\r
+                --boundary\r
+                Content-Disposition: form-data; name="parameterMap"\r
+                \r
+                {"Input1.text":"k0"}\r
+                --boundary\r
+                Content-Disposition: form-data; name="k0"; filename="blob"\r
+                Content-Type: text/plain\r
+                \r
+                {"name": "randomName", "data": "blob:12345678-1234-1234-1234-123456781234"}\r
+                --boundary\r
+                Content-Disposition: form-data; name="blob:12345678-1234-1234-1234-123456781234"; filename="blob"\r
+                Content-Type: text/plain\r
+                \r
+                xy\\nz\r
+                --boundary--""";
+
+        MockServerHttpRequest mock = MockServerHttpRequest
+                .method(HttpMethod.POST, URI.create("https://example.com"))
+                .contentType(new MediaType("multipart", "form-data", Map.of("boundary", "boundary")))
+                .body(partsWithBlobRefs);
+
+        final Flux<Part> partsFlux = BodyExtractors.toParts()
+                .extract(mock, this.context);
+
+        AtomicLong atomicLong = new AtomicLong();
+        ExecuteActionDTO executeActionDTO = new ExecuteActionDTO();
+        Mono<List<Param>> paramsListMono = newActionService.parsePartsAndGetParamsFlux(partsFlux, atomicLong, executeActionDTO).collectList().cache();
+
+        StepVerifier.create(paramsListMono)
+                .assertNext(paramsList -> {
+                    assertEquals(1, paramsList.size());
+                    Param param = paramsList.get(0);
+                    assertEquals("{\"name\": \"randomName\", \"data\": \"blob:12345678-1234-1234-1234-123456781234\"}", param.getValue());
+                })
+                .verifyComplete();
+    }
+
+    @Test
+    public void testEnrichExecutionParams_withBlobReference_performsSubstitutionCorrectly() {
+        AtomicLong atomicLong = new AtomicLong(45L);
+        ExecuteActionDTO executeActionDTO = new ExecuteActionDTO();
+        executeActionDTO.setActionId("testId");
+        executeActionDTO.setViewMode(false);
+        executeActionDTO.setParamProperties(Map.of("k0", new ParamProperty("string", List.of("blobId"))));
+        executeActionDTO.setParameterMap(Map.of("Input1.text", "k0"));
+        executeActionDTO.setBlobValuesMap(Map.of("blobId", "xy\\nz"));
+        Param param1 = new Param();
+        param1.setValue("{\"name\": \"randomName\", \"data\": \"blobId\"}");
+        param1.setPseudoBindingName("k0");
+        List<Param> params = List.of(param1);
+
+        Mono<ExecuteActionDTO> enrichedDto = newActionService.enrichExecutionParam(atomicLong, executeActionDTO, params);
+
+        StepVerifier.create(enrichedDto)
+                .assertNext(dto -> {
+                    assertEquals(45, dto.getTotalReadableByteCount());
+
+                    Param param = dto.getParams().get(0);
+                    assertEquals(ClientDataType.STRING, param.getClientDataType());
+                    assertEquals("{\"name\": \"randomName\", \"data\": \"xy\\\\nz\"}", param.getValue());
+                })
+                .verifyComplete();
+    }
 }