Revert "bring back publish throttling"

This reverts commit bc6a0a2.

Author: mukesh-ctds

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

@@ -16,9 +16,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Maps;
 import io.netty.buffer.ByteBuf;
-import io.netty.channel.ChannelHandlerContext;
 import io.netty.util.Recycler;
-import io.netty.util.concurrent.FastThreadLocal;
 import io.streamnative.pulsar.handlers.kop.KafkaServiceConfiguration;
 import io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager;
 import io.streamnative.pulsar.handlers.kop.KafkaTopicLookupService;
@@ -87,7 +85,6 @@
 import org.apache.kafka.common.record.RecordBatch;
 import org.apache.kafka.common.requests.FetchResponse;
 import org.apache.kafka.common.utils.Time;
-import org.apache.pulsar.broker.service.Topic;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.broker.service.plugin.EntryFilter;
 import org.apache.pulsar.common.naming.TopicName;
@@ -138,7 +135,6 @@ public CompletableFuture<KeyValueSchemaIds> getSchemaIds(String topic, BytesSche
     private final KafkaTopicLookupService kafkaTopicLookupService;
 
     private final List<EntryFilter> entryFilters;
-    private final boolean preciseTopicPublishRateLimitingEnable;
 
     private final ProducerStateManagerSnapshotBuffer producerStateManagerSnapshotBuffer;
 
@@ -172,7 +168,6 @@ public PartitionLog(KafkaServiceConfiguration kafkaConfig,
         this.time = time;
         this.topicPartition = topicPartition;
         this.fullPartitionName = fullPartitionName;
-        this.preciseTopicPublishRateLimitingEnable = kafkaConfig.isPreciseTopicPublishRateLimiterEnable();
         this.kafkaTopicLookupService = kafkaTopicLookupService;
         this.producerStateManagerSnapshotBuffer = producerStateManagerSnapshotBuffer;
         this.recoveryExecutor = recoveryExecutor.chooseThread(fullPartitionName);
@@ -898,8 +893,6 @@ private void publishMessages(final CompletableFuture<Long> appendFuture,
                                  final LogAppendInfo appendInfo,
                                  final EncodeResult encodeResult,
                                  final AppendRecordsContext appendRecordsContext) {
-        checkAndRecordPublishQuota(persistentTopic, appendInfo.validBytes(),
-                appendInfo.numMessages(), appendRecordsContext);
         if (persistentTopic.isSystemTopic()) {
             encodeResult.recycle();
             log.error("Not support producing message to system topic: {}", persistentTopic);
@@ -944,37 +937,6 @@ private void publishMessages(final CompletableFuture<Long> appendFuture,
         });
     }
 
-    private void checkAndRecordPublishQuota(Topic topic, int msgSize, int numMessages,
-                                              AppendRecordsContext appendRecordsContext) {
-        final boolean isPublishRateExceeded;
-        if (preciseTopicPublishRateLimitingEnable) {
-            boolean isPreciseTopicPublishRateExceeded =
-                    topic.isTopicPublishRateExceeded(numMessages, msgSize);
-            if (isPreciseTopicPublishRateExceeded) {
-                topic.disableCnxAutoRead();
-                return;
-            }
-            isPublishRateExceeded = topic.isBrokerPublishRateExceeded();
-        } else {
-            if (topic.isResourceGroupRateLimitingEnabled()) {
-                final boolean resourceGroupPublishRateExceeded =
-                        topic.isResourceGroupPublishRateExceeded(numMessages, msgSize);
-                if (resourceGroupPublishRateExceeded) {
-                    topic.disableCnxAutoRead();
-                    return;
-                }
-            }
-            isPublishRateExceeded = topic.isPublishRateExceeded();
-        }
-
-        if (isPublishRateExceeded) {
-            ChannelHandlerContext ctx = appendRecordsContext.getCtx();
-            if (ctx != null && ctx.channel().config().isAutoRead()) {
-                ctx.channel().config().setAutoRead(false);
-            }
-        }
-    }
-
     /**
      * Publish message to bookkeeper.
      * When the message is control message, then it will not do the message deduplication.