diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/HttpErrorLogSeverity.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/HttpErrorLogSeverity.java new file mode 100644 index 00000000..dc4170cc --- /dev/null +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/HttpErrorLogSeverity.java @@ -0,0 +1,43 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connector.http; + +/** Defines the SLF4J severity level for HTTP error logging. */ +public enum HttpErrorLogSeverity { + /** No error-level logging (DEBUG only, current behavior). */ + OFF, + /** Log errors at INFO level. */ + INFO, + /** Log errors at WARN level. */ + WARN, + /** Log errors at ERROR level (default for production). */ + ERROR; + + public static HttpErrorLogSeverity fromString(String level) { + if (level == null) { + return ERROR; + } + try { + return HttpErrorLogSeverity.valueOf(level.toUpperCase()); + } catch (IllegalArgumentException e) { + return ERROR; + } + } +} diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/HttpLogger.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/HttpLogger.java index b4a0e36d..ca4a8b54 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/HttpLogger.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/HttpLogger.java @@ -31,22 +31,28 @@ import java.util.Properties; import java.util.StringJoiner; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.ERROR_LOG_SEVERITY; import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.HTTP_LOGGING_LEVEL; /** * HttpLogger, this is a class to perform HTTP content logging based on a level defined in - * configuration. + * configuration. It also handles error logging with configurable severity levels. */ @Slf4j public class HttpLogger implements Serializable { private static final long serialVersionUID = 1L; + private static final int DEFAULT_MAX_BODY_SIZE = 1024; private final HttpLoggingLevelType httpLoggingLevelType; + private final HttpErrorLogSeverity errorLogSeverity; private HttpLogger(Properties properties) { String code = (String) properties.get(HTTP_LOGGING_LEVEL); this.httpLoggingLevelType = HttpLoggingLevelType.valueOfStr(code); + + String severityStr = (String) properties.get(ERROR_LOG_SEVERITY); + this.errorLogSeverity = HttpErrorLogSeverity.fromString(severityStr); } public static HttpLogger getHttpLogger(Properties properties) { @@ -60,7 +66,6 @@ public void logRequest(HttpRequest httpRequest) { } public void logResponse(HttpResponse response) { - if (log.isDebugEnabled()) { log.debug(createStringForResponse(response)); } @@ -78,6 +83,234 @@ public void logExceptionResponse(HttpLookupSourceRequestEntry request, Exception } } + /** + * Log HTTP lookup error with detailed context. + * + * @param request The HTTP request + * @param e The exception that occurred + * @param retryAttempt The retry attempt number (0 for first attempt) + * @param continueOnError Whether the error is tolerated (continue-on-error mode) + */ + public void logLookupError( + HttpRequest request, Exception e, int retryAttempt, boolean continueOnError) { + // If continue-on-error is true, always use DEBUG (error is tolerated) + if (continueOnError) { + if (log.isDebugEnabled()) { + log.debug( + "HTTP Lookup Error (tolerated) - Attempt {}: Method: {}, URL: {}, Exception: {}, Message: {}", + retryAttempt, + request.method(), + request.uri(), + e.getClass().getSimpleName(), + e.getMessage()); + } + return; + } + + // For actual errors, respect http.error.log.severity + String message = formatLookupErrorMessage(request, null, e, retryAttempt); + logWithSeverity(message); + } + + /** + * Log HTTP lookup error with response details. + * + * @param request The HTTP request + * @param response The HTTP response (may be null if error occurred before response) + * @param e The exception that occurred + * @param retryAttempt The retry attempt number + * @param continueOnError Whether the error is tolerated (continue-on-error mode) + */ + public void logLookupError( + HttpRequest request, + HttpResponse response, + Exception e, + int retryAttempt, + boolean continueOnError) { + + // If continue-on-error is true, always use DEBUG (error is tolerated) + if (continueOnError) { + if (log.isDebugEnabled()) { + log.debug( + "HTTP Lookup Error (tolerated) - Attempt {}: Method: {}, URL: {}, Exception: {}, Message: {}, Response Status: {}", + retryAttempt, + request.method(), + request.uri(), + e.getClass().getSimpleName(), + e.getMessage(), + response != null ? response.statusCode() : "N/A"); + } + return; + } + + // For actual errors, respect http.error.log.severity + String message = formatLookupErrorMessage(request, response, e, retryAttempt); + logWithSeverity(message); + } + + /** + * Log HTTP sink error with detailed context. + * + * @param request The HTTP request + * @param requestBody The request body (may be null) + * @param e The exception that occurred + * @param retryAttempt The retry attempt number + */ + public void logSinkError( + HttpRequest request, String requestBody, Exception e, int retryAttempt) { + String message = formatSinkErrorMessage(request, requestBody, null, e, retryAttempt); + logWithSeverity(message); + } + + /** + * Log HTTP sink error for non-successful status code responses. + * + * @param request The HTTP request + * @param requestBody The request body (may be null) + * @param response The HTTP response with error status code + * @param retryAttempt The retry attempt number + */ + public void logSinkError( + HttpRequest request, + String requestBody, + HttpResponse response, + int retryAttempt) { + String message = formatSinkErrorMessage(request, requestBody, response, null, retryAttempt); + logWithSeverity(message); + } + + /** + * Log HTTP sink error with response details. + * + * @param request The HTTP request + * @param requestBody The request body (may be null) + * @param response The HTTP response + * @param e The exception that occurred + * @param retryAttempt The retry attempt number + */ + public void logSinkError( + HttpRequest request, + String requestBody, + HttpResponse response, + Exception e, + int retryAttempt) { + String message = formatSinkErrorMessage(request, requestBody, response, e, retryAttempt); + logWithSeverity(message); + } + + private String formatLookupErrorMessage( + HttpRequest request, HttpResponse response, Exception e, int retryAttempt) { + StringBuilder message = + new StringBuilder( + String.format( + "HTTP Lookup Error - Attempt %d: Method: %s, URL: %s, Exception: %s, Message: %s", + retryAttempt, + request.method(), + request.uri(), + e.getClass().getSimpleName(), + e.getMessage())); + + // Add response details if available (based on logging level) + if (response != null) { + message.append(String.format(", Response Status: %d", response.statusCode())); + + if (httpLoggingLevelType != HttpLoggingLevelType.MIN) { + message.append( + String.format( + ", Response Headers: %s", getHeadersForLog(response.headers()))); + + if (httpLoggingLevelType == HttpLoggingLevelType.MAX && response.body() != null) { + message.append( + String.format( + ", Response Body: %s", + truncateBody(response.body().toString()))); + } + } + } + + // Add request headers based on logging level + if (httpLoggingLevelType != HttpLoggingLevelType.MIN) { + message.append( + String.format(", Request Headers: %s", getHeadersForLog(request.headers()))); + } + + return message.toString(); + } + + private String formatSinkErrorMessage( + HttpRequest request, + String requestBody, + HttpResponse response, + Exception e, + int retryAttempt) { + StringBuilder message = new StringBuilder(); + + if (e != null) { + message.append( + String.format( + "HTTP Sink Error - Attempt %d: Method: %s, URL: %s, Exception: %s, Message: %s", + retryAttempt, + request.method(), + request.uri(), + e.getClass().getSimpleName(), + e.getMessage())); + } else { + message.append( + String.format( + "HTTP Sink Error - Attempt %d: Method: %s, URL: %s", + retryAttempt, request.method(), request.uri())); + } + + // Add response details if available (based on logging level) + if (response != null) { + message.append(String.format(", Response Status: %d", response.statusCode())); + + if (httpLoggingLevelType != HttpLoggingLevelType.MIN) { + message.append( + String.format( + ", Response Headers: %s", getHeadersForLog(response.headers()))); + + if (httpLoggingLevelType == HttpLoggingLevelType.MAX && response.body() != null) { + message.append( + String.format(", Response Body: %s", truncateBody(response.body()))); + } + } + } + + // Add request body based on logging level + if (httpLoggingLevelType == HttpLoggingLevelType.MAX && requestBody != null) { + message.append(String.format(", Request Body: %s", truncateBody(requestBody))); + } + + // Add request headers based on logging level + if (httpLoggingLevelType != HttpLoggingLevelType.MIN) { + message.append( + String.format(", Request Headers: %s", getHeadersForLog(request.headers()))); + } + + return message.toString(); + } + + private void logWithSeverity(String message) { + switch (errorLogSeverity) { + case ERROR: + log.error(message); + break; + case WARN: + log.warn(message); + break; + case INFO: + log.info(message); + break; + case OFF: + // No error-level logging, fall back to DEBUG + if (log.isDebugEnabled()) { + log.debug(message); + } + break; + } + } + String createStringForRequest(HttpRequest httpRequest) { String headersForLog = getHeadersForLog(httpRequest.headers()); return String.format( @@ -96,12 +329,22 @@ private String getHeadersForLog(HttpHeaders httpHeaders) { if (this.httpLoggingLevelType == HttpLoggingLevelType.MAX) { StringJoiner headers = new StringJoiner(";"); for (Map.Entry> reqHeaders : headersMap.entrySet()) { - StringJoiner values = new StringJoiner(";"); - for (String value : reqHeaders.getValue()) { - values.add(value); + String headerName = reqHeaders.getKey().toLowerCase(); + + // Mask sensitive headers + if (headerName.contains("authorization") + || headerName.contains("cookie") + || headerName.contains("api-key") + || headerName.contains("x-api-key")) { + headers.add(reqHeaders.getKey() + ":[***]"); + } else { + StringJoiner values = new StringJoiner(";"); + for (String value : reqHeaders.getValue()) { + values.add(value); + } + String header = reqHeaders.getKey() + ":[" + values + "]"; + headers.add(header); } - String header = reqHeaders.getKey() + ":[" + values + "]"; - headers.add(header); } return headers.toString(); } @@ -148,4 +391,14 @@ String createStringForBody(String body) { } return bodyForLog; } + + private String truncateBody(String body) { + if (body == null || body.isEmpty()) { + return "None"; + } + if (body.length() <= DEFAULT_MAX_BODY_SIZE) { + return body; + } + return body.substring(0, DEFAULT_MAX_BODY_SIZE) + "... (truncated)"; + } } diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpConnectorConfigConstants.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpConnectorConfigConstants.java index 6dfb414e..23e7c262 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpConnectorConfigConstants.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpConnectorConfigConstants.java @@ -164,4 +164,14 @@ public final class HttpConnectorConfigConstants { SOURCE_RETRY_EXP_DELAY_PREFIX + "max-backoff"; public static final String SOURCE_RETRY_EXP_DELAY_MULTIPLIER = SOURCE_RETRY_EXP_DELAY_PREFIX + "backoff-multiplier"; + + // -------------- Error logging configuration -------------- + + /** + * SLF4J severity level for HTTP error logging. Valid values: OFF (DEBUG only), INFO, WARN, + * ERROR (default). This controls the log level for actual errors; continues-on-error always + * uses DEBUG. + */ + public static final String ERROR_LOG_SEVERITY = FLINK_CONNECTOR_HTTP + "error.log.severity"; + // ----------------------------------------------------- } diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClient.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClient.java index 0072b709..a9f53801 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClient.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClient.java @@ -126,6 +126,17 @@ private SinkHttpClientResponse prepareSinkHttpClientResponse( // TODO Add response processor here and orchestrate it with statusCodeChecker. if (optResponse.isEmpty() || statusCodeChecker.isErrorCode(optResponse.get().statusCode())) { + // Log failed response with detailed error context using HttpLogger + if (optResponse.isPresent()) { + var httpResponse = optResponse.get(); + httpLogger.logSinkError(sinkRequestEntry.httpRequest, null, httpResponse, 0); + } else { + httpLogger.logSinkError( + sinkRequestEntry.httpRequest, + null, + new Exception("HTTP request failed with no response"), + 0); + } failedResponses.add(sinkRequestEntry); } else { successfulResponses.add(sinkRequestEntry); diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java index 68b4eac7..c4de11c2 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java @@ -162,18 +162,29 @@ private HttpRowDataWrapper queryAndProcess(RowData lookupData) throws Exception // log if we fail for status code reasons. httpLogger.logResponse((HttpResponse) e.getResponse()); // Case 1 http non successful response + // Enhanced error logging with continueOnError awareness + httpLogger.logLookupError( + request.getHttpRequest(), + (HttpResponse) e.getResponse(), + e, + 0, + this.continueOnError); if (!this.continueOnError) { throw e; } + // If continueOnError is true, proceed with processing the error response // use the response in the Exception response = (HttpResponse) e.getResponse(); httpRowDataWrapper = processHttpResponse(response, request, true); } catch (Exception e) { httpLogger.logExceptionResponse(request, e); // Case 2 Exception occurred + // Enhanced error logging with continueOnError awareness + httpLogger.logLookupError(request.getHttpRequest(), e, 0, this.continueOnError); if (!this.continueOnError) { throw e; } + // If continueOnError is true, proceed with empty result String errMessage = e.getMessage(); // some exceptions do not have messages including the java.net.ConnectException we can // get here if