From 8dca9fa852c72984ac873eae9a96bbd739e502f3 Mon Sep 17 00:00:00 2001 From: Robert Metzger Date: Fri, 13 Nov 2020 15:37:24 +0100 Subject: [PATCH] [hotfix] Reduce logging verbosity from the checkpoint-related REST handlers 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. --- .../runtime/rest/handler/AbstractHandler.java | 2 +- .../rest/handler/RestHandlerException.java | 22 +++++++++++++++++++ .../checkpoints/CheckpointConfigHandler.java | 2 +- .../CheckpointingStatisticsHandler.java | 2 +- 4 files changed, 25 insertions(+), 3 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractHandler.java index 1f0b248fe4856..18d821fb875f9 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractHandler.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractHandler.java @@ -223,7 +223,7 @@ private CompletableFuture 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( diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/RestHandlerException.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/RestHandlerException.java index 7ae8939cb8ca5..e19750faa6a64 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/RestHandlerException.java +++ b/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 + } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointConfigHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointConfigHandler.java index 7e6f14dbd2ad9..137276eb1d641 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointConfigHandler.java +++ b/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(); diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointingStatisticsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointingStatisticsHandler.java index 06f9285af8749..eb5439028cbe5 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointingStatisticsHandler.java +++ b/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();