…pache#21062)

…#21048)

)

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.

…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.

Signed-off-by: tison <wander4096@gmail.com>
Co-authored-by: Alexander Preuß <alexander.preuss@streamnative.io>
Co-authored-by: tison <wander4096@gmail.com>

…deleting invalid underreplication nodes (apache#21059)

…che#21060)

… 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

…s limitation (apache#21065)

…21099)

…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.

…vice (apache#21033)

## 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.

…trategicTwoPhaseCompactor (apache#21091)

…fter RawReader reconnects (apache#21081)

…ing broker deletion event (apache#21083)

### 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.

apache#21116)

…ache#21023)

apache#21087)

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

… the fatal exception (apache#21079)

### Motivation

This PIP is to improve the current function exception handler. It will be applied to both Pulsar Function and Pulsar IO Connector.

For more details, please refer to `pip-297.md`

…apache#20614)

)

Signed-off-by: Jiwe Guo <technoboy@apache.org>

…ndex acks (apache#21126)

…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

…underloadbroker (apache#21025)

…21051)

pip: apache#21052
### Motivation

Introduce the `getLastMessageIds` API to Reader.

### Modifications

Implement getLastMessageIds API for Reader

…pache#21169)

apache#21155)

#### Issue 1
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`.

#### Issue 2
- 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.

### Modifications

- 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.

…etrics (apache#20720)

…21142)

…the fatal exception (apache#21143)

PIP: apache#21079 

### Motivation

Currently, the connector and function cannot terminate the function instance if there are fatal exceptions thrown
outside the function instance thread. The current implementation of the connector and Pulsar Function exception handler
cannot handle the fatal exceptions that are thrown outside the function instance thread.

For example, suppose we have a sink connector that uses its own threads to batch-sink the data to an external system. If
any fatal exceptions occur in those threads, the function instance thread will not be aware of them and will
not be able to terminate the connector. This will cause the connector to hang indefinitely. There is a related issue
here: apache#9464

The same problem exists for the source connector. The source connector may also use a separate thread to fetch data from
an external system. If any fatal exceptions happen in that thread, the connector will also hang forever. This issue has
been observed for the Kafka source connector: apache#9464. We have fixed it by adding
the notifyError method to the `PushSource` class in PIP-281: apache#20807. However, this
does not solve the same problem that all source connectors face because not all connectors are implemented based on
the `PushSource` class.

The problem is same for the Pulsar Function. Currently, the function can't throw fatal exceptions to the function
framework. We need to provide a way for the function developer to implement it.

We need a way for the connector and function developers to throw fatal exceptions outside the function instance
thread. The function framework should catch these exceptions and terminate the function accordingly.

### Modifications

 Introduce a new method `fatal` to the context. All the connector implementation code and the function code 
 can use this context and call the `fatal` method to terminate the instance while raising a fatal exception. 
  
 After the connector or function raises the fatal exception, the function instance thread will be interrupted. 
 The function framework then could catch the exception, log it, and then terminate the function instance.

…1168)

…ernalCallbacks.Processor (apache#21159)

… to 32MB (apache#21131)

… port if TLS is enabled (apache#21015)

…problem. (apache#21161)

…pache#21181)

Co-authored-by: 张浩 <zhanghao60@100.me>
Co-authored-by: tison <wander4096@gmail.com>

…ickly (apache#21207)

…opic (apache#21205)

… many requests (apache#21216)

Motivation: The Pulsar client will close the socket if it receives a `ServiceNotReady` error when doing a lookup. The Broker will respond to the client with a `TooManyRequests` error if there are too many lookup requests in progress, but the Pulsar Proxy responds to the client with a `ServiceNotReady` error in the same scenario.

Modifications: Make Pulsar Proxy respond to the client with a `TooManyRequests` error if there are too many lookup requests in progress.

…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.

…1222)

…apache#21237)

…oad-date CRUD (apache#21129)

…ache#21204)

### Motivation

We have introduced the `SinkContext.fatal` method in apache#21143
This PR use this fatal method to handle the elastic sink connector exception correctly.

### Modifications

- Use `SinkContext.fatal` to throw connector exception
- Add state to the ElasticSearchClient

…nd entry based on publishTime or index (apache#20867)

…ishContext (apache#21245)

…ding ack state (apache#21177)

…ndle unloading or metadata ex (apache#21211)

### Motivation

**Background**: The Pulsar client will close the socket if it receives a ServiceNotReady error when doing a lookup. 
Closing the socket causes the other consumer or producer to reconnect and does not make the lookup more efficient.

There are two cases that should be improved:
- If the broker gets a metadata read/write error, the broker responds with a `ServiceNotReady` error, but it should respond with a `MetadataError`
- If the topic is unloading, the broker responds with a `ServiceNotReady` error.

### Modifications
- Respond to the client with a `MetadataError` if the broker gets a metadata read/write error.
- Respond to the client with a `MetadataError` if the topic is unloading

…eper to Pulsar. (apache#21188)

### Motivation

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.

…ing queue is not empty (apache#21259)

### Motivation

Reproduce steps:
- Create a reader.
- Reader pulls messages into `incoming queue`, do not call `reader.readNext` now.
- Trim ledger task will delete the ledgers, then there is no in the topic.
- Now, you can get messages if you call `reader.readNext`, but the method `reader.hasMessageAvailable` return `false`

Note: the similar issue of `MultiTopicsConsumerImpl` has been fixed by apache#13332, current PR only trying to fix the issue of `ConsumerImpl`.

### Modifications

Make `reader.hasMessageAvailable` return `true` when `incoming queue` is not empty.

…er id when switch ledger (apache#21201)

### Modifications
- Print a warning log if the SSL handshake error 
- Print ledger ID when switching ledger

apache#21278)

…ear (apache#21302)

…#21273)

…o after bkclient initialize (apache#20944)

…g extensible load manager (apache#21213)

### Motivation

After trimming ledgers, the variable `lastConfirmedEntry` of the managed ledger might rely on a deleted ledger(the latest ledger which contains data).

There is a bug that makes pulsar allow users to set the start read position to an unexisting ledger or a deleted ledger when creating a subscription. This makes the `backlog` and `markDeletedPosition` wrong. 

### Modifications

Fix the bug.

Fix zookeeper related flacky test

)

### Motivation

See pip: apache#21129

### Modifications

Add  `BundleDataResources`

…gers (apache#21250)

### Background
- But after trimming ledgers, `ml.lastConfirmedPosition` relies on a deleted ledger when the current ledger of ML is empty. 
- Cursor prevents setting `markDeletedPosition` to a value larger than `ml.lastConfirmedPosition`, but there are no entries to read<sup>[1]</sup>.
- The code description in the method `advanceCursors` said: do not make `cursor.markDeletedPosition` larger than `ml.lastConfirmedPosition`<sup>[2]</sup>

### Issue
If there is no durable cursor, the `markDeletedPosition` might be set to `{current_ledger, -1}`, and `async mark delete` will be prevented by the `rule-2` above. So he `backlog`, `readPosition`, and `markDeletedPosition` of the cursor will be in an incorrect position after trimming the ledger. You can reproduce it by the test `testTrimLedgerIfNoDurableCursor`

### Modifications
Do not make `cursor.markDeletedPosition` larger than `ml.lastConfirmedPosition` when advancing non-durable cursors.

### 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.

…sagePublishContext (apache#21245) (apache#21309)

…pache#21329)