You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@streampark.apache.org by GitBox <gi...@apache.org> on 2022/09/13 12:25:21 UTC
[GitHub] [incubator-streampark] xujiangfeng001 opened a new pull request, #1594: [Bug] flinkclousterid cannot be saved in yarn-session mode
xujiangfeng001 opened a new pull request, #1594:
URL: https://github.com/apache/incubator-streampark/pull/1594
<!--
Thank you for contributing to StreamPark! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.-->
### What problem does this PR solve?
Issue Number: https://github.com/apache/incubator-streampark/issues/1593
Problem Summary:
### What is changed and how it works?
Proposal: https://github.com/apache/incubator-streampark/issues/1593
Feel free to ping committers for the review!
## Contribution Checklist
- Make sure that the pull request corresponds to a [GITHUB issue](https://github.com/apache/incubator-streampark/issues/1593).
- Name the pull request in the form "[Feature] [component] Title of the pull request", where *Feature* can be replaced by `Hotfix`, `Bug`, etc.
- Minor fixes should be named following this pattern: `[hotfix] [docs] Fix typo in README.md doc`.
## Purpose of this pull request
Fixed the loss of FlinkClusterId in the Flink session mode.
In case of loss, see the following table for the problems: https://github.com/apache/incubator-streampark/issues/1593
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-streampark] wolfboys commented on a diff in pull request #1594: [Bug] flinkclousterid cannot be saved in yarn-session mode
Posted by GitBox <gi...@apache.org>.
wolfboys commented on code in PR #1594:
URL: https://github.com/apache/incubator-streampark/pull/1594#discussion_r970877111
##########
streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkTrackingTask.java:
##########
@@ -737,7 +745,8 @@ private Overview httpOverview(Application application, FlinkCluster flinkCluster
private JobsOverview httpJobsOverview(Application application, FlinkCluster flinkCluster) throws Exception {
Review Comment:
you can improve this code :
```
private JobsOverview httpJobsOverview(Application application, FlinkCluster flinkCluster) throws Exception {
final String flinkUrl = "jobs/overview";
ExecutionMode execMode = application.getExecutionModeEnum();
if (ExecutionMode.REMOTE.equals(execMode) ||ExecutionMode.YARN_SESSION.equals(execMode)) {
if (application.getJobId() != null) {
String remoteUrl = flinkCluster.getActiveAddress().toURL() + "/" + flinkUrl;
JobsOverview jobsOverview = httpRestRequest(remoteUrl, JobsOverview.class);
if (jobsOverview != null) {
//过滤出当前job
List<JobsOverview.Job> jobs = jobsOverview.getJobs().stream().filter(x -> x.getId().equals(application.getJobId())).collect(Collectors.toList());
jobsOverview.setJobs(jobs);
}
return jobsOverview;
}
} else if (ExecutionMode.YARN_PER_JOB.equals(execMode) || ExecutionMode.YARN_APPLICATION.equals(execMode)) {
String format = "proxy/%s/" + flinkUrl;
String reqURL = String.format(format, application.getAppId());
JobsOverview jobsOverview = yarnRestRequest(reqURL, JobsOverview.class);
return jobsOverview;
}
return null;
}
```
##########
streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkTrackingTask.java:
##########
@@ -271,7 +271,8 @@ private void getFromFlinkRestApi(Application application, StopFrom stopFrom) thr
FlinkCluster flinkCluster = getFlinkCluster(application);
JobsOverview jobsOverview = httpJobsOverview(application, flinkCluster);
Optional<JobsOverview.Job> optional;
- if (ExecutionMode.isYarnMode(application.getExecutionMode())) {
+ if (ExecutionMode.isYarnMode(application.getExecutionMode()) &&
Review Comment:
change it to this, will be more clear : `
if (ExecutionMode.YARN_APPLICATION.equals(application.getExecutionMode()) || ExecutionMode.YARN_PER_JOB.equals(application.getExecutionMode()))
`
##########
streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkTrackingTask.java:
##########
@@ -764,11 +774,13 @@ private JobsOverview httpJobsOverview(Application application, FlinkCluster flin
private CheckPoints httpCheckpoints(Application application, FlinkCluster flinkCluster) throws IOException {
Review Comment:
you can improve this code :
```
private CheckPoints httpCheckpoints(Application application, FlinkCluster flinkCluster) throws IOException {
final String flinkUrl = "jobs/%s/checkpoints";
ExecutionMode execMode = application.getExecutionModeEnum();
if ( ExecutionMode.YARN_APPLICATION.equals(execMode) || ExecutionMode.YARN_PER_JOB.equals(execMode)) {
String format = "proxy/%s/" + flinkUrl;
String reqURL = String.format(format, application.getAppId(), application.getJobId());
return yarnRestRequest(reqURL, CheckPoints.class);
} else if (ExecutionMode.REMOTE.equals(execMode) || ExecutionMode.YARN_SESSION.equals(execMode) ) {
if (application.getJobId() != null) {
String remoteUrl = flinkCluster.getActiveAddress().toURL() + "/" + String.format(flinkUrl, application.getJobId());
return httpRestRequest(remoteUrl, CheckPoints.class);
}
}
return null;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-streampark] xujiangfeng001 closed pull request #1594: [Bug] flinkclousterid cannot be saved in yarn-session mode
Posted by GitBox <gi...@apache.org>.
xujiangfeng001 closed pull request #1594: [Bug] flinkclousterid cannot be saved in yarn-session mode
URL: https://github.com/apache/incubator-streampark/pull/1594
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-streampark] xujiangfeng001 commented on pull request #1594: [Bug] flinkclousterid cannot be saved in yarn-session mode
Posted by GitBox <gi...@apache.org>.
xujiangfeng001 commented on PR #1594:
URL: https://github.com/apache/incubator-streampark/pull/1594#issuecomment-1246124305
> Thanks for your contribution. I'll review later.
thank you. Let me describe why it is modified like this?
- For the problem that the yarn session cannot get the task status correctly:
1. In the yarn session mode, the cluster address needs to be specified to obtain the status of the task. In the code, "yarn pre job" and "yarn application" are used to obtain the status of the task in the same way, which leads to the failure of the task to obtain the correct status.
2. "FLINK_CLUSTER_ID" in "t_flink_app" in yarn session mode is null. I think this field should be applicable to "flash remote" and "yarn session" modes, so I saved the ID corresponding to yarn session in this field.
- For yarn session restart, the corresponding task needs to be reconfigured:
1. I don't think this is reasonable. I don't think the original task of the yarn session should be reconfigured after the yarn session is restarted. So when submitting a task, I changed "yarn. application. id" from "t_flink_app" to obtain the corresponding id from "t_flink_cluster".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-streampark] wolfboys commented on pull request #1594: [Bug] flinkclousterid cannot be saved in yarn-session mode
Posted by GitBox <gi...@apache.org>.
wolfboys commented on PR #1594:
URL: https://github.com/apache/incubator-streampark/pull/1594#issuecomment-1245575165
Thanks for your contribution. I'll review later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org