what time merge about this version ?

[startjava]

https://github.com/apache/pulsar/blob/4ee346693df8fc8314f94d53b00283f1c6079dc1/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java

now 2.10.1 version no top url api.
thank .

[liangyuanpeng]

~~Seems like the release/2.11.0 plan have not include it but it exist in milestone 2.11.0. ~~

PR: #16817

[liangyuanpeng]

I'm sorry I misunderstood.

[startjava]

what time release 2.11.0 RELEASE version?

…

------------------ 原始邮件 ------------------ 发件人: ***@***.***>; 发送时间: 2022年10月10日(星期一) 上午10:52 收件人: ***@***.***>; 抄送: ***@***.***>; ***@***.***>; 主题: Re: [apache/pulsar] what time merge about this version ? (Discussion #17978) Seems like the release/2.11.0 plan have not include it but it exist in milestone 2.11.0. PR: #16817 /cc @codelipenghui — Reply to this email directly, view it on GitHub, or unsubscribe. You are receiving this because you authored the thread.Message ID: ***@***.***>

[liangyuanpeng]

Do you mean you want to get this message from pulsar-admin?

[liangyuanpeng]

bin/pulsar-admin topics partitioned-stats {topic} or bin/pulsar-admin topics stats {topic}

and you can get this message:

"subscriptions" : {
    {subscriptionsName} : {
      "consumers": [{
            "msgRateOut" : 0.0,
           "msgThroughputOut" : 0.0,
         ...
      }]
 ...

[startjava]

thank you , but i want use java client get "unAckMsgCount" info 。 thank you too .

…

------------------ 原始邮件 ------------------ 发件人: ***@***.***>; 发送时间: 2022年10月10日(星期一) 中午11:11 收件人: ***@***.***>; 抄送: ***@***.***>; ***@***.***>; 主题: Re: [apache/pulsar] what time merge about this version ? (Discussion #17978) bin/pulsar-admin topics partitioned-stats {topic} or bin/pulsar-admin topics stats {topic} and you can get this message: "subscriptions" : { {subscriptionsName} : { "consumers": [{ "msgRateOut" : 0.0, "msgThroughputOut" : 0.0, ... }] ... — Reply to this email directly, view it on GitHub, or unsubscribe. You are receiving this because you authored the thread.Message ID: ***@***.***>

[liangyuanpeng]

pulsarAdmin.topics().getStats("").getSubscriptions().get("").getConsumers().get(index).getUnackedMessages()

[startjava]

pulsarAdmin.topics().getStats("").getSubscriptions().get("").getConsumers().get(index).getUnackedMessages()

producer:

    @RequestMapping("sendMessage2")
    public String sendMessage2() throws IOException, PulsarAdminException {
        for (int i = 1; i <= 5; i++) {
            String messageValue = "消息" + i;
            producer2.send(messageValue);
        }
        return "成功生产2";
    }

consumer:

@Component
public class MyMessageListener2 implements MessageListener {
    @Override
    public void received(Consumer consumer, Message msg) {
        try {
            String messageString = new String(msg.getData());
            if (messageString.equals("消息5")) {
                consumer.acknowledgeCumulative(msg);
                System.out.println("执行了acknowledgeCumulative");
            }
        } catch (PulsarClientException e) {
            throw new RuntimeException(e);
        }
    }
}

getinfo:

    @RequestMapping("getUnAckMsgCount")
    public String getUnAckMsgCount() throws PulsarClientException, PulsarAdminException {
        {
            List<? extends ConsumerStats> consumers = pulsarAdmin.topics().getStats(myTopic1).getSubscriptions().get(myTopic1_subscriptionName).getConsumers();
            for (int i = 0; i < consumers.size(); i++) {
                ConsumerStats consumerStats = consumers.get(i);
                System.out.println(myTopic1 + " getConsumerName=" + consumerStats.getConsumerName() + " getUnackedMessages=" + consumerStats.getUnackedMessages());
            }
        }
        System.out.println();
        {
            List<? extends ConsumerStats> consumers = pulsarAdmin.topics().getStats(myTopic2).getSubscriptions().get(myTopic2_subscriptionName).getConsumers();
            for (int i = 0; i < consumers.size(); i++) {
                ConsumerStats consumerStats = consumers.get(i);
                System.out.println(myTopic2 + " getConsumerName=" + consumerStats.getConsumerName() + " getUnackedMessages=" + consumerStats.getUnackedMessages());
            }
        }
        return "成功获得";
    }
}

result:

myTopic1 getConsumerName=消费者1 getUnackedMessages=0

myTopic2 getConsumerName=消费者2 getUnackedMessages=0

run acknowledge or acknowledgeCumulative method , getUnackedMessages result both 0 .

why ?? thank you !

[startjava]

@liangyuanpeng current version 2.10.1 getUnackedMessages() method availability??

[liangyuanpeng]

You can try to do not ack message and get the getUnackedMessages .

[liangyuanpeng]

Change consumer.acknowledgeCumulative(msg); to consumer.acknowledge(msg); also get what you want.

[startjava]

package com.ghy.www.cumulativeacknowledgement.consumer.listener;


import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageListener;
import org.apache.pulsar.client.api.PulsarClientException;
import org.springframework.stereotype.Component;

@Component
public class MyMessageListener1 implements MessageListener {
    @Override
    public void received(Consumer consumer, Message msg) {
        try {
            String messageString = new String(msg.getData());
            System.out.println(messageString);
            if (messageString.equals("acknowledge消息5")) {
                consumer.acknowledge(msg);
                System.out.println(messageString + "执行了acknowledge");
            }
        } catch (PulsarClientException e) {
            throw new RuntimeException(e);
        }
    }
}

two method both test , both return 0 .
i want implement Cumulative and Not Cumulative Diff 。

i code envirment has error ?

[startjava]

use bottom code print info :

    @RequestMapping("getUnAckMsgCount")
    public String getUnAckMsgCount() throws PulsarClientException, PulsarAdminException {
        {
            List<? extends ConsumerStats> consumers = pulsarAdmin.topics().getStats(myTopic1).getSubscriptions().get(myTopic1_subscriptionName).getConsumers();
            for (int i = 0; i < consumers.size(); i++) {
                ConsumerStats consumerStats = consumers.get(i);
                System.out.println(myTopic1 + " getConsumerName=" + consumerStats.getConsumerName() + " getUnackedMessages=" + consumerStats.getUnackedMessages());
            }
        }
        System.out.println();
        {
            List<? extends ConsumerStats> consumers = pulsarAdmin.topics().getStats(myTopic2).getSubscriptions().get(myTopic2_subscriptionName).getConsumers();
            for (int i = 0; i < consumers.size(); i++) {
                ConsumerStats consumerStats = consumers.get(i);
                System.out.println(myTopic2 + " getConsumerName=" + consumerStats.getConsumerName() + " getUnackedMessages=" + consumerStats.getUnackedMessages());
            }
        }
        return "成功获得";
    }

[startjava]

@liangyuanpeng use which version test ?