class DataSetService extends BaseDataSetService {

    @RequestMapping(value = "/datasets", method = POST, produces = TEXT_PLAIN_VALUE)
    @ApiOperation(value = "Create a data set", produces = TEXT_PLAIN_VALUE, notes = "Create a new data set based on content provided in POST body. For documentation purposes, body is typed as 'text/plain' but operation accepts binary content too. Returns the id of the newly created data set.")
    @Timed
    @VolumeMetered
    public String create(
            @ApiParam(value = "User readable name of the data set (e.g. 'Finance Report 2015', 'Test Data Set').") @RequestParam(defaultValue = "") String name,
            @ApiParam(value = "An optional tag to be added in data set metadata once created.") @RequestParam(defaultValue = "") String tag,
            @ApiParam(value = "Size of the data set, in bytes.") @RequestParam(required = false) Long size,
            @RequestHeader(CONTENT_TYPE) String contentType,
            @ApiParam(value = "content") InputStream content) {
        //@formatter:on
        checkDataSetName(name);

        final String id = UUID.randomUUID().toString();
        final Marker marker = Markers.dataset(id);
        LOG.debug(marker, "Creating...");

        // sanity check
        if (size != null && size < 0) {
            LOG.warn("invalid size provided {}", size);
            throw new TDPException(UNEXPECTED_CONTENT, build().put("size", size));
        }

        // check that the name is not already taken
        checkIfNameIsAvailable(name);

        // get the location out of the content type and the request body
        final DataSetLocation location;
        try {
            location = datasetLocator.getDataSetLocation(contentType, content);
        } catch (IOException e) {
            throw new TDPException(DataSetErrorCodes.UNABLE_TO_READ_DATASET_LOCATION, e);
        }
        DataSetMetadata dataSetMetadata = null;
        final TDPException hypotheticalException;
        try {

            // if the size is provided, let's check if the quota will not be exceeded
            if (size != null && size > 0) {
                quotaService.checkIfAddingSizeExceedsAvailableStorage(size);
            }

            dataSetMetadata = metadataBuilder //
                    .metadata() //
                    .id(id) //
                    .name(name) //
                    .author(security.getUserId()) //
                    .location(location) //
                    .created(System.currentTimeMillis()) //
                    .tag(tag) //
                    .build();

            dataSetMetadata.getLifecycle().setImporting(true); // Indicate data set is being imported

            // Save data set content
            LOG.debug(marker, "Storing content...");
            final long maxDataSetSizeAllowed = getMaxDataSetSizeAllowed();
            final StrictlyBoundedInputStream sizeCalculator =
                    new StrictlyBoundedInputStream(content, maxDataSetSizeAllowed);
            contentStore.storeAsRaw(dataSetMetadata, sizeCalculator);
            dataSetMetadata.setDataSetSize(sizeCalculator.getTotal());
            LOG.debug(marker, "Content stored.");

            // Create the new data set
            dataSetMetadataRepository.save(dataSetMetadata);
            LOG.debug(marker, "dataset metadata stored {}", dataSetMetadata);

            // Queue events (format analysis, content indexing for search...)
            analyzeDataSet(id, emptyList());

            LOG.debug(marker, "Created!");

            // TDP-6141 : Do not send `DataSteImportedEvent` synchronously because it will do the same analyse as before
            // a second time and can reach timeout
            asyncPublisher.multicastEvent(new DatasetImportedEvent(id));

            return id;
        } catch (StrictlyBoundedInputStream.InputStreamTooLargeException e) {
            hypotheticalException =
                    new TDPException(MAX_STORAGE_MAY_BE_EXCEEDED, e, build().put(STORAGE_LIMIT, e.getMaxSize()));
        } catch (TDPException e) {
            hypotheticalException = e;
        } catch (Exception e) {
            hypotheticalException = new TDPException(UNABLE_CREATE_DATASET, e);
        } finally {
            // because the client might still be writing the request content, closing the connexion right now
            // might end up in a 'connection reset' or a 'broken pipe' error in API.
            //
            // So, let's read fully the request content before closing the connection.
            dataSetContentToNull(content);
        }
        dataSetMetadataRepository.remove(id);
        if (dataSetMetadata != null) {
            try {
                contentStore.delete(dataSetMetadata);
            } catch (Exception e) {
                LOG.error("Unable to delete uploaded data.", e);
            }
        }
        throw hypotheticalException;
    }

}