pulsar

For example, could you start fresh by simply disabling retention across an entire namespace and then add it back later, and should you wait before doing so?

And how would this interact with offload policy, provided you had one set? I don't, but I am curious how it will work when I do.

For some more context, I have a namespace-wide retention policy set based on time, that I think might be naive. I set it to "four days" thinking that would be good for disaster recovery after a long weekend potentially, but now I am set to add a couple potentially chatty new topics, and I am considering my options. I know I can set topic-level policies, of course, but reading about orphaned ledgers has me wondering if that might have something to do with something I mentioned here month or so ago where there were messages in a topic that were WAY older than they needed to be.

• How do you detect such a thing?
• If I see it, how can I remediate?
• Is there a nuclear option (like disabling and then re-enabling) all retention and backlog policies, to get things consistent again?

<@UQ8NL176Y> - can you provide any additional insight here?

Did you ever solve the orphaned ledgers issue?

I wouldn't expect changing these settings to lead to orphaned ledgers.

I never solved the original issue and, as far as I know, it's still there, but, given my current knowledge, I'd have to do some sort of detailed audit of every topic to know for sure.

Unless there is a better way to detect these kind of edge cases. I figure there must be a way to detect data that will never expire somehow, but I am unaware of it.

And I'm still not sure what the best course to remediate it would be, provided I found it.

&gt;  I wouldn't expect changing these settings to lead to orphaned ledgers.
Yeah, and as far as I know that's a red herring, but since it seemed related and is also failure condition that is currently opaque to me, it's got me really curious about how to detect it and what symptoms it might cause.

Reading a bit more about it, it seems as if these ledgers would be untracked, which is to say invisible to pulsar, so would it be safe to say that it would really only be an issue of storage at that point?

When the ledgers are deleted, they should eventually get removed from bookkeeper. All of the relevant metadata is in zk if you need to verify some odd state... as always, the key is knowing where to look.

Here's some context you might find useful! I don't know if it will _fully_ answer the "why" of it, but it does indicate that getting precise backlog counts is an expensive operation.

<https://github.com/apache/pulsar/issues/7623>
<https://github.com/apache/pulsar/pull/16545>

The upshot is that there is now a `topics analyze-backlog` admin command provided by the following API endpoints:
`admin/v2/non-persistent/{tenant}/{namespace}/{topic}/subscription/{subName}/analyzeBacklog`
`admin/v2/persistent/{tenant}/{namespace}/{topic}/subscription/{subName}/analyzeBacklog`
Meaning you can at least _get_ those counts,  even though it's more expensive and requires safeguards to avoid DOS.

I don't have any deeper insight into why backlog isn't more precisely precomputed, however I imagine it has to do with the complexity of tracking subscription position as well as the "real" number of messages and keeping those deltas up to date and meaningful.

I'm fine with an estimate but I've got some topics that moved from unbatched to batched and the metrics looked very confusing until everybody recalibrated expectations

These are all client side latency metrics. Over at brokers, we do not see produce latency spikes. Even the max latencies are within 25ms

Do you have partitions on the topic? How many?

I'm checking if it was possible with `pulsar-perf` to reproduce the issue passing the `--message-key-generation-mode` , not sure if it will behave properly.

Have you reviewed if the issue happens too with version `2.10` ?

Do you have a small program to try to reproduce it? I will try to do it on my side and reproduce it.

Update on this, this was due to rate limiting set per partition where the producer was being stalled to next second due to permit exhausted

<@UQ8NL176Y> is direct memory king recently :slightly_smiling_face:

It looks like this PR fix it <https://github.com/apache/bookkeeper/pull/3746>

Could you use BookKeeper 4.14.7 to test whether this issue still exists?  BookKeeper 4.14.6 and  4.14.7 fixed a lot of memory leak issues
<https://github.com/apache/bookkeeper/releases/tag/release-4.14.7>
<https://github.com/apache/bookkeeper/releases/tag/release-4.14.6>

<@U01551PS4MA> <@UQ8NL176Y> Thanks a lot for giving this info . I will check on this and get back.

<@U05FPLUAHPX> Is there a specific reason to remain on Pulsar 2.8.2 ? We have fixed a lot of severe security issues and memory leaks after that. Also in Pulsar. Security advisories: <https://pulsar.apache.org/security/>

<@ULBJ01G6N> Thanks for your response .
Earlier our product were using 2.8.2 pulsar , Few month back we moved to pulsar 2.10.3 which has bookie 4.14.5. May be we have to cherry pick pulsar PR's for both 2.10.3 and 2.8.2.
Could I request you to suggest which PR is required to cherry pick for pulsar.

&gt; Could I request you to suggest which PR is required to cherry pick for pulsar.
Just curious, why do you use this type of maintenance process? There are a lot of bugs that have been fixed and if you run on an old version of Pulsar, the probability that you hit an already fixed issue is high.
Without understanding the context, I'd recommend keeping your Pulsar version as close as possible to the maintained versions. It is true that there is risk for regressions when upgrading, but if you would be doing that early, the bugs would come to the surface and get fixed in the opensource project earlier.

<@UQ8NL176Y> We upgraded bookie version 4.14.3(pulsar 2.8.2) to 4.14.7 (pulsar 2.10.3) and used used pulsar 2.10.4 which has  4.14.7 bookie , looks like same issue still persist :disappointed: .Basically I am upgrading from pulsar 2.8.2 to 2.10.4 pulsar with existing journal and ledgers.
Is it a preventive solution or we can recover from DirectMemory error?
```2023-07-07T09:51:06,998 [main] INFO  org.apache.bookkeeper.bookie.Bookie - Replaying journal 1644749554714 from position 708935680
2023-07-07T09:51:07,002 [main] INFO  org.apache.bookkeeper.bookie.JournalChannel - Opening journal data/bookkeeper/journal/current/17ef2b6841a.txn
2023-07-07T09:51:09,162 [main] ERROR org.apache.bookkeeper.common.component.AbstractLifecycleComponent - Failed to start Component: bookie-server
java.lang.OutOfMemoryError: Cannot reserve 1601533033 bytes of direct buffer memory (allocated: 554639373, limit: 1073741824)
        at java.nio.Bits.reserveMemory(Bits.java:178) ~[?:?]
        at java.nio.DirectByteBuffer.&lt;init&gt;(DirectByteBuffer.java:121) ~[?:?]
        at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:332) ~[?:?]
        at sun.nio.ch.Util.getTemporaryDirectBuffer(Util.java:243) ~[?:?]
        at sun.nio.ch.IOUtil.read(IOUtil.java:293) ~[?:?]
        at sun.nio.ch.IOUtil.read(IOUtil.java:273) ~[?:?]
        at sun.nio.ch.FileChannelImpl.read(FileChannelImpl.java:232) ~[?:?]
        at org.apache.bookkeeper.bookie.JournalChannel.read(JournalChannel.java:261) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.bookie.Journal.fullRead(Journal.java:1199) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.bookie.Journal.scanJournal(Journal.java:838) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.bookie.Bookie.replay(Bookie.java:997) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.bookie.Bookie.readJournal(Bookie.java:962) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.bookie.Bookie.start(Bookie.java:1017) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.proto.BookieServer.start(BookieServer.java:156) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.server.service.BookieService.doStart(BookieService.java:68) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.common.component.AbstractLifecycleComponent.start(AbstractLifecycleComponent.java:83) ~[org.apache.bookkeeper-bookkeeper-common-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.common.component.LifecycleComponentStack.lambda$start$4(LifecycleComponentStack.java:144) ~[org.apache.bookkeeper-bookkeeper-common-4.14.7.jar:4.14.7]
        at com.google.common.collect.ImmutableList.forEach(ImmutableList.java:422) ~[com.google.guava-guava-31.0.1-jre.jar:?]
        at org.apache.bookkeeper.common.component.LifecycleComponentStack.start(LifecycleComponentStack.java:144) ~[org.apache.bookkeeper-bookkeeper-common-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.common.component.ComponentStarter.startComponent(ComponentStarter.java:85) ~[org.apache.bookkeeper-bookkeeper-common-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.server.Main.doMain(Main.java:234) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.server.Main.main(Main.java:208) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
2023-07-07T09:51:09,173 [main] ERROR org.apache.bookkeeper.common.component.AbstractLifecycleComponent - Calling uncaughtExceptionHandler
2023-07-07T09:51:09,173 [main] ERROR org.apache.bookkeeper.common.component.ComponentStarter - Triggered exceptionHandler of Component: bookie-server because of Exception in Thread: Thread[main,5,main]
java.lang.OutOfMemoryError: Cannot reserve 1601533033 bytes of direct buffer memory (allocated: 554639373, limit: 1073741824)
        at java.nio.Bits.reserveMemory(Bits.java:178) ~[?:?]
        at java.nio.DirectByteBuffer.&lt;init&gt;(DirectByteBuffer.java:121) ~[?:?]
        at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:332) ~[?:?]
        at sun.nio.ch.Util.getTemporaryDirectBuffer(Util.java:243) ~[?:?]
        at sun.nio.ch.IOUtil.read(IOUtil.java:293) ~[?:?]
        at sun.nio.ch.IOUtil.read(IOUtil.java:273) ~[?:?]
        at sun.nio.ch.FileChannelImpl.read(FileChannelImpl.java:232) ~[?:?]
        at org.apache.bookkeeper.bookie.JournalChannel.read(JournalChannel.java:261) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.bookie.Journal.fullRead(Journal.java:1199) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.bookie.Journal.scanJournal(Journal.java:838) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.bookie.Bookie.replay(Bookie.java:997) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.bookie.Bookie.readJournal(Bookie.java:962) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.bookie.Bookie.start(Bookie.java:1017) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.proto.BookieServer.start(BookieServer.java:156) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.server.service.BookieService.doStart(BookieService.java:68) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.common.component.AbstractLifecycleComponent.start(AbstractLifecycleComponent.java:83) ~[org.apache.bookkeeper-bookkeeper-common-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.common.component.LifecycleComponentStack.lambda$start$4(LifecycleComponentStack.java:144) ~[org.apache.bookkeeper-bookkeeper-common-4.14.7.jar:4.14.7]
        at com.google.common.collect.ImmutableList.forEach(ImmutableList.java:422) ~[com.google.guava-guava-31.0.1-jre.jar:?]
        at org.apache.bookkeeper.common.component.LifecycleComponentStack.start(LifecycleComponentStack.java:144) ~[org.apache.bookkeeper-bookkeeper-common-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.common.component.ComponentStarter.startComponent(ComponentStarter.java:85) ~[org.apache.bookkeeper-bookkeeper-common-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.server.Main.doMain(Main.java:234) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
        at org.apache.bookkeeper.server.Main.main(Main.java:208) ~[org.apache.bookkeeper-bookkeeper-server-4.14.7.jar:4.14.7]
2023-07-07T09:51:09,179 [component-shutdown-thread] INFO  org.apache.bookkeeper.common.component.ComponentStarter - Closing component bookie-server in shutdown hook.
2023-07-07T09:51:09,222 [vert.x-eventloop-thread-0] INFO  org.apache.bookkeeper.http.vertx.VertxHttpServer - Starting Vertx HTTP server on port 8000
2023-07-07T09:51:09,231 [component-shutdown-thread] INFO  org.apache.zookeeper.ClientCnxnSocketNetty - channel is told closing
2023-07-07T09:51:09,237 [epollEventLoopGroup-17-1] INFO  org.apache.zookeeper.ClientCnxnSocketNetty - channel is disconnected: [id: 0x331b88c4, L:/172.16.2.41:45130 ! R:itomdipulsar-zookeeper/172.16.1.21:2281]
2023-07-07T09:51:09,242 [component-shutdown-thread] INFO  org.apache.zookeeper.ZooKeeper - Session: 0x3009a1951ba0031 closed
2023-07-07T09:51:09,242 [epollEventLoopGroup-17-1] INFO  org.apache.zookeeper.ClientCnxnSocketNetty - channel is told closing
2023-07-07T09:51:09,244 [main-EventThread] INFO  org.apache.zookeeper.ClientCnxn - EventThread shut down for session: 0x3009a1951ba0031
2023-07-07T09:51:09,298 [main] ERROR org.apache.bookkeeper.http.vertx.VertxHttpServer - Failed to start org.apache.bookkeeper.http server on port 8000
java.lang.IllegalStateException: executor not accepting a task
        at io.netty.resolver.AddressResolverGroup.getResolver(AddressResolverGroup.java:61) ~[io.netty-netty-resolver-4.1.86.Final.jar:4.1.86.Final]
        at io.vertx.core.impl.AddressResolver.resolveHostname(AddressResolver.java:82) ~[io.vertx-vertx-core-3.9.8.jar:3.9.8]
        at io.vertx.core.impl.VertxImpl.resolveAddress(VertxImpl.java:829) ~[io.vertx-vertx-core-3.9.8.jar:3.9.8]
        at io.vertx.core.net.impl.AsyncResolveConnectHelper.doBind(AsyncResolveConnectHelper.java:56) ~[io.vertx-vertx-core-3.9.8.jar:3.9.8]
        at io.vertx.core.http.impl.HttpServerImpl.listen(HttpServerImpl.java:290) ~[io.vertx-vertx-core-3.9.8.jar:3.9.8]
        at io.vertx.core.http.impl.HttpServerImpl.listen(HttpServerImpl.java:205) ~[io.vertx-vertx-core-3.9.8.jar:3.9.8]
        at io.vertx.core.http.impl.HttpServerImpl.listen(HttpServerImpl.java:201) ~[io.vertx-vertx-core-3.9.8.jar:3.9.8]
        at org.apache.bookkeeper.http.vertx.VertxHttpServer$2.start(VertxHttpServer.java:85) ~[org.apache.bookkeeper.http-vertx-http-server-4.14.7.jar:4.14.7]
        at io.vertx.core.AbstractVerticle.start(AbstractVerticle.java:108) ~[io.vertx-vertx-core-3.9.8.jar:3.9.8]
        at io.vertx.core.Verticle.start(Verticle.java:66) ~[io.vertx-vertx-core-3.9.8.jar:3.9.8]
        at io.vertx.core.impl.DeploymentManager.lambda$doDeploy$9(DeploymentManager.java:556) ~[io.vertx-vertx-core-3.9.8.jar:3.9.8]
        at io.vertx.core.impl.ContextImpl.executeTask(ContextImpl.java:366) ~[io.vertx-vertx-core-3.9.8.jar:3.9.8]
        at io.vertx.core.impl.EventLoopContext.lambda$executeAsync$0(EventLoopContext.java:38) ~[io.vertx-vertx-core-3.9.8.jar:3.9.8]
        at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174) ~[io.netty-netty-common-4.1.86.Final.jar:4.1.86.Final]
        at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167) ~[io.netty-netty-common-4.1.86.Final.jar:4.1.86.Final]
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470) ~[io.netty-netty-common-4.1.86.Final.jar:4.1.86.Final]
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:569) ~[io.netty-netty-transport-4.1.86.Final.jar:4.1.86.Final]
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997) ~[io.netty-netty-common-4.1.86.Final.jar:4.1.86.Final]
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[io.netty-netty-common-4.1.86.Final.jar:4.1.86.Final]
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[io.netty-netty-common-4.1.86.Final.jar:4.1.86.Final]
        at java.lang.Thread.run(Thread.java:833) ~[?:?]
Exception in thread "main" java.lang.IllegalStateException: Can't move to started state when closed
        at org.apache.bookkeeper.common.component.Lifecycle.moveToStarted(Lifecycle.java:134)
        at org.apache.bookkeeper.common.component.AbstractLifecycleComponent.start(AbstractLifecycleComponent.java:93)
        at org.apache.bookkeeper.common.component.LifecycleComponentStack.lambda$start$4(LifecycleComponentStack.java:144)
        at com.google.common.collect.ImmutableList.forEach(ImmutableList.java:422)
        at org.apache.bookkeeper.common.component.LifecycleComponentStack.start(LifecycleComponentStack.java:144)
        at org.apache.bookkeeper.common.component.ComponentStarter.startComponent(ComponentStarter.java:85)
        at org.apache.bookkeeper.server.Main.doMain(Main.java:234)
        at org.apache.bookkeeper.server.Main.main(Main.java:208)
2023-07-07T09:51:09,307 [vert.x-eventloop-thread-8] INFO  org.apache.bookkeeper.http.vertx.VertxHttpServer - HTTP server is shutdown
2023-07-07T09:51:09,311 [component-shutdown-thread] INFO  org.apache.bookkeeper.proto.BookieServer - Shutting down BookieServer```


<@ULBJ01G6N> we already upgraded to pulsar 2.10.3 (which is closer to pulsar latest release) and tried in pulsar 2.10.4 . Still seeing same issue while replaying a journal file, one of bookie is going to OutofDirectMemory.

:mega: The Apache Pulsar community meets every 2 weeks virtually. We discuss open Pull Requests, Proposals (PIP), future roadmaps and just chit chatting on everything Pulsar.

Join us!

It’s today at 15:30 UTC

All details are here: <https://github.com/apache/pulsar/wiki/Community-Meetings>

When using tiered storage, after data is offloaded, I noticed that the data in the data directory did not decrease. Why is this happening?

Another issue I encountered is that when I deleted the data of a topic, I noticed that the directory of that topic in my offload directory was not removed. Is this the expected behavior?

Here is <https://github.com/david-streamlio/sentiment-analysis/blob/main/sentiment-analysis-function/pom.xml|one>, and its parent <https://github.com/david-streamlio/sentiment-analysis/blob/main/pom.xml|pom> that contains the packaging plugin config, etc.

Hi David, I was hoping for a very simple or basic pom.xml file that creates a jar package for the the sample java functions in the pulsar repo.

It would basically be the parent pom file then. It includes the dependencies you need along with the NAR packaging config, which is the preferred packaging for Java Pulsar Functions. You can use the shade plugin module if you prefer bundling it as a JAR file,

```                 &lt;plugin&gt;
            		&lt;groupId&gt;org.apache.maven.plugins&lt;/groupId&gt;
        			&lt;artifactId&gt;maven-shade-plugin&lt;/artifactId&gt;
        			&lt;version&gt;${maven-shade-plugin.version}&lt;/version&gt;

            		&lt;executions&gt;
                		&lt;execution&gt;
              	      		&lt;phase&gt;package&lt;/phase&gt;
                	    	&lt;goals&gt;
                    	    	&lt;goal&gt;shade&lt;/goal&gt;
                    		&lt;/goals&gt;
                		&lt;/execution&gt;
            		&lt;/executions&gt;
            		&lt;configuration&gt;
              			&lt;filters&gt; 
                			&lt;filter&gt;
                  				&lt;artifact&gt;*:*&lt;/artifact&gt;
                  				&lt;excludes&gt;
					            	&lt;exclude&gt;META-INF/*.SF&lt;/exclude&gt;
                    				&lt;exclude&gt;META-INF/*.DSA&lt;/exclude&gt;
					                &lt;exclude&gt;META-INF/*.RSA&lt;/exclude&gt;
                    				&lt;exclude&gt;META-INF/MANIFEST.MF&lt;/exclude&gt;
					            &lt;/excludes&gt;
                			&lt;/filter&gt;
             			&lt;/filters&gt;
            		&lt;/configuration&gt;
        		&lt;/plugin&gt;```

David, I also found two pom files in the Pulsar site. One is for JAR another for NAR. You said NAR format is preferred. Could you elaborate on that. Thanks !!!

Sure, the <https://docs.cloudera.com/HDPDocuments/HDF3/HDF-3.0.1/bk_developer-guide/content/nars.html#:~:text=A%20NAR%20allows%20several%20components,NiFi%20components%20as%20NAR%20packages.|NAR> format is based on the package management developed as part of the Apache NiFi project. The packaging allows the pulsar functions framework to use a separate Classloader thread to load the dependencies for a Pulsar Function, which allows us to avoid package collisions when functions use different versions of the same library.

Hi, regarding the first question, the Pulsar function acts just like a consumer subscribed to multiple topics. What specifically is your question here?

For question 2 there are some limitations around the type: <https://github.com/apache/pulsar/blob/master/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Function.java#L27-L30>

<@U053NKRDBL0>, if your Function subscribes to multiple topics, then each of those topic must have schemas that are compatible with one another, i.e. the same schema. As Alexander pointed out, the Pulsar consumer feeding the Function will attempt to deserialize the incoming messages into the defined input class. If it cannot, then the messages will never be delivered to/processed by the Function

So for the input type one can use json object string and process it with org.json package. In this way I can handle any message types.

Yes, you can use a schema type of String for the topics and perform the deserialization yourself inside the Function.

Initially I was thinking Avro but that is not possible and not as flexible 

If you use `byte[]` then your function can consume any message. Maybe that is a path forward for you. I have a number of functions that rely on this. They consume `byte[]` , check the Message properties for a type hint we set, then do deserialization based on the type hint. In our case its protobuf.

If you are using a JSON based schema, then the String schema type will suffice. If you want to use Avro then you will need to use the byte[] schema type and ensure that your functions have access to the Avro library and type definition so they can deserialize the messages internally. You can also use the generic object <https://pulsar.incubator.apache.org/api/client/2.8.3/org/apache/pulsar/client/api/schema/GenericObject.html|class> as well

I like both suggestions. However I also thought about another. Graalvm has a JavaScript engine. It can run NodeJS applications natively. So I will wrap the Pulsar Functions with JavaScript and run my application logic in JavaScript. 

Good luck. That sounds like a lot of work just to deserialize json

I think I will opt for simpler solutions that you guys suggested. I will try to test both solutions and finally pick one. Thanks for your valuable inputs!

have anyone deployed the pulsar kafka connect adaptor in AWS? I see some source code in pulsar github, but nothing in documentation

Pulsar recommend by default to use upto 5MB of message size. Is there any limit from which point we should use streaming?
Also, will chunking will be available with 3.x for shared sub?

I am building a messaging app for RocketChat (<https://www.rocket.chat/>), sometimes the system create multiple rooms and messages so I am looking into using Pulsar to queue the messages from an external webhook&gt;
Now, implementing pulsar with nodejs pulsar-client but get his error each time I try deploying my app
``` ›   Error:    ⟿  Error: Deployment error: Cannot read property 'EventEmitter' of undefined```

Without code, it's tough to say, however it seems like this might be an execution context problem: <https://stackoverflow.com/questions/37307415/output-eventemitter-is-always-undefined>

<@U01AQHHNDS9> Here is my code

```import * as Pulsar from "pulsar-client";

class PulsarClient {
    public pc: any;
    public pulsarURL =
        "string";

    public pulsarJWT =
        "string"

    public pulsarTopic = "string";

    public subscriptionName = "sample";

    constructor() {
        const authentication = new Pulsar.AuthenticationToken({
            token: this.pulsarJWT,
        });

        this.pc = new Pulsar.Client({
            serviceUrl: this.pulsarURL,
            authentication,
        });
    }

    async consumer() {}

    async producer(payload: any) {
        const producer = await this.pc.createProducer({
            topic: this.pulsarTopic,
            // topic: topic,÷
        });

        producer.send({
            data:"data",
        });
    }
}

export default PulsarClient;```

<@U01AQHHNDS9> then I am instantiating the class in the a rocket chat life cycle method called `Initialize`  I don't know if you are familiar with the app engine

the zk fsync is taking too long (4+ seconds). what’s in the broker log?  in general, you can disable fsync or separate the zk data and datalog (journal) disks to avoid disk io issue.

I’ve got some response in github discussion, the main reason is that helm chart does not support k8s 1.24+ , and someone had already patched it. I’m going the try the latest helm charts. BTW, how to disable fsync with Pulsar?

Which Helm charts are you using? Are your Zookeeper pods all up and running?

Hi David, I'm using  chart pulsar-3.0.0    with APP version  2.10.2. and Yes, both pulsar-zookeeper-0 and pulsar-pulsar-manager-7d4f89db7b-tlnxg are running.  Thanks

StreamNative <https://github.com/streamnative/charts|Charts> or Apache <https://github.com/apache/pulsar-helm-chart|Charts>?

Apache <https://github.com/apache/pulsar-helm-chart|Charts>

Those have been deprecated, so they might not work as expected.

describe the pods, got the following errors: Warning Unhealthy              40m (x3 over 43m)     kubelet                 Readiness probe failed: Get "<http://100.64.4.129:8000/api/v1/bookie/is_ready><http://100.64.4.129:8000/api/v1/bookie/is_ready%22:|":> dial tcp 100.64.4.129:8000: connect: connection refused
 Warning Unhealthy              39m (x4 over 42m)     kubelet                 Liveness probe failed: Get "<http://100.64.4.129:8000/api/v1/bookie/state><http://100.64.4.129:8000/api/v1/bookie/state%22:|":> dial tcp 100.64.4.129:8000: Warning Unhealthy 52m                   kubelet           Liveness probe failed: Get "<http://100.64.4.130:8080/status.html><http://100.64.4.130:8080/status.html%22:|":> dial tcp 100.64.4.130:8080: connect: connection refused
 Warning Unhealthy 52m                   kubelet           Readiness probe failed: Get "<http://100.64.4.130:8080/status.html><http://100.64.4.130:8080/status.html%22:|":> dial tcp 100.64.4.130:8080: connect: connection refused

We were using old version 2.9.4, I'm trying to upgrade to latest version 3.0.0.  what you mean by deprecated? Which helm chart should I use instead? Thanks

Deprecated, as in no more releases, support etc.  <https://artifacthub.io/packages/helm/streamnative/pulsar>

“This Apache Pulsar Helm Chart has been deprecated and we recommend using <https://github.com/streamnative/charts/tree/master/charts/sn-platform|streamnative/sn-platform> to provision and manage the Apache Pulsar on Kubernetes.”

Good morning David! We had a introduction zoom meeting with Marshall Portwood yesterday. He mentioned that you can help me with trouble shooting the issues I am facing. Thank you in advance for your time and help!

I will forward you the email with the logos and describes I had for the issue.

Thanks for sharing the information / logs. I am reviewing them now.

Can you share the output of the following commands? `kubectl describe pulsar-bookie-0 -n pulsar` and `kubectl describe pulsar-broker-0 -n pulsar` Thanks

Those two should be in the attached email. I will run again

ok, let’s switch to communicating via email

Hi <@U03760LLTQW>, Currently there is one blocker issue for the 3.0.1 release. I think we will fix it this week and then start the candidate-1 release.

<@U0152Q766AE> what is the blocker issue? Could you please label the issue with `release/blocker` ? thanks

<https://github.com/apache/pulsar/issues?q=is%3Aissue+is%3Aopen+label%3Arelease%2Fblocker+> doesn't currently show any blockers related to 3.0.1 release

If you are referring to <https://docs.rapids.ai/api/cudf/stable/>, then I would say that it should be possible. Are you having issue using this library?

No, I have not attempted yet. I have a coworker who is curious about this and was wondering if anyone had tried this before? Yes, I am referring to the <http://rapids.ai|rapids.ai>

To my knowledge nobody has tried this, but in theory it should just work since it is “just” a Python library. Now you might want to package it differently in a separate docker image in order to exploit GPUs, etc. But that is doable. Step 1 would be running a simple function to validate it :smiley:

I would be curious to see how this works…..

We are going to look into it and hopefully I will be able to get back to you soon on how this can work! thanks for your help!

pulsar manager is basically abandonware at this point

So how can we see configurations such as namespaces, topics and etc of  our cluster using UI?

build your own, use one of non-"official" options, or spend the time to address the shortcomings of pulsar-manager

The `loadBalancerSheddingEnabled`  is broker level configuration, can you configure it at group level?

Please let me know how to set up password for Pulsar manager v0.3.0 and v0.4.0 version?
Version v0.2.0 seems to be set well with this command.

This is working for us with v0.4.0 as well

Thank you Moty,
It seems that the path is different from version v0.2.0.

CSRF_TOKEN=$(curl <http://172.20.29.205:9527/pulsar-manager/csrf-token>

are you using kubernetes or deployed it on some vm?

Thank you Moty,
We are using kubernetes (AWS EKS) with pulsar helm chart(valuse.yaml).
We use apachepulsar/pulsar-manager docker repository.
And, we try to use pulsar manager v0.4.0

Could someone help me understand what the blob files pushed by offload actually contains. I see 2 files per ledger, 1 file seems to have messages with some padding. Id like to be able to parse these blob offloads in an array of jsons if possible

Check out the document. <https://pulsar.apache.org/docs/3.0.x/admin-api-topics/#get-backlog-size>

Hi <@U01B9PMHVGX> it seems interesting. We are thinking of it. Just wanted to know if it is production ready?

Hi <@U0319FEUPBL> sorry for the delay, I was on vacation.
Our testing has proven it is production ready and we are in the process of rolling it out to all our Astra Streaming Pulsar production clusters.

Thanks <@U01B9PMHVGX> We would need help if faced any. Would appreciate if the suggestion keep coming.

No. I need a few dozen billion objects I've saved to be moved into pulsar

I'm not interested in any sort of change data, and it's more of the batch source rather than a streaming source, in that after the source load is done we won't trigger it again, but I am looking at a highly scalable way to bring in a huge data set.

We have looked at a few other etl type options, and the performance has not been all that impressive, though I'm open to suggestions. Ive done some other work writing some functions that broke the load job up into batches (published to Pulsar topics) and then processed the batches in a highly parallel way with Pulsar functions and got much closer to the performance targets I am looking for.

I wrote an <https://github.com/apache/flink/compare/master...alpreu:flink:SDK-152-elastic-source|Elasticsearch source for Flink> some time ago. You want to use `search_after` and create a `PIT` to ensure good performance. I think the biggest hurdle you will face is that the Pulsar IO framework does not provide abstractions that help with coordination of the connector instances. I’d be happy to provide some guidance if you start working on this

I was looking at the BatchSource interface and PIP: <https://github.com/apache/pulsar/wiki/PIP-65:-Adapting-Pulsar-IO-Sources-to-support-Batch-Sources>
Have you used that?

I was thinking of a sliced scroll (since I'm not interested in realtime data)

I've also achieved very high performance with sliced scrolls previously. I've not used PIT.

`BatchSource` will work if you are going to assign all work packages to the instances right at the beginning, yes

my initial idea on reading that PIP was to assign each slice of the scroll to one instance

The docs explicitly advise against sliced scrolls for larger datasets (see here <https://www.elastic.co/guide/en/elasticsearch/reference/current/paginate-search-results.html#scroll-search-results>)

specifically, I'm not interested in preserving index state, I'm not going to sort in elasticsearch, and I have not seen how you can do a PIT across multiple threads.

The design there would be to have a query that allowed it self to be partitioned to generate the parallel PITs

ok, I had missed how to do a sliced PIT.
<https://www.elastic.co/guide/en/elasticsearch/reference/current/point-in-time-api.html#search-slicing>

Same idea, however we _slice_ it. The idea would be to generate n number of tasks, where each task is a slice, then I believe BatchSource can help with managing the different instances, etc.

<@UFC5NB9FG> in creating the config for an ElasticSearchBatchSource, what is the preferred model for encapsulating common properties between the Sink and the Source? An interface with the full list of fields in each implementation? A common parent class?

Probably a common parent class may work.
it is not common to have both a source and a sink for the same system all together (they are usually quite different)

maybe you can take a look to KafkaSource and KafkaSink

Thank you. I've been working on creating a common parent so far today, with a few minor issues. I'll definitely look at how KafkaSink and Source are doing it. Its always nice to have a model.

In this case I want to be able to reuse all of the common ElasticSearch client code, because that is the same across both implementations. There are differences, certainly, but clearly a lot of work was done to support ElasticSearch, OpenSearch, and the different versions.

KafkaSink and Source Config classes are completely independent and just have the same fields.  If its ok with you I'd like to continue to try to go down the common parent path.

I implemented the sliced PIT search for Elasticsearch, but when working on OpenSearch, I noticed that it only supports PIT from 2.4 (and not at all in the version of the REST client used in Pulsar). <@UFC5NB9FG> <@U059A25BXMW> do you think I should also implement the scroll based search since that will have the most backwards compatibility? Also, is there a reason Pulsar isn't using the OpenSearch java client? That library is compatible back to OpenSearch 1.3.9 (<https://github.com/opensearch-project/opensearch-java/blob/main/COMPATIBILITY.md>). What should the goal for backwards compatibility be?

<@U01B9PMHVGX> may have a good answer here

Regarding backward compatibility, it is important that if an user upgrades Pulsar the system continues to work seamlessly.
We should avoid breaking changes as much as possible.

Ok. Given that the current functionality (bulk indexing) is basically the same going back to ES 5 or maybe even ES 2, I think we may have to think about a minimum supported version for using ES as a batch source.  I've implemented the core methods to do both PIT and Scroll searches for ElasticSearch and OpenSearch. Now I am at the point of working on what the common interface for all of this is in the `RestClient` to then be called by the `ElasticSearchBatchSource` . It is probably a good idea to implement some version checks if a user has configured PIT, but their ES / OS version doesn't support it.

Personal opinion: Even ES6 is ancient at this point, if there is no strong need in the community I think supporting versions going all the way back to ES5 is not necessary.

I’m not familiar with how the OpenSearch versions relate to the Elastic ones so I can’t really comment on that.

Implementing some checks on startup sounds like a good idea :+1:

<@U059A25BXMW> completely agree. I thought I would be able to use PIT for both, but given that 2.4 is when OpenSearch introduced PIT (latest OpenSearch version is 2.8), it felt like that might be too much of a constraint. Additionally, I've implemented sliced scroll searches dozens of times, so it didn't take me long and would provide the farthest back compatibility if needed by a user.

&gt; think we may have to think about a minimum supported version for using ES as a batch source.
Makes sense.

The versions supported by the Sink can be different from the versions supported by the Source

Sounds good. I keep plugging away at this.

Ok, so I built out the tests for PIT and Scroll searches, and found an interesting edge case. For ES 7 the code currently uses the OpenSearchHighLevelRestClient. Fine. Except for the fact that PIT searches use a different  endpoint in ES and OpenSearch. So using the OpenSearchHighLevelRestClient isn't going to work for ES 7 clusters.  I looked and the ES Java client works back to 7.14, so if you all think its prudent, I can make the change to use the ES java client if a supported version of ES is detected. I believe the OpenSearch rest client should be compatible with 7.10 and previous versions of ES...so I'm not really sure what to do for the 7.10 - 7.14 versions of ES.  Additionally, I believe scroll searches work just fine with the OpenSearch rest client against an ES cluster, as those have the same endpoints, so another option to just just not allow PIT searches on those and have the user configure a scroll search instead.

These client compatibility topics always seem to be a nightmare :grin:
Let’s say the implementation would only be based on the scrolled search right now - does this guarantee data isn’t read twice? IIRC this was one of the main reasons for using the PIT (besides the performance on large searches)

the one thing I've learned in my last 7 years of using ElasticSearch at scale is there is really no such thing as a guarantee.  Scroll search appears to be deprecated in ES 8, so we definitely need to leave the PIT search there. Both PIT and Scroll claim to effectively snapshot the index state at the start time. PIT has advantages that multiple queries can be run and you can move forwards and backwards in pages. Scroll only allows the one query to run, and you can only move forward in results.  There is also some documentation saying that the slicing mechanism used for PIT is more efficient than for Scroll.

<@U059A25BXMW> <@UFC5NB9FG> it gets worse. I wanted to add OpenSearch 2.4+  into testing so that I could validate that the PIT search works as expected on OpenSearch. Well, the OpenSearch 1.2.4 rest client being used isn't able to communicate with OpenSearch 2.x at all. I'm going to try to see if the later versions of the OpenSearch high level rest client can work with OpenSearch 1.X.  Otherwise, I guess I can look at implementing the OpenSearch java client (similar to the Elasticsearch Java client).

OpenSearch 2.7 is just the right version. The high level rest client works with 1.2.x as well as 2.x.  The gotcha there is that support for the es type has been completely removed. So again, we have to pick our poison on ES compatibility. I would say this might be where I copy everything into a new module, but honestly, the es sink should support OpenSearch 2.x as well, and types have been known to be a bad idea since ES 5 if not earlier, and deprecated for a long time.

Please let me know how youd like me to proceed. I can also push everything to a branch so you can see what it looks like so far.

<@UFC5NB9FG> Thank you. Will do. The main things I have left to finish are key generation from the ES result and any schema handling

<@UFC5NB9FG> <@U059A25BXMW> <@U01B9PMHVGX> Here is the draft PR in my repo : <https://github.com/Q6Cyber/pulsar/pull/2>  I greatly appreciate any feedback at this stage.  I still have a fair amount of work to do, but I think I've made some solid progress and hopefully have things on a good path.

I'm headed out on vacation for a week. Thank you for the help and guidance with this. I'm looking forward to getting it wrapped up when I come back.

I’ll have a look at the PR over the next few days. Enjoy your vacation!

Hi, we previously CDC'ed some data into our HA Pulsar Cluster but the storage replication settings set to:
```Ensembel Size - 1
Write Quorum Size - 1
Ack Quorum Size - 1```
We've now updated these to be after the initial data snapshot.
```Ensembel Size - 3
Write Quorum Size - 2
Ack Quorum Size - 2```
Is it possible to rereplicate the existing message across the bookies as the messages are unbalanced?

Is there a way to parse the blob files created by the Blobstore Offloader?

search jute.maxbuffer in <https://zookeeper.apache.org/doc/r3.8.0/zookeeperAdmin.html>
The explanation is under unsafe options

packet len 12644126 ~ 10MB. this means your znode size is bigger than 10MB. As a pulsar user, you need to increase jute.maxbuffer.  as a pulsar developer, you can try search which znode is over the limit and try to find how broker/bookkeeper store info in this znode and reduce the content size.

what does it mean that my znode is bigger than 10MB?

I’m deploying Pulsar on top of EKS and i’m giving zookeeper nodes a 20Gi EBS volume for each node, so 10MB shouldn’t be any problem.

do you mean: a message size that zookeeper is handling?

what does this line mean?
*When _jute.maxbuffer_ in the client side is less than the server side, the client wants to read the data exceeds _jute.maxbuffer_ in the client side, the client side will get java.io.IOException: Unreasonable length or Packet len is out of range!*

*who is the client in my case? pulsar broker?*

also, I see it’s not recommended to change that in production

The client is pulsar-broker, which use 10MB by default <https://github.com/apache/pulsar/blob/ecd16d68e29700cc9521e6d77d124a403c3486c9/bin/pulsar#L288>, you can try to change Pulsar Env <https://github.com/apache/pulsar/blob/ecd16d68e29700cc9521e6d77d124a403c3486c9/conf/pulsar_env.sh#L86> to increase the jute.maxbuffer size.

<@U053BJ8UUBZ>, Zookeeper is a tree-like system (like a linux file system, with directories and files). Like any tree in computer science, it is built of nodes, referred to as ZNode (Zookeeeper node). Each such node can hold data (an array of bytes) and list of child nodes.

 What does Jute.maxBuffer means? ChatGPT 4 (3.5 gives really bad result) to the rescue:

&gt; “jute.maxbuffer” is a configuration property in Apache ZooKeeper that specifies the maximum size of the data that ZooKeeper will accept from a client for processing. It’s particularly relevant when working with larger amounts of data.
&gt; This property is part of ZooKeeper’s underlying data serialization library, Jute. By default, the max buffer size is typically set to 1 megabyte (1MB), which means any attempt to write data larger than this limit to ZooKeeper will result in an exception.
Pulsar holds a ZK client since Pulsar stores information in ZK, like the list of ledger IDs per topic.

<@UC0BYCXKQ> Should Pulsar add to the error the node that was maxed out in size?

<@U053BJ8UUBZ> If you have the stacktrace, would like to contribute a fix to this?

<@U01551PS4MA> I don’t have the stacktrace and the change you are talking about is a change in pulsar code itself, not in the chart as I understand

Regarding my issue, do you think I should increase the Jute.maxBuffer for pulsar borker?

<@U053BJ8UUBZ> Yes, it’s in the Pulsar code.

Based on what was written here, yes, increase it.

jute.maxBuffer on server and client MUST match. make sure you update zk, broker and bookie (both are zk clients).