Copilot commented on code in PR #4282:
URL: https://github.com/apache/streampark/pull/4282#discussion_r2313245546
##########
streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkSavepointServiceImpl.java:
##########
@@ -140,24 +140,38 @@ public String getSavePointPath(FlinkApplication appParam)
throws Exception {
// 1) properties have the highest priority, read the properties are
set: -Dstate.savepoints.dir
String savepointPath =
getSavepointFromDynamicProps(application.getDynamicProperties());
- if (StringUtils.isNotBlank(savepointPath)) {
- return savepointPath;
- }
// Application conf configuration has the second priority. If it is a
streampark|flinksql type
// task, see if Application conf is configured when the task is
defined, if checkpoints are
// configured
// and enabled, read `state.savepoints.dir`
- savepointPath = getSavepointFromConfig(application);
- if (StringUtils.isNotBlank(savepointPath)) {
- return savepointPath;
+ if (StringUtils.isBlank(savepointPath)) {
+ savepointPath = getSavepointFromConfig(application);
}
// 3) If the savepoint is not obtained above, try to obtain the
savepoint path according to the
// deployment type (remote|on yarn)
// 3.1) At the remote mode, request the flink webui interface to get
the savepoint path
// 3.2) At the yarn or k8s mode, then read the savepoint in
flink-conf.yml in the bound flink
- return getSavepointFromDeployLayer(application);
+ if (StringUtils.isBlank(savepointPath)) {
+ savepointPath = getSavepointFromDeployLayer(application);
+ }
+
+ // 4) Supporting variables
+ if (StringUtils.isNotBlank(savepointPath)) {
+ savepointPath =
+ processPath(
+ savepointPath, application.getJobName(),
application.getId());
+ }
+ return savepointPath;
+ }
+
+ private String processPath(String path, String jobName, Long jobId) {
+ if (StringUtils.isNotBlank(path)) {
+ return path.replaceAll("\\$\\{job(Name|name)}|\\$job(Name|name)",
jobName)
+ .replaceAll("\\$\\{job(Id|id)}|\\$job(Id|id)",
jobId.toString());
Review Comment:
The regex pattern uses non-capturing groups incorrectly. The pattern
`\\$\\{job(Name|name)}` will match `${jobName}` but capture 'Name' or 'name' in
group 1, while `\\$job(Name|name)` will match `$jobName` and capture 'Name' or
'name'. This could lead to unexpected behavior. Consider using non-capturing
groups: `\\$\\{job(?:Name|name)}|\\$job(?:Name|name)` or separate the patterns
for clarity.
```suggestion
return
path.replaceAll("\\$\\{job(?:Name|name)}|\\$job(?:Name|name)", jobName)
.replaceAll("\\$\\{job(?:Id|id)}|\\$job(?:Id|id)",
jobId.toString());
```
##########
streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkSavepointServiceImpl.java:
##########
@@ -140,24 +140,38 @@ public String getSavePointPath(FlinkApplication appParam)
throws Exception {
// 1) properties have the highest priority, read the properties are
set: -Dstate.savepoints.dir
String savepointPath =
getSavepointFromDynamicProps(application.getDynamicProperties());
- if (StringUtils.isNotBlank(savepointPath)) {
- return savepointPath;
- }
// Application conf configuration has the second priority. If it is a
streampark|flinksql type
// task, see if Application conf is configured when the task is
defined, if checkpoints are
// configured
// and enabled, read `state.savepoints.dir`
- savepointPath = getSavepointFromConfig(application);
- if (StringUtils.isNotBlank(savepointPath)) {
- return savepointPath;
+ if (StringUtils.isBlank(savepointPath)) {
+ savepointPath = getSavepointFromConfig(application);
}
// 3) If the savepoint is not obtained above, try to obtain the
savepoint path according to the
// deployment type (remote|on yarn)
// 3.1) At the remote mode, request the flink webui interface to get
the savepoint path
// 3.2) At the yarn or k8s mode, then read the savepoint in
flink-conf.yml in the bound flink
- return getSavepointFromDeployLayer(application);
+ if (StringUtils.isBlank(savepointPath)) {
+ savepointPath = getSavepointFromDeployLayer(application);
+ }
+
+ // 4) Supporting variables
+ if (StringUtils.isNotBlank(savepointPath)) {
+ savepointPath =
+ processPath(
+ savepointPath, application.getJobName(),
application.getId());
+ }
+ return savepointPath;
+ }
+
+ private String processPath(String path, String jobName, Long jobId) {
+ if (StringUtils.isNotBlank(path)) {
+ return path.replaceAll("\\$\\{job(Name|name)}|\\$job(Name|name)",
jobName)
+ .replaceAll("\\$\\{job(Id|id)}|\\$job(Id|id)",
jobId.toString());
Review Comment:
The method doesn't handle the case where `jobName` or `jobId` could be null.
If `jobName` is null, calling `replaceAll` with a null replacement will throw a
NullPointerException. Similarly, if `jobId` is null, calling `jobId.toString()`
will throw a NullPointerException. Add null checks before processing.
```suggestion
String safeJobName = jobName != null ? jobName : "";
String safeJobId = jobId != null ? jobId.toString() : "";
return
path.replaceAll("\\$\\{job(Name|name)}|\\$job(Name|name)", safeJobName)
.replaceAll("\\$\\{job(Id|id)}|\\$job(Id|id)", safeJobId);
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
