in streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/SparkApplicationBuildPipelineServiceImpl.java [151:343]
public boolean buildApplication(@Nonnull Long appId, boolean forceBuild) {
// check the build environment
checkBuildEnv(appId, forceBuild);
SparkApplication app = applicationManageService.getById(appId);
ApplicationLog applicationLog = new ApplicationLog();
applicationLog.setJobType(EngineTypeEnum.SPARK.getCode());
applicationLog.setOptionName(RELEASE.getValue());
applicationLog.setAppId(app.getId());
applicationLog.setCreateTime(new Date());
applicationLog.setUserId(ServiceHelper.getUserId());
// check if you need to go through the build process (if the jar and pom have changed,
// you need to go through the build process, if other common parameters are modified,
// you don't need to go through the build process)
boolean needBuild = applicationManageService.checkBuildAndUpdate(app);
if (!needBuild) {
applicationLog.setSuccess(true);
applicationLogService.save(applicationLog);
return true;
}
// 1) spark sql setDependency
SparkSql newSparkSql = sparkSqlService.getCandidate(app.getId(), CandidateTypeEnum.NEW);
SparkSql effectiveSparkSql = sparkSqlService.getEffective(app.getId(), false);
if (app.isSparkSqlJob()) {
SparkSql sparkSql = newSparkSql == null ? effectiveSparkSql : newSparkSql;
AssertUtils.notNull(sparkSql);
app.setDependency(sparkSql.getDependency());
app.setTeamResource(sparkSql.getTeamResource());
}
// create pipeline instance
BuildPipeline pipeline = createPipelineInstance(app);
// clear history
removeByAppId(app.getId());
// register pipeline progress event watcher.
// save snapshot of pipeline to db when status of pipeline was changed.
pipeline.registerWatcher(
new PipeWatcher() {
@Override
public void onStart(PipelineSnapshot snapshot) {
ApplicationBuildPipeline buildPipeline = ApplicationBuildPipeline.fromPipeSnapshot(snapshot)
.setAppId(app.getId());
saveEntity(buildPipeline);
app.setRelease(ReleaseStateEnum.RELEASING.get());
applicationManageService.updateRelease(app);
if (sparkAppHttpWatcher.isWatchingApp(app.getId())) {
sparkAppHttpWatcher.init();
}
// 1) checkEnv
applicationInfoService.checkEnv(app);
// 2) some preparatory work
String appUploads = app.getWorkspace().APP_UPLOADS();
if (app.isSparkJarOrPySparkJob()) {
// spark jar and pyspark upload resource to appHome...
String appHome = app.getAppHome();
FsOperator fsOperator = app.getFsOperator();
fsOperator.delete(appHome);
if (app.isFromUploadJob()) {
String uploadJar = appUploads.concat("/").concat(app.getJar());
File localJar = new File(
String.format(
"%s/%d/%s",
Workspace.local().APP_UPLOADS(),
app.getTeamId(),
app.getJar()));
if (!localJar.exists()) {
Resource resource = resourceService.findByResourceName(app.getTeamId(),
app.getJar());
if (resource != null && StringUtils.isNotBlank(resource.getFilePath())) {
localJar = new File(resource.getFilePath());
uploadJar = appUploads.concat("/").concat(localJar.getName());
}
}
// upload jar copy to appHome
checkOrElseUploadJar(app.getFsOperator(), localJar, uploadJar, appUploads);
switch (app.getApplicationType()) {
case STREAMPARK_SPARK:
fsOperator.mkdirs(app.getAppLib());
fsOperator.copy(uploadJar, app.getAppLib(), false, true);
break;
case APACHE_SPARK:
fsOperator.mkdirs(appHome);
fsOperator.copy(uploadJar, appHome, false, true);
break;
default:
throw new IllegalArgumentException(
"[StreamPark] unsupported ApplicationType of FlinkJar: "
+ app.getApplicationType());
}
} else {
fsOperator.upload(app.getDistHome(), appHome);
}
} else {
if (!app.getDependencyObject().getJar().isEmpty()) {
String localUploads = Workspace.local().APP_UPLOADS();
// copy jar to local upload dir
for (String jar : app.getDependencyObject().getJar()) {
File localJar = new File(WebUtils.getAppTempDir(), jar);
File uploadJar = new File(localUploads, jar);
if (!localJar.exists() && !uploadJar.exists()) {
throw new ApiAlertException(
"Missing file: " + jar + ", please upload again");
}
if (localJar.exists()) {
checkOrElseUploadJar(
FsOperator.lfs(), localJar, uploadJar.getAbsolutePath(),
localUploads);
}
}
}
}
}
@Override
public void onStepStateChange(PipelineSnapshot snapshot) {
ApplicationBuildPipeline buildPipeline = ApplicationBuildPipeline.fromPipeSnapshot(snapshot)
.setAppId(app.getId());
saveEntity(buildPipeline);
}
@Override
public void onFinish(PipelineSnapshot snapshot, BuildResult result) {
ApplicationBuildPipeline buildPipeline = ApplicationBuildPipeline.fromPipeSnapshot(snapshot)
.setAppId(app.getId())
.setBuildResult(result);
saveEntity(buildPipeline);
if (result.pass()) {
// running job ...
if (app.isRunning()) {
app.setRelease(ReleaseStateEnum.NEED_RESTART.get());
} else {
app.setOptionState(OptionStateEnum.NONE.getValue());
app.setRelease(ReleaseStateEnum.DONE.get());
// If the current task is not running, or the task has just been added, directly
// set
// the candidate version to the official version
if (app.isSparkOnYarnJob()) {
applicationManageService.toEffective(app);
} else {
if (app.isStreamParkJob()) {
SparkApplicationConfig config =
applicationConfigService.getLatest(app.getId());
if (config != null) {
config.setToApplication(app);
applicationConfigService.toEffective(app.getId(),
app.getConfigId());
}
}
}
}
applicationLog.setSuccess(true);
app.setBuild(false);
} else {
Message message = new Message(
ServiceHelper.getUserId(),
app.getId(),
app.getAppName().concat(" release failed"),
ExceptionUtils.stringifyException(snapshot.error().exception()),
NoticeTypeEnum.EXCEPTION);
messageService.push(message);
app.setRelease(ReleaseStateEnum.FAILED.get());
app.setOptionState(OptionStateEnum.NONE.getValue());
app.setBuild(true);
applicationLog.setException(
ExceptionUtils.stringifyException(snapshot.error().exception()));
applicationLog.setSuccess(false);
}
applicationManageService.updateRelease(app);
applicationLogService.save(applicationLog);
if (sparkAppHttpWatcher.isWatchingApp(app.getId())) {
sparkAppHttpWatcher.init();
}
}
});
// save pipeline instance snapshot to db before release it.
ApplicationBuildPipeline buildPipeline =
ApplicationBuildPipeline.initFromPipeline(pipeline).setAppId(app.getId());
boolean saved = saveEntity(buildPipeline);
// async release pipeline
executorService.submit((Runnable) pipeline::launch);
return saved;
}