tillrohrmann commented on a change in pull request #14028:
URL: https://github.com/apache/flink/pull/14028#discussion_r529595194
##########
File path:
flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
##########
@@ -342,7 +343,7 @@ public void run() {
runner.join();
final Throwable t = errorRef.get();
- if (t != null) {
+ if (t != null && ((JobExecutionException) t).getStatus() ==
ApplicationStatus.FAILED) {
Review comment:
Same here?
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/client/JobExecutionException.java
##########
@@ -19,41 +19,67 @@
package org.apache.flink.runtime.client;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.clusterframework.ApplicationStatus;
import org.apache.flink.util.FlinkException;
/**
* This exception is the base exception for all exceptions that denote any
failure during
- * the execution of a job.
+ * the execution of a job, and signal the failure of an application with a
given
+ * {@link ApplicationStatus}.
*/
public class JobExecutionException extends FlinkException {
private static final long serialVersionUID = 2818087325120827525L;
private final JobID jobID;
+ private final ApplicationStatus status;
+
/**
* Constructs a new job execution exception.
*
- * @param jobID The job's ID.
- * @param msg The cause for the execution exception.
- * @param cause The cause of the exception
+ * @param jobID The job's ID.
+ * @param status The application status.
+ * @param msg The cause for the execution exception.
+ * @param cause The cause of the exception
*/
- public JobExecutionException(JobID jobID, String msg, Throwable cause) {
+ public JobExecutionException(JobID jobID, ApplicationStatus status,
String msg, Throwable cause) {
super(msg, cause);
this.jobID = jobID;
+ this.status = status;
Review comment:
Shouldn't we assert that `status` is not `SUCCEEDED`?
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraphBuilder.java
##########
@@ -221,7 +222,7 @@ public static ExecutionGraph buildGraph(
vertex.initializeOnMaster(classLoader);
}
catch (Throwable t) {
- throw new JobExecutionException(jobId,
+ throw new JobExecutionException(jobId,
ApplicationStatus.fromJobStatus(executionGraph.getState()),
Review comment:
I think it is not correct to already decide on the `ApplicationStatus`
here. If the initialization of the job fails, we might still be able to retry
it. Maybe it would be better to not use the `JobExecutionException` across the
client and server. Maybe it should stay a client side concept which is only
thrown by the client when obtaining the job result.
##########
File path:
flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterClientJobClientAdapter.java
##########
@@ -116,8 +115,7 @@ public JobID getJobID() {
try {
return
jobResult.toJobExecutionResult(classLoader);
} catch (Throwable t) {
- throw new
CompletionException(
- new
ProgramInvocationException("Job failed", jobID, t));
Review comment:
Aren't we changing here part of the public API by changing the return
type of the exceptions? How much of a contract is it that the user of the
JobClient expects an `ProgramInvocationException` @aljoscha?
##########
File path:
flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
##########
@@ -257,7 +258,7 @@ public void run() {
runner.join();
final Throwable t = errorRef.get();
- if (t != null) {
+ if (t != null && ((JobExecutionException) t).getStatus() ==
ApplicationStatus.FAILED) {
Review comment:
Why is this change necessary?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
