[BUG] NPE while enable BrokerPublishThrottling.

[jiazhai]

Describe the bug
while set

brokerPublisherThrottlingMaxByteRate

it will meet error:

08:38:34.452 [pulsar-broker-publish-rate-limiter-monitor-30-1] ERROR org.apache.bookkeeper.common.util.SafeRunnable - Unexpected throwable caught
java.lang.NullPointerException: null
        at org.apache.pulsar.broker.service.ServerCnx.enableCnxAutoRead(ServerCnx.java:1605) ~[org.apache.pulsar-pulsar-broker-2.5.0.jar:2.5.0]
        at org.apache.pulsar.broker.service.AbstractTopic.lambda$enableProducerRead$5(AbstractTopic.java:295) ~[org.apache.pulsar-pulsar-broker-2.5.0.jar:2.5.0]
        at java.util.concurrent.ConcurrentHashMap$ValuesView.forEach(ConcurrentHashMap.java:4707) ~[?:1.8.0_242]
        at org.apache.pulsar.broker.service.AbstractTopic.enableProducerRead(AbstractTopic.java:295) ~[org.apache.pulsar-pulsar-broker-2.5.0.jar:2.5.0]
        at org.apache.pulsar.broker.service.AbstractTopic.resetBrokerPublishCountAndEnableReadIfRequired(AbstractTopic.java:286) ~[org.apache.pulsar-pulsar-broker-2.5.0.jar:2.5.0]
        at org.apache.pulsar.broker.service.BrokerService.lambda$refreshBrokerPublishRate$39(BrokerService.java:1138) ~[org.apache.pulsar-pulsar-broker-2.5.0.jar:2.5.0]
        at org.apache.pulsar.broker.service.BrokerService.lambda$forEachTopic$40(BrokerService.java:1148) ~[org.apache.pulsar-pulsar-broker-2.5.0.jar:2.5.0]
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.forEach(ConcurrentOpenHashMap.java:385) ~[org.apache.pulsar-pulsar-common-2.5.0.jar:2.5.0]
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.forEach(ConcurrentOpenHashMap.java:159) ~[org.apache.pulsar-pulsar-common-2.5.0.jar:2.5.0]
        at org.apache.pulsar.broker.service.BrokerService.forEachTopic(BrokerService.java:1145) ~[org.apache.pulsar-pulsar-broker-2.5.0.jar:2.5.0]
        at org.apache.pulsar.broker.service.BrokerService.refreshBrokerPublishRate(BrokerService.java:1138) ~[org.apache.pulsar-pulsar-broker-2.5.0.jar:2.5.0]
        at org.apache.pulsar.broker.service.BrokerService.lambda$setupBrokerPublishRateLimiterMonitor$9(BrokerService.java:509) ~[org.apache.pu

This is caused by the mock Producer that we registered for bundle load/unload handling.

    /**
     * it sets cnx auto-readable if producer's cnx is disabled due to publish-throttling
     */
    protected void enableProducerRead() {
        if (producers != null) {
            producers.values().forEach(producer -> producer.getCnx().enableCnxAutoRead());
        }
    }


    public void enableCnxAutoRead() {
        // we can add check (&& pendingSendRequest < MaxPendingSendRequests) here but then it requires
        // pendingSendRequest to be volatile and it can be expensive while writing. also this will be called on if
        // throttling is enable on the topic. so, avoid pendingSendRequest check will be fine.
        if (!ctx.channel().config().isAutoRead() && autoReadDisabledRateLimiting) {
            // Resume reading from socket if pending-request is not reached to threshold
            ctx.channel().config().setAutoRead(true);
            // triggers channel read
            ctx.read();
            autoReadDisabledRateLimiting = false;
        }
    }

The code may meet error in "ctx.channel().config().isAutoRead()". we need to provide related fork. for InternalProducer and InternalServerCnx.