content: /*
   * 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.kafka.trogdor.workload;
  
  import org.apache.kafka.clients.consumer.ConsumerRecord;
  import org.apache.kafka.clients.consumer.ConsumerRecords;
  import org.apache.kafka.common.utils.Time;
  import org.apache.kafka.trogdor.common.JsonUtil;
  
  import com.fasterxml.jackson.annotation.JsonCreator;
  import com.fasterxml.jackson.annotation.JsonProperty;
  import com.fasterxml.jackson.databind.JsonNode;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.nio.ByteBuffer;
  import java.nio.ByteOrder;
  
  /**
   * This class will process records containing timestamps and generate a histogram based on the data.  It will then be
   * present in the status from the `ConsumeBenchWorker` class.  This must be used with a timestamped PayloadGenerator
   * implementation.
   *
   * Example spec:
   * {
   *    "type": "timestamp",
   *    "histogramMaxMs": 10000,
   *    "histogramMinMs": 0,
   *    "histogramStepMs": 1
   * }
   *
   * This will track total E2E latency up to 10 seconds, using 1ms resolution and a timestamp size of 8 bytes.
   */
  
  public class TimestampRecordProcessor implements RecordProcessor {
      private final int histogramMaxMs;
      private final int histogramMinMs;
      private final int histogramStepMs;
      private final ByteBuffer buffer;
      private final Histogram histogram;
  
      private static final Logger log = LoggerFactory.getLogger(TimestampRecordProcessor.class);
  
      static final float[] PERCENTILES = {0.5f, 0.95f, 0.99f};
  
      @JsonCreator
      public TimestampRecordProcessor(@JsonProperty("histogramMaxMs") int histogramMaxMs,
                                      @JsonProperty("histogramMinMs") int histogramMinMs,
                                      @JsonProperty("histogramStepMs") int histogramStepMs) {
          this.histogramMaxMs = histogramMaxMs;
          this.histogramMinMs = histogramMinMs;
          this.histogramStepMs = histogramStepMs;
          this.histogram = new Histogram((histogramMaxMs - histogramMinMs) / histogramStepMs);
          buffer = ByteBuffer.allocate(Long.BYTES);
          buffer.order(ByteOrder.LITTLE_ENDIAN);
      }
  
      @JsonProperty
      public int histogramMaxMs() {
          return histogramMaxMs;
      }
  
      @JsonProperty
      public int histogramMinMs() {
          return histogramMinMs;
      }
  
      @JsonProperty
      public int histogramStepMs() {
          return histogramStepMs;
      }
  
      private void putHistogram(long latency) {
          histogram.add(Long.max(0L, (latency - histogramMinMs) / histogramStepMs));
      }
  
      @Override
      public synchronized void processRecords(ConsumerRecords<byte[], byte[]> consumerRecords) {
          // Save the current time to prevent skew by processing time.
          long curTime = Time.SYSTEM.milliseconds();
          for (ConsumerRecord<byte[], byte[]> record : consumerRecords) {
              try {
                  buffer.clear();
                  buffer.put(record.value(), 0, Long.BYTES);
                  buffer.rewind();
                  putHistogram(curTime - buffer.getLong());
              } catch (RuntimeException e) {
                  log.error("Error in processRecords:", e);
              }
          }
      }
  
      @Override
      public JsonNode processorStatus() {
          Histogram.Summary summary = histogram.summarize(PERCENTILES);
          StatusData statusData = new StatusData(
                  summary.average() * histogramStepMs + histogramMinMs,
                  summary.percentiles().get(0).value() * histogramStepMs + histogramMinMs,
                  summary.percentiles().get(1).value() * histogramStepMs + histogramMinMs,
                  summary.percentiles().get(2).value() * histogramStepMs + histogramMinMs);
          return JsonUtil.JSON_SERDE.valueToTree(statusData);
      }
  
      private static class StatusData {
          private final float averageLatencyMs;
          private final int p50LatencyMs;
          private final int p95LatencyMs;
          private final int p99LatencyMs;
  
          /**
           * The percentiles to use when calculating the histogram data.
           * These should match up with the p50LatencyMs, p95LatencyMs, etc. fields.
           */
          static final float[] PERCENTILES = {0.5f, 0.95f, 0.99f};
  
          @JsonCreator
          StatusData(@JsonProperty("averageLatencyMs") float averageLatencyMs,
                     @JsonProperty("p50LatencyMs") int p50latencyMs,
                     @JsonProperty("p95LatencyMs") int p95latencyMs,
                     @JsonProperty("p99LatencyMs") int p99latencyMs) {
              this.averageLatencyMs = averageLatencyMs;
              this.p50LatencyMs = p50latencyMs;
              this.p95LatencyMs = p95latencyMs;
              this.p99LatencyMs = p99latencyMs;
          }
  
          @JsonProperty
          public float averageLatencyMs() {
              return averageLatencyMs;
          }
  
          @JsonProperty
          public int p50LatencyMs() {
              return p50LatencyMs;
          }
  
          @JsonProperty
          public int p95LatencyMs() {
              return p95LatencyMs;
          }
  
          @JsonProperty
          public int p99LatencyMs() {
              return p99LatencyMs;
          }
      }
  }
  