class SdxService implements ResourceIdProvider, ResourceBasedCrnProvider {

    public Pair<SdxCluster, FlowIdentifier> createSdx(final String userCrn, final String name, final SdxClusterRequest sdxClusterRequest,
            final StackV4Request internalStackV4Request) {
        LOGGER.info("Creating SDX cluster with name {}", name);
        validateSdxRequest(name, sdxClusterRequest.getEnvironment(), getAccountIdFromCrn(userCrn));
        DetailedEnvironmentResponse environment = getEnvironment(sdxClusterRequest.getEnvironment());
        validateInternalSdxRequest(internalStackV4Request, sdxClusterRequest.getClusterShape());
        validateEnv(environment);
        validateRazEnablement(sdxClusterRequest, environment);
        validateMediumDutySdxEnablement(sdxClusterRequest, environment);
        SdxCluster sdxCluster = new SdxCluster();
        sdxCluster.setInitiatorUserCrn(userCrn);
        sdxCluster.setCrn(createCrn(getAccountIdFromCrn(userCrn)));
        sdxCluster.setClusterName(name);
        sdxCluster.setAccountId(getAccountIdFromCrn(userCrn));
        sdxCluster.setClusterShape(sdxClusterRequest.getClusterShape());
        sdxCluster.setCreated(clock.getCurrentTimeMillis());
        sdxCluster.setEnvName(environment.getName());
        sdxCluster.setEnvCrn(environment.getCrn());
        sdxCluster.setRangerRazEnabled(sdxClusterRequest.isEnableRangerRaz());
        setTagsSafe(sdxClusterRequest, sdxCluster);

        CloudPlatform cloudPlatform = CloudPlatform.valueOf(environment.getCloudPlatform());

        if (isCloudStorageConfigured(sdxClusterRequest)) {
            validateCloudStorageRequest(sdxClusterRequest.getCloudStorage(), environment);
            String trimmedBaseLocation = StringUtils.stripEnd(sdxClusterRequest.getCloudStorage().getBaseLocation(), "/");
            sdxCluster.setCloudStorageBaseLocation(trimmedBaseLocation);
            sdxCluster.setCloudStorageFileSystemType(sdxClusterRequest.getCloudStorage().getFileSystemType());
            sdxClusterRequest.getCloudStorage().setBaseLocation(trimmedBaseLocation);
        } else if (!CloudPlatform.YARN.equalsIgnoreCase(cloudPlatform.name()) &&
                !CloudPlatform.GCP.equalsIgnoreCase(cloudPlatform.name()) &&
                !CloudPlatform.MOCK.equalsIgnoreCase(cloudPlatform.name()) &&
                internalStackV4Request == null) {
            throw new BadRequestException("Cloud storage parameter is required.");
        }

        String runtimeVersion = getRuntime(sdxClusterRequest, internalStackV4Request);
        sdxCluster.setRuntime(runtimeVersion);
        externalDatabaseConfigurer.configure(cloudPlatform, sdxClusterRequest.getExternalDatabase(), sdxCluster);
        updateStackV4RequestWithEnvironmentCrnIfNotExistsOnIt(internalStackV4Request, environment.getCrn());
        StackV4Request stackRequest = getStackRequest(sdxClusterRequest, internalStackV4Request, cloudPlatform, runtimeVersion);
        prepareCloudStorageForStack(sdxClusterRequest, stackRequest, sdxCluster, environment);
        prepareDefaultSecurityConfigs(internalStackV4Request, stackRequest, cloudPlatform);
        prepareProviderSpecificParameters(stackRequest, sdxClusterRequest, cloudPlatform);
        stackRequest.setResourceCrn(sdxCluster.getCrn());
        try {
            sdxCluster.setStackRequest(JsonUtil.writeValueAsString(stackRequest));
        } catch (JsonProcessingException e) {
            LOGGER.error("Can not parse internal stackrequest", e);
            throw new BadRequestException("Can not parse internal stackrequest", e);
        }

        MDCBuilder.buildMdcContext(sdxCluster);

        SdxCluster savedSdxCluster;
        try {
            savedSdxCluster = transactionService.required(() -> {
                SdxCluster created = sdxClusterRepository.save(sdxCluster);
                grpcUmsClient.assignResourceOwnerRoleIfEntitled(created.getInitiatorUserCrn(), created.getCrn(), created.getAccountId());
                sdxStatusService.setStatusForDatalakeAndNotify(DatalakeStatusEnum.REQUESTED, "Datalake requested", created);
                return created;
            });
        } catch (TransactionExecutionException e) {
            throw new TransactionRuntimeExecutionException(e);
        }

        FlowIdentifier flowIdentifier = sdxReactorFlowManager.triggerSdxCreation(savedSdxCluster);

        return Pair.of(savedSdxCluster, flowIdentifier);
    }

}