Master Issue: #6892
## Motivation

when config set-publish-rate on namespaces，then can limit publish rate, but when restart broker the limit has expired.

## Modifications

modify get the acquisition policy sync way, any namespaces will save a policy on zk.
```java
        try {
            policies = brokerService.pulsar().getConfigurationCache().policiesCache()
                    .get(AdminResource.path(POLICIES, TopicName.get(topic).getNamespace()))
                    .orElseGet(() -> new Policies());
        } catch (Exception e) {
            log.warn("[{}] Error getting policies {} and publish throttling will be disabled", topic, e.getMessage());
        }
```
(cherry picked from commit 1aaa1481)

Fixes #6841
Co-authored-by: NXiaopeng Zhang <xiaopengzhang@Xiaopengs-MacBook-Pro.local>
Co-authored-by: NJia Zhai <zhaijia@apache.org>
(cherry picked from commit 3198f4ff)

Master Issue: #6876

Motivation
the brokerPublisherThrPottlingTickTimeMillis config spelling mistake

Modifications
broker.conf

  # Tick time to schedule task that checks broker publish rate limiting across all topics
# Reducing to lower value can give more accuracy while throttling publish but
# it uses more CPU to perform frequent check. (Disable publish throttling with value 0)
brokerPublisherThrottlingTickTimeMillis=50

* fix brokerPublisherThrottlingTickTimeMillis in broker.conf

* fix brokerPublisherThrottlingTickTimeMillis in standalone.conf

Co-authored-by: dezhiliu <dezhiliu@tencent.com>(cherry picked from commit 65dc9d99)

Fixes #6887

### Motivation
The ttl for namespaces should be retrieved from broker's configuration if it is not configured at namespace policies. However, the current code only returns the value stored in namespace policies directly without judging if it is configured or not.

### Modifications

Added a condition to test if ttl is configured at namespace policies. If not, retrieve value stored in broker's configuration and return it as output.

* fixes "ttlDurationDefaultInSeconds is not applied"

* testcase added for getTtlDurationDefaultInSeconds
(cherry picked from commit a24203cc)

### Motivation

The subscription `InitialPosition` is not currently set when using the multi-topic or regex consumers in C++/Python.

That makes that if you try to start from `MessageId::earliest`, it would be ignored.

* [C++] Subscription InitialPosition is not correctly set on regex consumers

* fix test fail for topic name

* fix `make format`

Co-authored-by: Jia Zhai <zhaijia@apache.org>(cherry picked from commit 0b803a83)

[ISSUE 6563][Broker] Invalidate managed ledgers zookeeper cache instead of reloading on watcher triggered (#6659)

Fixes #6563

### Motivation
Frequent topics creation/deletion triggers zookeeper children cache reloading for z-nodes **/managed-ledgers/<tenant_name>/<cluster_name>/<namespace_name>/persistent** more than needed.
This creates additional load on zookeeper and broker, slows down brokers and makes them less stable. Also this causes scalability issues - adding more brokers increases operations duration even more.

* [ISSUE 6563][Broker] Invalidate managed ledgers zookeeper cache instead of reloading on watcher triggered

* [ISSUE 6563][Broker] Adding licence header to the new class

* [ISSUE 6563][Broker] Invalidate correct zk cache path

* [ISSUE 6563][Broker] Fix mocking issue
Co-authored-by: NPavel Tishkevich <pavel.tishkevich@onde.app>
Co-authored-by: penghui <penghui@apache.org>(cherry picked from commit f4fc7994)

(cherry picked from commit 90dba138)

related with #6853
add keystore tls config doc
(cherry picked from commit fd6f772d)

### Motivation

If messages were sent to a partitioned topic, the message id's `partition` field was always -1 because SendReceipt command only contains ledger id and entry id.

### Modifications

- Add a `partition` field to `ProducerImpl` and set the `MessageId`'s `partition` field with it in `ackReceived` method later.
- Add a test to check message id in send callback if messages were sent to a partitioned topic.

(cherry picked from commit 15cb920b)

### Motivation

We need to increase producers or consumers when partitions updated.

Java client has implemented this feature, see [#3513](https://github.com/apache/pulsar/pull/3513). This PR trys to implement the same feature in C++ client.

### Modifications

- Add a `boost::asio::deadline_timer` to `PartitionedConsumerImpl` and `PartitionedProducerImpl` to register lookup task to detect partitions changes periodly;
- Add an `unsigned int` configuration parameter to indicate the period seconds of detecting partitions change (default: 60 seconds);
- Unlock the `mutex_` in `PartitionedConsumerImpl::receive` after `state_` were checked.
> Explain: When new consumers are created, `handleSinglePartitionConsumerCreated` will be called finally, which tried to lock the `mutex_`. It may happen that `receive` acquire the lock again and again so that `handleSinglePartitionConsumerCreated`
are blocked in `lock.lock()` for a long time.

* auto update topic partitions extend for consumer and producer in c++ client

* add c++ unit test for partitions update

* format code with clang-format-5.0

* stop partitions update timer after producer/consumer called closeAsync()

* fix bugs when running gtest-parallel

* fix bug: Producer::flush() may cause deadlock

* use getters to read `numPartitions` with or without lock
(cherry picked from commit 30934e16)

* [Docs] Fix the Create subscribtion swagger of PersistentTopic
Signed-off-by: Nxiaolong.ran <rxl@apache.org>

* fix a little
Signed-off-by: Nxiaolong.ran <rxl@apache.org>

* fix a little
Signed-off-by: Nxiaolong.ran <rxl@apache.org>

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

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

* fix comments

Signed-off-by: xiaolong.ran <rxl@apache.org>(cherry picked from commit ce29135c)

Fixes #6838

### Motivation

Close producer when the topic does not exists.

### Modifications

1. Fix exception handle for the topic does not exist.
2. Change state to Close when producer got TopicDoesNotExists exception so that the producer can close the cnx and will no longer add send timeout tasks to HashedWheelTimer.
(cherry picked from commit 6eed2173)

Fixes #6749

### Motivation

Currently, the Pulsar SQL couldn't support AvroSchema use the `ByteBuffer` as the field type. For example, use the POJO class as below.

```
@data
public static class LogFile {
    int id;
    String name;
    ByteBuffer data;
}

Producer<LogFile> producer = pulsarClient.newProducer(Schema.AVRO(LogFile.class)).topic(topic).create();
```

Error Log
```
2020-05-08T23:34:47.079+0800	ERROR	SplitRunner-5-101	com.facebook.presto.execution.executor.TaskExecutor	Error processing Split 20200508_153445_00006_nxngm.1.0-1 PulsarSplit{splitId=1, connectorId='pulsar', originSchemaName='bytes-sql-test4', schemaName='public/default', tableName='bytes-sql-test4', splitSize=4, schema='{"type":"record","name":"LogFile","namespace":"com.ran.schema.KeyValueSchemaTest$","fields":[{"name":"id","type":"int"},{"name":"name","type":["null","string"],"default":null},{"name":"data","type":["null","bytes"],"default":null}]}', schemaType=AVRO, startPositionEntryId=5, endPositionEntryId=9, startPositionLedgerId=3359, endPositionLedgerId=3359, schemaInfoProperties={"__alwaysAllowNull":"true"}} (start = 4.34095226272178E8, wall = 546 ms, cpu = 0 ms, wait = 0 ms, calls = 1)
java.lang.ClassCastException: java.nio.HeapByteBuffer cannot be cast to [B
	at org.apache.pulsar.sql.presto.PulsarRecordCursor.getSlice(PulsarRecordCursor.java:516)
	at com.facebook.presto.spi.RecordPageSource.getNextPage(RecordPageSource.java:117)
	at com.facebook.presto.operator.TableScanOperator.getOutput(TableScanOperator.java:242)
	at com.facebook.presto.operator.Driver.processInternal(Driver.java:373)
	at com.facebook.presto.operator.Driver.lambda$processFor$8(Driver.java:282)
	at com.facebook.presto.operator.Driver.tryWithLock(Driver.java:672)
	at com.facebook.presto.operator.Driver.processFor(Driver.java:276)
	at com.facebook.presto.execution.SqlTaskExecution$DriverSplitRunner.processFor(SqlTaskExecution.java:973)
	at com.facebook.presto.execution.executor.PrioritizedSplitRunner.process(PrioritizedSplitRunner.java:162)
	at com.facebook.presto.execution.executor.TaskExecutor$TaskRunner.run(TaskExecutor.java:477)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
```

### Modifications

When the presto field record type is `VarbinaryType.VARBINARY`, check the record type is `ByteBuffer`, `byte[]`, `ByteBuf` or others, and to process the field record by the type.

* pulsar sql support avro schema `ByteBuffer` type

* add ByteBuf check and unit tests.
(cherry picked from commit 3aaed249)

Fix #6854

### Bug description
The consumer stuck due to `hasEnoughMessagesForBatchReceive` checking:
```
protected boolean hasEnoughMessagesForBatchReceive() {
        if (batchReceivePolicy.getMaxNumMessages() <= 0 && batchReceivePolicy.getMaxNumBytes() <= 0) {
            return false;
        }
        return (batchReceivePolicy.getMaxNumMessages() > 0 && incomingMessages.size() >= batchReceivePolicy.getMaxNumMessages())
                || (batchReceivePolicy.getMaxNumBytes() > 0 && INCOMING_MESSAGES_SIZE_UPDATER.get(this) >= batchReceivePolicy.getMaxNumBytes());
    }
```

### Changes
When batchReceivePolicy maxNumMessages > maxReceiverQueueSize, we force batch receivePolicy maxNumMessages to maxReceiverQueueSize to avoid consumer stuck when checking `hasEnoughMessagesForBatchReceive`

* fix consumer stuck when batchReceivePolicy maxNumMessages > maxReceiverQueueSize

* throw log warn and add test case
(cherry picked from commit 561868d4)

Fix #6839
### Motivation

The V3 endpoint that returns a list of all functions in a namespace always returns 404. The V2 version of the endpoint returns the actual list of functions in the namespace. It looks like during the switch from V2 to V3, the implementation for the endpoint was missed. This endpoint is part of the current API [documentation](https://pulsar.apache.org/functions-rest-api/?version=2.5.0#operation/listFunctions), so I don't think it was removed intentionally.

This endpoint is also used by the pulsar-admin command, so that always returns 404:

```
pulsar-admin functions list --tenant chris-kafkaesque-io --namespace local-useast2-aws
HTTP 404 Not Found

Reason: HTTP 404 Not Found
```

### Modifications

I have added the endpoint to `FunctionsApiV3Resource.java`. It is essentially a clone of the V2 version.

### Verifying this change

This is a pretty small change. I have confirmed that the V3 version of the endpoint now returns the same list of functions as the V2 version. I have also confirmed that the pulsar-admin command now works:

```
bin/pulsar-admin functions list --tenant chris-kafkaesque-io --namespace TTL
23:45:10.763 [main] INFO  org.apache.pulsar.common.util.SecurityUtility - Found and Instantiated Bouncy Castle provider in classpath BC
"exclaim"
"pulsar-functions-0.1"
```

(cherry picked from commit 714a776c)

Expose pulsar_out_bytes_total and pulsar_out_messages_total for namespace/subscription/consumer. (#6918)

Fixes #6891
Rated to #5802

Add pulsar_out_bytes_total and pulsar_out_messages_total for namespace/subscription/consumer.

New unit test added.
(cherry picked from commit 204f3271)

Fixes #6873

Use originalAuthMethod on originalAuthProvider check.
(cherry picked from commit 8381371a)

Fix typos in exceptions related to functions. The tests were updated as well.
(cherry picked from commit 20216d1c)

Fixes #6640

Add Tls with keystore type config.

Add Tls with keystore type config.

- Unit test passed

(cherry picked from commit 367ce782)

### Motivation
Right now, Pulsar-client provides tls authentication support and default TLS provider `AuthenticationTls` expects file path of cert and key files. However, there are usescases where it will be difficult for user-applications to store certs/key file locally for tls authentication.
eg:
1. Applications running on docker or K8s containers will not have certs at defined location and app uses KMS or various key-vault system whose API return streams of certs.
2. Operationally hard to manage key rotation in containers
3. Need to avoid storing key/trust store files on file system for stronger security

Therefore, it's good to have mechanism in default `AuthenticationTls` provider to read certs from memory/stream without storing certs on file-system.

### Modification
Add Stream support in `AuthenticationTls` to provide X509Certs and PrivateKey which also performs auto-refresh when stream changes in a given provider.
```
AuthenticationTls auth = new AuthenticationTls(certStreamProvider, keyStreamProvider);
```
It will be also address: #5241
(cherry picked from commit 3b48df15)

Fixes #6519

### Motivation

Currently, Pulsar Functions not support Async mode, e.g. user passed in a Function in format :
```
Function<I, CompletableFuture<O>>
```
This kind of function is useful if the function might use RPCs to call external systems.

e.g.
```java
public class AsyncFunction implements Function<String, CompletableFuture<O>> {
    CompletableFuture<O> apply (String input) {
        CompletableFuture future = new CompletableFuture();
        ...function compute...
        future.whenComplete(() -> {
            ...  call external system  ...
        });
        return future;
    }
```

### Modifications
- add support for Async Functions support.

### Verifying this change
current ut passed.

* support func: Function<I, CompletableFuture<O>>

* add 2 examples

* add limit to the max outstanding items
(cherry picked from commit 7cd28b9d)

(cherry picked from commit b9e96098)

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

### Motivation

Installation error of dashboard database postgres 11 resulted in startup failure.

### Modifications

* Update apachepulsar/pulsar-dashboard:2.5.1 image https://hub.docker.com/layers/apachepulsar/pulsar-dashboard/2.5.1/images/sha256-61b47a7302639aba1357d09ca69a842c4a67bff38b230753d6bd638df0461c6b?context=explore
* Update Docker file for fix postgresql version 11.

### Verifying this change

Local test pass
(cherry picked from commit c621d99b)

### Motivation

The validation of parameters for function's update was not properly implemented for the outputSerdeClassName parameter. It was checking the outputSchemaType field instead.

### Modifications

Updated the if conditions and added tests.
(cherry picked from commit bfec5231)

In #6489, a timeout was introduced to make sure calls into the
BrokerService finish or error out. However, this timeout is too low by
default when loading topics that have many replicated clusters.

Loading replicated topics is quite an expensive operation, involve
global ZK lookups and the start of many sub-processes. While we would
hope it finishes in 60 seconds we want to safe.

Long term, it may make sense to break out this operation into more
steps where each step can have it's own timeout

Co-authored-by: Addison Higham <ahigham@instructure.com>(cherry picked from commit 6854b007)

change log level to debug to avoid annoying.

(cherry picked from commit 4e6b2f09)

Fixes #6872
Fixes #6416

If a producer tries to create a producer to a topic that is currently
unloading, we can get a `RuntimeException` from
`BrokerService.checkTopicNsOwnership` which is bubbled up through
`topic.addProducer`. By only handling a `BrokerServiceException` this
results in a future that never completes and results in producers not
being able to be created if this topic is scheduled back to this broker.
(cherry picked from commit 30e26f84)

### Motivation

JDBC sink does not handle `null` fields. For example, the field `example` can potentially be null. The schema registered in Pulsar allows for it, and the table schema in MySQL has a column of the same name, is configured as double and also allows nulls. When messages are sent to the JDBC sink without that field, an exception like this is seen:

```
21:08:38.472 [pool-5-thread-1] ERROR org.apache.pulsar.io.jdbc.JdbcAbstractSink - Got exception
java.sql.SQLException: Data truncated for column 'example' at row 1
	at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:127) ~[mysql-connector-java-8.0.11.jar:8.0.11]
	at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:95) ~[mysql-connector-java-8.0.11.jar:8.0.11]
	at com.mysql.cj.jdbc.exceptions.SQLExceptionsMapping.translateException(SQLExceptionsMapping.java:122) ~[mysql-connector-java-8.0.11.jar:8.0.11]
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeInternal(ClientPreparedStatement.java:960) ~[mysql-connector-java-8.0.11.jar:8.0.11]
	at com.mysql.cj.jdbc.ClientPreparedStatement.execute(ClientPreparedStatement.java:388) ~[mysql-connector-java-8.0.11.jar:8.0.11]
	at org.apache.pulsar.io.jdbc.JdbcAbstractSink.flush(JdbcAbstractSink.java:202) ~[pulsar-io-jdbc-2.5.0.nar-unpacked/:?]
	at org.apache.pulsar.io.jdbc.JdbcAbstractSink.lambda$open$0(JdbcAbstractSink.java:108) ~[pulsar-io-jdbc-2.5.0.nar-unpacked/:?]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_232]
	at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) [?:1.8.0_232]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) [?:1.8.0_232]
```
Looking at the code for the JDBC sink, there was no handling of the case where the field was `null`. The PR adds code to handle that case. It also adds unit tests to cover this for both binary and JSON encoding of the schema.

### Modifications

When the sink encounters a `null` field value it uses the `setColumnNull` method to properly reflect this in the database row.
(cherry picked from commit c622de51)

(cherry picked from commit 466b0b89)

While offloading ledgers from bookies to 2nd storage, we could offload the ledgers in columnar format. Columnar data could accelerate analytical workloads' execution by skipping unnecessary columns or data blocks (also known as column pruning and filter push down in analytical systems).

The only blocker in Pulsar side is that offloaders cannot get the schema of the ledgers, this PR makes the schema storage accessible from offloaders.
(cherry picked from commit 32234771)

(cherry picked from commit 0951828d)

Fixes #6840

Motivation
Avoid creating partitioned topic for partition name

Verifying this change
New unit test added.
(cherry picked from commit bb417025)

Fixes #5395

This change introduces a new namespace policy `autoTopicCreationOverride`, which will enable an override of broker `autoTopicCreation` settings on the namespace level. You may keep `autoTopicCreation` disabled for the broker and allow it on a specific namespace using this feature.

- Add new namespace policy: `autoTopicCreationOverride` and associated API / CLI interface for setting and removing. Defaults to non-partitioned type, but also allows partitioned topics.
- Modifies BrokerService: when checking `autoTopicCreation` configuration, the broker first retrieves namespace policies from zookeeper. If the `autoTopicCreationOverride` policy exists for that namespace then it uses those settings. If not, falls back to broker configuration.
- Slight refactor to move `TopicType` enum to pulsar-common and add `autoTopicCreationOverride` to pulsar-common.
(cherry picked from commit fdc3a9bc)

### Motivation

In Python client, the serialization of enums when using the schema is currently broken, throwing error because it's not possible to directly serialize them into json.

Instead, for both Avro and JSON, we need to handle the enum serialization on its own way.
(cherry picked from commit 5ec9d7b0)

* Fix bug: sendCallback's 2nd argument was always the default MessageId

* Set batch index for each message's callback of batch

* Add test for message id in send callback

* Ensure all send callbacks completed before ASSERT_EQ
(cherry picked from commit fc69a628)

The messageReceiveTimeoutMs value in the PulsarConsumerSource class is hardcoded to 100ms and cannot be altered through configuration at present.
(cherry picked from commit 47b4dd07)