道君- Tao Jiuming

it seems not an issue, try set `brokerDeleteInactivePartitionedTopicMetadataEnabled` to true

@rdhabalia This PIP are approved, are we still working on the PR?

Reopen the issue since the PR is in processing https://github.com/apache/pulsar/pull/21759

It looks the issue fixed by https://github.com/apache/pulsar/pull/14667, closing... feel free to reopen if this PR still ongoing

Just add additional context to help understand it quickly. [CopyingEncoder](https://github.com/apache/pulsar/blob/master/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/ByteBufPair.java#L138-L158) is used to handle the case of `SslHandler`. Because `SslHandler` could compose ByteBufs to one(The first input ByteBuf). For instance,...

Yes, of cause, my bad, since I didn't explain myself correctly.

I maybe understand what caused the issue, try set `isAckReceiptEnabled(true)` to true when create consumer. ```java Consumer consumer = client.newConsumer(Schema.STRING) .topic(t) .subscriptionName("frozen") // enable ack receipt .isAckReceiptEnabled(true) .subscriptionType(SubscriptionType.Shared) .subscribe(); ```...

```go func test() { clientOps := pulsar.ClientOptions{ URL: "pulsar://localhost:6650", } client, err := pulsar.NewClient(clientOps) if err != nil { return } defer client.Close() consumerOps := pulsar.ConsumerOptions{ Topic: "persistent://public/default/my-topic", SubscriptionName: "test",...

The problem is Pulsar optimized `consumer.ack`. When you call `consumer.ack`, the `ack` request will not send to the broker immediately, but cached in the client. When the cached ack requests...

enable `AckWithResponse`, the consumer will wait until broker received the ack