Preconditions.checkNotNull(input);
ApplicationSpecification specification = input.getSpecification();
Id.Application appId = input.getApplicationId();
VerifyResult result = getVerifier(ApplicationSpecification.class).verify(appId, specification);
if (!result.isSuccess()) {
throw new RuntimeException(result.getMessage());
}
// NOTE: no special restrictions on dataset module names, etc
for (DatasetCreationSpec dataSetCreateSpec : specification.getDatasets().values()) {
result = getVerifier(DatasetCreationSpec.class).verify(appId, dataSetCreateSpec);
if (!result.isSuccess()) {
throw new RuntimeException(result.getMessage());
}
String dsName = dataSetCreateSpec.getInstanceName();
DatasetSpecification existingSpec = dsFramework.getDatasetSpec(dsName);
if (existingSpec != null && !existingSpec.getType().equals(dataSetCreateSpec.getTypeName())) {
// New app trying to deploy an dataset with same instanceName but different Type than that of existing.
throw new DataSetException
(String.format("Cannot Deploy Dataset : %s with Type : %s : Dataset with different Type Already Exists",
dsName, dataSetCreateSpec.getTypeName()));
}
}
for (StreamSpecification spec : specification.getStreams().values()) {
result = getVerifier(StreamSpecification.class).verify(appId, spec);
if (!result.isSuccess()) {
throw new RuntimeException(result.getMessage());
}
}
Iterable<ProgramSpecification> programSpecs = Iterables.concat(specification.getFlows().values(),
specification.getMapReduce().values(),
specification.getProcedures().values(),
specification.getWorkflows().values());
for (ProgramSpecification programSpec : programSpecs) {
result = getVerifier(programSpec.getClass()).verify(appId, programSpec);
if (!result.isSuccess()) {
throw new RuntimeException(result.getMessage());
}
}
// Emit the input to next stage.
emit(input);