nzw921rx commented on code in PR #11841:
URL: https://github.com/apache/seatunnel/pull/11841#discussion_r3913682105
##########
seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java:
##########
@@ -284,7 +285,7 @@ public synchronized void init(long initializationTimestamp,
boolean restart) thr
sink.getId()));
JobMaster.handleSaveMode(
((SinkAction<?, ?, ?, ?>)
sink).getSink(),
- logicalDag.isStartWithSavePoint());
Review Comment:
Hi, could you explain why `isStartWithSavePoint` in `LogicalDag` was
replaced with `jobImmutableInformation.getRestoreMode()`?
From what I understand, after a restore, the `LogicalDag` is generated from
`JobImmutableInformation`. If we directly rely on `JobImmutableInformation`
here, could that introduce some ambiguity in terms of responsibility or
separation of concerns?
What I’m really wondering is whether `RestoreMode` itself should also be
defined as part of `LogicalDag`.
```java
public static LogicalDag restoreLogicalDag(
JobImmutableInformation jobImmutableInformation,
SerializationService serializationService,
List<ClassLoader> classLoaders) {
LogicalDag logicalDag =
serializationService.toObject(jobImmutableInformation.getLogicalDag());
ClassLoader classLoader =
Thread.currentThread().getContextClassLoader();
try {
List<Data> logicalVertexDataList =
jobImmutableInformation.getLogicalVertexDataList();
for (int i = 0; i <
jobImmutableInformation.getLogicalVertexDataList().size(); i++) {
Thread.currentThread().setContextClassLoader(classLoaders.get(i));
logicalDag.addLogicalVertex(
serializationService.toObject(logicalVertexDataList.get(i)));
}
return logicalDag;
} finally {
Thread.currentThread().setContextClassLoader(classLoader);
}
}
```
##########
seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java:
##########
@@ -1394,9 +1396,12 @@ public PassiveCompletableFuture<Void> submitJob(
try {
JobImmutableInformation
submittedJobImmutableInformation =
deserializeJobImmutableInformation(jobImmutableInformation);
+ boolean isSavepointRestore =
Review Comment:
The submission mode is also defined here, but the `isStartWithSavePoint`
parameter passed into this method is silently overridden at a lower level by
the value derived from `submittedJobImmutableInformation`:
```java
boolean isSavepointRestore =
submittedJobImmutableInformation != null
&& submittedJobImmutableInformation.isSavepointRestore();
```
Should we make this behavior consistent by having the restore mode
explicitly passed in from the caller instead?
Otherwise, the current `isStartWithSavePoint` parameter effectively becomes
obsolete, even though it is still part of the method signature and is used by
many unit tests. I think unifying the semantics here would be more reasonable
and would also make the API behavior clearer.
Looking forward to your thoughts.
##########
seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/BaseService.java:
##########
@@ -1360,7 +1360,7 @@ protected JsonObject submitJobInternal(
new SubmitJobOperation(
Review Comment:
When this Hazelcast internal API is invoked, if `jobImmutableInformation`
does not carry `restoreMode` / `isStartWithSavePoint`, then the outer
`parameters.isStartWithSavePoint` value will effectively be ignored.
```java
public class SubmitJobTask
extends
AbstractSeaTunnelMessageTask<SeaTunnelSubmitJobCodec.RequestParameters, Void> {
protected SubmitJobTask(ClientMessage clientMessage, Node node,
Connection connection) {
super(
clientMessage,
node,
connection,
SeaTunnelSubmitJobCodec::decodeRequest,
x -> SeaTunnelSubmitJobCodec.encodeResponse());
}
@Override
protected Operation prepareOperation() {
return new SubmitJobOperation(
parameters.jobId,
parameters.jobImmutableInformation,
parameters.isStartWithSavePoint);
}
@Override
public String getMethodName() {
return "submitJob";
}
@Override
public Object[] getParameters() {
return new Object[] {};
}
}
```
```java
public class SubmitJobOperation extends AbstractJobAsyncOperation {
private Data jobImmutableInformation;
private boolean isStartWithSavePoint;
public SubmitJobOperation() {}
public SubmitJobOperation(
long jobId, @NonNull Data jobImmutableInformation, boolean
isStartWithSavePoint) {
super(jobId);
this.jobImmutableInformation = jobImmutableInformation;
this.isStartWithSavePoint = isStartWithSavePoint;
}
@Override
public int getClassId() {
return ClientToServerOperationDataSerializerHook.SUBMIT_OPERATOR;
}
@Override
protected void writeInternal(ObjectDataOutput out) throws IOException {
super.writeInternal(out);
IOUtil.writeData(out, jobImmutableInformation);
out.writeBoolean(isStartWithSavePoint); // write
}
@Override
protected void readInternal(ObjectDataInput in) throws IOException {
super.readInternal(in);
jobImmutableInformation = IOUtil.readData(in);
isStartWithSavePoint = in.readBoolean(); // read
}
```
So although `parameters.isStartWithSavePoint` is explicitly passed into
`SubmitJobOperation`, the value can become ineffective if the lower-level logic
only derives the restore state from `jobImmutableInformation`.
##########
seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/job/JobImmutableInformation.java:
##########
@@ -167,6 +168,14 @@ public boolean isRestoreJob() {
return restoreMode != null && restoreMode.isRestore();
}
+ public boolean isSavepointRestore() {
Review Comment:
I’m a bit puzzled by this scenario: since `JobImmutableInformation` gets
serialized for storage, if a user upgrades via a `savepoint`, after
deserialization these flags would all be false. How is backward compatibility
currently handled in that case?
--
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]
