class QualityAnalysis implements SynchronousDataSetAnalyzer {

    @Override
    public void analyze(String dataSetId) {
        if (StringUtils.isEmpty(dataSetId)) {
            throw new IllegalArgumentException("Data set id cannot be null or empty.");
        }
        DistributedLock datasetLock = repository.createDatasetMetadataLock(dataSetId);
        datasetLock.lock();
        try {
            DataSetMetadata metadata = repository.get(dataSetId);
            if (metadata == null) {
                LOGGER.info("Unable to analyze quality of data set #{}: seems to be removed.", dataSetId);
                return;
            }
            // e.g. excel multi sheet dataset when user has not choose the sheet yet
            if (!metadata.getLifecycle().isInProgress()) {
                LOGGER.debug("No need to recompute quality of data set #{} (statistics are completed).", dataSetId);
                return;
            }
            if (!metadata.getLifecycle().schemaAnalyzed()) {
                LOGGER.debug(
                        "Schema information must be computed before quality analysis can be performed, ignoring message");
                return; // no acknowledge to allow re-poll.
            }

            try (Stream<DataSetRow> stream = store.stream(metadata)) {

                LOGGER.debug("Analyzing quality of dataset #{}...", metadata.getId());
                // New data set, or reached the max limit of records for synchronous analysis, trigger a full scan (but
                // async).
                final long dataSetSize = metadata.getContent().getNbRecords();
                final boolean isNewDataSet = dataSetSize == 0;
                if (isNewDataSet || dataSetSize == maxRecord) {
                    // If data set size is maxRecord, performs a full scan, otherwise only take first maxRecord
                    // records.
                    computeQuality(metadata, stream, dataSetSize == maxRecord ? -1 : maxRecord);
                }
                // Turn on / off "in progress" flag
                if (isNewDataSet && metadata.getContent().getNbRecords() >= maxRecord) {
                    metadata.getLifecycle().setInProgress(true);
                } else {
                    metadata.getLifecycle().setInProgress(false);
                }
                // ... all quality is now analyzed, mark it so.
                metadata.getLifecycle().qualityAnalyzed(true);
                final DataSetMetadata savedDataSetMetadata = repository.get(metadata.getId());
                // in order to check that the dataset was not deleted during analysis
                if (savedDataSetMetadata != null) {
                    repository.save(metadata);
                    LOGGER.debug("Analyzed quality of dataset #{}.", dataSetId);
                }

            } catch (Exception e) {
                LOGGER.warn("dataset '{}' generate an error, message: {} ", dataSetId, e.getMessage());
                throw new TDPException(DataSetErrorCodes.UNABLE_TO_ANALYZE_DATASET_QUALITY, e);
            }
        } finally {
            datasetLock.unlock();
        }
    }

}