[Issue 3458: Tag Pulsar ledgers in order to distinguish from other leggers in the same Bookkeeper cluster (#3525)

Fixes #3458

### Motivation

See #3458 

### Modifications

Add a new LedgerMetadataUtils class which holds the logic for building "metadata" to be attached to 

### Verifying this change

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

* [go client] add SubscriptionInitPos option in ConsumerOptions
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* add comment for SubscriptionInitPos option
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

Signed-off-by: NVinceShieh <vinceshieh@gmail.com>

* [go client] add validation hostName for ClientOptions
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* [go client] add validation hostName for ClientOptions
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* rename ValidationHostName to TLSValidateHostname
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

### Motivation

Provides a builtin MongoDB Connector, in order to ease the storage of JSON formated message in MongoDB. It's a sink connector.


### Modifications

Add a new sub-module in the `pulsar-io`module.

### Verifying this change

This change added tests and can be verified as follows:
* deploy the connector with configuration file containing the following fields:
```
configs:
  mongoUri: mongodb://hostname:port
  database: pulsar
  collection: messages
```
* start a mongodb instance
* send messages in the topic declared when deploying the connector
* check in MongoDB if the messages have been stored in the collection `messages`

* moved all windowing related code into its own submodule.

* added pom

* [cpp-client] add multi topic for consumer #3494
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* use initial_position instead to multi-topics
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* Added docs for using schema in Python client lib

* Added info on required fields

### Motivation
Add a unit test for KafkaAbstractSink to check valid configs.

### Modifications
Add KafkaAbstractSinkTest.

*Motivation*

Add state support in python functions

*Changes*

- Bump bookkeeper version, so the table service has the changes to support python functions
- Add state to python function

*Motivation*

This change was part of #2714. Moving it out as a separate change to make reviews easier.

*Changes*

Bump bookkeeper version to 4.9.0

* [pulsar-broker][replicator] add remote cluster name in producer name

* add padding between remote and local cluster-names

* broker-producer can parse remote-cluster name in producer-name

*Motivation*

There are a few shaded modules introduced when functions was introduced.
They are not needed anymore. We should remove them.

*Modifications*

Remote those unused modules.

* Create SSL context in constructor of ChannelInitializer

* Set TLS ports only in tests that need to enable TLS

### Motivation

Remove compile and link dependency on JsonCPP. Instead use boost::property_tree with JSON parser. This will not require a link time dependency.

*Motivation*

Sometimes users would like to only execute a set of test suite in command line.

*Modifications*

Make the suite file name configurable in the pom file.
So people can run a specific test suite using
```
mvn -pl tests/integration -DredirectTestOutputToFile=false -DintegrationTestSuiteFile=pulsar-thread-suite.xml clean test -PintegrationTests
```

Also categorize the tests info a few test suites. This would allow us break down
the integration test job into multiple smaller test jobs.

### Motivation

Some runtime might require a minimum about of resources to run a function instance.  Currently functions don't support the check to determine if the resources requested is greater the the minimum required. 

### Modification

Add a worker config to set the minimum amount of resource a function must request

### Motivation

Sometimes while shutting down the broker, unknown uncompleted callback tries to update `managed-cursor` even if managed-cursor is closed and because of that topic which is already loaded by new broker sees below exception while updating managed-cursor-metadata and cursor is not able to update mark-delete position until it unloaded again.
```
09:22:02.641 [BookKeeperClientWorker-OrderedExecutor-8-0] INFO  org.apache.bookkeeper.mledger.impl.MetaStoreImplZookeeper - [sample/global/ns1/persistent/TopicBadVersion] [pulsar.repl.us-east] Updating cursor info ledgerId=234567890 mark-delete=12345678900:5306
00:00:05.642 [bookkeeper-ml-workers-OrderedExecutor-14-0] WARN  org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [sample/global/ns1/persistent/TopicBadVersion] Failed to update consumer pulsar.repl.us-east
org.apache.bookkeeper.mledger.ManagedLedgerException$BadVersionException: org.apache.zookeeper.KeeperException$BadVersionException: KeeperErrorCode = BadVersion
Caused by: org.apache.zookeeper.KeeperException$BadVersionException: KeeperErrorCode = BadVersion
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:118) ~[pulsar-broker-2.2.5-.jar:2.2.5-]
        at org.apache.bookkeeper.mledger.impl.MetaStoreImplZookeeper.lambda$null$103(MetaStoreImplZookeeper.java:287) ~[managed-ledger-original-2.2.5-.jar:2.2.5-]
        at org.apache.bookkeeper.mledger.util.SafeRun$1.safeRun(SafeRun.java:32) [managed-ledger-original-2.2.5-.jar:2.2.5-]
        at org.apache.bookkeeper.common.util.SafeRunnable.run(SafeRunnable.java:36) [bookkeeper-common-4.7.2.jar:4.7.2]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_131]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_131]
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [netty-all-4.1.22.Final.jar:4.1.22.Final]
```

It seems it always happens for replicator-cursor and not sure exact step to reproduce but it might happen when bookies might be in unstable state (not sure about the root cause). one of the example:

Behavior:
broker-1 unloads the topic by `09:21:59.451` and broker-2 loads the same topic at `09:22:00.135`. But broker-1 has log of updating metadata-znode at `09:22:05.207`. So, broker-2 fails to update cursor-metadata with exception `BadVersionException`
```
Previous-broker: broker-1
09:21:58.855 [shutdown-thread-49-1] INFO  org.apache.pulsar.broker.namespace.OwnedBundle - Disabling ownership: sample/global/ns1/0x2aaaaaa8_0x35555552
:
:
09:21:59.668 [shutdown-thread-49-1] INFO  org.apache.pulsar.broker.namespace.OwnedBundle - Unloading sample/global/ns1/0x2aaaaaa8_0x35555552 namespace-bundle with 345 topics completed in 812.0 ms
:
09:21:58.930 [shutdown-thread-49-1] INFO  org.apache.pulsar.broker.service.BrokerService - [persistent://sample/global/ns1/TopicBadVersion] Unloading topic
:
09:21:59.451 [bookkeeper-ml-workers-OrderedExecutor-14-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [sample/global/ns1/persistent/TopicBadVersion][pulsar.repl.us-east] Closed cursor at md-position=12345678900:5306
09:21:59.451 [bookkeeper-ml-workers-OrderedExecutor-14-0] INFO  org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://sample/global/ns1/TopicBadVersion] Topic closed
:
09:22:05.207 [BookKeeperClientWorker-OrderedExecutor-2-0] WARN  org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [sample/global/ns1/persistent/TopicBadVersion] Error updating cursor pulsar.repl.us-east position 12345678900:5306 in meta-ledger 14451380450: BookKeeper client is closed
09:22:05.207 [BookKeeperClientWorker-OrderedExecutor-2-0] INFO  org.apache.bookkeeper.mledger.impl.MetaStoreImplZookeeper - [sample/global/ns1/persistent/TopicBadVersion] [pulsar.repl.us-east] Updating cursor info ledgerId=-1 mark-delete=12345678900:5306
```

```
Current-broker: broker-2
09:22:00.135 [pulsar-ordered-OrderedExecutor-7-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - Opening managed ledger sample/global/ns1/persistent/TopicBadVersion
```

### Modification

It happens when someone tries to update cursor metadata by calling `persistPositionMetaStore(..)` so, it should have safe guard and should not update metadata if cursor is already closed.

### Result
It avoids any unexpected behavior while updating cursor-metadata in closed cursor state.

### Motivation

Added integration test to validate that a Java client with schema can interoperate with a Python client with equivalent schema declaration.

### Motivation

Fixes #3538. The second consumer was being created with `Exclusive` type and was actually failing to connect.

* Improved SimpleProducerConsumerTest to avoid receive timeout

* Use testDupConsumersOnSharedModeNotThrowsExcOnUnsubscribe to avoid interference with other tests running in parallel

*Motivation*

Currently bookie http server is disabled by default. Setting the http server port
to 8000 and include a vertx http server implementation to prepare enabling http server
on bookie.

*Modifications*

Currently bookkeeper only provides two http server bindings, one is vertx, the other one
is finagle server binding. both are netty based http server. so technically it doesn't
introduce additional dependencies. However if people prefers a jetty implementation,
I can add a jetty binding in bookkeeper.

* Added suppport for Kafka partitioner and explicit setting of partition on record

* Addressed comments

* Use StandardCharsets and fixed alignment

* Fixed initialization of custom partitioner

* Feed realistic cluster metadata for Kafka partitioner implementation

For integration tests, copy function logs out of container on shutdown for debugging purposes (#3529)

* Build static lib with bundled dependencies

* Added gitignore entry

* Fixed target_link_libraries when not linking statically

* Fixed linking on python wrapper