…apache#19275)

### Motivation

#### 1. `readPosition` point to a deleted ledger

When `trim ledgers` and `create new cursor` are executed concurrently, it will cause the `readPosition` of the cursor to point to a deleted ledger.

| time | `trim ledgers` | `create new cursor` |
| --- | --- | --- |
| 1 | | set read position and mark deleted position |
| 2 | delete ledger | |
| 3 | | add the cursor to `ManagedLedger.cursors` |

----

#### 2. Backlog wrong caused by `readPosition` wrong
<strong>(Highlight)</strong>Since the read position of the cursor is pointing at a deleted ledger, so deleted messages will never be consumed or acknowledged. Since the backlog in the API `topics stats` response is calculated as this: `managedLedger.entriesAddedCounter - cursor.messagesConsumedCounter`, the result is: Topics stats show `msgBacklog` but there is reality no backlog.
- `managedLedger.entriesAddedCounter`: Pulsar will set it to `0` when creating a new managed ledger, it will increment when adding entries.
- `cursor.messagesConsumedCounter`: Pulsar will set it to `0` when creating a new cursor, it will increment when acknowledging.

For example:
- write entries to the managed ledger: `{1:0~1:9}...{5:0~5:9}`
  - `managedLedger.entriesAddedCounter` is `50` now
- create a new cursor, and set the read position to `1:0`
  -  `cursor.messagesConsumedCounter` is `0` now
- delete ledgers `1~4`
- consume all messages
  - can only consume the messages {5:0~5:9}, so `cursor.messagesConsumedCounter` is `10` now
- the `backlog` in response of `topics stats` is `50 - 10 = 40`, but there reality no backlog

----

#### 3. Reproduce issue
Sorry, I spent 4 hours trying to write a non-invasive test, but failed. <strong>(Highlight)</strong>You can reproduce by `testBacklogIfCursorCreateConcurrentWithTrimLedger` in the PR apache#19274

https://github.com/apache/pulsar/blob/a2cdc759fc2710e4dd913eb0485d23ebcaa076a4/pulsar-broker/src/test/java/org/apache/bookkeeper/mledger/impl/StatsBackLogTest.java#L163

### Modifications

Avoid the race condition of `cursor.initializeCursorPosition` and `internalTrimLedgers`

(cherry picked from commit 4139fef)

Signed-off-by: Zixuan Liu <nodeces@gmail.com>

…dCursor(single subscription check) upon subscription (apache#19343)

…apache#19354)

…lower JVM versions (apache#19362)

…IP address (apache#19028)

(cherry picked from commit d8569cd)

…apache#19327)

(cherry picked from commit 3d8b52a)

Relates to: apache#17831 (comment)

### Motivation

When the `ProxyConnection` handles a `Connect` command, that is the time to go to `Connecting` state. There is no other time that makes sense to switch to connecting. The current logic will go to connecting in certain re-authentication scenarios, but those are incorrect. By moving the state change to earlier in the logic, we make the state transition clearer and prevent corrupted state.

### Modifications

* Remove `state = State.Connecting` from the `doAuthentication` method, which is called multiple times for various reasons
* Add `state = State.Connecting` to the start of the `handleConnect` method.

### Verifying this change

The existing tests will verify this change, and reading through the code makes it clear this is a correct change.

### Does this pull request potentially affect one of the following parts:

Not a breaking change.

### Documentation

- [x] `doc-not-needed`

It would be nice to map out the state transitions for our connection classes. That is our of the scope of this small improvement.

### Matching PR in forked repository

PR in forked repository: michaeljmarshall#21

(cherry picked from commit c8650ce)

…9377)

(cherry picked from commit 1481c74)

…pache#19288)

… close (apache#16756)

(cherry picked from commit 14912a6)

…pache#19423)

Co-authored-by: Michael Marshall <mmarshall@apache.org>

### Motivation

Cherry-pick apache#17123

### Verifying this change

- [ ] Make sure that the change passes the CI checks.

*(Please pick either of the following options)*

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

*(or)*

This change is already covered by existing tests, such as *(please describe tests)*.

*(or)*

This change added tests and can be verified as follows:

*(example:)*
  - *Added integration tests for end-to-end deployment with large payloads (10MB)*
  - *Extended integration test for recovery after broker failure*

### Does this pull request potentially affect one of the following parts:

<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->

*If the box was checked, please highlight the changes*

- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment

### Documentation

<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->

- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

### Matching PR in forked repository

PR in forked repository: <!-- ENTER URL HERE -->

<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.

apache/pulsar pull requests should be first tested in your own fork since the 
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in 
a forked repository.

The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->

Co-authored-by: fengwenzhi <fengwenzhi.max@bigo.sg>
(cherry picked from commit 41edd2e)

…8975)

### Motivation

fix multi invocation for ledger `createComplete`

### Modifications

Only call `createComplete` at the point of creating ledger timeout if the ledger is not created normally

(cherry picked from commit 8f1c1b1)

(cherry picked from commit 4129583)

…essage if enabled read compacted (apache#18877)

### Motivation

The method `consumer.getLastMessageId` will return the latest message which can be received.
- If disabled `read compacted`, will return the last confirmed position of `ManagedLedger`.
- If enabled `read compacted`, will return the latest message id which can be read from the compacted topic.

If we send a batch message like this:

```java
producer.newMessage().key("k1").value("v0").sendAsync(); // message-id is [3:1,-1:0]
producer.newMessage().key("k1").value("v1").sendAsync(); // message-id is [3:1,-1:1]
producer.newMessage().key("k1").value("v2").sendAsync(); // message-id is [3:1,-1:2]
producer.newMessage().key("k2").value("v0").sendAsync(); // message-id is [3:1,-1:3]
producer.newMessage().key("k2").value("v1").sendAsync(); // message-id is [3:1,-1:4]
producer.newMessage().key("k2").value(null).sendAsync(); // message-id is [3:1,-1:5]
producer.flush();
```

After the compaction task is done, the messages with key `k2` will be deleted by the compaction task. Then the latest message that can be received will be `[3:1:-1:2]`.

---
When we call `consumer.getLastMessageId`, the expected result is:

```
[3:1,-1:2]
```

---
But the actual result is:

```
[3:1,-1:5]
```

### Modifications
If enabled `read compacted` and the latest entry of the compacted topic is a batched message, extract the entry and calculate all internal messages, then return the latest message which is not marked `compacted out`.

(cherry picked from commit 83993ae)

…#18421)

(cherry picked from commit ac9742a)

…partition-` (apache#19230)

(cherry picked from commit fc4bca6)

…#19404)

(cherry picked from commit 33f40f6)

apache#19129)

(cherry picked from commit a6516a8)

…he#19258)

(cherry picked from commit cfd7e60)

(cherry picked from commit 96fb7da)

(cherry picked from commit 96fb7da)

…pic partition number. (apache#19223)

(cherry picked from commit 253e3e4)

…e#19446)

(cherry picked from commit 524288c)
(cherry picked from commit 4cbe68e)

…ion fails (apache#19129)"

This reverts commit 9f0e8e6.

…tent topic timeout (apache#19454)

Co-authored-by: Tao Jiuming <95597048+tjiuming@users.noreply.github.com>

…ache#12615)" (apache#19439)

This reverts commit 62e2547.

### Motivation

The motivation for apache#12615 relies on an incorrect understanding of Netty's threading model. The `ctx.executor()` is the context's event loop thread that is the same thread used to process messages. The `waitingForPingResponse` variable is only ever updated/read from the context's event loop, so there is no need to make the variable `volatile`.

### Modifications

* Remove `volatile` keyword for `waitingForPingResponse`

### Verifying this change

Read through all references to the variable.

### Documentation

- [x] `doc-not-needed`

### Matching PR in forked repository

PR in forked repository: Skipping for this trivial PR.

(cherry picked from commit fb28d83)

…ot aware rack info problem. (apache#18672)

(cherry picked from commit 43335fb)

(cherry picked from commit 11073fd)

…9346)

(cherry picked from commit e2851da)

…nt (apache#19308)

(cherry picked from commit 644be5f)

… txn race condition. (apache#19201)

Fixes apache#19200

transaction lasted for long time and will not be aborted, which cause TB's MaxReadPosition do not move and will not take snapshot. With an old snapshot, TB will read a lot of entry while doing recovery.
In worst cases, there are 30 minutes of unavailable time with Topics.

avoid concurrent execution.

(cherry picked from commit 96f4161)

…ache#19467) (apache#19473)

Co-authored-by: Nicolò Boschi <boschi1997@gmail.com>

…che#19199)

(cherry picked from commit 4d57828)

When a Pulsar topic is unloaded from a broker, certain metrics related to that topic will appear to remain active for the broker for 5 minutes. This is confusing for troubleshooting because it makes the topic appear to be owned by multiple brokers for a short period of time. See below for a way to reproduce this behavior.

In order to solve this "zombie" metric problem, I propose we remove the timestamps that get exported with each Prometheus metric served by the broker.

Since we introduced Prometheus metrics in #294, we have exported a timestamp along with most metrics. This is an optional, valid part of the spec defined [here](https://prometheus.io/docs/instrumenting/exposition_formats/#comments-help-text-and-type-information). However, after our adoption of Prometheus metrics, the Prometheus project released version 2.0 with a significant improvement to its concept of staleness. In short, before 2.0, a metric that was in the last scrape but not the next one (this often happens for topics that are unloaded) will essentially inherit the most recent value for the last 5 minute window. If there isn't one in the past 5 minutes, the metric becomes "stale" and isn't reported. Starting in 2.0, there was new logic to consider a value stale the very first time that it is not reported in a scrape. Importantly, this new behavior is only available if you do not export timestamps with metrics, as documented here: https://prometheus.io/docs/prometheus/latest/querying/basics/#staleness. We want to use the new behavior because it gives better insight into all topic metrics, which are subject to move between brokers at any time.

This presentation https://www.youtube.com/watch?v=GcTzd2CLH7I and slide deck https://promcon.io/2017-munich/slides/staleness-in-prometheus-2-0.pdf document the feature in detail. This blog post was also helpful: https://www.robustperception.io/staleness-and-promql/.

Additional motivation comes from mailing list threads like this one https://groups.google.com/g/prometheus-users/c/8OFAwp1OEcY. It says:

> Note, however, that adding timestamps is an extremely niche use
case. Most of the users who think the need it should actually not do
it.
>
> The main usecases within that tiny niche are federation and mirroring
the data from another monitoring system.

The Prometheus Go client also indicates a similar motivation: https://pkg.go.dev/github.com/prometheus/client_golang/prometheus#NewMetricWithTimestamp.

The OpenMetrics project also recommends against exporting timestamps: https://github.com/OpenObservability/OpenMetrics/blob/main/specification/OpenMetrics.md#exposing-timestamps.

As such, I think we are not a niche use case, and we should not add timestamps to our metrics.

1. Run any 2.x version of Prometheus (I used 2.31.0) along with the following scrape config:
```yaml
  - job_name: broker
    honor_timestamps: true
    scrape_interval: 30s
    scrape_timeout: 10s
    metrics_path: /metrics
    scheme: http
    follow_redirects: true
    static_configs:
      - targets: ["localhost:8080"]
```
2. Start pulsar standalone on the same machine. I used a recently compiled version of master.
3. Publish messages to a topic.
4. Observe `pulsar_in_messages_total` metric for the topic in the prometheus UI (localhost:9090)
5. Stop the producer.
6. Unload the topic from the broker.
7. Optionally, `curl` the metrics endpoint to verify that the topic’s `pulsar_in_messages_total` metric is no longer reported.
8. Watch the metrics get reported in prometheus for 5 additional minutes.

When you set `honor_timestamps: false`, the metric stops getting reported right after the topic is unloaded, which is the desired behavior.

* Remove all timestamps from metrics
* Fix affected tests and test files (some of those tests were in the proxy and the function worker, but no code was changed for those modules)

This change is accompanied by updated tests.

This is technically a breaking change to the metrics, though I would consider it a bug fix at this point. I will discuss it on the mailing list to ensure it gets proper visibility.

Given how frequently Pulsar changes which metrics are exposed between each scrape, I think this is an important fix that should be cherry picked to older release branches. Technically, we can avoid cherry picking this change if we advise users to set `honor_timestamps: false`. However, I think it is better to just remove them.

- [x] `doc-not-needed`

(cherry picked from commit 0bbc4e1)

…#16744)

(cherry picked from commit eb5725a)

…19412)

Signed-off-by: Zixuan Liu <nodeces@gmail.com>
(cherry picked from commit 016e7f0)

…ll messages. (apache#19428)

(cherry picked from commit c91303d)

…LQ (apache#19392)

(cherry picked from commit 39dd1cd)

Fixes: apache#19478

### Motivation

See issue for additional context. Essentially, we are doing a shallow clone when we needed a deep clone. The consequence is leaked labels, annotations, and tolerations.

### Modifications

* Add a `deepClone` method to the `BasicKubernetesManifestCustomizer.RuntimeOpts` method. Note that this method is not technically a deep clone for the k8s objects. However, based on the way we "merge" these objects, it is sufficient to copy references to the objects.

### Verifying this change

Added a test that fails before the change and passes afterwards.

### Documentation

- [x] `doc-not-needed`

This is an internal bug fix. No docs needed.

### Matching PR in forked repository

PR in forked repository: michaeljmarshall#27

(cherry picked from commit 0205148)

Fixes apache#19431

`authenticationData` is already `volatile`. We use `originalAuthData` when set, so we should match the style. In apache#19431, I proposed that we find a way to not use `volatile`. I still think this might be a "better" approach, but it will be a larger change, and since we already use `volatile` for `authenticationData`, I think this is the right change for now.

It's possible that this is not a bug, given that the `originalAuthData` does not change frequently. However, we always want to use up to date values for authorization.

* Add `volatile` keyword to `ServerCnx#originalAuthData`.

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

- [x] `doc-not-needed`

PR in forked repository: skipping test in fork.

(cherry picked from commit c4c1744)
(cherry picked from commit e1d9941)

…ache#19270)

(cherry picked from commit fd3ce8b)
(cherry picked from commit 2847dd1)

I broke all release branches when I cherry picked 2847dd1 to them. This change takes some of the underlying logic from apache#19409, without taking the async logic.

* Make changes to `ServerCnx` to make tests pass

Tests are currently failing, so passing tests will show that this solution is correct.

- [x] `doc-not-needed`

(cherry picked from commit 8246da2)
(cherry picked from commit 15e4198)

(cherry picked from commit 456d112)

…d incorrectly (apache#19355)

(cherry picked from commit fc9e8bf)

…in to prevent ZK event thread deadlock (apache#19539)

…pal (apache#19455)

Co-authored-by: Lari Hotari <lhotari@apache.org>
(cherry picked from commit aa63a55)

…he#19312)

PIP: apache#12105

When authentication fails in the `ServerCnx`, the state is left in `Start` if the primary `authData` fails authentication and in `Connecting` or `Connected` if the `originalAuthData` authentication fails. To prevent any kind of unexpected behavior, we should go to `Failed` state.

Note that the tests verify the current behavior where a failed `originalAuthData` results first in a `Connected` command from the broker and then an `Error` command. I documented that I think this is sub optimal here apache#19311.

* Update `ServerCnx` state to `Failed` when there is an authentication exception during `handleConnect` and during `handleAuthResponse`.
* Update `handleAuthResponse` reply to `"Unable to authenticate"` instead of the `AuthenticationState` exception.

A new test is added. The added test covers the change made in apache#19295 where we updated `ServerCnx` so that we call `AuthState#authenticate` instead of relying on the implementation detail that the initialization calls `authenticate`. That PR should have added a test.

This is not a breaking change.

- [x] `doc-not-needed`

PR in forked repository: michaeljmarshall#18

(cherry picked from commit 8049690)
(cherry picked from commit 3ef3bf1)

(cherry picked from commit 2225361)

…9430)

(cherry picked from commit aa7af10)
(cherry picked from commit c0f38a0)

This change was introduced by apache#19295.

That PR had more changes than are worth cherry-picking, though, so this
commit only has the additional call to authenticate the original auth data.
As a result, this commit is slightly less efficient because in some
implementations, the authdata will be validated twice.

(cherry picked from commit f9727ca)

In c0f38a0, we cherry picked a commit that relied on
a feature not yet implemented in this branch. This commit fixes what that commit
broke.

(cherry picked from commit 547dee2)

…oxy (apache#19576)

### Motivation

Cherry pick apache#19540 to branch 2.11

…ache#19594)

(cherry picked from commit f292bad)
(cherry picked from commit 6576231)

Co-authored-by: Lari Hotari <lhotari@users.noreply.github.com>
(cherry picked from commit 153e4d4)
(cherry picked from commit 161ec5a)

…Role (apache#19557)

I broke the Pulsar Proxy with apache#19455 because that PR requires that when `X-Original-Principal` is supplied, the auth role must be a proxy role. This is not always the case for proxied admin requests. This PR seeks to fix that incorrect assumption by changing the way verification is done for the roles. Specifically, when the two roles are the same and they are not a proxy role, we will consider it a valid combination.

Note that there is no inefficiency in this solution because When the `authenticatedPrincipal` is not a proxy role, that is the only role that is authenticated. Note also that we do not let the binary protocol authenticate this way, and that is consistent with the way the pulsar proxy forwards authentication data.

Currently, we do the following when authentication is enabled in the proxy:

1. Authenticate the client's http request and put the resulting role in the `X-Original-Principal` header for the call to the broker.
https://github.com/apache/pulsar/blob/38555851359f9cfc172650c387a58c5a03809e97/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java#L370-L373

2. Copy the `Authorization` header into the broker's http request:
https://github.com/apache/pulsar/blob/38555851359f9cfc172650c387a58c5a03809e97/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java#L232-L236

3. Configure the proxy's http client to use client TLS authentication (when configured):
https://github.com/apache/pulsar/blob/38555851359f9cfc172650c387a58c5a03809e97/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java#L269-L277

The problem with apache#19455 is that it assumes the proxy supplies its own authentication data. However, that only happens when using TLS authentication. Otherwise, the proxy forwards the client's authentication data in the `Authorization` header. As such, calls will fail because the `X-Original-Principal` header supplied without using a proxy role.

* Consider the `authenticatedPrincipal` and the `originalPrincipal` a valid pair when they are equal and are not a `proxyRole` for http requests.

I initially proposed that we only add the `X-Original-Principal` when we are using the proxy's authentication (see the first commit). I decided this solution is not ideal because it doesn't solve the problem, it doesn't make the brokers backwards compatible, and there isn't actually any inefficiency in passing the role as a header.

When cherry-picking apache#19455 to branch-2.9, I discovered that `PackagesOpsWithAuthTest#testPackagesOps` was consistently failing because of the way the proxy supplies authentication data when proxying http requests. That test was removed by apache#12771, which explains why I didn't catch the error sooner. This PR includes a test that fails  without this change.

Note that the primary issue must be that we didn't have any tests doing authentication forwarding through the proxy. Now we will have both relevant tests where the proxy is and is not authenticating.

This is not a breaking change.

- [x] `doc-required`

PR in forked repository: michaeljmarshall#31

(cherry picked from commit d4be954)
(cherry picked from commit 5f5551d)

…lti-dns names proxy (apache#19597)

(cherry picked from commit e286339)
(cherry picked from commit 14b070b)

…ServerCnx (apache#19517)

(cherry picked from commit 0bb0f6b)

Signed-off-by: Zixuan Liu <nodeces@gmail.com>

…ersistent policies cannot set to < 0 (apache#18999)

(cherry picked from commit 4cae20c)

…follow BookieServiceInfo updates (apache#18133)

* [fix] PulsarRegistrationClient - implement getAllBookies and follow BookieServiceInfo updates

* remove debug

(cherry picked from commit 0c326c0)

…ll (apache#16946)

(cherry picked from commit 046068a)

When calling the method `PulsarWebResource.getRedirectionUrl`, reuse the same `PulsarServiceNameResolver` instance.

(cherry picked from commit f9af424)

Signed-off-by: xiaolongran <xiaolongran@tencent.com>
(cherry picked from commit c0f89dc)

…eUrl (apache#19510)

(cherry picked from commit 0f025f3)

…t redevlier duplicate messages (apache#14327)

If individual ack batch message with transaction and abort this transaction, we will redeliver this message. but this batch message some bit sit are acked by another transaction and re consume this bit sit will produce `TransactionConflictException`, we don't need to redeliver this bit sit witch is acked by another transaction.

if batch have batch size 5

1. txn1 ack 0, 1     the ackSet is   00111
2. txn2 ack 2 3 4 the ack Set is  11000
3. abort txn2 redeliver this position is 00111
4. but now we don't filter txn1 ackSet so redeliver this position bitSet is 111111
When filter the message we should filter the bit sit witch is real ack or in pendingAck state
add the test

(cherry picked from commit e0c0d5e)

…pache#18211)

(cherry picked from commit 5b7c5c6)

…min client (apache#18358)

(cherry picked from commit 0f72a82)

…sForConsumer (apache#18729)

origin extract entry metadata logic is based on `Optional.map.orElseGet` which can be simplified by if condition and also has better performance on hot code path.

1. use if null check replace Optional code.
2. remove duplicate hasChunk check logic in `PersistentDispatcherMultipleConsumers.trySendMessagesToConsumers`

(cherry picked from commit a1e3b80)

…ope (apache#19234)

This PR is following up apache#19230

As @yuruguo mentioned apache#19230 (comment), we can tighten the validation scope.

I've checked the logic and found we have no way to create the partition topic with the `-partition-{index}` template. So we can righten the validation scope.

I will keep working on the partition topic section and try to clarify the concept and logic. Plus, ensuring compatibility.

- tighten the validation scope

(cherry picked from commit 246c270)

Co-authored-by: Cong Zhao <zhaocong@apache.org>
Co-authored-by: fengyubiao <9947090@qq.com>
Co-authored-by: AloysZhang <lofterzhang@gmail.com>

 (apache#19333)

(cherry picked from commit 71dafe8)

(cherry picked from commit 27d392d)

(cherry picked from commit 6bf4966)

(cherry picked from commit ca177b3)

This reverts commit e32f9ad.

Makes only ledgers removed from the meta of ledger info can be deleted from the BK.

(cherry picked from commit 3314d70)

…backlog-size of admin API topics stats true (apache#19302)

(cherry picked from commit e417fe7)

…ription-backlog-size of admin API topics stats true (apache#19302)"

This reverts commit a863e7d.

…) (apache#19637)

Co-authored-by: Yan Zhao <horizonzy@apache.org>

…pache#19674)

When we merged apache#15818 in order to make the broker's client configurable, we did not add an explicit config for hostname verification. This PR adds that config to the broker and the websocket proxy. I chose the name `tlsHostnameVerificationEnabled` because that is what is already used in the proxy. It diverges from the function worker's config of `tlsEnableHostnameVerification`.

Before this PR, you would have enabled hostname verification by configuring `brokerClient_tlsHostnameVerificationEnable=true` in the broker and WS proxy configs. (Note that the variable name is slightly different because the `ClientConfiguration` does not have a `d` at the end of its name.

The remaining follow up work will be to update the `ClusterData` objects to configure hostname verification there to make it easier to configure hostname verification for remote clusters.

* Add `tlsHostnameVerificationEnabled` to the `broker.conf` and the `proxy.conf`
* Update all of the relevant locations that were previously only relying on `brokerClient_tlsHostnameVerificationEnable`

I added a single test to ensure that the `WebSocketProxyConfiguration` properly converts to the `ServiceConfiguration` object.

Otherwise, I verified that anywhere we are using `"brokerClient_"`, this PR also adds the right configuration.

This PR introduces a "new" configuration key, but not a new concept. All underlying behaviors are unchanged.

- [x] `doc-not-needed`

Docs are automatically updated by these changes.

(cherry picked from commit 6621fd3)
(cherry picked from commit b8083b0)

This reverts commit d8763d4.

(cherry picked from commit e13865c)

… fence state (apache#19692)

Motivation : After a Topic close failure or a delete failure, the fence state will be reset to get the topic back to work,but it will not reset the fence state of the subscription, which will result in the consumer never being able to connect to the broker.
Modifications: Reset the fence state of subscriptions if the operation of topic close is failed.
(cherry picked from commit cdeef00)

…ot to work (apache#19754) (apache#19768)

Co-authored-by: Qiang Zhao <mattisonchao@apache.org>
Master apache#19754
### Motivation

This is a namespace bundle double-owners problem. We found it in the memory dumps.

The memory dumps show that the notification thread has been blocked for a long time by the leader election deadlock, And many notifications are blocked in the executor queue. This causes we can't to revalidate the locks, and they are still thinking them working well.

For private reasons, I can't share the namespace bundle snapshot, but the blocked thread easily explains it.

<img width="1061" alt="image" src="https://user-images.githubusercontent.com/74767115/223670419-c4319f44-f1e1-4361-8c79-04c7f0dabe3c.png"> ^^ blocked thread

<img width="949" alt="image" src="https://user-images.githubusercontent.com/74767115/223672000-fb4ce22c-6a45-4cb6-9d23-c36e3afbc93a.png">

^^ executor queue

### Modifications

- Avoid putting the new task to single thread executor causes deadlock.

(cherry picked from commit cbd799f)