[Issue 7864][docs]Update document for how to pick active consumer for failover subscription. (#7917)

Update document to specify how active consumer is picked for partitioned and non-partitoned topic for failover subscription mode. The change was push in 2.4.2 but document has not been updated.

Fixes #7864

**Motivation**

After #7813, the python client supports the oauth2 authentication as well. Add an example for it.

[Issue 7933][pulsar-broker-common] Add some logging to improve Authentication debugging and Fix typos in code "occured" -> "occurred" (#7934)

Fixes #7933 

### Motivation
Newbie trying to make his first contribution to this projection :-)

### Modifications

Added some logging to Authentication Service to help debugging when there are more than one AuthN provider.
When I did that I noticed a typo ('occured' should have two r's) so I decided to try to fix all of them

Modifications
Support set publish rate on topic level.
Support get publish rate on topic level.
Support remove publish rate on topic level.

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

* Add Pulsar Summit CFP blog

* update

* Update 2020-08-25-pulsar-summit-asia-2020-cfp.md

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

### Modifications

Fix the time of release 2.6.1 to `08/21`

Fixes #7481

Motivation
Default bookkeeper conf that ships with apache pulsar indicates that the unit of openLedgerRereplicationGracePeriod is in seconds, but official bookkeeper doc(and code) mentions that the unit is in milliseconds.

Pulsar should configure default as 30000 instead of 30 and bookkeeper.conf should mention that the unit of this period is milliseconds instead of seconds.

Modifications
Update the bookkeep.config file
Update the following Pulsar docs for releases 2.5.2, 2.6.0, 2.6.1, and master
Reference > Pulsar Config > BookKeeper

Motivation
Currently, we have to know the batchSize to generate BatchMessageAcker. If we could get the batch index ack bitSet from Broker we could generate the BatchMessageAcker by the bitSet, this is useful for consuming transaction messages, we don't need to change the protocol to get the total message number of one transaction.

Modifications
Add a new static method to generate the BatchMessageAcker by BitSet.

### Motivation

The transaction message consuming logic needs to be optimized. One transaction commit marker in the topic partition consists of many Entries in the TransactionBuffer, so one transaction could be seen as one batch.

### Modifications

1. When sending messages to consumers use the `batchIndex` in the MessageIdData to present the startBatchIndex of the Entry.

2. Calculate the transaction messages number.

3. When the first reading one transaction init it's batchDeletedIndexes in the ManagedCursorImpl.

Fix #7826 

### Motivation
Support compaction threshold on topic level.
Based on the system topic function.

### Modifications
Support set compaction threshold on topic level.
Support get compaction threshold on topic level.
Support remove compaction threshold on topic level.

Co-authored-by: NJerry Peng <jerryp@splunk.com>

Co-authored-by: NJerry Peng <jerryp@splunk.com>

Update AWS deployment in reference-configuration.

### Motivation

*Compatible with flink 1.11 need to use java8 date api in pulsar's primitive schemas.*

### Modifications

*Add Instant, LocalDate, LocalTime, LocalDateTime to pulsar's primitive schemas*

### Verifying this change

Add Instant, LocalDate, LocalTime, LocalDateTime types to the Schema type test

**Motivation**

Migrate the build website job to github actions.

### Motivation

 PR (https://github.com/apache/pulsar/pull/7821) supports set/get/remove the deduplication policy at the topic level. 


### Modifications

Update the doc based on code updates.

The following docs are updated:

- Messaging: message deduplication
- Manage message deduplication
- pulsar admin cli reference

Modifications
Sync the Pulsar project description with the latest version on home page.

### Motivation
Support message dispatch rate on topic level.
Based on the system topic function.

### Modifications
Support set message dispatch rate on topic level.
Support get message dispatch rate on topic level.
Support remove message dispatch rate on topic level.

Co-authored-by: NJerry Peng <jerryp@splunk.com>

Co-authored-by: NSanjeev Kulkarni <sanjeevk@splunk.com>

### Motivation

A deadlock occurred on our Pulsar 2.4.2 broker server. The cause is the following two threads:
```
"prometheus-stats-36-1" #410 prio=5 os_prio=0 tid=0x00007f4b70019800 nid=0x30ca waiting for monitor entry [0x00007f4bbe3b7000]
   java.lang.Thread.State: BLOCKED (on object monitor)
        at org.apache.pulsar.broker.service.persistent.PersistentSubscription.getNumberOfEntriesDelayed(PersistentSubscription.java:1013)
        - waiting to lock <0x00007f913d098dd0> (a org.apache.pulsar.broker.service.persistent.PersistentSubscription)
        at org.apache.pulsar.broker.stats.prometheus.NamespaceStatsAggregator.lambda$getTopicStats$8(NamespaceStatsAggregator.java:129)
        at org.apache.pulsar.broker.stats.prometheus.NamespaceStatsAggregator$$Lambda$523/2109257042.accept(Unknown Source)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.forEach(ConcurrentOpenHashMap.java:385)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.forEach(ConcurrentOpenHashMap.java:159)
        at org.apache.pulsar.broker.stats.prometheus.NamespaceStatsAggregator.getTopicStats(NamespaceStatsAggregator.java:122)
        at org.apache.pulsar.broker.stats.prometheus.NamespaceStatsAggregator.lambda$null$0(NamespaceStatsAggregator.java:64)
        at org.apache.pulsar.broker.stats.prometheus.NamespaceStatsAggregator$$Lambda$521/1017174654.accept(Unknown Source)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.forEach(ConcurrentOpenHashMap.java:385)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.forEach(ConcurrentOpenHashMap.java:159)
        at org.apache.pulsar.broker.stats.prometheus.NamespaceStatsAggregator.lambda$null$1(NamespaceStatsAggregator.java:63)
        at org.apache.pulsar.broker.stats.prometheus.NamespaceStatsAggregator$$Lambda$520/1098830264.accept(Unknown Source)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.forEach(ConcurrentOpenHashMap.java:385)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.forEach(ConcurrentOpenHashMap.java:159)
        at org.apache.pulsar.broker.stats.prometheus.NamespaceStatsAggregator.lambda$generate$2(NamespaceStatsAggregator.java:62)
        at org.apache.pulsar.broker.stats.prometheus.NamespaceStatsAggregator$$Lambda$316/212211274.accept(Unknown Source)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.forEach(ConcurrentOpenHashMap.java:385)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.forEach(ConcurrentOpenHashMap.java:159)
        at org.apache.pulsar.broker.stats.prometheus.NamespaceStatsAggregator.generate(NamespaceStatsAggregator.java:59)
        at org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsGenerator.generate(PrometheusMetricsGenerator.java:73)
        at org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsServlet.lambda$doGet$0(PrometheusMetricsServlet.java:70)
        at org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsServlet$$Lambda$315/1221766138.run(Unknown Source)
        at org.apache.bookkeeper.mledger.util.SafeRun$1.safeRun(SafeRun.java:32)
        at org.apache.bookkeeper.common.util.SafeRunnable.run(SafeRunnable.java:36)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.lang.Thread.run(Thread.java:748)

"ForkJoinPool.commonPool-worker-104" #953 daemon prio=5 os_prio=0 tid=0x00007f4dc8030800 nid=0x3b87 waiting on condition [0x00007f48f6ce1000]
   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00007f913d08b5c8> (a org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section)
        at java.util.concurrent.locks.StampedLock.acquireRead(StampedLock.java:1215)
        at java.util.concurrent.locks.StampedLock.readLock(StampedLock.java:428)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.forEach(ConcurrentOpenHashMap.java:377)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.forEach(ConcurrentOpenHashMap.java:159)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.values(ConcurrentOpenHashMap.java:174)
        at org.apache.pulsar.broker.service.persistent.PersistentTopic.getNumberOfConsumers(PersistentTopic.java:1227)
        at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.isConsumersExceededOnTopic(PersistentDispatcherMultipleConsumers.java:178)
        at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.addConsumer(PersistentDispatcherMultipleConsumers.java:144)
        - locked <0x00007f91120dc258> (a org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers)
        at org.apache.pulsar.broker.service.persistent.PersistentSubscription.addConsumer(PersistentSubscription.java:238)
        - locked <0x00007f913d098dd0> (a org.apache.pulsar.broker.service.persistent.PersistentSubscription)
        at org.apache.pulsar.broker.service.persistent.PersistentTopic.lambda$subscribe$11(PersistentTopic.java:590)
        at org.apache.pulsar.broker.service.persistent.PersistentTopic$$Lambda$451/1414070467.accept(Unknown Source)
        at java.util.concurrent.CompletableFuture.uniAccept(CompletableFuture.java:670)
        at java.util.concurrent.CompletableFuture.uniAcceptStage(CompletableFuture.java:683)
        at java.util.concurrent.CompletableFuture.thenAccept(CompletableFuture.java:2010)
        at org.apache.pulsar.broker.service.persistent.PersistentTopic.subscribe(PersistentTopic.java:584)
        at org.apache.pulsar.broker.service.ServerCnx.lambda$null$10(ServerCnx.java:699)
        at org.apache.pulsar.broker.service.ServerCnx$$Lambda$408/1168861154.apply(Unknown Source)
        at java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:995)
        at java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:2137)
        at org.apache.pulsar.broker.service.ServerCnx.lambda$null$13(ServerCnx.java:682)
        at org.apache.pulsar.broker.service.ServerCnx$$Lambda$406/1351396211.apply(Unknown Source)
        at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616)
        at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591)
        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)
        at java.util.concurrent.CompletableFuture.postFire(CompletableFuture.java:575)
        at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:943)
        at java.util.concurrent.CompletableFuture$Completion.exec(CompletableFuture.java:457)
        at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
        at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
        at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
        at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:163)
```

`prometheus-stats-36-1` was trying to lock `PersistentSubscription` (and `PersistentDispatcherMultipleConsumers`) after locking `PersistentTopic#subscriptions`, an instance of `ConcurrentOpenHashMap`.
https://github.com/apache/pulsar/blob/v2.4.2/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java#L122-L129

On the other hand, `ForkJoinPool.commonPool-worker-104` was trying to lock these instances in reverse order.
https://github.com/apache/pulsar/blob/v2.4.2/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java#L176-L237
https://github.com/apache/pulsar/blob/v2.4.2/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java#L127-L144
https://github.com/apache/pulsar/blob/v2.4.2/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java#L159-L175
https://github.com/apache/pulsar/blob/v2.4.2/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L1212-L1214

`PersistentSubscription#getNumberOfEntriesDelayed()` is no longer used in the master code, but it seems that this deadlock has not yet been resolved.
https://github.com/apache/pulsar/blob/e06e8726847584700d9e4fc98fd56a495eb05a23/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L1528-L1529
https://github.com/apache/pulsar/blob/17ae233a5d0fa364048b7c30ec90b8f7291d0d07/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java#L967-L1000
https://github.com/apache/pulsar/blob/6e7d1a83c3c2737610f01cb372f61e2b830a62f7/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java#L809

### Modifications

Moved the `isConsumersExceededOnTopic()` method to check the number of connected consumers when adding a consumer to the topic from the `Dispatcher` classes to the `AbstractTopic` class. Avoid the deadlock mentioned above by calling `PersistentTopic#getNumberOfConsumers()` before locking the `PersistentSubscription` instance.

Fixes #7554

### Motivation

As mentioned in #7554, the class of `individualDeletedMessages` is different between Pulsar 2.3.2 (and earlier) and 2.4.0 (and later). This causes some of ranges contained in `individualDeletedMessages` to be lost when the version of Pulsar is upgraded, and a large number of messages that have already been acked can be redelivered to consumers.

Also, even if the Pulsar version is 2.4.0 or later, the same phenomenon occurs when the value of `managedLedgerUnackedRangesOpenCacheSetEnabled` is switched from false to true.

### Modifications

If the list of individually deleted message ranges loaded from ZK contains ranges that span different ledgers, split the ranges by ledger ID and store them in `individualDeletedMessages`.

As a result, information about deleted message ranges is never lost and messages that have already been acked will not be redelivered.

### Motivation

Currently, the Pulsar SQL CI test is blocked. 

### Modifications

1. Build a new docker image before CI test to make sure the latest changes are covered.
2. Remove the config `distributed-joins-enabled` in the `$PULSAR_HOME/conf/presto/config.properties`, this config was discarded in presto version `332`.
3. Add the config `-Dpresto-temporarily-allow-java8=true` in `$PULSAR_HOME/conf/presto/jvm.config`, this config is needed if use Java 8 environment.
4. The ClientBuilder will be loaded by two different ClassLoaders, in the Pulsar SQL CI test the ClientBuilder will be created by unexpected ClassLoader, we need to specify the ClassLoader for it.

Fixes #7787

### Motivation

Throw std::exception instead of `const char*`

### Modifications

Should be self explanatory.

### Verifying this change

This change should be covered by existing tests, such as BatchMessageTest.  Not sure about the Auth test which catches `...` however.

Co-authored-by: NJerry Peng <jerryp@splunk.com>

* add 300 contributor blog

* update

* update

* update

* update date info

Fixes #7851 

### Motivation

`clear()` methods of `BatchAcknowledgementTracker` and `UnAckedMessageTrackerEnabled` are not thread-safe.

### Modifications

Acquire a mutex in these `clear()` methods.

Motivation
As a Pulsar newbie, the role of "partition key" and "ordering key" aren't entirely clear from the Doxygen comments. These names are not used in the public Pulsar documentation that I saw, so expanding the comments felt useful to me and may help other users.

Modifications
Expanded doc strings for MessageBuilder.setPartitionKey and setOrderingKey

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

Documentation
This change improves the C++ Doxygen output.

---

Fixes #7754

This CI job already be removed. And the check require is removed. So let's
remove this file.

Motivation
close() method never gets called in BatchSource

Motivation
Add auto partitions update interval setting for producer and consumer.

Modifications
add autoUpdatePartitionsInterval to partitioned producer and consumer

Return more informative error message when trying to create subscription on non-persistent throug Rest API or pulsar-admin CLI. (#7831)

Fixes #7397

Motivation
When use pulsar-admin to create a subscription on a non-persistent topic, get the server error
This change return more informative error message when trying to create subscript ion on non-persistent through Rest API or pulsar-admin CLI.

Modifications
Currently when creating subscription is called with non-persistent topic service will try to create the subscription which will fail with casting exception when trying to cast NonPersistentSubscription to PersistentSubscription and client will see internal error.
Add check if create subscription command is called for a non-persistent topic before actually

Verifying this change
This change added tests and can be verified as follows:

Added unit test
Verified with local standalone