Merge branch 'branch-3.0' into branch-3.0.0.1-rc1

Author: BewareMyPower

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

@@ -91,7 +91,7 @@ public void onComplete() {
         replicaManager.readFromLocalLog(
             readCommitted, fetchMaxBytes, maxReadEntriesNum, readPartitionInfo, context
         ).thenAccept(readRecordsResult -> {
-            this.context.getStatsLogger().getWaitingFetchesTriggered().add(1);
+            this.context.getStatsLogger().getWaitingFetchesTriggered().addCount(1);
             this.callback.complete(readRecordsResult);
         }).thenAccept(__ -> {
             // Ensure the old decode result are recycled.

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

@@ -189,7 +189,7 @@ protected boolean channelReady() {
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
         // Get a buffer that contains the full frame
         ByteBuf buffer = (ByteBuf) msg;
-        requestStats.getNetworkTotalBytesIn().add(buffer.readableBytes());
+        requestStats.getNetworkTotalBytesIn().addCount(buffer.readableBytes());
 
         // Update parse request latency metrics
         final BiConsumer<Long, Throwable> registerRequestParseLatency = (timeBeforeParse, throwable) -> {
@@ -448,7 +448,7 @@ protected void writeAndFlushResponseToClient(Channel channel) {
                         if (!future.isSuccess()) {
                             log.error("[{}] Failed to write {}", channel, request.getHeader(), future.cause());
                         } else {
-                            requestStats.getNetworkTotalBytesOut().add(resultSize);
+                            requestStats.getNetworkTotalBytesOut().addCount(resultSize);
                         }
                     });
                     requestStats.getRequestStatsLogger(apiKey, KopServerStats.REQUEST_QUEUED_LATENCY)
@@ -474,7 +474,7 @@ private void sendErrorResponse(KafkaHeaderAndRequest request, Channel channel, T
         final int resultSize = result.readableBytes();
         channel.writeAndFlush(result).addListener(future -> {
             if (future.isSuccess()) {
-                requestStats.getNetworkTotalBytesOut().add(resultSize);
+                requestStats.getNetworkTotalBytesOut().addCount(resultSize);
             }
         });
     }

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

@@ -99,7 +99,7 @@ public void updateConsumerStats(final TopicPartition topicPartition,
 
         final StatsLogger statsLoggerForThisPartition = statsLogger.getStatsLoggerForTopicPartition(topicPartition);
 
-        statsLoggerForThisPartition.getCounter(CONSUME_MESSAGE_CONVERSIONS).add(conversionCount);
+        statsLoggerForThisPartition.getCounter(CONSUME_MESSAGE_CONVERSIONS).addCount(conversionCount);
         statsLoggerForThisPartition.getOpStatsLogger(CONSUME_MESSAGE_CONVERSIONS_TIME_NANOS)
                 .registerSuccessfulEvent(conversionTimeNanos, TimeUnit.NANOSECONDS);
         final StatsLogger statsLoggerForThisGroup;
@@ -108,9 +108,9 @@ public void updateConsumerStats(final TopicPartition topicPartition,
         } else {
             statsLoggerForThisGroup = statsLoggerForThisPartition;
         }
-        statsLoggerForThisGroup.getCounter(BYTES_OUT).add(records.sizeInBytes());
-        statsLoggerForThisGroup.getCounter(MESSAGE_OUT).add(numMessages);
-        statsLoggerForThisGroup.getCounter(ENTRIES_OUT).add(entrySize);
+        statsLoggerForThisGroup.getCounter(BYTES_OUT).addCount(records.sizeInBytes());
+        statsLoggerForThisGroup.getCounter(MESSAGE_OUT).addCount(numMessages);
+        statsLoggerForThisGroup.getCounter(ENTRIES_OUT).addCount(entrySize);
 
     }
 

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

@@ -89,9 +89,9 @@ public void updateProducerStats(final TopicPartition topicPartition,
 
         final StatsLogger statsLoggerForThisPartition = requestStats.getStatsLoggerForTopicPartition(topicPartition);
 
-        statsLoggerForThisPartition.getCounter(BYTES_IN).add(numBytes);
-        statsLoggerForThisPartition.getCounter(MESSAGE_IN).add(numMessages);
-        statsLoggerForThisPartition.getCounter(PRODUCE_MESSAGE_CONVERSIONS).add(conversionCount);
+        statsLoggerForThisPartition.getCounter(BYTES_IN).addCount(numBytes);
+        statsLoggerForThisPartition.getCounter(MESSAGE_IN).addCount(numMessages);
+        statsLoggerForThisPartition.getCounter(PRODUCE_MESSAGE_CONVERSIONS).addCount(conversionCount);
         statsLoggerForThisPartition.getOpStatsLogger(PRODUCE_MESSAGE_CONVERSIONS_TIME_NANOS)
                 .registerSuccessfulEvent(conversionTimeNanos, TimeUnit.NANOSECONDS);
 

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

@@ -14,6 +14,7 @@
 package io.streamnative.pulsar.handlers.kop.stats;
 
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.LongAdder;
 import org.apache.bookkeeper.stats.Counter;
 
@@ -48,10 +49,15 @@ public void dec() {
     }
 
     @Override
-    public void add(long delta) {
+    public void addCount(long delta) {
         counter.add(delta);
     }
 
+    @Override
+    public void addLatency(long l, TimeUnit timeUnit) {
+        // No-op
+    }
+
     @Override
     public Long get() {
         return counter.sum();

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

@@ -88,7 +88,12 @@ public void dec() {
         }
 
         @Override
-        public void add(long delta) {
+        public void addCount(long delta) {
+            // nop
+        }
+
+        @Override
+        public void addLatency(long l, TimeUnit timeUnit) {
             // nop
         }
 

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

@@ -40,8 +40,7 @@
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Message;
-import org.apache.pulsar.client.impl.BatchMessageIdImpl;
-import org.apache.pulsar.client.impl.TopicMessageIdImpl;
+import org.apache.pulsar.client.api.MessageIdAdv;
 import org.apache.pulsar.common.policies.data.RetentionPolicies;
 import org.apache.pulsar.common.util.FutureUtil;
 import org.testng.annotations.AfterClass;
@@ -158,8 +157,7 @@ public void testKafkaProduceMessageOrder(int batchSize) throws Exception {
 
                     consumer.acknowledge(msg);
 
-                    BatchMessageIdImpl id =
-                            (BatchMessageIdImpl) ((TopicMessageIdImpl) msg.getMessageId()).getInnerMessageId();
+                    MessageIdAdv id = (MessageIdAdv) msg.getMessageId();
                     if (id.getBatchIndex() == 0) {
                         numBatches++;
                     }