[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to
programmatically execute streaming pipeline with savepoint restore
URL: https://github.com/apache/flink/pull/7249#discussion_r239453780
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
##
@@ -169,39 +170,46 @@ public RemoteStreamEnvironment(String host, int port,
Configuration clientConfig
}
}
- @Override
- public JobExecutionResult execute(String jobName) throws
ProgramInvocationException {
- StreamGraph streamGraph = getStreamGraph();
- streamGraph.setJobName(jobName);
- transformations.clear();
- return executeRemotely(streamGraph, jarFiles);
+ protected List getJarFiles() throws ProgramInvocationException {
+ return jarFiles;
}
/**
* Executes the remote job.
*
-* @param streamGraph
-*Stream Graph to execute
+* @param streamExecutionEnvironment
+*Execution Environment with Stream Graph to execute
* @param jarFiles
*List of jar file URLs to ship to the cluster
* @return The result of the job execution, containing elapsed time and
accumulators.
*/
- protected JobExecutionResult executeRemotely(StreamGraph streamGraph,
List jarFiles) throws ProgramInvocationException {
+ public static JobExecutionResult
executeRemotely(StreamExecutionEnvironment streamExecutionEnvironment,
Review comment:
Question: Why don't we pass the required dependencies directly here? i.e.
`StreamGraph` and `ExecutionConfig`
IMHO there is no need to pass the entire environment.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to
programmatically execute streaming pipeline with savepoint restore
URL: https://github.com/apache/flink/pull/7249#discussion_r239451677
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
##
@@ -169,39 +170,46 @@ public RemoteStreamEnvironment(String host, int port,
Configuration clientConfig
}
}
- @Override
- public JobExecutionResult execute(String jobName) throws
ProgramInvocationException {
- StreamGraph streamGraph = getStreamGraph();
- streamGraph.setJobName(jobName);
- transformations.clear();
- return executeRemotely(streamGraph, jarFiles);
+ protected List getJarFiles() throws ProgramInvocationException {
+ return jarFiles;
}
/**
* Executes the remote job.
*
-* @param streamGraph
-*Stream Graph to execute
+* @param streamExecutionEnvironment
+*Execution Environment with Stream Graph to execute
* @param jarFiles
*List of jar file URLs to ship to the cluster
* @return The result of the job execution, containing elapsed time and
accumulators.
*/
- protected JobExecutionResult executeRemotely(StreamGraph streamGraph,
List jarFiles) throws ProgramInvocationException {
+ public static JobExecutionResult
executeRemotely(StreamExecutionEnvironment streamExecutionEnvironment,
+
List jarFiles,
Review comment:
The "most readable option" depends a lot on preference :) But let's stick
with what is commonly used nowadays.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to
programmatically execute streaming pipeline with savepoint restore
URL: https://github.com/apache/flink/pull/7249#discussion_r239451179
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
##
@@ -169,39 +170,46 @@ public RemoteStreamEnvironment(String host, int port,
Configuration clientConfig
}
}
- @Override
- public JobExecutionResult execute(String jobName) throws
ProgramInvocationException {
- StreamGraph streamGraph = getStreamGraph();
- streamGraph.setJobName(jobName);
- transformations.clear();
- return executeRemotely(streamGraph, jarFiles);
+ protected List getJarFiles() throws ProgramInvocationException {
+ return jarFiles;
}
/**
* Executes the remote job.
*
-* @param streamGraph
-*Stream Graph to execute
+* @param streamExecutionEnvironment
+*Execution Environment with Stream Graph to execute
* @param jarFiles
*List of jar file URLs to ship to the cluster
* @return The result of the job execution, containing elapsed time and
accumulators.
*/
- protected JobExecutionResult executeRemotely(StreamGraph streamGraph,
List jarFiles) throws ProgramInvocationException {
+ public static JobExecutionResult
executeRemotely(StreamExecutionEnvironment streamExecutionEnvironment,
Review comment:
I agree that we can't do this. The method is `protected` but it would still
break subclasses relying on the `@Public` contract.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to
programmatically execute streaming pipeline with savepoint restore
URL: https://github.com/apache/flink/pull/7249#discussion_r239452431
##
File path:
flink-scala-shell/src/main/java/org/apache/flink/api/java/ScalaShellRemoteStreamEnvironment.java
##
@@ -68,30 +67,21 @@ public ScalaShellRemoteStreamEnvironment(
this.flinkILoop = flinkILoop;
}
- /**
-* Executes the remote job.
-*
-* @param streamGraph
-*Stream Graph to execute
-* @param jarFiles
-*List of jar file URLs to ship to the cluster
-* @return The result of the job execution, containing elapsed time and
accumulators.
-*/
- @Override
- protected JobExecutionResult executeRemotely(StreamGraph streamGraph,
List jarFiles) throws ProgramInvocationException {
+ protected List getJarFiles() throws ProgramInvocationException {
URL jarUrl;
try {
jarUrl =
flinkILoop.writeFilesToDisk().getAbsoluteFile().toURI().toURL();
} catch (MalformedURLException e) {
+ // TODO: we don't have the actual jobID at this point
throw new ProgramInvocationException("Could not write
the user code classes to disk.",
- streamGraph.getJobGraph().getJobID(), e);
+ new JobID(), e);
Review comment:
That will generate a random `JobID`, not sure about this. Let's use the
constructor without the JobID.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to
programmatically execute streaming pipeline with savepoint restore
URL: https://github.com/apache/flink/pull/7249#discussion_r239451409
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
##
@@ -169,39 +170,46 @@ public RemoteStreamEnvironment(String host, int port,
Configuration clientConfig
}
}
- @Override
- public JobExecutionResult execute(String jobName) throws
ProgramInvocationException {
- StreamGraph streamGraph = getStreamGraph();
- streamGraph.setJobName(jobName);
- transformations.clear();
- return executeRemotely(streamGraph, jarFiles);
+ protected List getJarFiles() throws ProgramInvocationException {
+ return jarFiles;
}
/**
* Executes the remote job.
*
-* @param streamGraph
-*Stream Graph to execute
+* @param streamExecutionEnvironment
+*Execution Environment with Stream Graph to execute
* @param jarFiles
*List of jar file URLs to ship to the cluster
* @return The result of the job execution, containing elapsed time and
accumulators.
*/
- protected JobExecutionResult executeRemotely(StreamGraph streamGraph,
List jarFiles) throws ProgramInvocationException {
+ public static JobExecutionResult
executeRemotely(StreamExecutionEnvironment streamExecutionEnvironment,
Review comment:
If we make this `static` it will also break the `@Public` contract. We can
introduce a new static method.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to
programmatically execute streaming pipeline with savepoint restore
URL: https://github.com/apache/flink/pull/7249#discussion_r239501341
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
##
@@ -169,39 +170,46 @@ public RemoteStreamEnvironment(String host, int port,
Configuration clientConfig
}
}
- @Override
- public JobExecutionResult execute(String jobName) throws
ProgramInvocationException {
- StreamGraph streamGraph = getStreamGraph();
- streamGraph.setJobName(jobName);
- transformations.clear();
- return executeRemotely(streamGraph, jarFiles);
+ protected List getJarFiles() throws ProgramInvocationException {
+ return jarFiles;
}
/**
* Executes the remote job.
*
-* @param streamGraph
-*Stream Graph to execute
+* @param streamExecutionEnvironment
+*Execution Environment with Stream Graph to execute
* @param jarFiles
*List of jar file URLs to ship to the cluster
* @return The result of the job execution, containing elapsed time and
accumulators.
*/
- protected JobExecutionResult executeRemotely(StreamGraph streamGraph,
List jarFiles) throws ProgramInvocationException {
+ public static JobExecutionResult
executeRemotely(StreamExecutionEnvironment streamExecutionEnvironment,
Review comment:
StreamGraph is exposed anyway because there is
`StreamExecutionEnvironment.getStreamGraph()` though I see it is marked
`@Internal`. I'm fine also with passing the environment.
>IMO any use-case where usage of the StreamGraph is appropriate (for which
tbh there are none as the JobGraph would be a better choice anyway) is already
a pretty low-level thing in which case you can work directly against a
ClusterClient.
So you are suggesting to move the code here to `ClusterClient`? I think that
would make sense.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to
programmatically execute streaming pipeline with savepoint restore
URL: https://github.com/apache/flink/pull/7249#discussion_r239777131
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
##
@@ -169,39 +170,46 @@ public RemoteStreamEnvironment(String host, int port,
Configuration clientConfig
}
}
- @Override
- public JobExecutionResult execute(String jobName) throws
ProgramInvocationException {
- StreamGraph streamGraph = getStreamGraph();
- streamGraph.setJobName(jobName);
- transformations.clear();
- return executeRemotely(streamGraph, jarFiles);
+ protected List getJarFiles() throws ProgramInvocationException {
+ return jarFiles;
}
/**
* Executes the remote job.
*
-* @param streamGraph
-*Stream Graph to execute
+* @param streamExecutionEnvironment
+*Execution Environment with Stream Graph to execute
* @param jarFiles
*List of jar file URLs to ship to the cluster
* @return The result of the job execution, containing elapsed time and
accumulators.
*/
- protected JobExecutionResult executeRemotely(StreamGraph streamGraph,
List jarFiles) throws ProgramInvocationException {
+ public static JobExecutionResult
executeRemotely(StreamExecutionEnvironment streamExecutionEnvironment,
Review comment:
@tweise The contract for `@Public` is that there won't be any breaking
changes to code using the the class. This also applies to subclasses, although
I get your point that `protected` is not as visible as `public`.
I don't think a setter on the environment is very nice. Savepoint restore
should be configured per execution. So let's just add another method here,
shall we?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to
programmatically execute streaming pipeline with savepoint restore
URL: https://github.com/apache/flink/pull/7249#discussion_r240559321
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
##
@@ -169,33 +173,66 @@ public RemoteStreamEnvironment(String host, int port,
Configuration clientConfig
}
}
- @Override
- public JobExecutionResult execute(String jobName) throws
ProgramInvocationException {
- StreamGraph streamGraph = getStreamGraph();
- streamGraph.setJobName(jobName);
- transformations.clear();
- return executeRemotely(streamGraph, jarFiles);
+ /**
+* Set savepoint restore settings that will be used when executing the
job.
+* @param savepointRestoreSettings savepoint restore settings
+*/
+ public void setSavepointRestoreSettings(SavepointRestoreSettings
savepointRestoreSettings) {
+ this.savepointRestoreSettings = savepointRestoreSettings;
Review comment:
Could we remove this setter? Savepoint restore should be configured
per-execution basis and not for the lifetime of the `RemoteStreamEnvironment`.
It is not necessary because we have an addtional `executeRemotely` method
now which takes the `SavepointRestoreSettings`. If you want to avoid too many
parameters, add an additional non-static method which takes the savepoint
settings.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to
programmatically execute streaming pipeline with savepoint restore
URL: https://github.com/apache/flink/pull/7249#discussion_r240560359
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
##
@@ -169,39 +170,46 @@ public RemoteStreamEnvironment(String host, int port,
Configuration clientConfig
}
}
- @Override
- public JobExecutionResult execute(String jobName) throws
ProgramInvocationException {
- StreamGraph streamGraph = getStreamGraph();
- streamGraph.setJobName(jobName);
- transformations.clear();
- return executeRemotely(streamGraph, jarFiles);
+ protected List getJarFiles() throws ProgramInvocationException {
+ return jarFiles;
}
/**
* Executes the remote job.
*
-* @param streamGraph
-*Stream Graph to execute
+* @param streamExecutionEnvironment
+*Execution Environment with Stream Graph to execute
* @param jarFiles
*List of jar file URLs to ship to the cluster
* @return The result of the job execution, containing elapsed time and
accumulators.
*/
- protected JobExecutionResult executeRemotely(StreamGraph streamGraph,
List jarFiles) throws ProgramInvocationException {
+ public static JobExecutionResult
executeRemotely(StreamExecutionEnvironment streamExecutionEnvironment,
Review comment:
Looks good but please see my comment above concerning the setter.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to
programmatically execute streaming pipeline with savepoint restore
URL: https://github.com/apache/flink/pull/7249#discussion_r240976627
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
##
@@ -283,6 +275,14 @@ public JobExecutionResult execute(String jobName) throws
ProgramInvocationExcept
return executeRemotely(streamGraph, jarFiles);
}
+ /**
+* Execute the job with savepoint restore.
+*/
+ public JobExecutionResult execute(String jobName,
SavepointRestoreSettings savepointRestoreSettings) throws
ProgramInvocationException {
+ this.savepointRestoreSettings = savepointRestoreSettings;
+ return execute(jobName);
Review comment:
This would still be an implicit setter. I don't think the settings should be
persisted across executions. I would also remove the field.
Could you just call `executeRemotely` directly here with the settings as
parameter?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore URL: https://github.com/apache/flink/pull/7249#discussion_r241822916 ## File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java ## @@ -65,6 +67,8 @@ /** The classpaths that need to be attached to each job. */ private final List globalClasspaths; + private SavepointRestoreSettings savepointRestoreSettings; Review comment: Could you document this field? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to
programmatically execute streaming pipeline with savepoint restore
URL: https://github.com/apache/flink/pull/7249#discussion_r241825068
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
##
@@ -283,6 +275,14 @@ public JobExecutionResult execute(String jobName) throws
ProgramInvocationExcept
return executeRemotely(streamGraph, jarFiles);
}
+ /**
+* Execute the job with savepoint restore.
+*/
+ public JobExecutionResult execute(String jobName,
SavepointRestoreSettings savepointRestoreSettings) throws
ProgramInvocationException {
+ this.savepointRestoreSettings = savepointRestoreSettings;
+ return execute(jobName);
Review comment:
I understand that it is one-time, but I would still prefer to not have this
side effect here. Could you either 1) make the SavepointSettings a `final`
field and initialize it in the constructor or 2) provide the settings here as a
parameter which is not stored in a field? Sorry for the hassle.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to
programmatically execute streaming pipeline with savepoint restore
URL: https://github.com/apache/flink/pull/7249#discussion_r242276315
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
##
@@ -169,39 +170,46 @@ public RemoteStreamEnvironment(String host, int port,
Configuration clientConfig
}
}
- @Override
- public JobExecutionResult execute(String jobName) throws
ProgramInvocationException {
- StreamGraph streamGraph = getStreamGraph();
- streamGraph.setJobName(jobName);
- transformations.clear();
- return executeRemotely(streamGraph, jarFiles);
+ protected List getJarFiles() throws ProgramInvocationException {
+ return jarFiles;
}
/**
* Executes the remote job.
*
-* @param streamGraph
-*Stream Graph to execute
+* @param streamExecutionEnvironment
+*Execution Environment with Stream Graph to execute
* @param jarFiles
*List of jar file URLs to ship to the cluster
* @return The result of the job execution, containing elapsed time and
accumulators.
*/
- protected JobExecutionResult executeRemotely(StreamGraph streamGraph,
List jarFiles) throws ProgramInvocationException {
+ public static JobExecutionResult
executeRemotely(StreamExecutionEnvironment streamExecutionEnvironment,
Review comment:
Do we still need this new `static` method now that we have the constructor?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to
programmatically execute streaming pipeline with savepoint restore
URL: https://github.com/apache/flink/pull/7249#discussion_r242884359
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
##
@@ -169,33 +173,66 @@ public RemoteStreamEnvironment(String host, int port,
Configuration clientConfig
}
}
- @Override
- public JobExecutionResult execute(String jobName) throws
ProgramInvocationException {
- StreamGraph streamGraph = getStreamGraph();
- streamGraph.setJobName(jobName);
- transformations.clear();
- return executeRemotely(streamGraph, jarFiles);
+ /**
+* Set savepoint restore settings that will be used when executing the
job.
+* @param savepointRestoreSettings savepoint restore settings
+*/
+ public void setSavepointRestoreSettings(SavepointRestoreSettings
savepointRestoreSettings) {
+ this.savepointRestoreSettings = savepointRestoreSettings;
Review comment:
To be fair the code above has been changed, it's not a setter anymore but
set in the constructor. You summed up quite nicely what is actually the state
of this PR now. Nothing is broken in terms of backwards-compatibility but the
structure changed a bit which makes it hard to review. The only difference is
that an additional constructor has been added for the savepoint settings
instead of this method you proposed:
```
public JobExecutionResult execute(String jobName, SavepointRestoreSettings
savepointRestoreSettings) throws ProgramInvocationException {
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to
programmatically execute streaming pipeline with savepoint restore
URL: https://github.com/apache/flink/pull/7249#discussion_r242884502
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
##
@@ -169,33 +173,66 @@ public RemoteStreamEnvironment(String host, int port,
Configuration clientConfig
}
}
- @Override
- public JobExecutionResult execute(String jobName) throws
ProgramInvocationException {
- StreamGraph streamGraph = getStreamGraph();
- streamGraph.setJobName(jobName);
- transformations.clear();
- return executeRemotely(streamGraph, jarFiles);
+ /**
+* Set savepoint restore settings that will be used when executing the
job.
+* @param savepointRestoreSettings savepoint restore settings
+*/
+ public void setSavepointRestoreSettings(SavepointRestoreSettings
savepointRestoreSettings) {
+ this.savepointRestoreSettings = savepointRestoreSettings;
Review comment:
Unless you prefer the method instead of the constructor, then this should be
ready to be merged :)
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to
programmatically execute streaming pipeline with savepoint restore
URL: https://github.com/apache/flink/pull/7249#discussion_r242907203
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
##
@@ -169,33 +173,66 @@ public RemoteStreamEnvironment(String host, int port,
Configuration clientConfig
}
}
- @Override
- public JobExecutionResult execute(String jobName) throws
ProgramInvocationException {
- StreamGraph streamGraph = getStreamGraph();
- streamGraph.setJobName(jobName);
- transformations.clear();
- return executeRemotely(streamGraph, jarFiles);
+ /**
+* Set savepoint restore settings that will be used when executing the
job.
+* @param savepointRestoreSettings savepoint restore settings
+*/
+ public void setSavepointRestoreSettings(SavepointRestoreSettings
savepointRestoreSettings) {
+ this.savepointRestoreSettings = savepointRestoreSettings;
Review comment:
This argument was countered by the fact that the other parameters are also
not per-execution. I don't mind, both the constructor or the method work for me.
Perhaps we could change the constructor to the method above @tweise? I think
we should have reached consensus then :)
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] mxm commented on a change in pull request #7249: [FLINK-11048] Ability to programmatically execute streaming pipeline with savepoint restore
mxm commented on a change in pull request #7249: [FLINK-11048] Ability to
programmatically execute streaming pipeline with savepoint restore
URL: https://github.com/apache/flink/pull/7249#discussion_r243359061
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
##
@@ -233,6 +301,36 @@ protected JobExecutionResult executeRemotely(StreamGraph
streamGraph, List
}
}
+ @Override
+ public JobExecutionResult execute(String jobName) throws
ProgramInvocationException {
+ StreamGraph streamGraph = getStreamGraph();
+ streamGraph.setJobName(jobName);
+ transformations.clear();
+ return executeRemotely(streamGraph, jarFiles);
+ }
+
+ /**
+* Executes the remote job.
+*
+* Note: This method exposes stream graph internal in the public
API, but cannot be removed for backward compatibility.
+* @param streamGraph
+*Stream Graph to execute
+* @param jarFiles
+*List of jar file URLs to ship to the cluster
+* @return The result of the job execution, containing elapsed time and
accumulators.
+*/
+ protected JobExecutionResult executeRemotely(StreamGraph streamGraph,
List jarFiles) throws ProgramInvocationException {
Review comment:
+1 Let's mark it deprecated.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
