…r. (apache#20880)

Main Issue: apache#20851
### Motivation
When the Proto version does not allow us to send TcClientConnectRequest to the broker, we should add a log to debug it.

### Modifications

Add a waining log.

…edup and set namespace disable deduplication. (apache#20905)

…20978)

(cherry picked from commit 63d9eaf)

(cherry picked from commit 3ab420c)

…e#20980)"

This reverts commit 2dc14d4.

(cherry picked from commit 3ab420c)

Fixes: apache#20997

### Motivation

Update the expired certs to get tests passing.

### Modifications

* Update all certs. See `README.md` in files for detailed steps.

### Verifying this change

This change is covered by tests.

### Documentation

- [x] `doc-not-needed`

(cherry picked from commit d6734b7)

…Horizon has been updated (apache#20984)

(cherry picked from commit 0cb1c78)

…ache#20914)

…currently (apache#20971)

### Motivation

**Background**: when calling `pulsar-admin topics stats --get-earliest-time-in-backlog <topic name>`, Pulsar will read the first entry which is not acknowledged, and respond with the entry write time. The flow is like this:
- get the mark deleted position of the subscription
- if no backlog, response `-1`
- else read the next position of the mark deleted position, and respond with the entry write time.

**Issue**: if the command `pulsar-admin topics stats --get-earliest-time-in-backlog <topic name>` and `consumer.acknowledge` are executed at the same time, the step 2 in above flow will get a position which is larger than the last confirmed position, lead a read entry error.

| time | `pulsar-admin topics stats --get-earliest-time-in-backlog <topic name>` | `consumer.acknowledge` |
| --- | --- | --- |
| 1 | mark deleted position is `3:1` and LAC is `3:2` now |
| 2 | the check `whether has backlog` is passed |
| 3 | | acknowledged `3:2`, mark deleted position is `3:2` now |
| 4 | calculate next position: `3:3` |
| 5 | Read `3:3` and get an error: `read entry failed` |

Note: the test in PR is not intended to reproduce the issue.

### Modifications

Respond `-1` if the next position of the mark deleted position is larger than the LAC

…tion check. (apache#20983)

…pache#20990)

…messages (apache#21032)

…data. sec ver. (apache#20620)

Co-authored-by: wangjinlong <wangjinlong@zhihu.com>

(cherry picked from commit d06cda6)

…e#21061)

(cherry picked from commit c05954e)

…ageId read reaches lastReadId (apache#20988)

(cherry picked from commit 9e2195c)

…letion of compaction (apache#21067)

(cherry picked from commit bb9c9b4)

…che#21070)

### Motivation

Current, when the producer resend the chunked message like this:
- M1: UUID: 0, ChunkID: 0
- M2: UUID: 0, ChunkID: 0 // Resend the first chunk
- M3: UUID: 0, ChunkID: 1

When the consumer received the M2, it will find that it's already tracking the UUID:0 chunked messages, and will then discard the message M1 and M2. This will lead to unable to consume the whole chunked message even though it's already persisted in the Pulsar topic.

Here is the code logic:
https://github.com/apache/pulsar/blob/44a055b8a55078bcf93f4904991598541aa6c1ee/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L1436-L1482

The bug can be easily reproduced using the testcase `testResendChunkMessages` introduced by this PR.

### Modifications

- When receiving the new duplicated first chunk of a chunked message, the consumer discard the current chunked message context and create a new context to track the following messages. For the case mentioned in Motivation, the M1 will be released and the consumer will assemble M2 and M3 as the chunked message.

(cherry picked from commit eb2e3a2)

…#21048)

(cherry picked from commit 99e3fea)

… was failed (apache#20935)

The progress Persist mark deleted position is like this:
- persist to BK
- If failed to persist to BK, try to persist to ZK

But in the current implementation: if the cursor ledger was created failed, Pulsar will not try to persist to ZK. It makes if the cursor ledger created fails, a lot of ack records can not be persisted, and we will get a lot of repeat consumption after the BK recover.

Modifications: Try to persist the mark deleted position to ZK if the cursor ledger was created failed

…ache#20948)

## Motivation
Make the chunk message function work properly when deduplication is enabled.
## Modification
### Only check and store the sequence ID of the last chunk in a chunk message.
 For example:
 ```markdown
     Chunk-1 sequence ID: 0, chunk ID: 0, total chunk: 2
     Chunk-2 sequence ID: 0, chunk ID: 1
     Chunk-3 sequence ID: 1, chunk ID: 0 total chunk: 3
     Chunk-4 sequence ID: 1, chunk ID: 1
     Chunk-5 sequence ID: 1, chunk ID: 1
     Chunk-6 sequence ID: 1, chunk ID: 2
```   
Only store check and store the sequence ID of Chunk-2 and Chunk-6.
**Add a property in the publishContext to determine whether this chunk is the last chunk when persistent completely.**
```java
publishContext.setProperty(IS_LAST_CHUNK, Boolean.FALSE);
```
### Filter and ack duplicated chunks in a chunk message instead of discarding ctx.
 For example:
 ```markdown
     Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1
     Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2
     Chunk-3 sequence ID: 0, chunk ID: 2, msgID: 1:3
     Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4
     Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5
     Chunk-6 sequence ID: 0, chunk ID: 3, msgID: 1:6
```   
We should filter and ack chunk-4 and chunk-5.

…n topic level (apache#21041)

…che#21060)

…s limitation (apache#21065)

)

Motivation: When deleting a namespace, we will delete znode under the path `/loadbalance/bundle-data` in `local metadata store` instead of `global metadata store`.

Modifications: Delete bundle data znode in local metadata store.

…fter RawReader reconnects (apache#21081)

…ing broker deletion event (apache#21083)

…trategicTwoPhaseCompactor (apache#21091)

…21099)

## Motivation
Handle ack hole case:
For example:
```markdown
                     Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1
                     Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2
                     Chunk-3 sequence ID: 0, chunk ID: 0, msgID: 1:3
                     Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4
                     Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5
```
 Consumer ack chunk message via ChunkMessageIdImpl that consists of all the chunks in this chunk
 message(Chunk-3, Chunk-4, Chunk-5). The Chunk-1 and Chunk-2 are not included in the
 ChunkMessageIdImpl, so we should process it here.
## Modification
Ack chunk-1 and chunk-2.

…ng timeout (apache#21028)

…deleting invalid underreplication nodes (apache#21059)

…pache#20952)

### Motivation

Minor improve `getIsolationGroup` by avoid creating arrayList. And it'  beneficial to GC

### Modifications

`new HashSet<>(Arrays.asList(...))` -> `Sets.newHashSet(...)`

### Modifications
When upgraded the pulsar version from 2.9.2 to 2.10.3, and the isolated group feature not work anymore.
Finally, we found the problem. In IsolatedBookieEnsemblePlacementPolicy, when it gets the bookie rack from the metadata store cache, uses future.isDone() to avoid sync operation. If the future is incomplete, return empty blacklists. 
The cache may expire due to the caffeine cache `getExpireAfterWriteMillis` config, if the cache expires, the future may be incomplete. (apache#21095 will correct the behavior)

In 2.9.2, it uses the sync to get data from the metadata store, we should also keep the behavior.

…ache#21023)

apache#21116)

…and schema. (apache#21093)

Fixes apache#21075 

### Motivation

When the topic is loaded, it will delete the topic-level policy if it is enabled. But if the topic is not loaded, it will directly delete through managed ledger factory. But then we will leave the topic policy there. When the topic is created next time, it will use the old topic policy

### Modifications

When deleting the topic, delete the schema and topic policies even if the topic is not loaded.

apache#21087)

…ndex acks (apache#21126)

…eystore TLS and admin API is called (apache#21077)

…n per broker (apache#21144)

Motivation: Pulsar has two mechanisms to guarantee that a producer connects to the broker multiple times the result is still correct.

- In a connection, the second connection waits for the first connection to complete.
- In a topic, the second connection will override the previous one.

However, if a producer can use different connections to connect to the broker, these two mechanisms will not work.

When the config `connectionsPerBroker` of `PulsarClient` is larger than `1`, a producer could use more than one connection, leading to the error above. You can reproduce this issue by the test `testSelectConnectionForSameProducer.`

Modifications: Make the same producer/consumer usage the same connection

…roker and underloadbroker (apache#21180)

…ernalCallbacks.Processor (apache#21159)

…1168)

… port if TLS is enabled (apache#21015)

…problem. (apache#21161)

…pache#21181)

…ns wrong port if TLS is enabled (apache#21015)"

This reverts commit ad69dec.

…opic (apache#21205)

…apache#21237)

…1232)

Motivation: Multiple consumers and producers can be maintained by the same Pulsar Client. In some cases, multiple consumers or producers might attempt to connect to the same topic. To optimize the process, it is recommended to perform the topic lookup only once for each topic.

Modifications:
- Merge lookup requests for the same topic.
- Merge get partitioned metadata request for the same partitioned topic.

…ding ack state (apache#21177)

(cherry picked from commit 1bf7371)

apache#21278)

(cherry picked from commit 682eb36)

# Conflicts:
#	pom.xml

…ear (apache#21302)

(cherry picked from commit 70d086f)

(cherry picked from commit e76a86e)

# Conflicts:
#	pulsar-common/src/test/java/org/apache/pulsar/common/nar/NarUnpackerTest.java

…pache#21329)

(cherry picked from commit c883f50)

…AuthorizationProvider plugin (apache#20939)

Fixed apache#20938

…apache#21338)

Co-authored-by: Lari Hotari <lhotari@apache.org>

…1222)

(cherry picked from commit be4bcac)

…g from BookKeeper to Pulsar. (apache#21340)

There is no testing for AutoRecovery replication in Pulsar's current test suite, and we need to cover it. So migrate the replication testing from BookKeeper to Pulsar.

Fix zookeeper related flacky test

(cherry picked from commit 9ab7417)

…re building (apache#21365)

(cherry picked from commit 421c98a)

apache#21155)

The client assumed the connection was inactive, but the Broker assumed the connection was fine. The Client tried to  use a new connection to reconnect a producer, then got an error `Producer with name 'st-0-5' is already connected to topic`.

- In a connection, the second connection waits for the first connection to complete\. But there is a bug that causes this mechanism to fail\.
- If a producer uses a default name, the second registration will override the first one. But it can not override the first one if it uses a specified producer name\. I think this mechanism is to prevent a client from creating two producers with the same name. However, method `Producer.isSuccessorTo` has checked the `producer-id`, and the `producer-id` of multiple producers created by the same client are different. So this mechanism can be deleted.

- For `issue 1`: If a producer with the same name tries to use a new connection, async checks the old connection is available. The producers related to the connection that is not available are automatically cleaned up.

- For `issue 2`:
  -  Fix the bug that causes a complete producer future will be removed from `ServerCnx`.
  - Remove the mechanism that prevents a producer with a specified name from overriding the previous producer.

(cherry picked from commit bda16b6)

(cherry picked from commit 39235ed)

# Conflicts:
#	pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ClusterMigrationTest.java

### Motivation
As https://pulsar.apache.org/docs/3.1.x/sql-overview/, Pulsar SQL is based on [Trino (formerly Presto SQL)](https://trino.io/), which supports UUID type. But now, the UUID field in Avro or JSON schema will be interpreted as VARCHAR.

### Modifications

Support decoding UUID form AVRO or JSON schema.

(cherry picked from commit 8c70943)

(cherry picked from commit e6cd11f)

)

(cherry picked from commit ffc083b)

…le when using extensible load manager (apache#21213) (apache#21314)

…r migration tests (apache#21391)

(cherry picked from commit e2c6c08)

…o after bkclient initialize (apache#20944)

…ic (apache#21348)