Fix NPE when send a large message and don't release batchedMessageMetadataAndPayload when discard in batch message container. (#5748)

Fixes #5746 #5747

### Motivation

Fix NPE and release an already released ByteBuf when publish an oversize message.

Here is error log:
```
io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1
	at io.netty.util.internal.ReferenceCountUpdater.toLiveRealRefCnt(ReferenceCountUpdater.java:74) ~[netty-common-4.1.43.Final.jar:4.1.43.Final]
	at io.netty.util.internal.ReferenceCountUpdater.release(ReferenceCountUpdater.java:138) ~[netty-common-4.1.43.Final.jar:4.1.43.Final]
	at io.netty.buffer.AbstractReferenceCountedByteBuf.release(AbstractReferenceCountedByteBuf.java:100) ~[netty-buffer-4.1.43.Final.jar:4.1.43.Final]
	at io.netty.util.ReferenceCountUtil.release(ReferenceCountUtil.java:88) ~[netty-common-4.1.43.Final.jar:4.1.43.Final]
	at io.netty.util.ReferenceCountUtil.safeRelease(ReferenceCountUtil.java:113) [netty-common-4.1.43.Final.jar:4.1.43.Final]
	at org.apache.pulsar.client.impl.BatchMessageKeyBasedContainer$KeyedBatch.discard(BatchMessageKeyBasedContainer.java:244) [classes/:?]
	at org.apache.pulsar.client.impl.BatchMessageKeyBasedContainer.createOpSendMsg(BatchMessageKeyBasedContainer.java:125) [classes/:?]
	at org.apache.pulsar.client.impl.BatchMessageKeyBasedContainer.createOpSendMsgs(BatchMessageKeyBasedContainer.java:145) [classes/:?]
	at org.apache.pulsar.client.impl.ProducerImpl.batchMessageAndSend(ProducerImpl.java:1426) [classes/:?]
	at org.apache.pulsar.client.impl.ProducerImpl.triggerFlush(ProducerImpl.java:1411) [classes/:?]
	at org.apache.pulsar.client.impl.ProducerBase.send(ProducerBase.java:112) [classes/:?]
	at org.apache.pulsar.client.impl.TypedMessageBuilderImpl.send(TypedMessageBuilderImpl.java:89) [classes/:?]
	at org.apache.pulsar.client.impl.ProducerBase.send(ProducerBase.java:63) [classes/:?]
	at org.apache.pulsar.broker.service.BatchMessageTest.testSendOverSizeMessage(BatchMessageTest.java:875) [test-classes/:?]
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_201]
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_201]
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_201]
	at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_201]
	at org.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:124) [testng-6.14.3.jar:?]
	at org.testng.internal.Invoker.invokeMethod(Invoker.java:583) [testng-6.14.3.jar:?]
	at org.testng.internal.Invoker.invokeTestMethod(Invoker.java:719) [testng-6.14.3.jar:?]
	at org.testng.internal.Invoker.invokeTestMethods(Invoker.java:989) [testng-6.14.3.jar:?]
	at org.testng.internal.TestMethodWorker.invokeTestMethods(TestMethodWorker.java:125) [testng-6.14.3.jar:?]
	at org.testng.internal.TestMethodWorker.run(TestMethodWorker.java:109) [testng-6.14.3.jar:?]
	at org.testng.TestRunner.privateRun(TestRunner.java:648) [testng-6.14.3.jar:?]
	at org.testng.TestRunner.run(TestRunner.java:505) [testng-6.14.3.jar:?]
	at org.testng.SuiteRunner.runTest(SuiteRunner.java:455) [testng-6.14.3.jar:?]
	at org.testng.SuiteRunner.runSequentially(SuiteRunner.java:450) [testng-6.14.3.jar:?]
	at org.testng.SuiteRunner.privateRun(SuiteRunner.java:415) [testng-6.14.3.jar:?]
	at org.testng.SuiteRunner.run(SuiteRunner.java:364) [testng-6.14.3.jar:?]
	at org.testng.SuiteRunnerWorker.runSuite(SuiteRunnerWorker.java:52) [testng-6.14.3.jar:?]
	at org.testng.SuiteRunnerWorker.run(SuiteRunnerWorker.java:84) [testng-6.14.3.jar:?]
	at org.testng.TestNG.runSuitesSequentially(TestNG.java:1208) [testng-6.14.3.jar:?]
	at org.testng.TestNG.runSuitesLocally(TestNG.java:1137) [testng-6.14.3.jar:?]
	at org.testng.TestNG.runSuites(TestNG.java:1049) [testng-6.14.3.jar:?]
	at org.testng.TestNG.run(TestNG.java:1017) [testng-6.14.3.jar:?]
	at org.testng.IDEARemoteTestNG.run(IDEARemoteTestNG.java:73) [testng-plugin.jar:?]
	at org.testng.RemoteTestNGStarter.main(RemoteTestNGStarter.java:123) [testng-plugin.jar:?]



16:19:13.850 [main:org.apache.pulsar.client.impl.ProducerImpl@1439] WARN  org.apache.pulsar.client.impl.ProducerImpl - [persistent://prop/ns-abc/testSendOverSizeMessage-623833fc-d9f7-4b28-aead-27955928fae9] [test-0-0] error while create opSendMsg by batch message container
java.lang.NullPointerException: null
	at org.apache.pulsar.client.impl.ProducerImpl.releaseSemaphoreForSendOp(ProducerImpl.java:858) ~[classes/:?]
	at org.apache.pulsar.client.impl.ProducerImpl.processOpSendMsg(ProducerImpl.java:1477) ~[classes/:?]
	at org.apache.pulsar.client.impl.ProducerImpl.batchMessageAndSend(ProducerImpl.java:1432) [classes/:?]
	at org.apache.pulsar.client.impl.ProducerImpl.triggerFlush(ProducerImpl.java:1411) [classes/:?]
	at org.apache.pulsar.client.impl.ProducerBase.send(ProducerBase.java:112) [classes/:?]
	at org.apache.pulsar.client.impl.TypedMessageBuilderImpl.send(TypedMessageBuilderImpl.java:89) [classes/:?]
	at org.apache.pulsar.client.impl.ProducerBase.send(ProducerBase.java:63) [classes/:?]
	at org.apache.pulsar.broker.service.BatchMessageTest.testSendOverSizeMessage(BatchMessageTest.java:875) [test-classes/:?]
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_201]
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_201]
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_201]
	at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_201]
	at org.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:124) [testng-6.14.3.jar:?]
	at org.testng.internal.Invoker.invokeMethod(Invoker.java:583) [testng-6.14.3.jar:?]
	at org.testng.internal.Invoker.invokeTestMethod(Invoker.java:719) [testng-6.14.3.jar:?]
	at org.testng.internal.Invoker.invokeTestMethods(Invoker.java:989) [testng-6.14.3.jar:?]
	at org.testng.internal.TestMethodWorker.invokeTestMethods(TestMethodWorker.java:125) [testng-6.14.3.jar:?]
	at org.testng.internal.TestMethodWorker.run(TestMethodWorker.java:109) [testng-6.14.3.jar:?]
	at org.testng.TestRunner.privateRun(TestRunner.java:648) [testng-6.14.3.jar:?]
	at org.testng.TestRunner.run(TestRunner.java:505) [testng-6.14.3.jar:?]
	at org.testng.SuiteRunner.runTest(SuiteRunner.java:455) [testng-6.14.3.jar:?]
	at org.testng.SuiteRunner.runSequentially(SuiteRunner.java:450) [testng-6.14.3.jar:?]
	at org.testng.SuiteRunner.privateRun(SuiteRunner.java:415) [testng-6.14.3.jar:?]
	at org.testng.SuiteRunner.run(SuiteRunner.java:364) [testng-6.14.3.jar:?]
	at org.testng.SuiteRunnerWorker.runSuite(SuiteRunnerWorker.java:52) [testng-6.14.3.jar:?]
	at org.testng.SuiteRunnerWorker.run(SuiteRunnerWorker.java:84) [testng-6.14.3.jar:?]
	at org.testng.TestNG.runSuitesSequentially(TestNG.java:1208) [testng-6.14.3.jar:?]
	at org.testng.TestNG.runSuitesLocally(TestNG.java:1137) [testng-6.14.3.jar:?]
	at org.testng.TestNG.runSuites(TestNG.java:1049) [testng-6.14.3.jar:?]
	at org.testng.TestNG.run(TestNG.java:1017) [testng-6.14.3.jar:?]
	at org.testng.IDEARemoteTestNG.run(IDEARemoteTestNG.java:73) [testng-plugin.jar:?]
	at org.testng.RemoteTestNGStarter.main(RemoteTestNGStarter.java:123) [testng-plugin.jar:?]
```

### Modifications

Add check for processOpSendMsg, if the op is null, just return.
Don't release the batchedMessageMetadataAndPayload since it is already released in getCompressedBatchMetadataAndPayload() method.

### Motivation

Remove cursor from cursors of managed ledger while remove non-durable subscription.  The data deletion is depends the mark delete position of all cursors, if left a unused cursor in the cursors of managed ledger, data can't be delete as expected.

### Modifications

Remove cursor while remove non-durable subscription

Fixes #3009

*Motivation*

There are no REST APIs usage docs for the presto SQL.

* Fix Issue 5614, Deprecate brokerServicePurgeInactiveFrequencyInSeconds

* [Issue 5614] Deprecate brokerServicePurgeInactiveFrequencyInSeconds

Updated Athenz libraries to the latest version. The latest version of Athenz seems to include some bug fixes.
https://github.com/yahoo/athenz/releases

[Pulsar-sql]Using pulsar SQL query messages will appear `NoSuchLedger` when zk root directory changed (#5001)

Fixes #4715 

*Motivation*

When zookeeper ledgers root path is changed, using pulsar-sql to query messages will cause `BKNoSuchLedgerExistsException`.

*Modifications*

Make bookie using `setMetadataServiceUri` to set zookeeper uri.

Fixes #5729 

### Motivation

Picture and text are not match.

### Modifications
changed the ```Consumer-A``` to ```Consumer A-0```

### Motivation
Currently, the following features are not available in Go Functions: SerDe, metrics, state storage, user config. 

### Modifications
Add the language-specific tab for Go, and add a sentence "the feature is not available in Go".

### Modifications

Out-of-date instructions in pulsar-client-cpp/README.md about how to start test server

Broker publish limiter was not start at broker start time, and only init when dynamic system config happens. 
This change init broker publish limiter at broker start time, and add a test for it.

### Motivation

Added the class SensorReading and class ProducerSparkWithPojo  to showcase writing Pojo as Json Object to Pulsar Queue 

### Modifications

Based on the documentation written example class to write Pojos to Pulsar 
This might be more real life examples for application than bytes stream 

Fixes #5287 

### Motivation
Users do not know how to specify cluster (multi broker/zk) connection for presto.

### Modifications
Add content on how to configure multiple hosts for `pulsar.broker-service-url`.

# Motivation

A deadlock on PulsarClient.close() can happen if there are producers/consumers that are not closed prior to calling PulsarClient.close() and we have to close them in the routine.  The deadlock happens be cause we piggy pack off of a "pulsar-io" thread that is used to shutdown the connection in producer/consumer close, to also shutdown the EventLoopGroup.  "pulsar-io" thread is part of the EventLoopGroup thus it tries to shutdown itself and causes a deadlock.  

Below is a stacktrace of what it look like:

```
"pulsar-client-io-1-1" #20 prio=5 os_prio=31 tid=0x00007fc312a78800 nid=0x9a03 in Object.wait() [0x000070000384e000]
   java.lang.Thread.State: WAITING (on object monitor)
    at java.lang.Object.wait(Native Method)
    - waiting on <0x00000007b8532640> (a io.netty.util.concurrent.DefaultPromise)
    at java.lang.Object.wait(Object.java:502)
    at io.netty.util.concurrent.DefaultPromise.await(DefaultPromise.java:252)
    - locked <0x00000007b8532640> (a io.netty.util.concurrent.DefaultPromise)
    at io.netty.util.concurrent.DefaultPromise.sync(DefaultPromise.java:403)
    at io.netty.util.concurrent.DefaultPromise.sync(DefaultPromise.java:35)
    at org.apache.pulsar.client.impl.ConnectionPool.close(ConnectionPool.java:286)
    at org.apache.pulsar.client.impl.PulsarClientImpl.shutdown(PulsarClientImpl.java:578)
    at org.apache.pulsar.client.impl.PulsarClientImpl.lambda$closeAsync$18(PulsarClientImpl.java:560)
    at org.apache.pulsar.client.impl.PulsarClientImpl$$Lambda$82/878861517.run(Unknown Source)
    at java.util.concurrent.CompletableFuture.uniRun(CompletableFuture.java:705)
    at java.util.concurrent.CompletableFuture$UniRun.tryFire(CompletableFuture.java:687)
    at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
    at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962)
    at org.apache.pulsar.client.impl.ProducerImpl.lambda$closeAsync$9(ProducerImpl.java:735)
    at org.apache.pulsar.client.impl.ProducerImpl$$Lambda$80/1123226989.apply(Unknown Source)
    at java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:822)
    at java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:797)
    at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
    at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962)
    at org.apache.pulsar.client.impl.ClientCnx.handleSuccess(ClientCnx.java:406)
    at org.apache.pulsar.common.protocol.PulsarDecoder.channelRead(PulsarDecoder.java:222)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
    at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:326)
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:300)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1422)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:931)
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:700)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:635)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:552)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:514)
    at io.netty.util.concurrent.SingleThreadEventExecutor$6.run(SingleThreadEventExecutor.java:1050)
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.lang.Thread.run(Thread.java:748)
```

This deadlock went unnoticed for a long time because in the past we never waited for the shutdown of the EventLoopGroup to complete

Also #5628 does not solve the issue.  This PR should supersede that one.

Fixes #5513 

### Motivation

Through #3985, user could set the publish rate for each topic, but the topic number for each broker is not limited, so there is case that a lot of topics served in same broker, and if each topic send too many message, it will cause the messages not able to send to BookKeeper in time,  and messages be hold in the direct memory of broker, and cause Broker out of direct memory.

### Modifications

- add broker publish rate limit base on #3985,
- add unit test.

### Verifying this change
unit test passed.

In TypedMessageBuilderImpl::beforeSend function, when setting TxnIdMostBits field of msgMetadataBuilder, it sets into TxnidLeastBits field and override the original TxnIdLeastBits value.

Resolved issue #5728

[Issues 5692]Add document for mongodb connector

Fixes #5692 

Fixes #5676

### Motivation


Expose the redelivery count in the C++ client.

### Modifications

Exposed the redelivery count from the broker in `Message` and `MessageImpl`. Set the counter when receiving messages. Added test.

### Motivation

Even after the synchronous close() method is called on a PulsarClient, netty thread(s) from the EventLoopGroup used in the ConnectionPool may still linger around for another 2 seconds afterwards due to not wait for the the graceful shutdown to complete.

While this usually doesn't cause any problems for a normal use case, it does cause problems in the Pulsar Flink sources and sinks.  Exceptions like the following can happen when the Flink job with a Pulsar source or sink gets closed:

```
Exception in thread "pulsar-client-io-1-1" java.lang.NoClassDefFoundError: org/apache/pulsar/shade/io/netty/buffer/PoolArena$1
	at org.apache.pulsar.shade.io.netty.buffer.PoolArena.freeChunk(PoolArena.java:293)
	at org.apache.pulsar.shade.io.netty.buffer.PoolThreadCache$MemoryRegionCache.freeEntry(PoolThreadCache.java:460)
	at org.apache.pulsar.shade.io.netty.buffer.PoolThreadCache$MemoryRegionCache.free(PoolThreadCache.java:430)
	at org.apache.pulsar.shade.io.netty.buffer.PoolThreadCache$MemoryRegionCache.free(PoolThreadCache.java:422)
	at org.apache.pulsar.shade.io.netty.buffer.PoolThreadCache.free(PoolThreadCache.java:279)
	at org.apache.pulsar.shade.io.netty.buffer.PoolThreadCache.free(PoolThreadCache.java:270)
	at org.apache.pulsar.shade.io.netty.buffer.PoolThreadCache.free(PoolThreadCache.java:241)
	at org.apache.pulsar.shade.io.netty.buffer.PooledByteBufAllocator$PoolThreadLocalCache.onRemoval(PooledByteBufAllocator.java:450)
	at org.apache.pulsar.shade.io.netty.buffer.PooledByteBufAllocator$PoolThreadLocalCache.onRemoval(PooledByteBufAllocator.java:426)
	at org.apache.pulsar.shade.io.netty.util.concurrent.FastThreadLocal.remove(FastThreadLocal.java:271)
	at org.apache.pulsar.shade.io.netty.util.concurrent.FastThreadLocal.removeAll(FastThreadLocal.java:67)
	at org.apache.pulsar.shade.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:32)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.ClassNotFoundException: org.apache.pulsar.shade.io.netty.buffer.PoolArena$1
	at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
	at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$ChildFirstClassLoader.loadClass(FlinkUserCodeClassLoaders.java:129)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
	... 13 more
```

This is because threads from the EventLoopGroup lingers after the pulsar client is closed and the Flink job is unloaded.

### Modifications

Shorten the graceful shutdown period and wait for it to complete

### Motivation

Expose lastExpireTimestamp for subscription stats. Can be used to troubleshooting TTL related issues

### Verifying this change

Added new unit tests

Fixes #5658 

### Motivation
Fix prometheus access control not enough in kubernetes

### Modifications
Add service account for pulsar cluster prometheus, service account binding cluster role, cluster role have  access control.

### Motivation
The structure and content needs improvement.

### Modifications
1. Improve the structure.
2. Improve the deployment steps.

Fixes https://github.com/apache/pulsar/issues/5634


Master Issue: https://github.com/apache/pulsar/issues/5634

### Motivation

In java clients, when we call the `receive`, we will block it until a message arrives. in python clients, when we call the `receive` function, we add a delay parameter of 100ms. when the queue size is 0, the `receive` will have a strict check on the queue size, causing the following exception to be thrown

```
Traceback (most recent call last):
  File "tst.py", line 10, in <module>
    msg = consumer.receive()
  File "/python3.7/site-packages/pulsar/__init__.py", line 930, in receive
    msg = self._consumer.receive()
Exception: Pulsar error: InvalidConfiguration
```
### Modifications

* Removing timeout parameter in synchronous `receive`
* Add test for queue size is 0

### Verifying this change

Add Test

Fixes #5698

Motivation
Since #5571 handle the generated producer name, the replicator producer was created by broker, only one producer for a replicated topic.

So, we can handle it simple by considered replicator producer as generated name producer.

Fix https://github.com/apache/pulsar/issues/5667

Get configurations from [HbaseSinkConfig.java](https://github.com/apache/pulsar/blob/master/pulsar-io/hbase/src/main/java/org/apache/pulsar/io/hbase/sink/HbaseSinkConfig.java) and [HbaseAbstractConfig.java](https://github.com/apache/pulsar/blob/master/pulsar-io/hbase/src/main/java/org/apache/pulsar/io/hbase/HbaseAbstractConfig.java)

Get configuration examples from [sinkConfig.yaml](https://github.com/apache/pulsar/blob/master/pulsar-io/hbase/src/test/resources/sinkConfig.yaml)

### Motivation
Currently when running functions as kubernetes pods, the grpc/metrics port that the container binds to is hardcoded. This pr allows this to be configurable.

### Motivation

Services started in integration tests are using a very aggressive GC config, with many threads and trading more memory for pause times. 

Additionally, multiple variables are being passed in the wrong way to the bookie container, resulting in bookies using the default Xmx of 2 GB.

### Motivation
There are some redundant and unclear description in the Overview.

### Modifications
1. Refine it to be clear and concise.
2. Adopt minimalism writing style.

### Motivation
The structure is not quite clear.

### Modifications
Refine the structure, title and content, make it more clear to follow.

*Motivation*

ProxyLookupThrottlingTest relies on the number of permits for testing the behavior of current lookup.
We should reset the proxyService for each run.

*Modifications*

Change the setup and cleanup methods from `BeforeClass/AfterClass` to `BeforeMethod/AfterMethod`.

Fix https://github.com/apache/pulsar/issues/5667

Motivation
Support messages batch receiving, some application scenarios can be made simpler. Users often increase application throughput through batch operations. For example, batch insert or update database.

At present, we provide the ability to receive a single message. If users want to take advantage of batch operating advantages, need to implement a message collector him self. So this proposal aims to provide a universal interface and mechanism for batch receiving messages.

For example:
```
Messages messages = consumer.batchReceive();
insertToDB(messages);
consumer.acknowledge(messages);
Verifying this change
Added new UT to verify this change.
```

when i start bookie using `bin/pulsar-daemon start bookie` command, i found the bookie environment variable defined in conf/bkenv.sh has no effect, but using $PULSAR_MEM and $PULSAR_GC instead.

Then i check the bin/pulsar shell script, i found as follow:
```
elif [ $COMMAND == "bookie" ]; then
    PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"bookkeeper.log"}
    # Pass BOOKIE_EXTRA_OPTS option defined in pulsar_env.sh
    OPTS="$OPTS $BOOKIE_EXTRA_OPTS"
    exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.bookkeeper.proto.BookieServer --conf $PULSAR_BOOKKEEPER_CONF $@
```
but in pulsar_env.sh, the `$BOOKIE_EXTRA_OPTS` defined as follow:
```
# Set BOOKIE_EXTRA_OPTS option here to ensure that all pulsar scripts can work seamless with bookkeeper

# Extra options to be passed to the jvm
BOOKIE_EXTRA_OPTS="${BOOKIE_EXTRA_OPTS} -Dio.netty.leakDetectionLevel=disabled -Dio.netty.recycler.maxCapacity.default=1000 -Dio.netty.recycler.linkCapacity=1024"
```
`$BOOKIE_EXTRA_OPTS` do not define `$BOOKIE_MEM` and `$BOOKIE_GC`, then the pulsar script will using `$PULSAR_MEM` and `$PULSAR_GC` instead to start bookkeeper.

So, i suggest to load conf/bkenv.sh in bin/pulsar script and using `$BOOKIE_EXTRA_OPTS` in conf/bkenv.sh instread of `$BOOKIE_EXTRA_OPTS` in bin/pulsar_env.sh

Fixes #5533 

*Motivation*

When we using command-line tools, it hangs after type an invalid args.

*Modifications*

We don't need to parse the command, just show the command usage.