SchemaAnalysis 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 schema of data set #{}: seems to be removed.", dataSetId); return; } try (Stream<DataSetRow> stream = store.stream(metadata)) { LOGGER.info("Analyzing schema in dataset #{}...", dataSetId); final List<ColumnMetadata> columns = metadata.getRowMetadata().getColumns(); try (Analyzer<Analyzers.Result> analyzer = analyzerService.schemaAnalysis(columns)) { stream.limit(100).map(row -> row.toArray(DataSetRow.SKIP_TDP_ID)).forEach(analyzer::analyze); adapter.adapt(columns, analyzer.getResult()); LOGGER.info("Analyzed schema in dataset #{}.", dataSetId); metadata.getLifecycle().schemaAnalyzed(true); DataSetMetadata savedDataSetMetadata = repository.get(dataSetId); if (savedDataSetMetadata != null) { repository.save(metadata); } else { LOGGER.info("Data set #{} no longer exists.", dataSetId); } } } catch (Exception e) { LOGGER.error("Unable to analyse schema for dataset " + dataSetId + ".", e); TDPException.rethrowOrWrap(e, UNABLE_TO_ANALYZE_COLUMN_TYPES); } } finally { datasetLock.unlock(); } }  @Override void analyze(String dataSetId); @Override int order();  }
@Test public void testNoDataSetFound() { schemaAnalysis.analyze("1234"); assertThat(dataSetMetadataRepository.get("1234"), nullValue()); }