[GitHub] [flink] curcur commented on a change in pull request #12525: [FLINK-17769] [runtime] Fix the order of log events on a task failure

[GitBox <gi...@apache.org>]

curcur commented on a change in pull request #12525:
URL: https://github.com/apache/flink/pull/12525#discussion_r437299439



##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -702,11 +704,14 @@ private void disposeAllOperators(boolean logOnlyErrors) throws Exception {
 						operator.dispose();
 					}
 					catch (Exception e) {
-						LOG.error("Error during disposal of stream operator.", e);
+						disposalException = ExceptionUtils.firstOrSuppressed(e, disposalException);

Review comment:
       > Isn't `e` and thus `newException` always non-null?
   
   yep, i agree. I think the assumption here is you want to suppress an exception only if you catch a real exception. But I guess no hurt to add the check.

##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -702,11 +704,14 @@ private void disposeAllOperators(boolean logOnlyErrors) throws Exception {
 						operator.dispose();
 					}
 					catch (Exception e) {
-						LOG.error("Error during disposal of stream operator.", e);
+						disposalException = ExceptionUtils.firstOrSuppressed(e, disposalException);

Review comment:
       > Isn't `e` and thus `newException` always non-null?
   
   yep, i agree. I think the assumption here is you want to suppress an exception only if you catch a real exception.

##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -702,11 +704,14 @@ private void disposeAllOperators(boolean logOnlyErrors) throws Exception {
 						operator.dispose();
 					}
 					catch (Exception e) {
-						LOG.error("Error during disposal of stream operator.", e);
+						disposalException = ExceptionUtils.firstOrSuppressed(e, disposalException);

Review comment:
       > Isn't `e` and thus `newException` always non-null?
   
   yep, i agree. I think the assumption here is suppressing an exception only if catching a real exception. And the `checkNotNull(newException, "newException");'
   explicitly requires the newException non-null.

##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -702,11 +704,14 @@ private void disposeAllOperators(boolean logOnlyErrors) throws Exception {
 						operator.dispose();
 					}
 					catch (Exception e) {
-						LOG.error("Error during disposal of stream operator.", e);
+						disposalException = ExceptionUtils.firstOrSuppressed(e, disposalException);

Review comment:
       > Isn't `e` and thus `newException` always non-null?
   
   yep, i agree. I think the assumption here is suppressing an exception only if catching a real exception. And the `checkNotNull(newException, "newException");`
   explicitly requires the newException non-null.

##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -692,6 +693,7 @@ private void shutdownAsyncThreads() throws Exception {
 	 */
 	private void disposeAllOperators(boolean logOnlyErrors) throws Exception {
 		if (operatorChain != null && !disposedOperators) {
+			Exception disposalException = null;
 			for (StreamOperatorWrapper<?, ?> operatorWrapper : operatorChain.getAllOperators(true)) {
 				StreamOperator<?> operator = operatorWrapper.getStreamOperator();
 				if (!logOnlyErrors) {

Review comment:
       > I think this `!logOnlyErrors` branch could also gain from your change.
   > Currently, it disposes operators until 1st failure.
   
   Yes, I agree. However, I was a little afraid to make the change from the safety perspective. It changes the behavior of the code, like what exception to throw, and the exit procedure.
   
   For example, in some cases, the system is preferring a faster crash as soon as the first disposal occurs (it is just an example, not saying it would be like that).

##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -702,11 +704,14 @@ private void disposeAllOperators(boolean logOnlyErrors) throws Exception {
 						operator.dispose();
 					}
 					catch (Exception e) {
-						LOG.error("Error during disposal of stream operator.", e);
+						disposalException = ExceptionUtils.firstOrSuppressed(e, disposalException);
 					}
 				}
 			}
 			disposedOperators = true;
+			if (disposalException != null) {
+				throw disposalException;

Review comment:
       That's true, but what gonna happen to the suppressed exception? You still need to throw it otherwise the exception is swallowed I guess.
   
   cc @pnowojski
   
   Is `disposeAllOperators(true);` designed not to throw any exception just force shut down? If that's the case, I guess the best way is just logging saying that `dispose error is not the rootcause?' 

##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -537,6 +537,7 @@ public final void invoke() throws Exception {
 			afterInvoke();

Review comment:
       This basically says we want to log the first encountered disposal error.
   
   In the catch block, we do an enforce one and clean up no matter what. So I guess `disposeAllOperators(true)` was intended to not throw exceptions, which I guess the proposed option 3 (logging "disposal error is not the root cause") might be a 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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org