Modify the logical in producer carry a schema to connect broker with isAllowAutoUpdateSchema.  (#5782)

### Motivation
Now, the logical is if isAllowAutoUpdateSchema = false, producer can't connect broker. So modify the logical, when isAllowAutoUpdateSchema = false, the first to check the schema whether is registered.

### Verifying this change
Add the tests for it

Signed-off-by: Nxiaolong.ran <rxl@apache.org>

* Update website for 2.4.2 release
Signed-off-by: Nxiaolong.ran <rxl@apache.org>

* update website docs for release 2.4.2
Signed-off-by: Nxiaolong.ran <rxl@apache.org>

### Motivation

fix potential NPE and repeated conditional test

### Modifications

local changes

Signed-off-by: Nxiaolong.ran <rxl@apache.org>

Fixes #5687 

### Motivation

When the user wants to add new keys for Env, adding fails if no prefix is added. 

Currently, add new keys for Env use the script of [apply-config-from-env.py](https://github.com/apache/pulsar/commits/master/docker/pulsar/scripts/apply-config-from-env.py), to ensure that the env set by the user can take effect, add the prefix(**PULSAR_PREFIX_**) for all keys.

### Modifications

- Add prefix for new keys from Env

Signed-off-by: Nxiaolong.ran <rxl@apache.org>
Signed-off-by: Nxiaolong.ran <rxl@apache.org>

Fixes #5787 

### Motivation

When we creating a K8S cluster on Minikube, due to the different versions of Minikube in the local environment, the installation fails on `--kubernetes-version=v1.10.5`.

### Modifications

- Remove the `--kubernetes-version=v1.10.5` in docs.

Fixes #5562 

Update the schema compatibility check documentation to follow the latest changes from #5227 

Updated jackson libraries to the latest version. There is a security vulnerability in `jackson-databind` currently used by Pulsar.
https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2019-17531

### Motivation
Modify the logical for getting schema compatibility strategy, if schemaCompatibilityStrategy = SchemaCompatibilityStrategy.UNDEFINED return the SchemaAutoUpdateCompatibilityStrategy. 

### Verifying this change
Add the tests for it

StrEntry variable should be cleared after finish addLogTopicHandler()
function regardless the log messages are appended to logger or not. If
it is not cleared, it causes memory leak because StrEntry has grown
indefinitely. Moreover, if the function set --log-topic, then the topic
could get accumulated huge messages that cause ProducerQueueIsFull
error.

* Update security-authorization.md

Fix the link issue #34

see ["Enabling TLS Authentication on Proxies"](security-tls-authentication.md#on-proxies)

* Fixes #5753

Fixes the link for "Enabling TLS Authentication on Proxies".

* Update security-authorization.md

* [Issue 5597][pulsar-client-java]retry when getPartitionedTopicMetadata failed
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* remove unuse code
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* fix ci error

If `killAfterIdleMs` is 0, time.After() function call makes memory leak.
Refer to https://golang.org/pkg/time/#After. According to the document,
the underlying Timer object is not recovered by the garbage collector if
the timer is not fired. And It seems that if the duration is 0, it will
not be released. Therefore, use one Timer object for idle time checking
explicitly instead of the time.After() function.

* Fixes the dead link

changed to: 
see ["Enabling TLS Authentication on Proxies"](security-tls-authentication.md#enable-tls-authentication-on-proxies)

Fixes #5772 

### Motivation

I would like to be able to build Pulsar with at least JDK12.

### Modifications

Bumping lombok dep.

### Verifying this change


This change is a trivial rework / code cleanup without any test coverage.

*Motivation*

Add a tool to collect load report on a specific broker.
This is used for troubleshooting problems in a production cluster.

*Modifications*

Add a broker tool for operations of a specific broker.

### Motivation

As a sub-project of the pulsar, pulsar-manager should update relevant content on the website after the first release is completed, so as to facilitate users' use.

### Modifications

* Add pulsar-manager release notes page
* Add pulsar-manager download page
* Add a document in sidebar

### Verifying this change

yarn build pass

*Motivation*

ManagedLedgerFactoryChangeLedgerPathTest sets the zk session timeout to 10 ms.

10ms is too low for a zookeeper client to connect to the zookeeper server.

*Modifications*

Remove zkSessionTimeout and use the default

*Motivation*

ManagedLedgerFactoryChangeLedgerPathTest sets the zk session timeout to 10 ms.

10ms is too low for a zookeeper client to connect to the zookeeper server.

*Modifications*

Remove zkSessionTimeout and use the default

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