pulsar

These are the stats written to the log if that helps:
```2023-05-17T20:19:35,142+0000 [pulsar-timer-6-1] INFO  org.apache.pulsar.client.impl.ConsumerStatsRecorderImpl - [<persistent://hyperflow/demo/demo2_sink_topic>] [hyperflow/demo/demo2_sink] [05dee] Prefetched messages: 0 --- Consume throughput received: 0.02 msgs/s --- 0.00 Mbit/s --- Ack sent rate: 0.00 ack/s --- Failed messages: 0 --- batch messages: 0 ---Failed acks: 0```

<https://pulsar.apache.org/docs/2.11.x/admin-api-topics/#get-internal-stats>

pulsar-admin topics stats-internal <persistent://test-tenant/ns1/tp1>

Oh awesome! Thats new to me so thanks for showing it to me. Here is the output:
```{
  "entriesAddedCounter" : 1,
  "numberOfEntries" : 1,
  "totalSize" : 90,
  "currentLedgerEntries" : 1,
  "currentLedgerSize" : 90,
  "lastLedgerCreatedTimestamp" : "2023-05-17T20:16:35.195Z",
  "waitingCursorsCount" : 1,
  "pendingAddEntriesCount" : 0,
  "lastConfirmedEntry" : "311:0",
  "state" : "LedgerOpened",
  "ledgers" : [ {
    "ledgerId" : 311,
    "entries" : 0,
    "size" : 0,
    "offloaded" : false,
    "underReplicated" : false
  } ],
  "cursors" : {
    "hyperflow%2Fdemo%2Fdemo2_sink" : {
      "markDeletePosition" : "311:-1",
      "readPosition" : "311:1",
      "waitingReadOp" : true,
      "pendingReadOps" : 0,
      "messagesConsumedCounter" : 0,
      "cursorLedger" : 312,
      "cursorLedgerLastEntry" : 0,
      "individuallyDeletedMessages" : "[]",
      "lastLedgerSwitchTimestamp" : "2023-05-17T20:16:35.224Z",
      "state" : "Open",
      "numberOfEntriesSinceFirstNotAckedMessage" : 2,
      "totalNonContiguousDeletedMessagesRange" : 0,
      "subscriptionHavePendingRead" : true,
      "subscriptionHavePendingReplayRead" : false,
      "properties" : { }
    }
  },
  "schemaLedgers" : [ {
    "ledgerId" : 307,
    "entries" : 1,
    "size" : 80,
    "offloaded" : false,
    "underReplicated" : false
  }, {
    "ledgerId" : 308,
    "entries" : 1,
    "size" : 342,
    "offloaded" : false,
    "underReplicated" : false
  }, {
    "ledgerId" : 313,
    "entries" : 1,
    "size" : 1121,
    "offloaded" : false,
    "underReplicated" : false
  } ],
  "compactedLedger" : {
    "ledgerId" : -1,
    "entries" : -1,
    "size" : -1,
    "offloaded" : false,
    "underReplicated" : false
  }
}```

I managed to get it working but it required rebuilding the postgres jdbc sink with `_properties_.*setProperty*("stringtype", "unspecified");` so it would convert strings to `text`rather than `character varying`  (idea stolen from this thread: <https://postgrespro.com/list/thread-id/1939980>) and then sending the datetime as a string rather than a datatime object.
it works but I would rather use Pulsar built packages rather than my own so if anyone knows a way to do it please let me know!!

<https://twitter.com/jerrinot/status/1658766743215517698>

<@UDZPKSJSK> is the author of PIP-193 and can help here

Great. I'm working on using a transform function to transform from our protobuf to String for saving with the ElasticSearchSink

I think I have it mostly working, though I'm having some schema issues between the function and the sink I think.

I am getting an error in the ElasticSearchSink that says `Record does not carry message information` however when I'm looking at the transform function output, it is always producing something. I'm producing my record like this:
```return context.newOutputRecordBuilder(Schema.STRING)
                .key(id)
                .value(jsonObject.toString())
                .properties(properties)
                .build();```
which follows the function example pretty closely.  Any help

This is because you have `schemaEnable`  to false in the ES Sink config, so it tries to write the raw bytes data of the message. But as you have a transform, there's no serialized message.
Is it possible for you to set `schemaEnable`  to true ? This is a much better option as it uses the record instead of the message. In general, sinks should use the record abstraction, not the message. The default of this config option is false only for backward compatibility reasons.

The error message could probably be improved.

BTW, have you checked <https://github.com/datastax/pulsar-transformations> ?
There's a `cast`  operation that could be used to transform to STRING.
At the moment, Protobuf is not supported but the pieces are here, just need to add the conversion logic.
If you'd like to contribute by a PR, I would be very happy to help.

Thanks for the help. I can't use schema enabled because all of our schemas are protobuf. My transform function outputs a json object as a string that is properly formatted for our ES indexes.

I can handle it all as bytes very easily, but when I tried that I think I got a different set of errors...maybe it was schema bytes not supported.

The transform function uses a byte[] input already because it handles multiple types with a hint in the properties to properly convert to json.

If you transform your protobuf to schema STRING with your transform function, I think it should work with schemaEnable

Ok, thanks! I'll give that a try. I'm out of the office today, but I'll get after that on Monday morning.

I tried turning on schemaEnable this morning with my existing setup (separate proto -&gt; json function (byte[] -&gt; String) ), and the ElasticSearchSink wasn't thrilled about it.
<https://github.com/apache/pulsar/blob/b7f0004313ea4565717cc6d3c0b99aee5c079c6c/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java#L380|https://github.com/apache/pulsar/blob/b7f0004313ea4565717cc6d3c0b99aee5c079c6c/pul[…]/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java>
that seems to show that really only JSON or AVRO schemas are supported in the sink. It seems like it would be an easy lift to support STRING or BYTES (as long as they both contained utf-8 String data that could be parsed as valid JSON).

```java.lang.ClassCastException: class java.lang.String cannot be cast to class org.apache.pulsar.client.api.schema.GenericObject (java.lang.String is in module java.base of loader 'bootstrap'; org.apache.pulsar.client.api.schema.GenericObject is in unnamed module of loader 'app')
	at org.apache.pulsar.io.elasticsearch.ElasticSearchSink.extractIdAndDocument(ElasticSearchSink.java:197) ~[8HdTpyUyRQuezR6JAN4j-g/:?]
	at org.apache.pulsar.io.elasticsearch.ElasticSearchSink.write(ElasticSearchSink.java:113)```
That is specifically the error I'm getting.

My function is publishing a Record&lt;String&gt;, I'm going to switch it to a key, value and see if that cleans this up.

You're right, the ES Sink only supports JSON and AVRO atm...

So it seems the ES Sink could be improved to support more schema types, and records that don't have a message when schemaEnable is false.

I have a branch where I've done the basic work. All of the existing tests pass, I'm just working on how to best test my changes within the existing test framework

I've finally found the time to finish testing my changes and have created an issue (<https://github.com/apache/pulsar/issues/20619>) to track them. I'm going to submit the PR momentarily.

You can look at the docker images used in the broker pods.

or you can run command to check version in pod
```pulsar version```


you can put bookie config here. <https://github.com/apache/pulsar-helm-chart/blob/b3c19fd7fc7067e67940244777dd4241c8295dd4/charts/pulsar/values.yaml#L559>

Hello Yu Wei Sung,
Thanks, i use the same config, you can see a bookkeeper.conf (part without comments)

```## DB Ledger storage configuration
dbStorage_writeCacheMaxSizeMb=32
dbStorage_readAheadCacheMaxSizeMb=32
dbStorage_readAheadCacheBatchSize=1000
dbStorage_rocksDB_blockCacheSize=8388608

# Other RocksDB specific tunables
dbStorage_rocksDB_writeBufferSizeMB=8
dbStorage_rocksDB_sstSizeInMB=64
dbStorage_rocksDB_blockSize=65536
dbStorage_rocksDB_bloomFilterBitsPerKey=10
dbStorage_rocksDB_numLevels=-1
dbStorage_rocksDB_numFilesInLevel0=4
dbStorage_rocksDB_maxSizeInLevel1MB=256

# Settings below are used by stream/table service

extraServerComponents=org.apache.bookkeeper.stream.server.StreamStorageLifecycleComponent
storageserver.grpc.port=4181
dlog.bkcEnsembleSize=3
dlog.bkcWriteQuorumSize=2
dlog.bkcAckQuorumSize=2
storage.range.store.dirs=data/bookkeeper/ranges
storage.serve.readonly.tables=false```
In fact, the only difference is that I had to add the variable "extraServerComponents=org.apache.bookkeeper.stream.server.StreamStorageLifecycleComponent" because otherwise port 4181 is never open.

Sorry, I didn't specify the correct error, the main error is: (Caused by: java.nio.file.AccessDeniedException: /pulsar/data/bookkeeper/ranges)
```2023-05-19T13:10:25,997+0000 [io-write-scheduler-OrderedScheduler-0-0] INFO  org.apache.bookkeeper.statelib.impl.journal.AbstractStateStoreWithJournal - Initializing the local state for mvcc store 000000000000000000/000000000000000000/000000000000000000
2023-05-19T13:10:25,999+0000 [io-write-scheduler-OrderedScheduler-0-0] ERROR org.apache.bookkeeper.statelib.impl.kv.RocksdbKVStore - Failed to restore checkpoint: Checkpoint{ID='6e1c6fd3-8992-4bfe-b580-bb0e918332e0', createdAt: 0 null}
org.apache.bookkeeper.statelib.api.exceptions.StateStoreException: Failed to create dir 000000000000000000/000000000000000000/000000000000000000
...
...
Caused by: java.nio.file.AccessDeniedException: /pulsar/data/bookkeeper/ranges
...
..```

I will try to add volume /pulsar/data/bookkeeper/ranges

access denied. can you shell into bookie pod and check the owner of /pullsar/data?

Thanks ! its ok when i add volume /pulsar/data/bookkeeper/ranges
```$ bin/pulsar-admin functions querystate --tenant public --namespace default --name wordcount --key test --watch
{
  "key": "test",
  "stringValue": "hello pulsar",
  "version": 0
}```

Finally, I don't use the datastax chart because they use their own docker images, and they don't have the Pulsar 2.11.0 and 3.0.0 version yet.. however, I was strongly inspired by the chart to succeed in operate stateful functions. Do you know if I can use the official Pulsar 3.0.0 images with the datastax chart?

I don’t use helm chart because it is hard to control the version changes. 

Apologies for hijacking this thread, but I'm also having trouble enabling stream/state storage on the bookie in k8s. <@U02S6GPTREH> - you commented that the bookie config can be included in the values.yml `configData` - would you mind providing an example of the format? I've tried the following (also tried replacing `.` with `_` in the property names). (I'm using a replica count of 1 to test on Minikube)
```bookkeeper:
  replicaCount: 1 # default 4
  configData:
    extraServerComponents: "org.apache.bookkeeper.stream.server.StreamStorageLifecycleComponent"
    storageserver.grpc.port: "4181"
    dlog.bkcEnsembleSize: "1"
    dlog.bkcWriteQuorumSize: "1"
    dlog.bkcAckQuorumSize: "1"
    storage.range.store.dirs: "data/bookkeeper/ranges"
    storage.serve.readonly.tables: "false"
    <http://storage.cluster.controller.schedule.interval.ms|storage.cluster.controller.schedule.interval.ms>: "30000"```
... but only `extraServerComponents` gets picked up.

I had what I believe is a similar issue configuring it in Docker because `apply-config-from-env.py` only applies values which already have defaults in `bookkeeper.conf` and the various storage server properties are not in `bookkeeper.conf`. Running in Docker I had to 'add' them to the conf file as follows:
```    command: &gt;
      bash -c "
        bin/apply-config-from-env.py conf/bookkeeper.conf &amp;&amp;
        echo 'storageserver.grpc.port=4181' &gt;&gt; conf/bookkeeper.conf &amp;&amp;
        echo 'dlog.bkcEnsembleSize=1' &gt;&gt; conf/bookkeeper.conf &amp;&amp;
        echo 'dlog.bkcWriteQuorumSize=1' &gt;&gt; conf/bookkeeper.conf &amp;&amp;
        echo 'dlog.bkcAckQuorumSize=1' &gt;&gt; conf/bookkeeper.conf &amp;&amp;
        echo 'storage.range.store.dirs=data/bookkeeper/ranges' &gt;&gt; conf/bookkeeper.conf &amp;&amp;
        echo 'storage.serve.readonly.tables=false' &gt;&gt; conf/bookkeeper.conf &amp;&amp;
        echo '<http://storage.cluster.controller.schedule.interval.ms|storage.cluster.controller.schedule.interval.ms>=30000' &gt;&gt; conf/bookkeeper.conf &amp;&amp;
        exec bin/pulsar bookie
      " ```
Many thanks
(oh, and I'm using Pulsar 3.0.0 Docker images. So weird that Helm chart version 3.0.0 doesn't use image versions 3.0.0?!)

I made progress. This works
```bookkeeper:
  replicaCount: 1 # default 4
  configData:
    extraServerComponents: "org.apache.bookkeeper.stream.server.StreamStorageLifecycleComponent"
  additionalCommand: &gt; 
    echo 'storageserver.grpc.port=4181' &gt;&gt; conf/bookkeeper.conf &amp;&amp; 
    echo 'dlog.bkcEnsembleSize=1' &gt;&gt; conf/bookkeeper.conf &amp;&amp; 
    echo 'dlog.bkcWriteQuorumSize=1' &gt;&gt; conf/bookkeeper.conf &amp;&amp; 
    echo 'dlog.bkcAckQuorumSize=1' &gt;&gt; conf/bookkeeper.conf &amp;&amp; 
    echo 'storage.range.store.dirs=data/bookkeeper/ranges' &gt;&gt; conf/bookkeeper.conf &amp;&amp; 
    echo 'storage.serve.readonly.tables=false' &gt;&gt; conf/bookkeeper.conf &amp;&amp; 
    echo '<http://storage.cluster.controller.schedule.interval.ms|storage.cluster.controller.schedule.interval.ms>=30000' &gt;&gt; conf/bookkeeper.conf &amp;&amp;```
The final `&amp;&amp;` is required, otherwise bookie exec command gets written into `bookkeeper.conf` as well, rather than executed!

I think you need to add “PULSAR” as prefix to the new variables.  like
PULSAR_storageserver.grpc.port: “4181"

Can't seem to make that work, or see anything in the `apply-config-from-env` script which handles that pattern, just `PULSAR_PREFIX_` for env variables. (edit: I noticed a `-with-prefix` version in the repo which does in fact add/replace based on a prefix, but the helm chart uses `apply-config-from-env` version)

Now apologies <@U057X0JF1P0> - I've got to the same
```Caused by: java.nio.file.AccessDeniedException: /pulsar/data/bookkeeper/ranges
...
Cannot invoke "org.rocksdb.RocksDB.get(org.rocksdb.ColumnFamilyHandle, byte[])" because "this.db" is null```
errors you were seeing. When you say `i add volume /pulsar/data/bookkeeper/ranges`, would you mind explaining how you did that for the bookie in your setup? I'm struggling to work out how to do it, am fairly new to k8s. Found <https://github.com/apache/pulsar/pull/15769/files|this PR> which looked like fix for similar issue, but not working here. Trying to run in Minikube on an M2 Pro Mac, using official Pulsar Helm chart 3.0.0 - overriding to use Pulsar 3.0.0 Docker image. I have `volumes.persistence: false` in case that affects my ability to configure volumes. Thank you.

Since I've so far gone with `additionalCommand` solves everything :wink: I tried adding
```    mkdir /pulsar/data/bookkeeper/ranges &amp;&amp;
    chmod -R g+w /pulsar/data/bookkeeper/ranges &amp;&amp;```
to the above, but no joy

I don't know of any specific resources, but this section of the helm chart readme has some info on making sure the security context is set up correctly: <https://github.com/apache/pulsar-helm-chart/tree/master#upgrading-to-apache-pulsar-2100-and-above-or-helm-chart-version-300-and-above>

That error is typically part of a larger stack trace. I often see that error when there are issues with the client verifying the server's certificates or when hostname verification fails.

image.png

The pulsar helm chart is versioned independently of pulsar itself.

:tada: :zap: :party-parrot: Pulsar Virtual Summit Europe is starting in 10 hours....!
All, you’re invited to <https://events.zoom.us/ev/Ap6rsDg9LeVfmdajJ_eB13HH026J1d_o8OoTKkQnl_jzVl-srhwB~AggLXsr32QYFjq8BlYLZ5I06Dg|register and attend> either the live event starting at 9:00 AM CEST Tuesday, May 23rd or jump into the lobby and sessions as your work allows over the next 2 days to check out the recordings and event space.

Anyone have tried to deploy pulsar on AWS ECS ?

Hi everyone,
during today’s Pulsar summit, many speakers shared the tools they’ve crafted to enhance Pulsar’s functionalities. This highlighted the perfect opportunity to share our latest project, Streamvisor. *Just as Lenses and Conduktor amplified the Apache Kafka ecosystem, we aim to do the same for Pulsar, with a professional management UI that eliminates the need to build these tools from scratch.*

Excitingly, we’re launching our beta program and would be thrilled to have you on board! *Visit <http://streamvisor.com|streamvisor.com> to sign up and stay up to date. At this stage your feedback is instrumental in shaping our tool, so please drop a comment or PM me with your thoughts or questions.* We’re eager to hear from you!

A good UI for Pulsar definitely should be born this year.
In my opinion, it makes more sense to collaborate on a single one, instead of making the same things in parallel.
It will be fun. :sweat_smile:

<https://www.streemo.io/>

The Pulsar Metadata Store abstraction introduced in PIP 45 doesn't really provide a way to paginate, so this kind of change would probably be large. Just mentioning the feature in case someone is interested in working on it.

<@U01GAUEA9TR> With the current implementation, at what amount of topics in a namespace do you expect to start facing some problems?

I don't have any numbers. It'll be dependent on several factors, like the topic name length and the number of topics. My main point in posting is that the lack of pagination can make it take longer to get the topics and that makes page loads more expensive than they would be if you could request only the content you wanted to display

Screen Recording 2023-05-23 at 7.51.05 PM.mov

Not a perfect test, but for ~8000 topics with `topic-nn` names its about 200ms and 500kb (not gzipped).

Can’t say it’s too slow for such amount of topics, but of course, can be improved as you said. :)

Nice. I haven't looked at this in a while, but the other problem is likely when trying to get topic stats, which has a lot more data.

In case we need to display the data without sorting, we can make a limited amount of requests (10-20) to load stats only for visible topics in a list.

In case we need sorting on UI, it makes more sense to periodically poll Prometheus metrics from all brokers and preprocess them on the server side. Unfortunately, it looks like not all metrics can be calculated correctly this way, and I didn’t found a good solution yet.

Actually, with the OTel PIP by <@U01551PS4MA> it probably could be implemented in a better way. Need to dive into it a bit.

If someone could help with a few other questions, we could get a good Pulsar UI pretty fast. :smiley:

For now, server side pre-processing is the only way to go.
One option is to add a batch call to get certain metrics for many topics at once, so everything you need in one call.

<@U01551PS4MA> you mean to implement a Pulsar broker admin API endpoint for this use case?

If your intuition tells you that it may work fast enough and such a contribution may be accepted, I can try.

Should be a good Java and Pulsar codebase exercise for me.

For the batch stats - yes. The idea is that the topic stats can be requested not only for a single topic, and you can specify a filter, listing only the metrics you care about.
I’m pretty sure all the people that scrape those to compose their own special dashboards would appreciate it.

This endpoint can delegate calls to other brokers for you.

Ok. Actually great idea!
Will take a look in a few days.
Most likely will have some questions, be ready. :slightly_smiling_face:

&gt; This endpoint can delegate calls to other brokers for you.
I don't remember, do we do this already? This kind of operation seems like it could get expensive in large clusters.

Is a cluster of 100 brokers may be considered large enough?
I actually don’t see a problem with 100 parallel HTTP requests if the new endpoint won’t be called too frequently.

Probably I should make some tests first on a smaller scale first to better understand the data amount and latency, then multiply to get approximate results.

I’ve done some work with this in Go for work by hooking into both Prometheus and the `pulsarctl` libs. It’s not too difficult, but it would be a decent amount of work to flesh out all the calls, and to add caching layer and such.  Another option is to use Prometheus for coarse stats, making API calls as necessary when drilling down. 

hm cannot find the PIP which sounds related

could you help me find it?
also could you direct me towards who could help we with this undertaking?

<https://lists.apache.org/thread/k31ngxn298okztnxtnndqx8l2b58tsz7>

You can ping the people mentioned in this discussion via Slack or GitHub discussions or DEV mailing list.

cc <@U015AARD76F> <@U6D1PLAUT> <@U6E5YMRKR> - pls have a look on the issue raised above, you are mor ethan. welcome to comment.

in a nutshell it's about creating a BC FIPS implementation of MessageCrypto. The big diff when moving to FIPS is the ECDSA key handling code (I could not exactly replicate it - but I am not an expert), but BC FIPS is not supporting using ECDSA keys anyways - so that part could be specifically left in only for the non-FIPS code.

I have brought this issue to the attention of <@U03S6FZPT5F> as well

cc <@U03B2LX35R7> <@U02S6GPTREH> <@U01GAUEA9TR>
• who participated in related discussions as well

I have created a PR: <https://github.com/apache/pulsar/pull/20542>

a bump on this, so far seems there is not enough interest / bandwidth to review this PR: <https://github.com/apache/pulsar/pull/20542>

Hi everyone,
Sorry some how I missed this message.
<@U6E5YMRKR> please let me know if I should take a look or if this has been resolved already.

Hey, does anyone have any idea about if being able to use state with Go functions is on the road map?

Well this ns is now 404 not found. But its policies still exists in global zk, with deleted:false :confused:.

Ok, so im manually recreating them and deleting, then it clean global zk :slightly_smiling_face:

Do you have a chance to see if the problem is reproducible with Pulsar 2.10.4 (and with the PR you referenced)?

Unfortunately no :confused: But will do if I encounter it again.

hello everyone,
i am trying to install prometheus operator (using <https://prometheus-operator.dev/docs/prologue/quick-start/>) in a K8 cluster with apache pulsar already installed using helm chart. but after standalone prometheus installation, some of the pods like '<https://jp-tok.containers.cloud.ibm.com/kubeproxy/clusters/chj22rht0cab0fcrd35g/service/#/serviceaccount/pulsar/pulsar-prometheus-node-exporter?namespace=pulsar|pulsar-prometheus-node-exporter>' stopped running.
any idea if there will be any conflict if we install prometheus and pulsar in the same K8 cluster?
thanks in advance...

It looks like an issue in the MsgBacklog stats. If you are interested in the part, you can help look it over.

I suppose you have authentication enabled, right? In that case, what `authenticationProviders` are registered in the `broker.conf`?

Yeah, `pulsar-admin` works just fine for example, without specifying the provider or token

```# Authentication provider name list, which is comma separated list of class names
authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderToken```

Does it work using `pulsar-client produce` ?
The `pulsar-admin` goes against a different endpoint so it’s not comparable

Ah so it’s using token authentication. You should be able to make it work using the following:

```./bin/pulsar-perf produce \
--auth_plugin org.apache.pulsar.client.impl.auth.AuthenticationToken \
--auth-params token:&lt;YOUR-TOKEN&gt; \
test-topic```

So `/pulsar-client produce test-topic -m "tesT"` this seemed to work as `test-topic` has been created

It still seems to hang, not sure how long its expected to take to start up?

Weird, it should start immediately giving some info logs about its config

Is there still sufficient memory available on the broker?

i tried that `exit-on-failure` flag too but no luck there

Yeah that should only help if it’s already trying to produce, but it sounds like there is already an issue on startup

I think I’m out of ideas too. Can you try running the tool from your local machine instead?

Sure, should i just copy the script off of the instance?

You can just download the respective binaries from <https://pulsar.apache.org/download/|here>

hmmm, works locally it seems :thinking_face:

i think thats ok for my use case but just weird

You could try running with `--debug` , might see some more output <@U054Z6P6QLF>

Thanks both, think its in a pretty good spot now

I added a new page to the Pulsar website with a list of books.
Preview: <https://pulsar-site-git-books-and-videos-teal-tools.vercel.app/books/>
PR: <https://github.com/apache/pulsar-site/pull/586>

Please suggest your favorite books on the related topics.
It’s definitely not a goal to promote Kafka on the Pulsar website, please suggest something more-less technology-neutral.

How can I change the log level for the Java Pulsar Client?

<@U053NKRDBL0> :arrow_up: this is due to the fact that the client uses SLF4J for logging. So if your application is using log4j then your log4j config will control which levels. If you application is using Logback then your logback config will control which levels are logged.

ahhh `list-partitioned-topics` returns them.

Have you enabled topic auto-creation in the target cluster?

No, we’re controlling it from one of our wrapper application. Only admins have the access to create the topic. Producer/Consumer don’t have this access. We want to geo replicate topics created by admin across clusters. 

Do you mean you want the geo-replicate topics to be created by pulsar-admin by the source cluster ?

Would you please create an issue in the Pulsar github repo? It can be done from technical side, but we need more discussion

Pulsar with ZK currently supports up to 1M topics.

30M topics is too much :slightly_smiling_face:

&gt; One way we could break it apart would create 30 or 40 million topics
It maybe hard to support too much with Zookeeper, the new metadata impl Oxia may support. But we’d better not use it in this way, because it will cost a lot of resoruces

&gt; Another way may create a few thousand to a few hundred thousand topics, some of which may be very large, and others may be very small.
Some topics may be very large, and `large` means high throughput or high data storage? I think it will be OK

<@UQ8NL176Y> Sorry for the delayed response. Large in terms of storage size. Throughput is relatively modest overall.

&gt; Another way may create a few thousand to a few hundred thousand topics, some of which may be very large, and others may be very small.
The offloader can support large storage size, and this way will work for you

the highest throughput demands in our system are on recent data, which should generally be cached on NVMe local disks before being eventually moved to GCS. Am I over thinking this? Do we need to worry about the overall size of the topic in terms of number of messages?

You are right. We can keep the recent data both in BookKeeper and GCS, and Pulsar can control read data from which storage system when data is stored in both systems, and you can set the reading to BookKeeper to get high throughput and low latency.

&gt; Do we need to worry about the overall size of the topic in terms of number of messages?
Do not need to worry about it.

Thank you! I really appreciate the help Hang.

&gt; a) Backlog for the topics that we produced messages to is showing 0 , however the `storageSize`(from TopicStats) is not. Shows a sizable number ~ 130 MB on those topics
Even though the backlog is 0, but the cursor’s mark-delete position still point to the last ledger of the topic, and it won’t be deleted. So the 130MB storage size maybe the last ledger’s size. You can use `bin/pulsar-admin topics stats-internal &lt;topic&gt;` to check it

&gt; b) The disk space is not getting reclaimed even after almost 2 days. We have tweaked the `gcWaitTime` &amp; `majorCompactionInterval/threshold` configuration properties in hopes of it clearing the disk space.
After the disk runs into read-only (95% of usage), the garbage collection on Bookie won’t work. You can make it work by following this PR: <https://github.com/apache/bookkeeper/pull/3205>

&gt; Even after running consumers to drain of any (hidden) backlog, the `disk utilization %` hasn’t come down.
The root cause is the same with b)

&gt; d) We have also tried running `clear-backlog` command from CLI at the namespace level without any help in clearing out the disks
The root cause is the same with b)

<@UQ8NL176Y> Thank you very much for your inputs and suggestions. I see the above PR changes are in `Bookkeeper v4.15.0`  .. Would upgrade to `Pulsar v2.11.0` address this issue since it uses `4.15.4` ? Thanks

Also, `forceAllowCompaction`  seems to be a broker_server.conf property in bookkeeper project. Is that exposed as part of Pulsar configuration in `bookkeeper.conf` ?

Yes, you can set `forceAllowCompaction=true` in conf/bookkeeper.conf. The difference between `bk_server.conf` in the BookKeeper repo and `conf/bookkeeper.conf` in the Pulsar repo is Pulsar overrides some configurations for BookKeeper.

So.. After
a) Upgrading to Pulsar `v2.11.1` (Bookkeeper v`4.15.4`)
b) Setting `forceAllowCompaction` to true in conf/bookkeeper.conf
I keep getting `HTTP 500` Internal Server Error (details below)
``` curl -v -XPUT <http://localhost:8000/api/v1/bookie/gc>
*   Trying 127.0.0.1:8000...
* TCP_NODELAY set
* Connected to localhost (127.0.0.1) port 8000 (#0)
PUT /api/v1/bookie/gc HTTP/1.1
Host: localhost:8000
User-Agent: curl/7.68.0
Accept: /

* Mark bundle as not supporting multiuse
&lt; HTTP/1.1 500 Internal Server Error
&lt; content-length: 21```
The majorCompactionCounter keeps increasing in gc_details response
```curl -XGET <http://localhost:8000/api/v1/bookie/gc_details>
[ {
  "forceCompacting" : false,
  "majorCompacting" : false,
  "minorCompacting" : false,
  "lastMajorCompactionTime" : 1685458132776,
  "lastMinorCompactionTime" : 1685458132776,
  "majorCompactionCounter" : 1745,
  "minorCompactionCounter" : 0
}]```
Any pointers on why ? I couldn't find anything in the logs

&gt; ```500 Internal Server Error```
&gt; 
Could you share the bookie server logs, it is a server error

bookkeeper-server.log file is empty and only available in one of  3 bookie pods :sob: . I `exec` 'd into a bookie pod to check for it .. Am I supposed to be looking for it elsewhere ?

Yup the df command outputs are also consistent with what I am seeing :
`/dev/nvme3n1    1.5T  1.4T   85G  95% /pulsar/data/bookkeeper/ledgers`
`/dev/nvme2n1     99G  120K   99G   1% /pulsar/data/bookkeeper/journal`

No .. Nothing specific that I am aware of. Just a namespace level ttl to keep unacknowledged messages for 14 days.

```May 25, 2023 08:14:27.253
2023-05-25T15:14:27,253+0000 [LedgerDirsMonitorThread] WARN  org.apache.bookkeeper.bookie.LedgerDirsMonitor - LedgerDirsMonitor check process: All ledger directories are non writable
 
May 25, 2023 08:14:27.253
2023-05-25T15:14:27,253+0000 [LedgerDirsMonitorThread] ERROR org.apache.bookkeeper.util.DiskChecker - Space left on device /pulsar/data/bookkeeper/ledgers/current : 78524096512, Used space fraction: 0.95162326 &gt; threshold 0.95.
 
May 25, 2023 08:14:27.291
2023-05-25T15:14:27,291+0000 [SyncThread-7-1] INFO  org.apache.bookkeeper.bookie.LedgerDirsManager - No writable ledger dirs below diskUsageThreshold. But Dirs that can accommodate 1073741824 are: [/pulsar/data/bookkeeper/ledgers/current]
 
May 25, 2023 08:14:27.292
[1275304.251s][info ][safepoint    ] Application time: 0.7264509 seconds
 
May 25, 2023 08:14:27.292
[1275304.251s][info ][safepoint    ] Entering safepoint region: BulkRevokeBias
 
May 25, 2023 08:14:27.292
[1275304.252s][info ][safepoint    ] Leaving safepoint region
 
May 25, 2023 08:14:27.292
[1275304.252s][info ][safepoint    ] Total time for which application threads were stopped: 0.0004699 seconds, Stopping threads took: 0.0000186 seconds
 
May 25, 2023 08:14:27.408
2023-05-25T15:14:27,408+0000 [GarbageCollectorThread-11-1] INFO  org.apache.bookkeeper.bookie.GarbageCollectorThread - Disk almost full, suspend major compaction to slow down filling disk.
 
May 25, 2023 08:14:27.408
2023-05-25T15:14:27,408+0000 [GarbageCollectorThread-11-1] INFO  org.apache.bookkeeper.bookie.GarbageCollectorThread - Disk full, suspend minor compaction to slow down filling disk.```
:point_up:  is what I found ..Anything particular that I could look for around GarbageCollection

for the Proxy they are named Proxy Extensions

there is a Kafka Proxy extension in Statlight for Kafka, that is a fork of KOP