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

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