pulsar

thing is, you wan't know where in the past :slightly_smiling_face: if you need to iterate from now to previous 500 messages for example. The only way would be to index MessageIDs so that we know what MessageID should be the starting point in the right reading order.

```public static void main(String[] args) {
        List&lt;Message&gt; msgs = getMessagesBetween(Long.parseLong(args[0]), Long.parseLong(args[1]));
    }

    private static List&lt;Message&gt; getMessagesBetween(long start, long finish) {
        List&lt;Message&gt; msgs = new LinkedList&lt;&gt;();
        boolean finished = false;
        try {
            Consumer&lt;byte[]&gt; consumer = getClient()
                    .newConsumer()
                    .topic(TOPIC)
                    .subscribe();

            consumer.seek(start);

            while (!finished) {
                Message&lt;byte[]&gt; msg = consumer.receive();
                if (msg.getEventTime() &lt;= finish) {
                    msgs.add(msg);
                } else {
                    finished = true;
                }
                
                consumer.acknowledge(msg);
            }
        }  catch (PulsarClientException e) {
            System.err.println(e.getMessage());
        }

        Collections.reverse(msgs);
        return msgs;
    }```

This example does a time-range reversal of the messages in a topic. A similar pattern could be used for a “count after a point in time” range as well.

Hello. I had a question regarding topic level replication. I have posted my question here:
<https://github.com/apache/pulsar/discussions/19682>
Please can someone help with it.

VeryCapture_20230302114438.jpg

VeryCapture_20230302114414.jpg

This is a bug and the related issue is <https://github.com/apache/pulsar/issues/6437>

this occur when broker is not the owner of topic. In case if you have configured ns-isolation policy. You gotta configure same for kafka ns too

Thanks for replying,
But we haven't configured any ns-isolation policy on cluster.

Ok then broker must have redirected it to the correct broker.
You can check ownership with command
pulsar-admin topics lookup topicname

You might ask your <#C0106RFKAHM|> on that channel. FYI - KoP is an OpenSource project from StreamNative and not from the Apache Pulsar project. We are happy to host their discussion in our community channels, but I want to be clear about the state of things.
Also, I'll mention that DataStax has a fork of KoP called Starlight for Kafka and we contribute back to KoP - see <https://github.com/datastax/starlight-for-kafka/>

Not that I know of. The go client has built in prometheus metrics that it exports. I wonder if we could add something similar to the java client. Are you looking for metrics that would be generally useful to all pulsar client users or just for your application?

Just general producer &amp; consumer metrics with a micrometer registry. It would be nice if the client kept a public method to retrieve every ProducerStats &amp; ConsumerStats object that it generates

Or, `ConsumerStats` could have another field to reference the `ProducerStats` for its retry &amp; deadLetter producers

I think it might make sense to add a method on the consumer interface to get those dlq producer stats.

as in `Consumer.getDlqProducerStats` and `Consumer.getRetryProducerStats`? Or `ConsumerStats.getDlqProducerStats` / `ConsumerStats.getRetryProducerStats`

Yes, that's what I am thinking. We'd need a PIP to get it added. The other option is to embed the stats in the `ConsumerStats`, as you proposed. I could see it going either way

Alright. I will create a github issue to discuss. It should be a PIP either way?

Yes, I would expect a PIP for either implementation

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

Hi, is there any other tool that can be used to generate go files from avro like gogen-avro ?

Extended: The CFP for Pulsar Virtual Summit Europe 2023 now closes Friday, March 10th! Submit here: <https://sessionize.com/pulsar-virtual-summit-europe-2023/>

Hello David, don't think this operation can be done

I contributed Pulsar client support to OpenTelemetry java agent. The PR has been merged and should be available in the next release. Please try it if you need.
<https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/5926>

Which Pulsar client version are you using? How about the Pulsar broker version?

Hi Lari,

Thank you for getting back to me.
The broker version is <https://pulsar.apache.org/release-notes/versioned/pulsar-2.10.1/>
And the client is : <https://javadoc.io/doc/org.apache.pulsar/pulsar-client-api/2.10.3/index.html>

This looks like a bug. <@U03CRM6DC58> Please report an issue at <https://github.com/apache/pulsar/issues/new/choose> to get things going.

Do you know how many CPU cores does the test environment have?

One possible workaround could be to increase the client's numIoThreads (`ioThreads` method in ClientBuilder) to a value more than one, for example 4.
The possible reason for the deadlock is that it's waiting for a result on the same thread that the result would complete on. With numIoThreads of 1, this would be an issue.
<https://github.com/apache/pulsar/blob/b69f4efa6058c3f51885a61a2b3acb46f8b730f4/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java#L112|numIoThreads defaults to 1 in 2.10.3 client.>

Thanks Lari, I will try this and get back to you

The reason I asked about the CPU cores is that in master branch, the default value for numIoThreads is the amount of CPU cores. source code: <https://github.com/apache/pulsar/blob/7f37670fc96e4b629234c242257fe1d42480e5cd/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java#L113|https://github.com/apache/pulsar/blob/7f37670fc96e4b629234c242257fe1d42480e5cd/pul[…]org/apache/pulsar/client/impl/conf/ClientConfigurationData.java>
However, this change (<https://github.com/apache/pulsar/pull/19059>) isn't part of any release client versions, so my question about the CPU cores isn't that relevant in this case.

I opened a ticket on GitHub <https://github.com/apache/pulsar/issues/19784>

thank you <@U03CRM6DC58>. did you have a chance to test the workaround?

not yet but we will run tests during the weekend and let you know then

FWIW, the standalone startup for desktop also changed. If you run `bin/pulsar standalone` you now get the following. `ERROR org.apache.pulsar.PulsarStandaloneStarter - The following option is required: [-c | --config]` . Not sure, but it might be related.

I’m sure it has something to do with that, I’ll look into it

are you asking about the acktimeout on the consumer side?

I don't see a mention of it in the source code..l;et me cross check..However, by default the acktimeout is disabled in the client side..

```By default, the acknowledge timeout is disabled and that means that messages delivered to a consumer will not be re-delivered unless the consumer crashes.```

Ok, this is the equivalent in the go client

```// NackBackoffPolicy is a interface for custom message negativeAcked policy, users can specify a NackBackoffPolicy
// for a consumer.
//
// &gt; Notice: the consumer crashes will trigger the redelivery of the unacked message, this case will not respect the
// &gt; NackBackoffPolicy, which means the message might get redelivered earlier than the delay time
// &gt; from the backoff.```

<https://github.com/apache/pulsar-client-go/search?q=NackBackoffPolicy>

<https://github.com/apache/pulsar-client-go/issues/217>

Feel free to reply there if you think the issue should be reopened

Thanks for the additional pointers <@U01GAUEA9TR>

Thank you, I got the idea :raised_hands: 

Hi! In the exclusive and failover subscription types, consumers only negatively acknowledge the last message they have received. So the order is preserved.

Yes, but I found that in my testing, a message can be delivered right after the consumer nacks, meaning it received messages out-of-order

I think this thread might have some relevant context: <https://lists.apache.org/thread/zz68xx3gydbtggfrm9t2lf215qssp3z8>

Actually, I might have linked the wrong discussion. Negative acknowledgement will not work correctly if you have a receive queue greater than 0 because the consumer pre-fetches messages.

reconsumeLater is similar, but not the same

is it supposed to maintain order if you use `receiverQueue` of 0? I didn’t find that to be the case in my tests

I don't remember the semantics for a zero queue consumer well enough to say if it'll work

It's a great question. It might be worth asking what the expected semantics are on the dev list.

I think I've generally assumed that negative acknowledgement changes the semantics for order, but that might not be substantiated.

The main question is what constitutes order. Is it that messages are delivered in order the first time or no matter what.

Yeah, I believe that since nacking isn’t actually synchronous (it is tracked in the consumer, then nacked on a timer), it won’t be done soon enough to prevent the next message from arriving out of order

Right, the async nature makes it more complicated. I could imagine a scenario where nack actually empties the queue, disconnects the consumer, and then that would probably retain order (at the cost of messages getting redelivered)

I'm not sure if that design would truly make sense, just thinking out loud

It'd be good to ask though what the expected behavior is for a zero queue consumer. This kind of edge case might not have been defined yet, so the behavior might not be "ideal"

Well, in the docs it explicitly says that nacking with an exclusive consumer can result in out-of-order messages. I was just wondering if there’s a way to prevent that. But I think if we need to do that, we should be using a reader

Sure, but the zero queue consumer could be a special case, and yes, there is always the option to use a reader.

Out of curiosity, what is the intention of a nack when you have an exclusive consumer? If you remove the ack timeout, you can hold on to the message indefinitely. (If you use anything other than a zero queue consumer, a disconnect can lead to message duplication in the receive queue, but with a zero queue consumer, that risk is eliminated.)

We are sharing listener threads across multiple exclusive consumers, and one of the messages is encountering (retryable) errors. In this case, we’d like to re-process that message with a delay, but free up the threads for the other consumers. And ordering is important, we don’t want to receive the next message from the same consumer before we retry this one

Cool, makes sense to me. In this case, are you relying on pulsar for the delay using reconsumeLater or are you doing a normal negative ack?

I’m doing a normal negative ack, with the `negativeAckRedeliveryDelay` setting

Last day to submit! Looking forward to your amazing submissions!

hello, has anything changed in 2.10.x regarding the topic schema upload by the producers/consumers when connecting. We rely that the producer/consumer will upload a schema to the brokers on a topic that has no schema defined. What we saw changed in 2.10.x is that the producer can indeed upload a schema but a consumer can't. Is this a new feature or a bug ?

Hello Ric, Pulsar messages can be produced with header yes..Also, I believe  you could allow consumer to inspect the message header to determine if it needs to skip deserializing the message payload when it's not needed.

thx <@U03BQF7P61Z> for confirming point 1. how about point 2? thx

so you can take a look at Starlight for JMS entry filter to see if could fit your use case

<https://docs.datastax.com/en/streaming/starlight-for-jms/3.2/examples/pulsar-jms-server-side-filters.html>

out of the box you can use  Startlight for JMS EntryFilter that allow you to filter messages on the broker with a simple query language that is evaluated against the header of the messages

Sure, feel free to ping me (I am one of the authors of the server side filtering JMS features) and I have already helped a few production cases with big workloads.

thx <@U9NHC1ETG> u mean this URL? Thx

<https://streamnative.io/videos/pulsar-summit-san-francisco-2022-ecosystem-simplify-pulsar-functions-development-with-sql|https://streamnative.io/videos/pulsar-summit-san-francisco-2022-ecosystem-simplify-pulsar-functions-development-with-sql>

started a GitHub discussion <https://github.com/apache/pulsar/discussions/19805>

What version of pulsar is this?. Do you get this error only when creating a tenant?. Does tenant list work

image.png

I'm trying create goroutine for consumer. It created 3 consumer for main queue and 3 for retry queue. Then I tried to push an item and it stuck the app after first retry (need 3 times to push to dead queue). It will be ok if I only use 1 subscribe. So do you know what is the problem here?

You can modify prometheus configuration to drop the metric.

would i be doing it via the relabel section?

yes, u can use `metric_relabel_configs` to drop the specified metric

thanks! checking the prometheus documentation on the same.

Hello <@U049R495EAX> pls check the scrape_config section to filter metrics being sent..

<https://prometheus.io/docs/prometheus/latest/configuration/configuration/#scrape_config>

```Modify the Prometheus configuration file: In the prometheus.yml configuration file, you can use the metric_relabel_configs or relabel_configs section to define rules for filtering the metrics.```


The problem when increasing the number of partitions is that we need to pre-create the existing subscriptions in the newly added partitions, otherwise there would be a risk of data loss.

While not impossible, doing it across clusters is definitely more challenging.

This line caught my eye in the stacktrace. `Could not connect to bookie: null/compute6,` null doesn’t look right.

Have you enabled rack-awareness previously? If not, perhaps you need to configure the data placement policy of the bookie instances as described <https://pulsar.apache.org/docs/2.11.x/administration-isolation-bookie/#configure-data-placement-policy-on-bookie-instances|here>:

after updating the config following the doc and running
```pulsar-admin bookies set-bookie-rack --bookie compute4:3181 --hostname <http://compute4.pulsar.com:3181|compute4.pulsar.com:3181> --group group1 --rack RegionA/rack1
pulsar-admin bookies set-bookie-rack --bookie compute5:3181 --hostname <http://compute5.pulsar.com:3181|compute5.pulsar.com:3181> --group group1 --rack RegionB/rack1
pulsar-admin bookies set-bookie-rack --bookie compute6:3181 --hostname <http://compute6.pulsar.com:3181|compute6.pulsar.com:3181> --group group1 --rack RegionC/rack1```
I still have the same error.

```Mar 14 08:54:01 compute4 pulsar-broker-start[25846]: 08:54:01.294 [BKClientMetaDataPollScheduler-0] ERROR org.apache.bookkeeper.proto.PerChannelBookieClient - Could not connect to bookie: null/compute5, current state CONNECTING :```

on the bookie side, I get this error
```Mar 14 08:46:33 compute4 pulsar-bookie-start[25576]: 08:46:33.330 [main-EventThread] ERROR org.apache.curator.framework.imps.EnsembleTracker - Invalid config event received: {server.2=172.16.1.75:2888:3888:participant, server.1=172.16.1.43:2888:3888:participant, server.0=172.16.1.5:2888:3888:participant, version=0}```

I started a GitHub discussion <https://github.com/apache/pulsar/discussions/19805>

`Using the Reader interface, the application can store the message id along side with the received data. If the state change from processing the _data_ and the _message id_ associated with the data are updated _atomically_, then the application can make sure that the state transitions triggered by the messages read from the topic are only applied once.`

what mean about "`If the state change from processing the _data_ and the _message id_ associated with the data are updated _atomically_, then the application can make sure that the state transitions triggered by the messages read from the topic are only applied once.`"