[hotfix] Reduce logging verbosity from the checkpoint-related REST ha…

…ndlers

Before the change, everytime the REST UI tried to access the checkpointing statistics, an ERROR message was logged.

When using a BATCH job, this leads to log files looking like this:

2020-11-13 15:27:38,785 ERROR org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointConfigHandler [] - Exception occurred in REST handler: Checkpointing is not enabled for this job (6df1c639d1904f8b7a54cf6a649ab567).
2020-11-13 15:27:38,788 ERROR org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointConfigHandler [] - Exception occurred in REST handler: Checkpointing is not enabled for this job (6df1c639d1904f8b7a54cf6a649ab567).
2020-11-13 15:27:38,788 ERROR org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointingStatisticsHandler [] - Exception occurred in REST handler: Checkpointing has not been enabled.
2020-11-13 15:27:38,793 ERROR org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointingStatisticsHandler [] - Exception occurred in REST handler: Checkpointing has not been enabled.
2020-11-13 15:27:38,793 ERROR org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointConfigHandler [] - Exception occurred in REST handler: Checkpointing is not enabled for this job (6df1c639d1904f8b7a54cf6a649ab567).
2020-11-13 15:27:38,797 ERROR org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointingStatisticsHandler [] - Exception occurred in REST handler: Checkpointing has not been enabled.
2020-11-13 15:27:38,797 ERROR org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointConfigHandler [] - Exception occurred in REST handler: Checkpointing is not enabled for this job (6df1c639d1904f8b7a54cf6a649ab567).
2020-11-13 15:27:38,802 ERROR org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointingStatisticsHandler [] - Exception occurred in REST handler: Checkpointing has not been enabled.

With this change, these log messages are surpressed.

flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractHandler.java

@@ -223,7 +223,7 @@ private CompletableFuture<Void> handleException(Throwable throwable, ChannelHand
  String truncatedStackTrace = Ascii.truncate(stackTrace, maxLength, "...");
  if (log.isDebugEnabled()) {
  log.error("Exception occurred in REST handler.", rhe);
- } else {
+ } else if (rhe.logException()) {
  log.error("Exception occurred in REST handler: {}", rhe.getMessage());
  }
  return HandlerUtils.sendErrorResponse(

flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/RestHandlerException.java

@@ -30,17 +30,39 @@ public class RestHandlerException extends FlinkException {
 
  private final int responseCode;
 
+ // This flag indicates whether the AbstractHandler should log about this exception on INFO level or not.
+ private final LoggingBehavior loggingBehavior;
+
  public RestHandlerException(String errorMessage, HttpResponseStatus httpResponseStatus) {
  super(errorMessage);
  this.responseCode = httpResponseStatus.code();
+ this.loggingBehavior = LoggingBehavior.LOG;
+ }
+
+ public RestHandlerException(String errorMessage, HttpResponseStatus httpResponseStatus, LoggingBehavior loggingBehavior) {
+ super(errorMessage);
+ this.responseCode = httpResponseStatus.code();
+ this.loggingBehavior = loggingBehavior;
  }
 
  public RestHandlerException(String errorMessage, HttpResponseStatus httpResponseStatus, Throwable cause) {
  super(errorMessage, cause);
  this.responseCode = httpResponseStatus.code();
+ this.loggingBehavior = LoggingBehavior.LOG;
  }
 
  public HttpResponseStatus getHttpResponseStatus() {
  return HttpResponseStatus.valueOf(responseCode);
  }
+
+ public boolean logException() {
+ return LoggingBehavior.LOG == loggingBehavior;
+ }
+
+ /**
+ * Enum to control logging behavior of RestHandlerExceptions.
+ */
+ public enum LoggingBehavior {
+ LOG, IGNORE
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointConfigHandler.java

@@ -92,7 +92,7 @@ private static CheckpointConfigInfo createCheckpointConfigInfo(AccessExecutionGr
  if (checkpointCoordinatorConfiguration == null) {
  throw new RestHandlerException(
  "Checkpointing is not enabled for this job (" + executionGraph.getJobID() + ").",
- HttpResponseStatus.NOT_FOUND);
+ HttpResponseStatus.NOT_FOUND, RestHandlerException.LoggingBehavior.IGNORE);
  } else {
  CheckpointRetentionPolicy retentionPolicy = checkpointCoordinatorConfiguration.getCheckpointRetentionPolicy();
 

flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointingStatisticsHandler.java

@@ -91,7 +91,7 @@ private static CheckpointingStatistics createCheckpointingStatistics(AccessExecu
  final CheckpointStatsSnapshot checkpointStatsSnapshot = executionGraph.getCheckpointStatsSnapshot();
 
  if (checkpointStatsSnapshot == null) {
- throw new RestHandlerException("Checkpointing has not been enabled.", HttpResponseStatus.NOT_FOUND);
+ throw new RestHandlerException("Checkpointing has not been enabled.", HttpResponseStatus.NOT_FOUND, RestHandlerException.LoggingBehavior.IGNORE);
  } else {
  final CheckpointStatsCounts checkpointStatsCounts = checkpointStatsSnapshot.getCounts();