pulsar

Hi <@U053BJ8UUBZ>, to better understand, do you want to create a new admin user using the REST API or do you want to make the change directly in the database of pulsar-manager?

Currently i’m in a situation that I can’t login to pulsar-manager with: pulsar:pulsar creds

To start with, I just would like to be able to login..
I already tried running this:
`CSRF_TOKEN=$(curl <http://localhost:7750/pulsar-manager/csrf-token>)`
`curl \`
   `-H 'X-XSRF-TOKEN: $CSRF_TOKEN' \`
   `-H 'Cookie: XSRF-TOKEN=$CSRF_TOKEN;' \`
   `-H "Content-Type: application/json" \`
   `-X PUT <http://localhost:7750/pulsar-manager/users/superuser> \`
   `-d '{"name": "admin", "password": "apachepulsar", "description": "test", "email": "<mailto:username@test.org|username@test.org>"}'`

But since there’s no backend or it isn’t configured, this doesn’t work for me

Yeah unfortunately, pulsar-manager has a lot of issues and is not really stable.
You should be able to run the curl command you posted above against `localhost:9527`

It’s possible that the version of pulsar-manager in the helm chart is not the latest one

Once the pod has started you might have to wait for 1-2mins until the backend is fully initialized in order to retrieve the CSRF token. You can check if it works by just running `curl <http://localhost:7750/pulsar-manager/csrf-token>` a few times

you mean to run it from within the pod (or port-forward)

`curl: (7) Failed to connect to localhost port 7750: Connection refused`

and when trying to run it with port 9527 i’m getting 404

Assuming you have k8s access, can you take a look at the containers in the pod? There should be 2, one for the UI and one for the backend. Maybe the backend one has an issue on startup for some reason

The user experience with pulsar-manager is very unfortunate, which is why we are working on an enterprise grade replacement at <http://streamvisor.com|streamvisor.com>

and in the chart there’s no mentioning of another pod which acts as a backend

No, it’s a single pod, but if I remember correctly there are 2 containers running inside. Or it could be that it was 1 container running with 2 processes, not sure anymore

I asked for an early access <http://streamvisor.com|streamvisor.com> :slightly_smiling_face:

root         1     0  0 07:12 ?        00:00:00 /bin/bash /pulsar-manager/entrypoint.sh
pulsar      64     1  0 07:12 ?        00:00:00 /usr/lib/postgresql/13/bin/postgres -D /data/postgresql
pulsar      66    64  0 07:12 ?        00:00:00 postgres: checkpointer
pulsar      67    64  0 07:12 ?        00:00:00 postgres: background writer
pulsar      68    64  0 07:12 ?        00:00:00 postgres: walwriter
pulsar      69    64  0 07:12 ?        00:00:00 postgres: autovacuum launcher
pulsar      70    64  0 07:12 ?        00:00:00 postgres: stats collector
pulsar      71    64  0 07:12 ?        00:00:00 postgres: logical replication launcher
root        85     1  0 07:12 ?        00:00:00 nginx: master process nginx
www-data    86    85  0 07:12 ?        00:00:00 nginx: worker process
www-data    88    85  0 07:12 ?        00:00:00 nginx: worker process
www-data    89    85  0 07:12 ?        00:00:00 nginx: worker process
www-data    90    85  0 07:12 ?        00:00:00 nginx: worker process
www-data    91    85  0 07:12 ?        00:00:00 nginx: worker process
www-data    93    85  0 07:12 ?        00:00:00 nginx: worker process
www-data    94    85  0 07:12 ?        00:00:00 nginx: worker process
www-data    95    85  0 07:12 ?        00:00:00 nginx: worker process
root        96     1  0 07:12 ?        00:00:01 /usr/bin/python3 /usr/bin/supervisord -c /etc/supervisord.conf -n
root       439     0  0 07:13 pts/0    00:00:00 bash
root      1357     0  0 08:39 pts/1    00:00:00 bash
root      1455   439  0 08:50 pts/0    00:00:00 ps -ef

also netstat doesn’t show that it listens on 7750

Active Internet connections (only servers)
Proto Recv-Q Send-Q Local Address           Foreign Address         State       PID/Program name
tcp        0      0 0.0.0.0:80              0.0.0.0:*               LISTEN      85/nginx: master pr
tcp        0      0 0.0.0.0:9527            0.0.0.0:*               LISTEN      85/nginx: master pr
tcp        0      0 127.0.0.1:5432          0.0.0.0:*               LISTEN      -
tcp6       0      0 :::80                   :::*                    LISTEN      85/nginx: master pr

Hm, looks like the backend did not start or failed. In the container (maybe under `/pulsar-manager/`) there should be a `pulsar-manager.log` showing the backend logs

Either there or in the supervisord logs there could be a hint to what’s wrong

`2023-06-27 07:12:06,908 CRIT Supervisor is running as root.  Privileges were not dropped because no user is specified in the config file.  If you intend to run as root, you can set user=root in the config file to avoid this message.`
`2023-06-27 07:12:06,910 INFO supervisord started with pid 96`
`2023-06-27 07:12:07,913 INFO spawned: 'pulsar-manager-backend' with pid 97`
`2023-06-27 07:12:09,015 INFO success: pulsar-manager-backend entered RUNNING state, process has stayed up for &gt; than 1 seconds (startsecs)`
`2023-06-27 07:12:09,015 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:10,019 INFO spawned: 'pulsar-manager-backend' with pid 117`
`2023-06-27 07:12:10,150 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:11,153 INFO spawned: 'pulsar-manager-backend' with pid 137`
`2023-06-27 07:12:11,311 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:13,316 INFO spawned: 'pulsar-manager-backend' with pid 157`
`2023-06-27 07:12:13,464 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:16,471 INFO spawned: 'pulsar-manager-backend' with pid 177`
`2023-06-27 07:12:17,587 INFO success: pulsar-manager-backend entered RUNNING state, process has stayed up for &gt; than 1 seconds (startsecs)`
`2023-06-27 07:12:17,587 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:18,589 INFO spawned: 'pulsar-manager-backend' with pid 197`
`2023-06-27 07:12:19,775 INFO success: pulsar-manager-backend entered RUNNING state, process has stayed up for &gt; than 1 seconds (startsecs)`
`2023-06-27 07:12:19,775 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:20,778 INFO spawned: 'pulsar-manager-backend' with pid 217`
`2023-06-27 07:12:21,899 INFO success: pulsar-manager-backend entered RUNNING state, process has stayed up for &gt; than 1 seconds (startsecs)`
`2023-06-27 07:12:21,899 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:22,908 INFO spawned: 'pulsar-manager-backend' with pid 237`
`2023-06-27 07:12:23,045 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:24,048 INFO spawned: 'pulsar-manager-backend' with pid 257`
`2023-06-27 07:12:25,163 INFO success: pulsar-manager-backend entered RUNNING state, process has stayed up for &gt; than 1 seconds (startsecs)`
`2023-06-27 07:12:25,163 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:26,167 INFO spawned: 'pulsar-manager-backend' with pid 277`
`2023-06-27 07:12:26,300 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:27,302 INFO spawned: 'pulsar-manager-backend' with pid 297`
`2023-06-27 07:12:27,417 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:29,421 INFO spawned: 'pulsar-manager-backend' with pid 317`
`2023-06-27 07:12:29,524 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:32,529 INFO spawned: 'pulsar-manager-backend' with pid 337`
`2023-06-27 07:12:33,647 INFO success: pulsar-manager-backend entered RUNNING state, process has stayed up for &gt; than 1 seconds (startsecs)`
`2023-06-27 07:12:33,648 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:34,654 INFO spawned: 'pulsar-manager-backend' with pid 357`
`2023-06-27 07:12:34,771 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:35,774 INFO spawned: 'pulsar-manager-backend' with pid 377`
`2023-06-27 07:12:35,869 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:37,874 INFO spawned: 'pulsar-manager-backend' with pid 397`
`2023-06-27 07:12:37,977 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:40,982 INFO spawned: 'pulsar-manager-backend' with pid 417`
`2023-06-27 07:12:41,131 INFO exited: pulsar-manager-backend (exit status 1; not expected)`
`2023-06-27 07:12:42,132 INFO gave up: pulsar-manager-backend entered FATAL state, too many start retries too quickly`

Did you find a file with the backend log too?

<https://github.com/apache/pulsar-helm-chart/blob/master/charts/pulsar/templates/pulsar-manager-deployment.yaml>

- name: PULSAR_MANAGER_OPTS
   value: “$(PULSAR_MANAGER_OPTS) -Dlog4j2.formatMsgNoLookups=true”

This is how it looks in line 92-93 and the $(PULSAR_MANAGER_OPTS) causes an issue

BTW: now that I had managed to access the pulsar-manager
I see no data in it, but my pulsar cluster has many namespaces configured there

<https://streamnative.io/blog/pulsar-summit-na-2023-call-for-speakers>

Let's announce it on the pulsar mailing lists too!

Could anyone please help me clear my understanding? Based on results, I would also like to contribute to docs if needed.

does anyone have a reference/examples of installing Pulsar on aws in multi-zone settings using helm for production workload? I see many Helm charts, but want to know if anyone have any experience or recommendation.

Hi,

When deploying Pulsar on EKS in production environment, which machines are recommended for it?

I’m mostly talking about:
ZooKeeper
Bookies
Brokers

Protobuf native backward and forward compatability validation almost doesn’t exists.

That’s one downside protobuf native have.

Pulsar PROTOBUF is actually using Avro to describe  the schema instead of using the normal protobuf descriptor

PROTOBUF_NATIVE aimed to fix that by allowing you to specify a schema using protobuf descriptor

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

Unfortunately there's not more docs than the javadocs in the project. pulsar-client-reactive is used in Spring Pulsar's Reactive support and that contains documentation, <https://docs.spring.io/spring-pulsar/docs/current-SNAPSHOT/reference/html/#reactive-pulsar> .

What kind of documentation would you find useful? I believe that <https://github.com/apache/pulsar-client-reactive/graphs/contributors|existing contributors of pulsar-client-reactive> will be eager to improve the areas that are valuable for the community. /cc <@UDZPKSJSK> <@U03RM6GQ2M9> <@UE71CHRPS>

I was trying to understand, difference between `ReactivePulsarListener` ,`ReactivePulsarMessageListenerContainer`, `ReactiveMessagePipeline` and `ReactivePulsarMessageHandler` not really clear on the spring pulsar docs

The doc is here : <https://docs.spring.io/spring-pulsar/docs/current-SNAPSHOT/reference/html/#reactive-message-listener-container>
Basically `ReactivePulsarListener` is an annotation to create a `ReactivePulsarMessageListenerContainer` that will call the `ReactivePulsarMessageHandler` which in turn will call the code you annotated with `ReactivePulsarListener` .

Internally the `ReactivePulsarMessageListenerContainer`  starts a Reactive Pulsar client's `ReactiveMessagePipeline` and sets the `ReactivePulsarMessageHandler` as handler for the messages.
This is all internal stuff and you probably just need to use `ReactivePulsarListener` from a high level.

I missed this thread while on PTO. Hi all.

<@U03BWT1C65S> it sounds like you are interested in introducing Spring for Apache Pulsar (Reactive support) to your company, I just wanted to point out the<https://docs.spring.io/spring-pulsar/docs/current-SNAPSHOT/reference/html/#_project_status| current project status> . Basically, the “full” GA version (`1.0.0`) is targeted to be available when Spring Boot `3.2.0` releases in November. Until that time we have an “experimental” GA version (`0.2.x`) that we recommend using. The link above explains what is full vs exp GA.

Also, the Pulsar reactive client can be used w/o Spring Pulsar. Spring Pulsar just uses the Pulsar reactive client internally (already mentioned above as well).

Feel free to reach out to me on DM if you have any question around Spring Pulsar.

The short answer is no. MessageIds in Pulsar aren’t single counters, instead they are composed of multiple elements that indicate the batch id, entry id,  ledger id, etc. This makes it impossible to consistently determine the message id of the current message id plus N because any of those elements may have changed if they are in a different batch or ledger, etc., e.g. `ledgerId:entryId:partitionIndex:batchIndex`

The exact number of messages is hard to determine especially when an entry can consist of many messages and the entry data it self is compressed. Though I wonder if we can attach metadata to each entry that describes how many messages are in this entry.

 What we can determine is the size, so we can have an client API that asks for x bytes starting from a certain message id

I notice that most pulsar source and sink tests are dropped in this commit <https://github.com/streamnative/pulsar-spark/commit/c467f0996a0e5e338175ccd56db072facfe08e1e>. Now the test coverage is insufficient, can we add it back?(or add other testing) <@U0276C5H1SA> <@U680ZCXA5> <@U9ZQ34LDU>

yes, we should. But the previous test case may need update

Do you have plan to add the tests back in the near future?

given our current resource is tight (need to prepare various things for Pulsar Summit in Q3), we will probably start to work on that in Q4.

I can create a PR to add back the testing, could you elaborate on what update is necessary? <@U0276C5H1SA>

just make sure it follows some general scala test guideline.

No problem! Can I simply add back all the old tests except the continuous one that is no longer supported?

BTW, did you guys observe the spark connector creates too many connections on the pulsar topic issue from your customer?

Also, can someone point me to the pulsar client interface that we can leverage to implement admission control? I remember it was about polling the size of the next few batches of message and stop when the accumulated size is larger that read limit, but I don't know which interface I can use to achieve this. <@U0276C5H1SA> <@U680ZCXA5> <@U9ZQ34LDU>

We have not heard from customer about any performance issue of pulsar spark connector. Is there any improvement you have in mind?

there might be a producer reuse optimization, but we are still investigating an issue reported by one of our users.

I saw that pulsar source RDD use preferred location to maximize reader reuse. I don't find similar optimization in the sink side.

This is the implementation of cached kafka producer pool in spark, we can do something similar for pulsar sink <https://github.com/apache/spark/blob/master/connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/producer/InternalKafkaProducerPool.scala|https://github.com/apache/spark/blob/master/connector/kafka-0-10-sql/src/main/scal[…]che/spark/sql/kafka010/producer/InternalKafkaProducerPool.scala>

kafka producer is thread safe, so it can be put in a cache and reused by multiple worker threads safely.

Can we prioritize this fix? We have a customer that is being impacted by this. <@U0276C5H1SA> <@U05F423HY3B>

it seems that pulsar producer is reused in the executor, so the only problem is locality? <https://github.com/streamnative/pulsar-spark/blob/master/src/main/scala/org/apache/spark/sql/pulsar/PulsarWriteTask.scala#L137> <@U0276C5H1SA> <@U9NHC1ETG>

Thanks <@U05F423HY3B>, is there an eviction thread for this Map similar to the one for Kafka?  We are seeing an ever increasing number of connections created to the Pulsar broker for the topic and are trying to determine the root cause

I don't see any eviction. Why would connection keeps increasing if the output topics are fixed?

Actually, if singleProducer is true, then a new producer get created every call, correct?

```if (null != singleProducer) {
      return singleProducer.asInstanceOf[Producer[T]]
    }```

```protected lazy val singleProducer =
    if (topic.isDefined) {
      PulsarSinks.createProducer(clientConf, producerConf, topic.get, pulsarSchema)
    } else null```

I am not very scala fluent, so I could be wrong. But in the getProducer method the first call is to check if `singleProducer` is not null. If the topic IS defined, then a producer is created and returned.

This created/returned object is never added to the Map for reuse, or am I just missing something?

I think you are correct. A new PulsarWriteTask is created for each write and neither singleProducer or topic2Producer get reused.

We would need a singleton cache for this.

since write task may be scheduled to different executors, will a cache work in this case?

We don't enforce write tasks to be on the same executor explicitly, but they are usually scheduled on the same executor because we set locality for read task and stateful partition, and write tasks following them should also have certain level of locality.

<https://github.com/streamnative/pulsar-spark/issues/136>

Can we have a simple fix before implementing the producer pool? Just call close() for the producer when the write task finished. <@U0276C5H1SA> <@U9NHC1ETG>

I guess that is better than the current scenario. But there would be a lot of network connections opened and then closed right away.

Why not just add the created producer to the map after it is created? Subsequent calls would then just reuse it rather than having to create a new one, use it, and then close a producer over and over again

The map is owned by the writer task, which is a short live object that doesn't get reused. We would need to implement a singleton cache with cache eviction, which would require some design and discussion.

<https://github.com/streamnative/pulsar-spark/blob/master/src/main/scala/org/apache/spark/sql/pulsar/PulsarWriteTask.scala#L137>

Probably can reuse much of the code for the consumer and producer cache pool from kafka source/sink

<https://github.com/streamnative/pulsar-spark/pull/137> &lt;-- I draft a PR to fix the producer spam issue.

This fix looks good to me(at least no longer leak the connection), we can design and implement producer cache later.

We would also like to add pyspark support to pulsar spark connector, are you ok with that? <@U0276C5H1SA> <@U680ZCXA5>

Strange. All acked consumed messages are supposed to be deleted by default

fyi, in Pulsar, TTL and retention each solve different problems. Maybe u would you like to check the documentation for reference.  <https://pulsar.apache.org/docs/cookbooks-retention-expiry/|https://pulsar.apache.org/docs/cookbooks-retention-expiry/>

In the above doc, there's a single bullet point referring to BookKeeper garbage collection. That's what eventually releases disk space when ledgers get compacted and possibly deleted too. A Pulsar topic is like a linked list of "segments" which are Bookkeeper ledgers. The ledger would have to be closed before BookKeeper Garbage Collection can happen. The ledger is closed when the topic segment roll over happens.

This is the essential part of the doc:
&gt; • Segment rollover period: basically, the segment rollover period is how often a new segment is created. Once a new segment is created, the old segment will be deleted. By default, this happens either when you have written 50,000 entries (messages) or have waited 240 minutes. You can tune this in your broker.
&gt; • Entry log rollover period: multiple ledgers in BookKeeper are interleaved into an <https://bookkeeper.apache.org/docs/4.11.1/getting-started/concepts/#entry-logs|entry log>. In order for a ledger that has been deleted, the entry log must all be rolled over. The entry log rollover period is configurable but is purely based on the entry log size. For details, see <https://bookkeeper.apache.org/docs/4.11.1/reference/config/#entry-log-settings|here>. Once the entry log is rolled over, the entry log can be garbage collected.
&gt; • Garbage collection interval: because entry logs have interleaved ledgers, to free up space, the entry logs need to be rewritten. The garbage collection interval is how often BookKeeper performs garbage collection. which is related to minor compaction and major compaction of entry logs. For details, see <https://bookkeeper.apache.org/docs/4.11.1/reference/config/#entry-log-compaction-settings|here>.
Bookkeeper compaction is described here: <https://bookkeeper.apache.org/docs/getting-started/concepts#data-compaction>

There are multiple tunable parameters for rollover and multiple tunable parameters for BookKeeper compaction / garbage collection.

you could test these settings as per your requirement

```managedLedgerMaxEntriesPerLedger=10000
managedLedgerMinLedgerRolloverTimeMinutes=5
managedLedgerMaxLedgerRolloverTimeMinutes=10
managedLedgerMaxSizePerLedgerMbytes=512```
Or Throttle the compaction bytes rate to 10MB/s on bookkeeper.conf to get faster cleanup

```isThrottleByBytes=true
compactionRateByBytes=10485760```


Also found the following line to be insightful if you are trying to figure out whats occupying the disks
```If you do not have any retention period and you never have much of a backlog, the upper limit for retained messages, which are acknowledged, equals the Pulsar segment rollover period + entry log rollover period + (garbage collection interval * garbage collection ratios).```

Hola <@U05ERLZ44TG>,
In the logs of the connector, do you see producer-related logs (like producer-stats) that indicate the connector writes the CDC records somewhere?

Another thing you could give a try is changing `snapshot.mode` to `always`  to see if topics are created

Hola Alexander, thank you very much for taking the time to read this question.

On the other hand I am looking for logs about "producer-stats" in the logs of the main log directory I did not find any information, however I found that the connector logs were located in "logs/functions/public/default/debezium-oracle-source/debezium-oracle-source-0.log" .

The file shows the creation of the two topics that are shown when you request to list them using the command ("pulsar-admin topics list public/default") but it does not generate the ones for each of the tables shown in the schema(TABLE1,TESTS,TESTS2).

`Configure offset backing store on pulsar topic public/default/debezium-oracle-source-debezium-offset-topic".`

`Successfully created producer to produce updates to topic public/default/debezium/debezium-oracle-source-debezium-offset-topic`

`Successfully created reader to replay updates from topic public/default/debezium/debezium-oracle-source-debezium-offset-topic"`

Also when I set the snapshot.mode attribute to always, the errors appeared in log indicating that this alternative is not supported by the connector:

`ERROR [public/default/debezium-oracle-source-0] [instance: 0] BaseSourceTask - The 'snapshot.mode' value 'always' is invalid: Value must be one of initial_only, initial, schema_only, schema_only_recovery`

Finally now only this topic is shown:

<persistent://public/default/debezium-oracle-source-debezium-offset-topic>

Thank you very much for your help.

Oh you’re right `snapshot.mode=always` is for newer Debezium versions, sorry for the confusion. Can you try setting it to `schema_only`?

Hi Alexander, again thank you very much.

Finally I made the parameter change to "schema_only" and now again the log (`"logs/functions/public/default/debezium-oracle-source/debezium-oracle-source-0.log"` ) shows no errors when running the connector.

However using the command `"pulsar-admin topics list public/default"` shows again the two previous topics (in the same way as when using "initial"):
<persistent://public/default/debezium-oracle-source-history-topic>
<persistent://public/default/debezium-oracle-source-debezium-offset-topic>

Do you think that before using this connector I should do some configuration on pulsar, actually I just start the server and run the script for the connector, previously I have configured the Oracle DB engine using the recommendations for (<https://debezium.io/blog/2022/09/30/debezium-oracle-series-part-1>).

Thank you very much for your cooperation.

How to avoid error `ProducerBlockedQuotaExceededError` for pulsar producer

Anyone with a good understanding of consumer rebalancing in failover sub? I posted a big question above and in dev mailing list, but it might have scared people :sweat_smile:. No replies in 3 days.

Connectors are completely self-contained, platform-independent pieces of code that can be deployed on any Pulsar platform. If you want to deploy your own or a third-party connector, then the steps are outlined here. I wouldn’t recommend dropping the nar file into the `/pulsar/connectors` folder. The preferred method is to use the admin REST API to deploy/update/delete your <https://pulsar.apache.org/source-rest-api/?version=3.0.0|sources> and <https://pulsar.apache.org/sink-rest-api/?version=3.0.0|sinks>. Using the REST API allows you to make changes, update the code etc w/o having to stop/restart the brokers.

I guess putting them in the image would effectively make them immutable.

Just like the built-ins. Much less flexible. Thanks for the advice.

The source code logging the stacktrace seems to serve no specific purpose. I think we can simplify logging. Can you wait for next release tho?

It makes it very difficult to use the software from my work machine, but I guess I can squeak by using docker for now. I am also willing to try and hack it in which seems easier than trying to compile conscrypt for my arch (which I got about 90% of the way through and gave up on)

But it appears that I’m kind of stuck. Other projects seem to be switching to bouncy castle or other methods since conscrypt seems to have abandoned this issue

If there’s a way I can build the shell such that this doesn’t happen, I’m all for it. I can build from source for now, but am not familiar enough with Java build tools to know what steps to take myself

I tried the forked repo mentioned on the conscrypt repo <https://github.com/google/conscrypt/issues/1034|here>, but, even after resolving all of the painful gradle issues, it runs into .cpp calling convention problems in boringssl, and that's where I draw the line. :slightly_smiling_face:

<https://github.com/sigstore/sigstore-java/pull/82|Sigstore> and <https://github.com/wiremock/wiremock/issues/1671|wiremock> just ended up removing Conscrypt altogether.

<@U01AQHHNDS9> Okay, discussions in sigstore and wiremock helps to identify what the problem is. I will try to organize things so that others can be informed for consideration on removal.

Meanwhile, <https://github.com/apache/pulsar/pull/20705|this PR> might suffice your need, <@U01AQHHNDS9>?

Yes, that would definitely help!  What's the `mvn` invocation for building just the user tools again?

I think I've done it a grand total of one time.

It would also be nice to just avoid the message entirely based on an environment variable for now.

But I think I can probably manage to hack that in myself

&gt; But I think I can probably manage to hack that in myself
For sure, but just in case :+1::skin-tone-2: you may refer to <https://github.com/apache/pulsar/tree/master#readme>

Yeah, I was just curious what module this would be. I would think `pulsar-client-tools`, but:
``` mvn -pl pulsar-client-tools,-main -DskipTests
[INFO] Scanning for projects...
[ERROR] [ERROR] Could not find the selected project in the reactor: main @
[INFO] 0 goals, 0 executed
[ERROR] Could not find the selected project in the reactor: main -&gt; [Help 1]
[ERROR]
[ERROR] To see the full stack trace of the errors, re-run Maven with the -e switch.
[ERROR] Re-run Maven using the -X switch to enable full debug logging.
[ERROR]
[ERROR] For more information about the errors and possible solutions, please read the following articles:
[ERROR] [Help 1] <http://cwiki.apache.org/confluence/display/MAVEN/MavenExecutionException>```

like, is there a  way for me to `mvn package` just the shell?

image.png

Single Threaded Producer is likely the issue. Try having multiple threads produce to the same partition.

I need to ensure the order of data within a single partition, so I can only use a single producer.

Can you send the messages async? - <https://github.com/apache/pulsar-client-python/blob/766db9e0420954798d8d24ef4dc55fa250270921/pulsar/__init__.py#L1100>

The send retuns when it is Acknowledge. So it is slower

I have tried using `send_async` to send data, and the QPS can reach 30,000, but I’m worried that the data will be out of order. I don’t understand why `send` is so slow.

Send waits for the ACK from the broker.

It will be ordered if there is only one thread.

Check it <https://pulsar.apache.org/api/python/3.2.x/pulsar.Producer.html#send_async|here>

I can achieve fast performance when using `ack='all'` in Kafka as well.

Could it be because I did not allocate enough resources to Pulsar? I deployed Pulsar in Kubernetes

I will test whether increasing the resources allocated to the broker will improve the write performance.

The problem is that as your python is blocked until one message is ACK by the broker. So the slowness is the network connectivity between your cient and the broker.

In your use case, sending async with just one thread will avoid the blocking of the thread and it will be keep the order.

there is no ack='all'  on the send in Pulsar. You have to use `send_async`

If three messages are sent using `send_async`, and middle of them fails to be processed by Pulsar, will the data be lost or unordered?
I

SO. Is it normal for the QPS of the `send()` method to be around 500 in the case of a single partition and a single producer?

&gt; Is it normal for the QPS of the send() method to be around 500 
yes. :slightly_smiling_face:. each send is incurring the network round trip time cost. it has nothing to do with pulsar. kafka's send_sync() will perform the same.

How are you achieving ordering in kafka with high throughput? As far as I remember, Kafka client can also reorder messages due to its retry policy in the face of failures. Check <https://kafka.apache.org/documentation/#producerconfigs_retries>. This should limit throughput.

You’re missing the “reason” part of the exception

Check the logs and I’ll help you with that - I know that code quite well

after restarting the broker sts I don’t see this error anymore, thanks <@U01551PS4MA>

But now i’m seeing some other interesting issue, from one of our apps that is publishing to pulsar I see these errors:

`"Pulsar.Client.Api.LookupException: {\"errorMsg\":\"Failed due to too many pending lookup requests\"`

In the broker, there is a in-flight protection mechanism, to avoid having too many concurrent requests for lookups

```    @FieldContext(
        dynamic = true,
        category = CATEGORY_SERVER,
        doc = "Max number of concurrent lookup request broker allows to throttle heavy incoming lookup traffic")
    private int maxConcurrentLookupRequest = 50000;```

Are you using reg-exp based subscriptions perhars?

We didn’t touch the in-flight protection

<@U053BJ8UUBZ> You seem to be hitting some bug that might be fixed in more recent versions of Pulsar. There are <https://github.com/apache/pulsar/compare/v2.11.0...v2.11.1|214 commits between 2.11.0 and 2.11.1> .
One possible cause is a deadlock in the broker. When you encounter an issue, a thread dump from each broker could help reveal deadlocks.
If you run Pulsar with Kubernetes, this is a sample script that collects diagnostics information from a Pulsar pod: <https://github.com/lhotari/pulsar-contributor-toolbox/blob/master/scripts/collect_jvm_diagnostics_from_pod.sh>
It also retrieves a heap dump (which contains sensitive information so don't share that publicly).

<@U053BJ8UUBZ> do you have a chance to upgrade to 2.11.1 and see if the problem reproduces?

hmmm… I see.. then we will upgrade our cluster and see if it solves our issue

6750 is the default port for a Function Worker or 8080?

I have no idea what the port 6750 is for. If you go to this link <https://pulsar.apache.org/docs/next/functions-quickstart> It says to check the pulsar Function cluster by running: bin/pulsar-admin functions-worker get-cluster command. And the output should be: [{"workerId":"c-standalone-fw-localhost-6750","workerHostname":"localhost","port":6750}]

<@U053NKRDBL0> could you check the `conf/standalone.conf` file? the service port is defined here: `webServicePort`

Hello Ali, thanks for reaching to the pulsar community. What version of pulsar do you use and what is the output you see when you run the -a -h command..

I'm testing this on 3.0, however there was a problem on 2.11

while an upgrade was in progress from 2.10, bookies or PVC was deleted

```$ bin/bookkeeper shell listbookies -a -
.....
2023-07-04T16:44:31,877+0000 [main] INFO  org.apache.zookeeper.ZooKeeper - Client environment:java.library.path=/usr/java/packages/lib:/usr/lib64:/lib64:/lib:/usr/lib
2023-07-04T16:44:31,878+0000 [main] INFO  org.apache.zookeeper.ZooKeeper - Client environment:java.io.tmpdir=/tmp
2023-07-04T16:44:31,878+0000 [main] INFO  org.apache.zookeeper.ZooKeeper - Client environment:java.compiler=&lt;NA&gt;
2023-07-04T16:44:31,878+0000 [main] INFO  org.apache.zookeeper.ZooKeeper - Client environment:os.name=Linux
2023-07-04T16:44:31,878+0000 [main] INFO  org.apache.zookeeper.ZooKeeper - Client environment:os.arch=amd64
2023-07-04T16:44:31,878+0000 [main] INFO  org.apache.zookeeper.ZooKeeper - Client environment:os.version=5.4.238-148.347.amzn2.x86_64
2023-07-04T16:44:31,878+0000 [main] INFO  org.apache.zookeeper.ZooKeeper - Client environment:user.name=?
2023-07-04T16:44:31,878+0000 [main] INFO  org.apache.zookeeper.ZooKeeper - Client environment:user.home=?
2023-07-04T16:44:31,878+0000 [main] INFO  org.apache.zookeeper.ZooKeeper - Client environment:user.dir=/pulsar
2023-07-04T16:44:31,878+0000 [main] INFO  org.apache.zookeeper.ZooKeeper - Client environment:os.memory.free=111MB
2023-07-04T16:44:31,878+0000 [main] INFO  org.apache.zookeeper.ZooKeeper - Client environment:os.memory.max=256MB
2023-07-04T16:44:31,878+0000 [main] INFO  org.apache.zookeeper.ZooKeeper - Client environment:os.memory.total=130MB
2023-07-04T16:44:31,882+0000 [main] INFO  org.apache.zookeeper.ZooKeeper - Initiating client connection, connectString=pulsar-apache-zookeeper:2181 sessionTimeout=30000 watcher=org.apache.bookkeeper.zookeeper.ZooKeeperWatcherBase@5be067de
2023-07-04T16:44:31,886+0000 [main] INFO  org.apache.zookeeper.common.X509Util - Setting -D jdk.tls.rejectClientInitiatedRenegotiation=true to disable client-initiated TLS renegotiation
2023-07-04T16:44:31,906+0000 [main] INFO  org.apache.zookeeper.ClientCnxnSocket - jute.maxbuffer value is 1048575 Bytes
2023-07-04T16:44:31,914+0000 [main] INFO  org.apache.zookeeper.ClientCnxn - zookeeper.request.timeout value is 0. feature enabled=false
2023-07-04T16:44:31,924+0000 [main-SendThread(pulsar-apache-zookeeper:2181)] INFO  org.apache.zookeeper.ClientCnxn - Opening socket connection to server pulsar-apache-zookeeper/10.104.86.163:2181.
2023-07-04T16:44:31,925+0000 [main-SendThread(pulsar-apache-zookeeper:2181)] INFO  org.apache.zookeeper.ClientCnxn - SASL config status: Will not attempt to authenticate using SASL (unknown error)
2023-07-04T16:44:31,932+0000 [main-SendThread(pulsar-apache-zookeeper:2181)] INFO  org.apache.zookeeper.ClientCnxn - Socket connection established, initiating session, client: /10.104.44.32:46836, server: pulsar-apache-zookeeper/10.104.86.163:2181
2023-07-04T16:44:31,940+0000 [main-SendThread(pulsar-apache-zookeeper:2181)] INFO  org.apache.zookeeper.ClientCnxn - Session establishment complete on server pulsar-apache-zookeeper/10.104.86.163:2181, session id = 0x100d33a1a4d0063, negotiated timeout = 30000
2023-07-04T16:44:31,943+0000 [main-EventThread] INFO  org.apache.bookkeeper.zookeeper.ZooKeeperWatcherBase - ZooKeeper client is connected now.
2023-07-04T16:44:32,044+0000 [main-EventThread] INFO  org.apache.bookkeeper.discover.ZKRegistrationClient - Update BookieInfoCache (writable bookie) pulsar-apache-bookie-2.pulsar-apache-bookie.pulsar-apache.svc.cluster.local:3181 -&gt; BookieServiceInfo{properties={}, endpoints=[EndpointInfo{id=httpserver, port=8000, host=0.0.0.0, protocol=http, auth=[], extensions=[]}, EndpointInfo{id=bookie, port=3181, host=pulsar-apache-bookie-2.pulsar-apache-bookie.pulsar-apache.svc.cluster.local, protocol=bookie-rpc, auth=[], extensions=[]}]}
2023-07-04T16:44:32,045+0000 [main-EventThread] INFO  org.apache.bookkeeper.discover.ZKRegistrationClient - Update BookieInfoCache (writable bookie) pulsar-apache-bookie-0.pulsar-apache-bookie.pulsar-apache.svc.cluster.local:3181 -&gt; BookieServiceInfo{properties={}, endpoints=[EndpointInfo{id=httpserver, port=8000, host=0.0.0.0, protocol=http, auth=[], extensions=[]}, EndpointInfo{id=bookie, port=3181, host=pulsar-apache-bookie-0.pulsar-apache-bookie.pulsar-apache.svc.cluster.local, protocol=bookie-rpc, auth=[], extensions=[]}]}
2023-07-04T16:44:32,045+0000 [main-EventThread] INFO  org.apache.bookkeeper.discover.ZKRegistrationClient - Update BookieInfoCache (writable bookie) pulsar-apache-bookie-3.pulsar-apache-bookie.pulsar-apache.svc.cluster.local:3181 -&gt; BookieServiceInfo{properties={}, endpoints=[EndpointInfo{id=httpserver, port=8000, host=0.0.0.0, protocol=http, auth=[], extensions=[]}, EndpointInfo{id=bookie, port=3181, host=pulsar-apache-bookie-3.pulsar-apache-bookie.pulsar-apache.svc.cluster.local, protocol=bookie-rpc, auth=[], extensions=[]}]}
2023-07-04T16:44:32,055+0000 [main] INFO  org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand - All Bookies :
2023-07-04T16:44:32,062+0000 [main] INFO  org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand - BookieID:pulsar-apache-bookie-2.pulsar-apache-bookie.pulsar-apache.svc.cluster.local:3181, IP:10.104.44.32, Port:3181, Hostname:pulsar-apache-bookie-2.pulsar-apache-bookie.pulsar-apache.svc.cluster.local
2023-07-04T16:44:32,062+0000 [main] INFO  org.apache.bookkeeper.client.DefaultBookieAddressResolver - Cannot resolve pulsar-apache-bookie-1.pulsar-apache-bookie.pulsar-apache.svc.cluster.local:3181, bookie is unknown org.apache.bookkeeper.client.BKException$BKBookieHandleNotAvailableException: Bookie handle is not available
2023-07-04T16:44:32,063+0000 [main] INFO  org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand - BookieID:pulsar-apache-bookie-1.pulsar-apache-bookie.pulsar-apache.svc.cluster.local:3181, IP:UNKNOWN, Port:0, Hostname:UNKNOWN
2023-07-04T16:44:32,064+0000 [main] INFO  org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand - BookieID:pulsar-apache-bookie-0.pulsar-apache-bookie.pulsar-apache.svc.cluster.local:3181, IP:10.104.95.113, Port:3181, Hostname:pulsar-apache-bookie-0.pulsar-apache-bookie.pulsar-apache.svc.cluster.local
2023-07-04T16:44:32,065+0000 [main] INFO  org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand - BookieID:pulsar-apache-bookie-3.pulsar-apache-bookie.pulsar-apache.svc.cluster.local:3181, IP:10.104.81.102, Port:3181, Hostname:pulsar-apache-bookie-3.pulsar-apache-bookie.pulsar-apache.svc.cluster.local
2023-07-04T16:44:32,170+0000 [main] INFO  org.apache.zookeeper.ZooKeeper - Session: 0x100d33a1a4d0063 closed
2023-07-04T16:44:32,170+0000 [main-EventThread] INFO  org.apache.zookeeper.ClientCnxn - EventThread shut down for session: 0x100d33a1a4d0063
[1.409s][info ][gc,heap,exit] Heap
[1.409s][info ][gc,heap,exit]  garbage-first heap   total 133120K, used 30410K [0x00000000f0000000, 0x0000000100000000)
[1.409s][info ][gc,heap,exit]   region size 1024K, 22 young (22528K), 8 survivors (8192K)
[1.409s][info ][gc,heap,exit]  Metaspace       used 17854K, committed 18112K, reserved 1064960K
[1.409s][info ][gc,heap,exit]   class space    used 2066K, committed 2176K, reserved 1048576K```

If you have auto recovery pod set up, it should be able to rebuild ledgers from the new empty pvc

I realize you said the PVC is deleted. Is the PV deleted? If you have the PV’s persistentVolumeReclaimPolicy set as `Retain` or it’s not deleted, you can bind the PVC to the retained PV. The pv is the volumename in the pvc manifest file. There is no need to all the above steps.

Hmm, I wasnt sure how to set up a websocket reader that allowed you to do time seeking of messages to play back events: <https://pulsar.apache.org/docs/3.0.x/client-libraries-websocket/#reader-endpoint>

It looks like you connect just like a consumer. Did you manage to do any message seeking?

All I did was listen for messages and send an ACK and then display message on the UI

Yes, you need one dedicated subscription for each consumer when you are using the Pulsar Consumer API.

one possible solution is to use the subscription expiration feature to cleanup the subscriptions.
<https://stackoverflow.com/a/69036785>

Another choice would be to use the reader API and store the latest consumed position in some external storage. This would only work in a model where there's no need for per-message state. The position would be handled like an offset.

Thanks a lot, I did not we can specify expiration on defined namespace.

Just bumping this as I am still facing this issue. Any help on the matter would be really appriciated!

Could you check out this issue and see if the symptoms are similar?

<https://github.com/apache/pulsar/issues/20714>

Thats my issue... The problem i described in that has been resolved by the PR related to the issue

Ah, I see. I totally missed that part. Thanks

<@U054Z6P6QLF> Did you get this worked? Unfortunately, I couldn’t spare time to really look into it yesterday.

I've had to resort to specifying the `-partition-\d` in the topic regex which isn't ideal due to the regex length limits

Just linking to my similar question for reference - <https://apache-pulsar.slack.com/archives/C5Z4T36F7/p1682014140836679>

If you use the correct “command” as appears in <https://pulsar.apache.org/docs/3.0.x/getting-started-docker-compose/>, then you can configure any property using env variables with prefix `PULSAR_PREFIX`. Like this:

```    environment:
      - PULSAR_MEM=-Xms512m -Xmx512m -XX:MaxDirectMemorySize=256m
      - PULSAR_PREFIX_metadataStoreUrl=zk:zookeeper:2181```

Gotcha, so it may not work for non-broker config like Presto wants. Ill give it a shot though, thanks :slightly_smiling_face:

I also note the docs say:

```To enable it, set the following configurations in the ${project.root}/trino/conf/catalog/pulsar.properties properties file:```
`pulsar-all:3.0.0` image doesnt have a trino conf, though I managed to find a `/pulsar/conf/presto/catalog/pulsar.properties`

Is the documentation incorrect here, or am I missing some other location?

This message contains interactive elements.

Have you considered upgrading to 2.11.1 ? There are <https://github.com/apache/pulsar/compare/v2.11.0...v2.11.1|over 200 commits between 2.11.0 and 2.11.1> and a lot of bugs were fixed.
It's possible that the problem has been fixed.

we are upgrading from 2.8.3 to 2.11.0. We had done on dev and one more lower environment and did not get issue. I'll check with the team if they agrees with 2.11.1 now.

In most projects I saw `x.y.0` version has most bugs :thinking_face: Maybe latest one from 2.11.x might be most stable and also, maybe if lucky, there might be possible backport (cherrypicking backward) for huge bugs.

We are getting this error also. Sending 100,000 jobs over an hour. Is this known error on 2.11.0?
2023-07-07 12:31:12.676 ERROR [140206523020864] ClientConnection:469 | [&lt;none&gt; -&gt; <pulsar://xx-pulsar-211-broker-2.xx-pulsar-211-broker.pulsar.svc.cluster.local:6650>] Resolve error: asio.netdb:1 : Host not found (authoritative)

Getting this error.

```[ERROR]   ZipException opening "guava-32.1.1-jre.jar": zip END header not found
[ERROR] /Users/kimjoohyuk/dev/JooHyukKim/primary/pulsar/buildtools/src/main/java/org/apache/pulsar/tests/MockitoThreadLocalStateCleaner.java:[26,19] cannot access org.mockito.internal
[ERROR]   ZipException opening "guava-32.1.1-jre.jar": zip END header not found
[ERROR] /Users/kimjoohyuk/dev/JooHyukKim/primary/pulsar/buildtools/src/main/java/org/apache/pulsar/tests/MockitoThreadLocalStateCleaner.java:[26,28] cannot access org.mockito.internal.stubbing
[ERROR]   ZipException opening "guava-32.1.1-jre.jar": zip END header not found
[ERROR] /Users/kimjoohyuk/dev/JooHyukKim/primary/pulsar/buildtools/src/main/java/org/apache/pulsar/tests/MockitoThreadLocalStateCleaner.java:[27,28] cannot access org.mockito.internal.util
[ERROR]   ZipException opening "guava-32.1.1-jre.jar": zip END header not found
[ERROR] -&gt; [Help 1]
[ERROR] 
[ERROR] To see the full stack trace of the errors, re-run Maven with the -e switch.
[ERROR] Re-run Maven using the -X switch to enable full debug logging.```

Digging <https://issues.apache.org/jira/browse/MNG-7425>

try to remove the guava from your local repo
rm -rf ~/.m2/repository/com/google/guava/

<@U01B9PMHVGX> Thanks :+1::skin-tone-2:
It worked by
• rm -rf ~/.m2/repository/com/google/guava/
• download `guava-32.1.1-jre.jar`
• reload IDE


My general understanding is that you should be able to update these policies and have it "just work" such that decreasing retention leads to messages getting deleted while increasing retention leads to messages getting stored longer (with the obvious note that already deleted messages will be gone).

<https://pulsar.apache.org/docs/3.0.x/cookbooks-retention-expiry>

If the docs don't answer your questions, I'm happy to grab some code references

For sure. I'm familiar with the broad concepts, how to set them and what they _mean_, but I was more curious about what happens when they're changed and how long that takes, whether there's a concern with accidentally changing them again while the change is ongoing, whether or not that matters or might lead to orphaned ledgers. Best practices stuff like that.  The admin API makes it super easy to set these policies and get super granular

but it's not clear what could go wrong and how to know what to do if it does

like, for example, if I have a retention set only on time and then I convert it to 50% of its former value, only to realize that 80% would be better, can I just issue the change to 80% immediately, or should I wait, and are there any risks and followup steps I need to take after such an action? Or, more broadly, are there checks I can be doing to ensure that retention and backlog are behaving within bounds and not just filling up disk with orphaned stuff?

When you update the retention policy, the update is applied to each affected topic (via its managed ledger) in an eventually consistent way. The first update is to the metadata store (zookeeper), and then all of the broker's will get notified of the updated state and will update the managed ledger configuration. The managed ledger then takes the retention policy into consideration by checking once every `retentionCheckIntervalInSeconds` (a broker setting that defaults to 120 seconds).

That's great, thanks, that's a big part of what I was curious about!

Just trying to fill out my understanding.