[FLINK-39367][connector-http] Add configurable error logging with detailed context

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;
+        }
+    }
+}

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<String> 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<String> 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<String> 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<String> 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<String, List<String>> 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)";
+    }
 }

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";
+    // -----------------------------------------------------
 }

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);