@Test public void testNoDataSetFound() { qualityAnalysis.analyze("1234"); assertThat(dataSetMetadataRepository.get("1234"), nullValue()); }
@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; } 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; } try (Stream<DataSetRow> stream = store.stream(metadata)) { LOGGER.debug("Analyzing quality of dataset #{}...", metadata.getId()); final long dataSetSize = metadata.getContent().getNbRecords(); final boolean isNewDataSet = dataSetSize == 0; if (isNewDataSet || dataSetSize == maxRecord) { computeQuality(metadata, stream, dataSetSize == maxRecord ? -1 : maxRecord); } if (isNewDataSet && metadata.getContent().getNbRecords() >= maxRecord) { metadata.getLifecycle().setInProgress(true); } else { metadata.getLifecycle().setInProgress(false); } metadata.getLifecycle().qualityAnalyzed(true); final DataSetMetadata savedDataSetMetadata = repository.get(metadata.getId()); 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(); } }
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; } 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; } try (Stream<DataSetRow> stream = store.stream(metadata)) { LOGGER.debug("Analyzing quality of dataset #{}...", metadata.getId()); final long dataSetSize = metadata.getContent().getNbRecords(); final boolean isNewDataSet = dataSetSize == 0; if (isNewDataSet || dataSetSize == maxRecord) { computeQuality(metadata, stream, dataSetSize == maxRecord ? -1 : maxRecord); } if (isNewDataSet && metadata.getContent().getNbRecords() >= maxRecord) { metadata.getLifecycle().setInProgress(true); } else { metadata.getLifecycle().setInProgress(false); } metadata.getLifecycle().qualityAnalyzed(true); final DataSetMetadata savedDataSetMetadata = repository.get(metadata.getId()); 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(); } } }
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; } 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; } try (Stream<DataSetRow> stream = store.stream(metadata)) { LOGGER.debug("Analyzing quality of dataset #{}...", metadata.getId()); final long dataSetSize = metadata.getContent().getNbRecords(); final boolean isNewDataSet = dataSetSize == 0; if (isNewDataSet || dataSetSize == maxRecord) { computeQuality(metadata, stream, dataSetSize == maxRecord ? -1 : maxRecord); } if (isNewDataSet && metadata.getContent().getNbRecords() >= maxRecord) { metadata.getLifecycle().setInProgress(true); } else { metadata.getLifecycle().setInProgress(false); } metadata.getLifecycle().qualityAnalyzed(true); final DataSetMetadata savedDataSetMetadata = repository.get(metadata.getId()); 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(); } }  }
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; } 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; } try (Stream<DataSetRow> stream = store.stream(metadata)) { LOGGER.debug("Analyzing quality of dataset #{}...", metadata.getId()); final long dataSetSize = metadata.getContent().getNbRecords(); final boolean isNewDataSet = dataSetSize == 0; if (isNewDataSet || dataSetSize == maxRecord) { computeQuality(metadata, stream, dataSetSize == maxRecord ? -1 : maxRecord); } if (isNewDataSet && metadata.getContent().getNbRecords() >= maxRecord) { metadata.getLifecycle().setInProgress(true); } else { metadata.getLifecycle().setInProgress(false); } metadata.getLifecycle().qualityAnalyzed(true); final DataSetMetadata savedDataSetMetadata = repository.get(metadata.getId()); 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(); } }  @Override void analyze(String dataSetId); @Override int order(); void computeQuality(DataSetMetadata dataset, Stream<DataSetRow> records, long limit); }
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; } 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; } try (Stream<DataSetRow> stream = store.stream(metadata)) { LOGGER.debug("Analyzing quality of dataset #{}...", metadata.getId()); final long dataSetSize = metadata.getContent().getNbRecords(); final boolean isNewDataSet = dataSetSize == 0; if (isNewDataSet || dataSetSize == maxRecord) { computeQuality(metadata, stream, dataSetSize == maxRecord ? -1 : maxRecord); } if (isNewDataSet && metadata.getContent().getNbRecords() >= maxRecord) { metadata.getLifecycle().setInProgress(true); } else { metadata.getLifecycle().setInProgress(false); } metadata.getLifecycle().qualityAnalyzed(true); final DataSetMetadata savedDataSetMetadata = repository.get(metadata.getId()); 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(); } }  @Override void analyze(String dataSetId); @Override int order(); void computeQuality(DataSetMetadata dataset, Stream<DataSetRow> records, long limit);  }