diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KopServerStats.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KopServerStats.java index 8e8324d1d6..724df9fa1c 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KopServerStats.java +++ b/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. diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/AbstractEntryFormatter.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/AbstractEntryFormatter.java index b08ab28301..f2a7bfef8d 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/AbstractEntryFormatter.java +++ b/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 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 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 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 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 entries, byte magic) { return DecodeResult.get( MemoryRecords.readableRecords(ByteBufUtils.getNioBuffer(batchedByteBuf)), batchedByteBuf, - conversionCount); + conversionCount, + conversionTimeNanos); } protected static boolean isKafkaEntryFormat(final MessageMetadata messageMetadata) { diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/DecodeResult.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/DecodeResult.java index c7390077ad..f065a5a485 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/DecodeResult.java +++ b/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 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); diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/EncodeResult.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/EncodeResult.java index ed410b52bd..6d2cb124ca 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/EncodeResult.java +++ b/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 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); } diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/KafkaMixedEntryFormatter.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/KafkaMixedEntryFormatter.java index 38061d74ca..601f277eb8 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/KafkaMixedEntryFormatter.java +++ b/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 diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/KafkaV1EntryFormatter.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/KafkaV1EntryFormatter.java index 8f106af6bc..63fd7afa2e 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/KafkaV1EntryFormatter.java +++ b/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 diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/PulsarEntryFormatter.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/PulsarEntryFormatter.java index f64609819b..31adff73a0 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/PulsarEntryFormatter.java +++ b/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 diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/ValidationAndOffsetAssignResult.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/ValidationAndOffsetAssignResult.java index 95fb8ae388..271858570c 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/ValidationAndOffsetAssignResult.java +++ b/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 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); } diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ByteBufUtils.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ByteBufUtils.java index 9f372d7fec..19c058b62a 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ByteBufUtils.java +++ b/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 diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KopLogValidator.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KopLogValidator.java index eafc108580..99be6b8269 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KopLogValidator.java +++ b/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 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) { diff --git a/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/format/NoHeaderKafkaEntryFormatter.java b/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/format/NoHeaderKafkaEntryFormatter.java index 1c42278065..892c9965d1 100644 --- a/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/format/NoHeaderKafkaEntryFormatter.java +++ b/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 diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/MetricsProviderTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/MetricsProviderTest.java index 28221b2217..18b2d4c481 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/MetricsProviderTest.java +++ b/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,6 +192,10 @@ 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")); @@ -199,6 +203,10 @@ public void testMetricsProvider() throws Exception { + "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)