[FEATURE] Add message conversions metrics for prometheus metrics

docs/reference-metrics.md

@@ -58,6 +58,7 @@ The KoP metrics are exposed under "/metrics" at port `8000` along with Pulsar me
 | kop_server_BYTES_IN | Counter | The producer bytes in stats. <br> Available labels: *topic*, *partition*. </br> <ul><li>*topic*: the topic name to produce.</li><li>*partition*: the partition id for the topic to produce</li></ul>|
 | kop_server_MESSAGE_IN | Counter | The producer message in stats. <br> Available labels: *topic*, *partition*. </br> <ul><li>*topic*: the topic name to produce.</li><li>*partition*: the partition id for the topic to produce</li></ul>|
 | kop_server_BATCH_COUNT_PER_MEMORYRECORDS | Gauge | The number of batches in each memory records|
+| kop_server_PRODUCE_MESSAGE_CONVERSIONS | Counter | The producer message conversions in stats. <br> Available labels: *topic*, *partition*. </br> <ul><li>*topic*: the topic name to produce.</li><li>*partition*: the partition id for the topic to produce</li></ul>|
 
 ### Consumer metrics
 
@@ -70,6 +71,7 @@ The KoP metrics are exposed under "/metrics" at port `8000` along with Pulsar me
 | kop_server_BYTES_OUT | Counter | The consumer bytes out stats. <br> Available labels: *topic*, *partition*, *group*. </br> <ul><li>*topic*: the topic name to consume.</li><li>*partition*: the partition id for the topic to consume</li><li>*group*: the group id for consumer to consumer message from topic-partition</li></ul>|
 | kop_server_MESSAGE_OUT | Counter | The consumer message out stats. <br> Available labels: *topic*, *partition*, *group*. </br> <ul><li>*topic*: the topic name to consume.</li><li>*partition*: the partition id for the topic to consume</li><li>*group*: the group id for consumer to consumer message from topic-partition</li></ul>|
 | kop_server_ENTRIES_OUT | Counter | The consumer entries out stats. <br> Available labels: *topic*, *partition*, *group*. </br> <ul><li>*topic*: the topic name to consume.</li><li>*partition*: the partition id for the topic to consume</li><li>*group*: the group id for consumer to consumer message from topic-partition</li></ul>|
+| kop_server_CONSUME_MESSAGE_CONVERSIONS | Counter | The consumer message conversions in stats. <br> Available labels: *topic*, *partition*. </br> <ul><li>*topic*: the topic name to consume.</li><li>*partition*: the partition id for the topic to consume</li></ul>|
 
 ### Kop event metrics
 

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

@@ -17,10 +17,6 @@
 import static com.google.common.base.Preconditions.checkState;
 import static io.streamnative.pulsar.handlers.kop.KafkaServiceConfiguration.TENANT_ALLNAMESPACES_PLACEHOLDER;
 import static io.streamnative.pulsar.handlers.kop.KafkaServiceConfiguration.TENANT_PLACEHOLDER;
-import static io.streamnative.pulsar.handlers.kop.KopServerStats.BYTES_IN;
-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.TOPIC_SCOPE;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.apache.kafka.common.internals.Topic.GROUP_METADATA_TOPIC_NAME;
 import static org.apache.kafka.common.internals.Topic.TRANSACTION_STATE_TOPIC_NAME;
@@ -167,7 +163,6 @@
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.pulsar.broker.PulsarService;
-import org.apache.pulsar.broker.service.Producer;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminException;
@@ -917,10 +912,7 @@ private void publishMessages(final Optional<PersistentTopic> persistentTopicOpt,
 
         topicManager.registerProducerInPersistentTopic(partitionName, persistentTopic);
         // collect metrics
-        final Producer producer = KafkaTopicManager.getReferenceProducer(partitionName);
-        producer.updateRates(numMessages, byteBuf.readableBytes());
-        producer.getTopic().incrementPublishCount(numMessages, byteBuf.readableBytes());
-        updateProducerStats(topicPartition, numMessages, byteBuf.readableBytes());
+        encodeResult.updateProducerStats(topicPartition, requestStats);
 
         // publish
         final CompletableFuture<Long> offsetFuture = new CompletableFuture<>();
@@ -2584,22 +2576,6 @@ private static MemoryRecords validateRecords(short version, TopicPartition topic
         return validRecords;
     }
 
-    private void updateProducerStats(final TopicPartition topicPartition, final int numMessages, final int numBytes) {
-        requestStats.getStatsLogger()
-                .scopeLabel(TOPIC_SCOPE, topicPartition.topic())
-                .scopeLabel(PARTITION_SCOPE, String.valueOf((topicPartition.partition())))
-                .getCounter(BYTES_IN)
-                .add(numBytes);
-
-        requestStats.getStatsLogger()
-                .scopeLabel(TOPIC_SCOPE, topicPartition.topic())
-                .scopeLabel(PARTITION_SCOPE, String.valueOf((topicPartition.partition())))
-                .getCounter(MESSAGE_IN)
-                .add(numMessages);
-
-        RequestStats.BATCH_COUNT_PER_MEMORY_RECORDS_INSTANCE.set(numMessages);
-    }
-
     @VisibleForTesting
     protected CompletableFuture<Boolean> authorize(AclOperation operation, Resource resource) {
         Session session = authenticator != null ? authenticator.session() : null;

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

@@ -61,6 +61,7 @@ public interface KopServerStats {
     String BYTES_IN = "BYTES_IN";
     String MESSAGE_IN = "MESSAGE_IN";
     String BATCH_COUNT_PER_MEMORYRECORDS = "BATCH_COUNT_PER_MEMORYRECORDS";
+    String PRODUCE_MESSAGE_CONVERSIONS = "PRODUCE_MESSAGE_CONVERSIONS";
 
     /**
      * FETCH stats.
@@ -81,6 +82,7 @@ public interface KopServerStats {
     String BYTES_OUT = "BYTES_OUT";
     String MESSAGE_OUT = "MESSAGE_OUT";
     String ENTRIES_OUT = "ENTRIES_OUT";
+    String CONSUME_MESSAGE_CONVERSIONS = "CONSUME_MESSAGE_CONVERSIONS";
 
     /**
      * Kop event queue stats.

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

@@ -13,12 +13,6 @@
  */
 package io.streamnative.pulsar.handlers.kop;
 
-import static io.streamnative.pulsar.handlers.kop.KopServerStats.BYTES_OUT;
-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;
-import static io.streamnative.pulsar.handlers.kop.KopServerStats.PARTITION_SCOPE;
-import static io.streamnative.pulsar.handlers.kop.KopServerStats.TOPIC_SCOPE;
 import static org.apache.kafka.common.protocol.CommonFields.THROTTLE_TIME_MS;
 
 import com.google.common.collect.Lists;
@@ -28,7 +22,6 @@
 import io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionCoordinator;
 import io.streamnative.pulsar.handlers.kop.exceptions.KoPMessageMetadataNotFoundException;
 import io.streamnative.pulsar.handlers.kop.format.DecodeResult;
-import io.streamnative.pulsar.handlers.kop.format.EntryFormatter;
 import io.streamnative.pulsar.handlers.kop.security.auth.Resource;
 import io.streamnative.pulsar.handlers.kop.security.auth.ResourceType;
 import io.streamnative.pulsar.handlers.kop.utils.GroupIdUtils;
@@ -447,7 +440,7 @@ private void handleEntries(final List<Entry> entries,
                 MathUtils.elapsedNanos(startDecodingEntriesNanos), TimeUnit.NANOSECONDS);
         decodeResults.add(decodeResult);
 
-        MemoryRecords kafkaRecords = decodeResult.getRecords();
+        final MemoryRecords kafkaRecords = decodeResult.getRecords();
 
         CompletableFuture<String> groupNameFuture = requestHandler
                 .getCurrentConnectedGroup()
@@ -476,10 +469,10 @@ private void handleEntries(final List<Entry> entries,
                 groupName = "";
             }
             // collect consumer metrics
-            updateConsumerStats(topicPartition,
-                    kafkaRecords,
+            decodeResult.updateConsumerStats(topicPartition,
                     entries.size(),
-                    groupName);
+                    groupName,
+                    statsLogger);
             final List<FetchResponse.AbortedTransaction> abortedTransactions =
                     (readCommitted ? tc.getAbortedIndexList(partitionData.fetchOffset) : null);
             responseData.put(topicPartition, new PartitionData<>(
@@ -598,30 +591,4 @@ public void markDeleteFailed(ManagedLedgerException e, Object ctx) {
             }
         }, null);
     }
-
-    private void updateConsumerStats(final TopicPartition topicPartition, final MemoryRecords records,
-                                     int entrySize, final String groupId) {
-        int numMessages = EntryFormatter.parseNumMessages(records);
-
-        statsLogger.getStatsLogger()
-                .scopeLabel(TOPIC_SCOPE, topicPartition.topic())
-                .scopeLabel(PARTITION_SCOPE, String.valueOf(topicPartition.partition()))
-                .scopeLabel(GROUP_SCOPE, groupId)
-                .getCounter(BYTES_OUT)
-                .add(records.sizeInBytes());
-
-        statsLogger.getStatsLogger()
-                .scopeLabel(TOPIC_SCOPE, topicPartition.topic())
-                .scopeLabel(PARTITION_SCOPE, String.valueOf(topicPartition.partition()))
-                .scopeLabel(GROUP_SCOPE, groupId)
-                .getCounter(MESSAGE_OUT)
-                .add(numMessages);
-
-        statsLogger.getStatsLogger()
-                .scopeLabel(TOPIC_SCOPE, topicPartition.topic())
-                .scopeLabel(PARTITION_SCOPE, String.valueOf(topicPartition.partition()))
-                .scopeLabel(GROUP_SCOPE, groupId)
-                .getCounter(ENTRIES_OUT)
-                .add(entrySize);
-    }
 }

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

@@ -45,6 +45,7 @@ public abstract class AbstractEntryFormatter implements EntryFormatter {
     @Override
     public DecodeResult decode(List<Entry> entries, byte magic) {
         int totalSize = 0;
+        int conversionCount = 0;
         // batched ByteBuf should be released after sending to client
         ByteBuf batchedByteBuf = PulsarByteBufAllocator.DEFAULT.directBuffer(totalSize);
         for (Entry entry : entries) {
@@ -63,6 +64,7 @@ public DecodeResult decode(List<Entry> entries, byte magic) {
                         // down converted, batch magic will be set to client magic
                         ConvertedRecords<MemoryRecords> convertedRecords =
                                 memoryRecords.downConvert(magic, startOffset, time);
+                        conversionCount += convertedRecords.recordConversionStats().numRecordsConverted();
 
                         final ByteBuf kafkaBuffer = Unpooled.wrappedBuffer(convertedRecords.records().buffer());
                         totalSize += kafkaBuffer.readableBytes();
@@ -83,6 +85,7 @@ public DecodeResult decode(List<Entry> entries, byte magic) {
                 } else {
                     final DecodeResult decodeResult =
                             ByteBufUtils.decodePulsarEntryToKafkaRecords(metadata, byteBuf, startOffset, magic);
+                    conversionCount += decodeResult.getConversionCount();
                     final ByteBuf kafkaBuffer = decodeResult.getOrCreateByteBuf();
                     totalSize += kafkaBuffer.readableBytes();
                     batchedByteBuf.writeBytes(kafkaBuffer);
@@ -101,7 +104,9 @@ public DecodeResult decode(List<Entry> entries, byte magic) {
         }
 
         return DecodeResult.get(
-                MemoryRecords.readableRecords(ByteBufUtils.getNioBuffer(batchedByteBuf)), batchedByteBuf);
+                MemoryRecords.readableRecords(ByteBufUtils.getNioBuffer(batchedByteBuf)),
+                batchedByteBuf,
+                conversionCount);
     }
 
     protected static boolean isKafkaEntryFormat(final MessageMetadata messageMetadata) {

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

@@ -13,11 +13,22 @@
  */
 package io.streamnative.pulsar.handlers.kop.format;
 
+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.ENTRIES_OUT;
+import static io.streamnative.pulsar.handlers.kop.KopServerStats.GROUP_SCOPE;
+import static io.streamnative.pulsar.handlers.kop.KopServerStats.MESSAGE_OUT;
+import static io.streamnative.pulsar.handlers.kop.KopServerStats.PARTITION_SCOPE;
+import static io.streamnative.pulsar.handlers.kop.KopServerStats.TOPIC_SCOPE;
+
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.util.Recycler;
+import io.streamnative.pulsar.handlers.kop.RequestStats;
+import io.streamnative.pulsar.handlers.kop.stats.StatsLogger;
 import lombok.Getter;
 import lombok.NonNull;
+import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.record.MemoryRecords;
 
 /**
@@ -28,18 +39,22 @@ public class DecodeResult {
     @Getter
     private MemoryRecords records;
     private ByteBuf releasedByteBuf;
+    @Getter
+    private int conversionCount;
 
     private final Recycler.Handle<DecodeResult> recyclerHandle;
 
     public static DecodeResult get(MemoryRecords records) {
-        return get(records, null);
+        return get(records, null, 0);
     }
 
     public static DecodeResult get(MemoryRecords records,
-                                   ByteBuf releasedByteBuf) {
+                                   ByteBuf releasedByteBuf,
+                                   int conversionCount) {
         DecodeResult decodeResult = RECYCLER.get();
         decodeResult.records = records;
         decodeResult.releasedByteBuf = releasedByteBuf;
+        decodeResult.conversionCount = conversionCount;
         return decodeResult;
     }
 
@@ -60,6 +75,7 @@ public void recycle() {
             releasedByteBuf.release();
             releasedByteBuf = null;
         }
+        conversionCount = -1;
         recyclerHandle.recycle(this);
     }
 
@@ -71,4 +87,24 @@ public void recycle() {
         }
     }
 
+    public void updateConsumerStats(final TopicPartition topicPartition,
+                                    int entrySize,
+                                    final String groupId,
+                                    RequestStats statsLogger) {
+        final int numMessages = EntryFormatter.parseNumMessages(records);
+
+        final StatsLogger statsLoggerForThisPartition = statsLogger.getStatsLogger()
+                .scopeLabel(TOPIC_SCOPE, topicPartition.topic())
+                .scopeLabel(PARTITION_SCOPE, String.valueOf(topicPartition.partition()));
+
+        statsLoggerForThisPartition.getCounter(CONSUME_MESSAGE_CONVERSIONS).add(conversionCount);
+
+        final StatsLogger statsLoggerForThisGroup = statsLoggerForThisPartition.scopeLabel(GROUP_SCOPE, groupId);
+
+        statsLoggerForThisGroup.getCounter(BYTES_OUT).add(records.sizeInBytes());
+        statsLoggerForThisGroup.getCounter(MESSAGE_OUT).add(numMessages);
+        statsLoggerForThisGroup.getCounter(ENTRIES_OUT).add(entrySize);
+
+    }
+
 }

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

@@ -13,10 +13,22 @@
  */
 package io.streamnative.pulsar.handlers.kop.format;
 
+import static io.streamnative.pulsar.handlers.kop.KopServerStats.BYTES_IN;
+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.TOPIC_SCOPE;
+
 import io.netty.buffer.ByteBuf;
 import io.netty.util.Recycler;
+import io.streamnative.pulsar.handlers.kop.KafkaTopicManager;
+import io.streamnative.pulsar.handlers.kop.RequestStats;
+import io.streamnative.pulsar.handlers.kop.stats.StatsLogger;
+import io.streamnative.pulsar.handlers.kop.utils.KopTopic;
 import lombok.Getter;
+import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.pulsar.broker.service.Producer;
 
 /**
  * Result of encode in entry formatter.
@@ -27,16 +39,19 @@ public class EncodeResult {
     private MemoryRecords records;
     private ByteBuf encodedByteBuf;
     private int numMessages;
+    private int conversionCount;
 
     private final Recycler.Handle<EncodeResult> recyclerHandle;
 
     public static EncodeResult get(MemoryRecords records,
                                    ByteBuf encodedByteBuf,
-                                   int numMessages) {
+                                   int numMessages,
+                                   int conversionCount) {
         EncodeResult encodeResult = RECYCLER.get();
         encodeResult.records = records;
         encodeResult.encodedByteBuf = encodedByteBuf;
         encodeResult.numMessages = numMessages;
+        encodeResult.conversionCount = conversionCount;
         return encodeResult;
     }
 
@@ -58,7 +73,27 @@ public void recycle() {
             encodedByteBuf = null;
         }
         numMessages = -1;
+        conversionCount = -1;
         recyclerHandle.recycle(this);
     }
 
+    public void updateProducerStats(final TopicPartition topicPartition,
+                                    final RequestStats requestStats) {
+        final int numBytes = encodedByteBuf.readableBytes();
+
+        final Producer producer = KafkaTopicManager.getReferenceProducer(KopTopic.toString(topicPartition));
+        producer.updateRates(numMessages, numBytes);
+        producer.getTopic().incrementPublishCount(numMessages, numBytes);
+
+        final StatsLogger statsLoggerForThisPartition = requestStats.getStatsLogger()
+                .scopeLabel(TOPIC_SCOPE, topicPartition.topic())
+                .scopeLabel(PARTITION_SCOPE, String.valueOf(topicPartition.partition()));
+
+        statsLoggerForThisPartition.getCounter(BYTES_IN).add(numBytes);
+        statsLoggerForThisPartition.getCounter(MESSAGE_IN).add(numMessages);
+        statsLoggerForThisPartition.getCounter(PRODUCE_MESSAGE_CONVERSIONS).add(conversionCount);
+
+        RequestStats.BATCH_COUNT_PER_MEMORY_RECORDS_INSTANCE.set(numMessages);
+    }
+
 }

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

@@ -52,15 +52,19 @@ public EncodeResult encode(final EncodeRequest encodeRequest) {
         final KopLogValidator.CompressionCodec sourceCodec = getSourceCodec(records);
         final KopLogValidator.CompressionCodec targetCodec = getTargetCodec(sourceCodec);
 
-        final MemoryRecords validRecords = KopLogValidator.validateMessagesAndAssignOffsets(records,
-                offset,
-                System.currentTimeMillis(),
-                sourceCodec,
-                targetCodec,
-                false,
-                RecordBatch.MAGIC_VALUE_V2,
-                TimestampType.CREATE_TIME,
-                Long.MAX_VALUE);
+        final ValidationAndOffsetAssignResult validationAndOffsetAssignResult =
+                KopLogValidator.validateMessagesAndAssignOffsets(records,
+                        offset,
+                        System.currentTimeMillis(),
+                        sourceCodec,
+                        targetCodec,
+                        false,
+                        RecordBatch.MAGIC_VALUE_V2,
+                        TimestampType.CREATE_TIME,
+                        Long.MAX_VALUE);
+
+        MemoryRecords validRecords = validationAndOffsetAssignResult.getRecords();
+        int conversionCount = validationAndOffsetAssignResult.getConversionCount();
 
         final int numMessages = EntryFormatter.parseNumMessages(validRecords);
         final ByteBuf recordsWrapper = Unpooled.wrappedBuffer(validRecords.buffer());
@@ -69,8 +73,9 @@ public EncodeResult encode(final EncodeRequest encodeRequest) {
                 getMessageMetadataWithNumberMessages(numMessages),
                 recordsWrapper);
         recordsWrapper.release();
+        validationAndOffsetAssignResult.recycle();
 
-        return EncodeResult.get(validRecords, buf, numMessages);
+        return EncodeResult.get(validRecords, buf, numMessages, conversionCount);
     }
 
     @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);
+        return EncodeResult.get(records, buf, numMessages, 0);
     }
 
     @Override