[FEATURE] Add metric about the time spent on message conversions

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KopServerStats.java

@@ -62,6 +62,7 @@ public interface KopServerStats {
     String MESSAGE_IN = "MESSAGE_IN";
     String BATCH_COUNT_PER_MEMORYRECORDS = "BATCH_COUNT_PER_MEMORYRECORDS";
     String PRODUCE_MESSAGE_CONVERSIONS = "PRODUCE_MESSAGE_CONVERSIONS";
+    String PRODUCE_MESSAGE_CONVERSIONS_TIME_NANOS = "PRODUCE_MESSAGE_CONVERSIONS_TIME_NANOS";
 
     /**
      * FETCH stats.
@@ -84,6 +85,7 @@ public interface KopServerStats {
     String MESSAGE_OUT = "MESSAGE_OUT";
     String ENTRIES_OUT = "ENTRIES_OUT";
     String CONSUME_MESSAGE_CONVERSIONS = "CONSUME_MESSAGE_CONVERSIONS";
+    String CONSUME_MESSAGE_CONVERSIONS_TIME_NANOS = "CONSUME_MESSAGE_CONVERSIONS_TIME_NANOS";
 
     /**
      * Kop event queue stats.

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/AbstractEntryFormatter.java

@@ -24,6 +24,7 @@
 import java.io.IOException;
 import java.util.List;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.common.util.MathUtils;
 import org.apache.bookkeeper.mledger.Entry;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.record.ConvertedRecords;
@@ -45,6 +46,7 @@ public abstract class AbstractEntryFormatter implements EntryFormatter {
     public DecodeResult decode(List<Entry> entries, byte magic) {
         int totalSize = 0;
         int conversionCount = 0;
+        long conversionTimeNanos = 0L;
         // batched ByteBuf should be released after sending to client
         ByteBuf batchedByteBuf = PulsarByteBufAllocator.DEFAULT.directBuffer(totalSize);
         for (Entry entry : entries) {
@@ -59,11 +61,13 @@ public DecodeResult decode(List<Entry> entries, byte magic) {
                     // batch magic greater than the magic corresponding to the version requested by the client
                     // need down converted
                     if (batchMagic > magic) {
+                        long startConversionNanos = MathUtils.nowInNano();
                         MemoryRecords memoryRecords = MemoryRecords.readableRecords(ByteBufUtils.getNioBuffer(byteBuf));
                         // down converted, batch magic will be set to client magic
                         ConvertedRecords<MemoryRecords> convertedRecords =
                                 memoryRecords.downConvert(magic, startOffset, time);
                         conversionCount += convertedRecords.recordConversionStats().numRecordsConverted();
+                        conversionTimeNanos += MathUtils.elapsedNanos(startConversionNanos);
 
                         final ByteBuf kafkaBuffer = Unpooled.wrappedBuffer(convertedRecords.records().buffer());
                         totalSize += kafkaBuffer.readableBytes();
@@ -85,6 +89,7 @@ public DecodeResult decode(List<Entry> entries, byte magic) {
                     final DecodeResult decodeResult =
                             ByteBufUtils.decodePulsarEntryToKafkaRecords(metadata, byteBuf, startOffset, magic);
                     conversionCount += decodeResult.getConversionCount();
+                    conversionTimeNanos += decodeResult.getConversionTimeNanos();
                     final ByteBuf kafkaBuffer = decodeResult.getOrCreateByteBuf();
                     totalSize += kafkaBuffer.readableBytes();
                     batchedByteBuf.writeBytes(kafkaBuffer);
@@ -105,7 +110,8 @@ public DecodeResult decode(List<Entry> entries, byte magic) {
         return DecodeResult.get(
                 MemoryRecords.readableRecords(ByteBufUtils.getNioBuffer(batchedByteBuf)),
                 batchedByteBuf,
-                conversionCount);
+                conversionCount,
+                conversionTimeNanos);
     }
 
     protected static boolean isKafkaEntryFormat(final MessageMetadata messageMetadata) {

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/DecodeResult.java

@@ -15,6 +15,7 @@
 
 import static io.streamnative.pulsar.handlers.kop.KopServerStats.BYTES_OUT;
 import static io.streamnative.pulsar.handlers.kop.KopServerStats.CONSUME_MESSAGE_CONVERSIONS;
+import static io.streamnative.pulsar.handlers.kop.KopServerStats.CONSUME_MESSAGE_CONVERSIONS_TIME_NANOS;
 import static io.streamnative.pulsar.handlers.kop.KopServerStats.ENTRIES_OUT;
 import static io.streamnative.pulsar.handlers.kop.KopServerStats.GROUP_SCOPE;
 import static io.streamnative.pulsar.handlers.kop.KopServerStats.MESSAGE_OUT;
@@ -26,6 +27,7 @@
 import io.netty.util.Recycler;
 import io.streamnative.pulsar.handlers.kop.RequestStats;
 import io.streamnative.pulsar.handlers.kop.stats.StatsLogger;
+import java.util.concurrent.TimeUnit;
 import lombok.Getter;
 import lombok.NonNull;
 import org.apache.kafka.common.TopicPartition;
@@ -41,20 +43,24 @@ public class DecodeResult {
     private ByteBuf releasedByteBuf;
     @Getter
     private int conversionCount;
+    @Getter
+    private long conversionTimeNanos;
 
     private final Recycler.Handle<DecodeResult> recyclerHandle;
 
     public static DecodeResult get(MemoryRecords records) {
-        return get(records, null, 0);
+        return get(records, null, 0, 0L);
     }
 
     public static DecodeResult get(MemoryRecords records,
                                    ByteBuf releasedByteBuf,
-                                   int conversionCount) {
+                                   int conversionCount,
+                                   long conversionTimeNanos) {
         DecodeResult decodeResult = RECYCLER.get();
         decodeResult.records = records;
         decodeResult.releasedByteBuf = releasedByteBuf;
         decodeResult.conversionCount = conversionCount;
+        decodeResult.conversionTimeNanos = conversionTimeNanos;
         return decodeResult;
     }
 
@@ -76,6 +82,7 @@ public void recycle() {
             releasedByteBuf = null;
         }
         conversionCount = -1;
+        conversionTimeNanos = -1L;
         recyclerHandle.recycle(this);
     }
 
@@ -98,6 +105,8 @@ public void updateConsumerStats(final TopicPartition topicPartition,
                 .scopeLabel(PARTITION_SCOPE, String.valueOf(topicPartition.partition()));
 
         statsLoggerForThisPartition.getCounter(CONSUME_MESSAGE_CONVERSIONS).add(conversionCount);
+        statsLoggerForThisPartition.getOpStatsLogger(CONSUME_MESSAGE_CONVERSIONS_TIME_NANOS)
+                .registerSuccessfulEvent(conversionTimeNanos, TimeUnit.NANOSECONDS);
 
         final StatsLogger statsLoggerForThisGroup = statsLoggerForThisPartition.scopeLabel(GROUP_SCOPE, groupId);
 

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/EncodeResult.java

@@ -17,6 +17,7 @@
 import static io.streamnative.pulsar.handlers.kop.KopServerStats.MESSAGE_IN;
 import static io.streamnative.pulsar.handlers.kop.KopServerStats.PARTITION_SCOPE;
 import static io.streamnative.pulsar.handlers.kop.KopServerStats.PRODUCE_MESSAGE_CONVERSIONS;
+import static io.streamnative.pulsar.handlers.kop.KopServerStats.PRODUCE_MESSAGE_CONVERSIONS_TIME_NANOS;
 import static io.streamnative.pulsar.handlers.kop.KopServerStats.TOPIC_SCOPE;
 
 import io.netty.buffer.ByteBuf;
@@ -25,6 +26,7 @@
 import io.streamnative.pulsar.handlers.kop.RequestStats;
 import io.streamnative.pulsar.handlers.kop.stats.StatsLogger;
 import io.streamnative.pulsar.handlers.kop.utils.KopTopic;
+import java.util.concurrent.TimeUnit;
 import lombok.Getter;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.record.MemoryRecords;
@@ -40,18 +42,21 @@ public class EncodeResult {
     private ByteBuf encodedByteBuf;
     private int numMessages;
     private int conversionCount;
+    private long conversionTimeNanos;
 
     private final Recycler.Handle<EncodeResult> recyclerHandle;
 
     public static EncodeResult get(MemoryRecords records,
                                    ByteBuf encodedByteBuf,
                                    int numMessages,
-                                   int conversionCount) {
+                                   int conversionCount,
+                                   long conversionTimeNanos) {
         EncodeResult encodeResult = RECYCLER.get();
         encodeResult.records = records;
         encodeResult.encodedByteBuf = encodedByteBuf;
         encodeResult.numMessages = numMessages;
         encodeResult.conversionCount = conversionCount;
+        encodeResult.conversionTimeNanos = conversionTimeNanos;
         return encodeResult;
     }
 
@@ -74,6 +79,7 @@ public void recycle() {
         }
         numMessages = -1;
         conversionCount = -1;
+        conversionTimeNanos = -1L;
         recyclerHandle.recycle(this);
     }
 
@@ -94,6 +100,8 @@ public void updateProducerStats(final TopicPartition topicPartition,
         statsLoggerForThisPartition.getCounter(BYTES_IN).add(numBytes);
         statsLoggerForThisPartition.getCounter(MESSAGE_IN).add(numMessages);
         statsLoggerForThisPartition.getCounter(PRODUCE_MESSAGE_CONVERSIONS).add(conversionCount);
+        statsLoggerForThisPartition.getOpStatsLogger(PRODUCE_MESSAGE_CONVERSIONS_TIME_NANOS)
+                .registerSuccessfulEvent(conversionTimeNanos, TimeUnit.NANOSECONDS);
 
         RequestStats.BATCH_COUNT_PER_MEMORY_RECORDS_INSTANCE.set(numMessages);
     }

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/KafkaMixedEntryFormatter.java

@@ -58,6 +58,7 @@ public EncodeResult encode(final EncodeRequest encodeRequest) {
 
         MemoryRecords validRecords = validationAndOffsetAssignResult.getRecords();
         int conversionCount = validationAndOffsetAssignResult.getConversionCount();
+        long conversionTimeNanos = validationAndOffsetAssignResult.getConversionTimeNanos();
 
         final int numMessages = appendInfo.numMessages();
         final ByteBuf recordsWrapper = Unpooled.wrappedBuffer(validRecords.buffer());
@@ -68,7 +69,7 @@ public EncodeResult encode(final EncodeRequest encodeRequest) {
         recordsWrapper.release();
         validationAndOffsetAssignResult.recycle();
 
-        return EncodeResult.get(validRecords, buf, numMessages, conversionCount);
+        return EncodeResult.get(validRecords, buf, numMessages, conversionCount, conversionTimeNanos);
     }
 
     @Override

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/KafkaV1EntryFormatter.java

@@ -41,7 +41,7 @@ public EncodeResult encode(final EncodeRequest encodeRequest) {
                 recordsWrapper);
         recordsWrapper.release();
 
-        return EncodeResult.get(records, buf, numMessages, 0);
+        return EncodeResult.get(records, buf, numMessages, 0, 0L);
     }
 
     @Override

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/PulsarEntryFormatter.java

@@ -21,6 +21,7 @@
 import java.util.List;
 import java.util.stream.StreamSupport;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.common.util.MathUtils;
 import org.apache.bookkeeper.mledger.Entry;
 import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.record.ControlRecordType;
@@ -49,6 +50,7 @@ public EncodeResult encode(final EncodeRequest encodeRequest) {
         final int numMessages = encodeRequest.getAppendInfo().numMessages();
         long currentBatchSizeBytes = 0;
         int numMessagesInBatch = 0;
+        long startConversionNanos = MathUtils.nowInNano();
 
         long sequenceId = -1;
 
@@ -107,7 +109,8 @@ public EncodeResult encode(final EncodeRequest encodeRequest) {
 
         batchedMessageMetadataAndPayload.release();
 
-        return EncodeResult.get(records, buf, numMessages, numMessagesInBatch);
+        return EncodeResult.get(records, buf, numMessages, numMessagesInBatch,
+                MathUtils.elapsedNanos(startConversionNanos));
     }
 
     @Override

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/ValidationAndOffsetAssignResult.java

@@ -25,14 +25,17 @@ public class ValidationAndOffsetAssignResult {
 
     private MemoryRecords records;
     private int conversionCount;
+    private long conversionTimeNanos;
 
     private final Recycler.Handle<ValidationAndOffsetAssignResult> recyclerHandle;
 
     public static ValidationAndOffsetAssignResult get(MemoryRecords records,
-                                                      int conversionCount) {
+                                                      int conversionCount,
+                                                      long conversionTimeNanos) {
         ValidationAndOffsetAssignResult validationAndOffsetAssignResult = RECYCLER.get();
         validationAndOffsetAssignResult.records = records;
         validationAndOffsetAssignResult.conversionCount = conversionCount;
+        validationAndOffsetAssignResult.conversionTimeNanos = conversionTimeNanos;
         return validationAndOffsetAssignResult;
     }
 
@@ -52,6 +55,7 @@ protected ValidationAndOffsetAssignResult newObject(
     public void recycle() {
         records = null;
         conversionCount = -1;
+        conversionTimeNanos = -1L;
         recyclerHandle.recycle(this);
     }
 

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ByteBufUtils.java

@@ -24,6 +24,7 @@
 import java.util.List;
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.common.util.MathUtils;
 import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.header.internals.RecordHeader;
 import org.apache.kafka.common.record.CompressionType;
@@ -121,6 +122,7 @@ public static DecodeResult decodePulsarEntryToKafkaRecords(final MessageMetadata
                     new EndTransactionMarker(controlRecordType, 0)
                     ));
         }
+        long startConversionNanos = MathUtils.nowInNano();
         final int uncompressedSize = metadata.getUncompressedSize();
         final CompressionCodec codec = CompressionCodecProvider.getCompressionCodec(metadata.getCompression());
         final ByteBuf uncompressedPayload = codec.decode(payload, uncompressedSize);
@@ -196,7 +198,10 @@ public static DecodeResult decodePulsarEntryToKafkaRecords(final MessageMetadata
 
         final MemoryRecords records = builder.build();
         uncompressedPayload.release();
-        return DecodeResult.get(records, directBufferOutputStream.getByteBuf(), conversionCount);
+        return DecodeResult.get(records,
+                directBufferOutputStream.getByteBuf(),
+                conversionCount,
+                MathUtils.elapsedNanos(startConversionNanos));
     }
 
     @NonNull

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KopLogValidator.java

@@ -19,6 +19,7 @@
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.Locale;
+import org.apache.bookkeeper.common.util.MathUtils;
 import org.apache.kafka.common.errors.InvalidTimestampException;
 import org.apache.kafka.common.errors.UnsupportedForMessageFormatException;
 import org.apache.kafka.common.record.AbstractRecords;
@@ -99,6 +100,7 @@ private static ValidationAndOffsetAssignResult convertAndAssignOffsetsNonCompres
                                                                                         TimestampType timestampType,
                                                                                         long timestampDiffMaxMs,
                                                                                         byte toMagicValue) {
+        long startConversionNanos = MathUtils.nowInNano();
         int sizeInBytesAfterConversion = AbstractRecords.estimateSizeInBytes(toMagicValue, offsetCounter.value(),
                 CompressionType.NONE, records.records());
 
@@ -136,7 +138,8 @@ private static ValidationAndOffsetAssignResult convertAndAssignOffsetsNonCompres
         MemoryRecords memoryRecords = builder.build();
         int conversionCount = builder.numRecords();
 
-        return ValidationAndOffsetAssignResult.get(memoryRecords, conversionCount);
+        return ValidationAndOffsetAssignResult.get(memoryRecords,
+                conversionCount, MathUtils.elapsedNanos(startConversionNanos));
     }
 
     private static ValidationAndOffsetAssignResult assignOffsetsNonCompressed(MemoryRecords records,
@@ -178,7 +181,7 @@ private static ValidationAndOffsetAssignResult assignOffsetsNonCompressed(Memory
             }
         }
 
-        return ValidationAndOffsetAssignResult.get(records, 0);
+        return ValidationAndOffsetAssignResult.get(records, 0, 0L);
     }
 
     /**
@@ -330,7 +333,7 @@ private static ValidationAndOffsetAssignResult buildInPlaceAssignment(MemoryReco
             batch.setPartitionLeaderEpoch(RecordBatch.NO_PARTITION_LEADER_EPOCH);
         }
 
-        return ValidationAndOffsetAssignResult.get(records, 0);
+        return ValidationAndOffsetAssignResult.get(records, 0, 0L);
     }
 
     private static ValidationAndOffsetAssignResult buildRecordsAndAssignOffsets(byte magic,
@@ -340,6 +343,7 @@ private static ValidationAndOffsetAssignResult buildRecordsAndAssignOffsets(byte
                                                                                 long logAppendTime,
                                                                                 ArrayList<Record> validatedRecords,
                                                                                 MutableRecordBatch first) {
+        long startConversionNanos = MathUtils.nowInNano();
         long producerId = first.producerId();
         short producerEpoch = first.producerEpoch();
         int baseSequence = first.baseSequence();
@@ -367,7 +371,8 @@ private static ValidationAndOffsetAssignResult buildRecordsAndAssignOffsets(byte
         MemoryRecords memoryRecords = builder.build();
         int conversionCount = builder.numRecords();
 
-        return ValidationAndOffsetAssignResult.get(memoryRecords, conversionCount);
+        return ValidationAndOffsetAssignResult.get(memoryRecords,
+                conversionCount, MathUtils.elapsedNanos(startConversionNanos));
     }
 
     private static void validateBatch(RecordBatch batch, byte toMagic) {

kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/format/NoHeaderKafkaEntryFormatter.java

@@ -28,7 +28,8 @@ public EncodeResult encode(final EncodeRequest encodeRequest) {
         final MemoryRecords records = encodeRequest.getRecords();
         final int numMessages = encodeRequest.getAppendInfo().numMessages();
         // The difference from KafkaEntryFormatter is here we don't add the header
-        return EncodeResult.get(records, Unpooled.wrappedBuffer(records.buffer()), numMessages, 0);
+        return EncodeResult.get(records, Unpooled.wrappedBuffer(records.buffer()),
+                numMessages, 0, 0L);
     }
 
     @Override

tests/src/test/java/io/streamnative/pulsar/handlers/kop/MetricsProviderTest.java

@@ -46,7 +46,7 @@
  * test for kop prometheus metrics.
  */
 @Slf4j
-public class MetricsProviderTest extends KopProtocolHandlerTestBase{
+public class MetricsProviderTest extends KopProtocolHandlerTestBase {
 
     @BeforeMethod
     @Override
@@ -76,7 +76,7 @@ public void testMetricsProvider() throws Exception {
 
         int totalMsgs = 10;
 
-        String messageStrPrefix = "Message_Kop_KafkaProducePulsarConsume_"  + partitionNumber + "_";
+        String messageStrPrefix = "Message_Kop_KafkaProducePulsarConsume_" + partitionNumber + "_";
 
         for (int i = 0; i < totalMsgs; i++) {
             String messageStr = messageStrPrefix + i;
@@ -192,13 +192,21 @@ public void testMetricsProvider() throws Exception {
         Assert.assertTrue(sb.toString().contains("kop_server_BYTES_OUT{group=\"DemoKafkaOnPulsarConsumer\","
                 + "partition=\"0\",topic=\"kopKafkaProducePulsarMetrics1\"} 1130"));
         Assert.assertTrue(sb.toString().contains("kop_server_BYTES_OUT"));
+        Assert.assertTrue(sb.toString().contains("kop_server_CONSUME_MESSAGE_CONVERSIONS"));
+        Assert.assertTrue(sb.toString().contains("kop_server_CONSUME_MESSAGE_CONVERSIONS{partition=\"0\","
+                + "topic=\"kopKafkaProducePulsarMetrics1\"} 10"));
+        Assert.assertTrue(sb.toString().contains("kop_server_CONSUME_MESSAGE_CONVERSIONS_TIME_NANOS"));
 
         // producer stats
         Assert.assertTrue(sb.toString().contains("kop_server_BATCH_COUNT_PER_MEMORYRECORDS"));
         Assert.assertTrue(sb.toString().contains("kop_server_MESSAGE_IN{partition=\"0\","
                 + "topic=\"kopKafkaProducePulsarMetrics1\"} 10"));
         Assert.assertTrue(sb.toString().contains("kop_server_BYTES_IN{partition=\"0\","
                 + "topic=\"kopKafkaProducePulsarMetrics1\"} 1170"));
+        Assert.assertTrue(sb.toString().contains("kop_server_PRODUCE_MESSAGE_CONVERSIONS"));
+        Assert.assertTrue(sb.toString().contains("kop_server_PRODUCE_MESSAGE_CONVERSIONS{partition=\"0\","
+                + "topic=\"kopKafkaProducePulsarMetrics1\"} 10"));
+        Assert.assertTrue(sb.toString().contains("kop_server_PRODUCE_MESSAGE_CONVERSIONS_TIME_NANOS"));
     }
 
     @Test(timeOut = 20000)