You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by "abzymeatsjtu (via GitHub)" <gi...@apache.org> on 2025/09/04 09:21:35 UTC
[PR] [Improvement-16994][TaskPlugin] support retry for every api call for serverless spark [dolphinscheduler]
abzymeatsjtu opened a new pull request, #17476:
URL: https://github.com/apache/dolphinscheduler/pull/17476
support retry for every api call for EMR Serverless Spark, this will improve the robustness of this task plugin against some temporarily malfunction of remote service
part of #16994
--
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: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Improvement-16994][TaskPlugin] support retry for every api call for serverless spark [dolphinscheduler]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #17476:
URL: https://github.com/apache/dolphinscheduler/pull/17476#issuecomment-3253791059
## [](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476) **Quality Gate failed**
Failed conditions
 [C Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
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: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Improvement-16994][TaskPlugin] support retry for every api call for serverless spark [dolphinscheduler]
Posted by "ruanwenjun (via GitHub)" <gi...@apache.org>.
ruanwenjun commented on code in PR #17476:
URL: https://github.com/apache/dolphinscheduler/pull/17476#discussion_r2326549930
##########
dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/task/aliyunserverlessspark/AliyunServerlessSparkTask.java:
##########
@@ -126,60 +126,66 @@ public void init() {
@Override
public void handle(TaskCallBack taskCallBack) throws TaskException {
- try {
- GetTemplateResponse getTemplateResponse = aliyunServerlessSparkClient.getTemplate(
- aliyunServerlessSparkParameters.getWorkspaceId(),
- buildGetTemplateRequest());
-
- if (getTemplateResponse != null) {
- templateConf = getTemplateResponse.getBody()
- .getData()
- .getSparkConf()
- .stream()
- .map(item -> "--conf " + item.getKey() + "=" + item.getValue())
- .collect(Collectors.joining(" "));
-
- templateDisplayReleaseVersion = getTemplateResponse.getBody().getData().getDisplaySparkVersion();
- templateFusion = getTemplateResponse.getBody().getData().getFusion();
+ GetTemplateResponse getTemplateResponse = RetryUtils.retryFunction(() -> {
+ try {
+ return aliyunServerlessSparkClient.getTemplate(
+ aliyunServerlessSparkParameters.getWorkspaceId(),
+ buildGetTemplateRequest());
+ } catch (Exception e) {
+ throw new TaskException("Failed to get template info", e);
}
- } catch (Exception e) {
- throw new AliyunServerlessSparkTaskException("Failed to get serverless spark template!");
+ }, retryPolicy);
+
+ if (getTemplateResponse != null) {
+ templateConf = getTemplateResponse.getBody()
+ .getData()
+ .getSparkConf()
+ .stream()
+ .map(item -> "--conf " + item.getKey() + "=" + item.getValue())
+ .collect(Collectors.joining(" "));
+
+ templateDisplayReleaseVersion = getTemplateResponse.getBody().getData().getDisplaySparkVersion();
+ templateFusion = getTemplateResponse.getBody().getData().getFusion();
}
- try {
- StartJobRunRequest startJobRunRequest = buildStartJobRunRequest(aliyunServerlessSparkParameters);
- RuntimeOptions runtime = new RuntimeOptions();
- Map<String, String> headers = new HashMap<>();
- StartJobRunResponse startJobRunResponse = aliyunServerlessSparkClient.startJobRunWithOptions(
- aliyunServerlessSparkParameters.getWorkspaceId(), startJobRunRequest, headers, runtime);
- jobRunId = startJobRunResponse.getBody().getJobRunId();
- setAppIds(jobRunId);
- log.info("Successfully submitted serverless spark job, jobRunId - {}", jobRunId);
-
- while (!RunState.isFinal(currentState)) {
- GetJobRunRequest getJobRunRequest = buildGetJobRunRequest();
-
- GetJobRunResponse getJobRunResponse = RetryUtils.retryFunction(() -> {
- try {
- return aliyunServerlessSparkClient
- .getJobRun(aliyunServerlessSparkParameters.getWorkspaceId(), jobRunId,
- getJobRunRequest);
- } catch (Exception e) {
- throw new AliyunServerlessSparkTaskException("Failed to get job run!", e);
- }
- }, new RetryUtils.RetryPolicy(10, 1000L));
-
- currentState = RunState.valueOf(getJobRunResponse.getBody().getJobRun().getState());
- log.info("job - {} state - {}", jobRunId, currentState);
- Thread.sleep(10 * 1000L);
+ StartJobRunRequest startJobRunRequest = buildStartJobRunRequest(aliyunServerlessSparkParameters);
+ StartJobRunResponse startJobRunResponse = RetryUtils.retryFunction(() -> {
+ try {
+ return aliyunServerlessSparkClient.startJobRun(
+ aliyunServerlessSparkParameters.getWorkspaceId(), startJobRunRequest);
+ } catch (Exception e) {
+ throw new AliyunServerlessSparkTaskException("Failed to start job run!");
Review Comment:
Ah, I missed that. If the server-side is capable of implementing idempotency handling via tokens, that would be a great solution.
--
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: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Improvement-16994][TaskPlugin] support retry for every api call for serverless spark [dolphinscheduler]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #17476:
URL: https://github.com/apache/dolphinscheduler/pull/17476#issuecomment-3253800296
## [](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476) **Quality Gate failed**
Failed conditions
 [C Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
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: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Improvement-16994][TaskPlugin] support retry for every api call for serverless spark [dolphinscheduler]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #17476:
URL: https://github.com/apache/dolphinscheduler/pull/17476#issuecomment-3264472505
## [](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476) **Quality Gate failed**
Failed conditions
 [C Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
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: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Improvement-16994][TaskPlugin] support retry for every api call for serverless spark [dolphinscheduler]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #17476:
URL: https://github.com/apache/dolphinscheduler/pull/17476#issuecomment-3256962148
## [](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476) **Quality Gate failed**
Failed conditions
 [C Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
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: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Improvement-16994][TaskPlugin] support retry for every api call for serverless spark [dolphinscheduler]
Posted by "EricGao888 (via GitHub)" <gi...@apache.org>.
EricGao888 merged PR #17476:
URL: https://github.com/apache/dolphinscheduler/pull/17476
--
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: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Improvement-16994][TaskPlugin] support retry for every api call for serverless spark [dolphinscheduler]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #17476:
URL: https://github.com/apache/dolphinscheduler/pull/17476#issuecomment-3256963078
## [](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476) **Quality Gate failed**
Failed conditions
 [C Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
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: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Improvement-16994][TaskPlugin] support retry for every api call for serverless spark [dolphinscheduler]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #17476:
URL: https://github.com/apache/dolphinscheduler/pull/17476#issuecomment-3264470211
## [](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476) **Quality Gate failed**
Failed conditions
 [C Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache-dolphinscheduler&pullRequest=17476)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
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: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Improvement-16994][TaskPlugin] support retry for every api call for serverless spark [dolphinscheduler]
Posted by "ruanwenjun (via GitHub)" <gi...@apache.org>.
ruanwenjun commented on code in PR #17476:
URL: https://github.com/apache/dolphinscheduler/pull/17476#discussion_r2324046942
##########
dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/task/aliyunserverlessspark/AliyunServerlessSparkTask.java:
##########
@@ -126,60 +126,66 @@ public void init() {
@Override
public void handle(TaskCallBack taskCallBack) throws TaskException {
- try {
- GetTemplateResponse getTemplateResponse = aliyunServerlessSparkClient.getTemplate(
- aliyunServerlessSparkParameters.getWorkspaceId(),
- buildGetTemplateRequest());
-
- if (getTemplateResponse != null) {
- templateConf = getTemplateResponse.getBody()
- .getData()
- .getSparkConf()
- .stream()
- .map(item -> "--conf " + item.getKey() + "=" + item.getValue())
- .collect(Collectors.joining(" "));
-
- templateDisplayReleaseVersion = getTemplateResponse.getBody().getData().getDisplaySparkVersion();
- templateFusion = getTemplateResponse.getBody().getData().getFusion();
+ GetTemplateResponse getTemplateResponse = RetryUtils.retryFunction(() -> {
+ try {
+ return aliyunServerlessSparkClient.getTemplate(
+ aliyunServerlessSparkParameters.getWorkspaceId(),
+ buildGetTemplateRequest());
+ } catch (Exception e) {
+ throw new TaskException("Failed to get template info", e);
}
- } catch (Exception e) {
- throw new AliyunServerlessSparkTaskException("Failed to get serverless spark template!");
+ }, retryPolicy);
+
+ if (getTemplateResponse != null) {
+ templateConf = getTemplateResponse.getBody()
+ .getData()
+ .getSparkConf()
+ .stream()
+ .map(item -> "--conf " + item.getKey() + "=" + item.getValue())
+ .collect(Collectors.joining(" "));
+
+ templateDisplayReleaseVersion = getTemplateResponse.getBody().getData().getDisplaySparkVersion();
+ templateFusion = getTemplateResponse.getBody().getData().getFusion();
}
- try {
- StartJobRunRequest startJobRunRequest = buildStartJobRunRequest(aliyunServerlessSparkParameters);
- RuntimeOptions runtime = new RuntimeOptions();
- Map<String, String> headers = new HashMap<>();
- StartJobRunResponse startJobRunResponse = aliyunServerlessSparkClient.startJobRunWithOptions(
- aliyunServerlessSparkParameters.getWorkspaceId(), startJobRunRequest, headers, runtime);
- jobRunId = startJobRunResponse.getBody().getJobRunId();
- setAppIds(jobRunId);
- log.info("Successfully submitted serverless spark job, jobRunId - {}", jobRunId);
-
- while (!RunState.isFinal(currentState)) {
- GetJobRunRequest getJobRunRequest = buildGetJobRunRequest();
-
- GetJobRunResponse getJobRunResponse = RetryUtils.retryFunction(() -> {
- try {
- return aliyunServerlessSparkClient
- .getJobRun(aliyunServerlessSparkParameters.getWorkspaceId(), jobRunId,
- getJobRunRequest);
- } catch (Exception e) {
- throw new AliyunServerlessSparkTaskException("Failed to get job run!", e);
- }
- }, new RetryUtils.RetryPolicy(10, 1000L));
-
- currentState = RunState.valueOf(getJobRunResponse.getBody().getJobRun().getState());
- log.info("job - {} state - {}", jobRunId, currentState);
- Thread.sleep(10 * 1000L);
+ StartJobRunRequest startJobRunRequest = buildStartJobRunRequest(aliyunServerlessSparkParameters);
+ StartJobRunResponse startJobRunResponse = RetryUtils.retryFunction(() -> {
+ try {
+ return aliyunServerlessSparkClient.startJobRun(
+ aliyunServerlessSparkParameters.getWorkspaceId(), startJobRunRequest);
+ } catch (Exception e) {
+ throw new AliyunServerlessSparkTaskException("Failed to start job run!");
Review Comment:
There seems to exist timeout issue here, if the http is timeout, then client will retry, but server side might already handle the previous, then will cause the request be handled twice. I'm unsure whether the service side has implemented idempotency handling. Because a new token is passed here each time, so the server side cannot know the second request are retry.
--
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: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Improvement-16994][TaskPlugin] support retry for every api call for serverless spark [dolphinscheduler]
Posted by "EricGao888 (via GitHub)" <gi...@apache.org>.
EricGao888 commented on code in PR #17476:
URL: https://github.com/apache/dolphinscheduler/pull/17476#discussion_r2324651870
##########
dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/task/aliyunserverlessspark/AliyunServerlessSparkTask.java:
##########
@@ -126,60 +126,66 @@ public void init() {
@Override
public void handle(TaskCallBack taskCallBack) throws TaskException {
- try {
- GetTemplateResponse getTemplateResponse = aliyunServerlessSparkClient.getTemplate(
- aliyunServerlessSparkParameters.getWorkspaceId(),
- buildGetTemplateRequest());
-
- if (getTemplateResponse != null) {
- templateConf = getTemplateResponse.getBody()
- .getData()
- .getSparkConf()
- .stream()
- .map(item -> "--conf " + item.getKey() + "=" + item.getValue())
- .collect(Collectors.joining(" "));
-
- templateDisplayReleaseVersion = getTemplateResponse.getBody().getData().getDisplaySparkVersion();
- templateFusion = getTemplateResponse.getBody().getData().getFusion();
+ GetTemplateResponse getTemplateResponse = RetryUtils.retryFunction(() -> {
+ try {
+ return aliyunServerlessSparkClient.getTemplate(
+ aliyunServerlessSparkParameters.getWorkspaceId(),
+ buildGetTemplateRequest());
+ } catch (Exception e) {
+ throw new TaskException("Failed to get template info", e);
}
- } catch (Exception e) {
- throw new AliyunServerlessSparkTaskException("Failed to get serverless spark template!");
+ }, retryPolicy);
+
+ if (getTemplateResponse != null) {
+ templateConf = getTemplateResponse.getBody()
+ .getData()
+ .getSparkConf()
+ .stream()
+ .map(item -> "--conf " + item.getKey() + "=" + item.getValue())
+ .collect(Collectors.joining(" "));
+
+ templateDisplayReleaseVersion = getTemplateResponse.getBody().getData().getDisplaySparkVersion();
+ templateFusion = getTemplateResponse.getBody().getData().getFusion();
}
- try {
- StartJobRunRequest startJobRunRequest = buildStartJobRunRequest(aliyunServerlessSparkParameters);
- RuntimeOptions runtime = new RuntimeOptions();
- Map<String, String> headers = new HashMap<>();
- StartJobRunResponse startJobRunResponse = aliyunServerlessSparkClient.startJobRunWithOptions(
- aliyunServerlessSparkParameters.getWorkspaceId(), startJobRunRequest, headers, runtime);
- jobRunId = startJobRunResponse.getBody().getJobRunId();
- setAppIds(jobRunId);
- log.info("Successfully submitted serverless spark job, jobRunId - {}", jobRunId);
-
- while (!RunState.isFinal(currentState)) {
- GetJobRunRequest getJobRunRequest = buildGetJobRunRequest();
-
- GetJobRunResponse getJobRunResponse = RetryUtils.retryFunction(() -> {
- try {
- return aliyunServerlessSparkClient
- .getJobRun(aliyunServerlessSparkParameters.getWorkspaceId(), jobRunId,
- getJobRunRequest);
- } catch (Exception e) {
- throw new AliyunServerlessSparkTaskException("Failed to get job run!", e);
- }
- }, new RetryUtils.RetryPolicy(10, 1000L));
-
- currentState = RunState.valueOf(getJobRunResponse.getBody().getJobRun().getState());
- log.info("job - {} state - {}", jobRunId, currentState);
- Thread.sleep(10 * 1000L);
+ StartJobRunRequest startJobRunRequest = buildStartJobRunRequest(aliyunServerlessSparkParameters);
+ StartJobRunResponse startJobRunResponse = RetryUtils.retryFunction(() -> {
+ try {
+ return aliyunServerlessSparkClient.startJobRun(
+ aliyunServerlessSparkParameters.getWorkspaceId(), startJobRunRequest);
+ } catch (Exception e) {
+ throw new AliyunServerlessSparkTaskException("Failed to start job run!");
Review Comment:
> There seems to exist timeout issue here, if the http is timeout, then client will retry, but server side might already handle the previous, then will cause the request be handled twice. I'm unsure whether the service side has implemented idempotency handling. Because a new token is passed here each time, so the server side cannot know the second request are retry.
@ruanwenjun @abzymeatsjtu Looks like the token is generated and set when initializing the request line#257, therefore, I assume the idempotency is alright here?
--
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: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org