Merge pull request #1300 from confluentinc/KNET-12362

[KNET-12362] Produce V3 API - Improve visibility around produce-record errors

Author: wcheng92

kafka-rest/src/main/java/io/confluent/kafkarest/KafkaRestApplication.java

@@ -130,7 +130,10 @@ private static RequestLog createRequestLog(
       return new CustomLog(
           requestLogWriter,
           requestLogFormat,
-          new String[] {CustomLogRequestAttributes.REST_ERROR_CODE});
+          new String[] {
+            CustomLogRequestAttributes.REST_ERROR_CODE,
+            CustomLogRequestAttributes.REST_PRODUCE_RECORD_ERROR_CODE_COUNTS
+          });
     }
     // Return null, as Application's ctor would set-up a default request-logger.
     return null;

kafka-rest/src/main/java/io/confluent/kafkarest/requestlog/CustomLog.java

@@ -15,6 +15,9 @@
 
 package io.confluent.kafkarest.requestlog;
 
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.stream.Collectors;
 import org.eclipse.jetty.server.CustomRequestLog;
 import org.eclipse.jetty.server.Request;
 import org.eclipse.jetty.server.RequestLog;
@@ -39,6 +42,8 @@ public class CustomLog extends AbstractLifeCycle implements RequestLog {
 
   private final String[] requestAttributesToLog;
 
+  public static final String PRODUCE_ERROR_CODE_LOG_PREFIX = "Codes=";
+
   public CustomLog(RequestLog.Writer writer, String formatString, String[] requestAttributesToLog) {
     for (String attr : requestAttributesToLog) {
       // Add format-specifier to log request-attributes as response-headers in Jetty's
@@ -63,6 +68,27 @@ protected void doStop() throws Exception {
     }
   }
 
+  /**
+   * This class aggregates error-codes for produce-records within a single (http)produce-request.
+   * This implements toString() method which is used by CustomLog to get a message to log for the
+   * aggregated error counts.
+   */
+  public static class ProduceRecordErrorCounter {
+    private final Map<Integer, Integer> produceErrorCodeCountMap = new TreeMap<>();
+
+    public synchronized void incrementErrorCount(int httpErrorCode) {
+      produceErrorCodeCountMap.merge(httpErrorCode, 1, Integer::sum);
+    }
+
+    @Override
+    public synchronized String toString() {
+      return PRODUCE_ERROR_CODE_LOG_PREFIX
+          + produceErrorCodeCountMap.entrySet().stream()
+              .map(entry -> entry.getKey() + ":" + entry.getValue())
+              .collect(Collectors.joining(","));
+    }
+  }
+
   @Override
   public void log(Request request, Response response) {
     // The configured request-attributes are converted to response-headers so Jetty can log them.

kafka-rest/src/main/java/io/confluent/kafkarest/requestlog/CustomLogRequestAttributes.java

@@ -24,4 +24,6 @@ public final class CustomLogRequestAttributes {
   private CustomLogRequestAttributes() {}
 
   public static final String REST_ERROR_CODE = "REST_ERROR_CODE";
+  public static final String REST_PRODUCE_RECORD_ERROR_CODE_COUNTS =
+      "REST_PRODUCE_RECORD_ERROR_CODE_COUNTS";
 }

kafka-rest/src/main/java/io/confluent/kafkarest/resources/v3/ProduceAction.java

@@ -38,6 +38,8 @@
 import io.confluent.kafkarest.extension.ResourceAccesslistFeature.ResourceName;
 import io.confluent.kafkarest.ratelimit.DoNotRateLimit;
 import io.confluent.kafkarest.ratelimit.RateLimitExceededException;
+import io.confluent.kafkarest.requestlog.CustomLog.ProduceRecordErrorCounter;
+import io.confluent.kafkarest.requestlog.CustomLogRequestAttributes;
 import io.confluent.kafkarest.resources.v3.V3ResourcesModule.ProduceResponseThreadPool;
 import io.confluent.kafkarest.response.JsonStream;
 import io.confluent.kafkarest.response.StreamingResponseFactory;
@@ -147,13 +149,19 @@ public void produce(
       throw Errors.invalidPayloadException("Request body is empty. Data is required.");
     }
 
+    ProduceRecordErrorCounter produceRecordErrorCounter = new ProduceRecordErrorCounter();
+
     ProduceController controller = produceControllerProvider.get();
     streamingResponseFactory
         .from(requests)
         .compose(
             request ->
                 produce(clusterId, topicName, request, controller, producerMetricsProvider.get()))
-        .resume(asyncResponse);
+        .resume(asyncResponse, produceRecordErrorCounter);
+
+    httpServletRequest.setAttribute(
+        CustomLogRequestAttributes.REST_PRODUCE_RECORD_ERROR_CODE_COUNTS,
+        produceRecordErrorCounter);
   }
 
   private CompletableFuture<ProduceResponse> produce(

kafka-rest/src/main/java/io/confluent/kafkarest/response/StreamingResponse.java

@@ -33,6 +33,7 @@
 import io.confluent.kafkarest.exceptions.StatusCodeException;
 import io.confluent.kafkarest.exceptions.v3.ErrorResponse;
 import io.confluent.kafkarest.exceptions.v3.V3ExceptionMapper;
+import io.confluent.kafkarest.requestlog.CustomLog.ProduceRecordErrorCounter;
 import io.confluent.rest.entities.ErrorMessage;
 import io.confluent.rest.exceptions.KafkaExceptionMapper;
 import io.confluent.rest.exceptions.RestConstraintViolationException;
@@ -159,7 +160,8 @@ public final <O> StreamingResponse<O> compose(
    * <p>This method will block until all requests are read in. The responses are computed and
    * written to {@code asyncResponse} asynchronously.
    */
-  public final void resume(AsyncResponse asyncResponse) {
+  public final void resume(
+      AsyncResponse asyncResponse, ProduceRecordErrorCounter produceRecordErrorCounter) {
     log.debug("Resuming StreamingResponse");
     AsyncResponseQueue responseQueue = new AsyncResponseQueue(chunkedOutputFactory);
     responseQueue.asyncResume(asyncResponse);
@@ -186,7 +188,11 @@ public final void resume(AsyncResponse asyncResponse) {
                               "Streaming connection open for longer than allowed",
                               "Connection will be closed.")))));
         } else if (!closingStarted) {
-          responseQueue.push(next().handle(this::handleNext));
+          responseQueue.push(
+              next()
+                  .handle(
+                      (result, exception) ->
+                          handleNext(result, exception, produceRecordErrorCounter)));
         } else {
           break;
         }
@@ -218,11 +224,17 @@ private void closeAll(AsyncResponseQueue responseQueue) {
     responseQueue.close();
   }
 
-  private ResultOrError handleNext(T result, @Nullable Throwable error) {
+  private ResultOrError handleNext(
+      T result, @Nullable Throwable error, ProduceRecordErrorCounter produceRecordErrorCounter) {
     if (error == null) {
       return ResultOrError.result(result);
     } else {
       log.debug("Error processing streaming operation.", error);
+      if (error.getCause() == null) {
+        throw new IllegalArgumentException("Error cause is null", error);
+      }
+      int errorCode = EXCEPTION_MAPPER.toErrorResponse(error.getCause()).getErrorCode();
+      produceRecordErrorCounter.incrementErrorCount(errorCode);
       return ResultOrError.error(EXCEPTION_MAPPER.toErrorResponse(error.getCause()));
     }
   }