You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@pulsar.apache.org by Apache Pulsar Slack <ap...@gmail.com> on 2018/06/09 09:11:04 UTC
Slack digest for #general - 2018-06-09
2018-06-08 09:11:14 UTC - Idan: @Idan uploaded a file: <https://apache-pulsar.slack.com/files/UALJD8929/FB53PNCJ3/-.java|Untitled>
----
2018-06-08 09:12:46 UTC - Idan: so using maxPendingmessages(1) turns the async mode, true?
----
2018-06-08 09:16:42 UTC - Sijie Guo: Overall looks good. However I think the max pending messages is misunderstood. Max pending messages isn’t related to asynchronous or sync. It means how many messages are allowed to be pending on acknowledgments. That says :-
If you set max to 1,
You #sendAsync first, before it’s callback completed, you try to sendAsync again, it would wait until the first callback completed.
----
2018-06-08 09:17:30 UTC - Sijie Guo: For your user case, I don’t think you need to set that value.
----
2018-06-08 09:19:17 UTC - Idan: so how do I “turn-on” the async send? without blocking the producer thread and also send immediately.
----
2018-06-08 09:20:23 UTC - Sijie Guo: CompletableFuture<MessageId> sendAsync(T message);
----
2018-06-08 09:20:34 UTC - Sijie Guo: Use sendAsync
----
2018-06-08 09:20:57 UTC - Idan: @Idan uploaded a file: <https://apache-pulsar.slack.com/files/UALJD8929/FB53V1JEB/Untitled.pl|Untitled>
----
2018-06-08 09:23:00 UTC - Sijie Guo: Actually let me ask a question do you need to wait for the acknowledgments or receipt for producing a message?
----
2018-06-08 09:33:42 UTC - Sijie Guo: @Idan
so a pulsar producer can produce message asynchronously or synchronusly.
when you want to produce synchronously, you can use Producer#send() - <https://github.com/apache/incubator-pulsar/blob/master/pulsar-client/src/main/java/org/apache/pulsar/client/api/Producer.java#L58>
when you want to produce asynchronously, you can use Producer#sendAsync() - <https://github.com/apache/incubator-pulsar/blob/master/pulsar-client/src/main/java/org/apache/pulsar/client/api/Producer.java#L75>
using sendAsync, it will not block the thread calling this method.
Batching or not is more about when pulsar will send the data to network to brokers. It controls how fast a message can be reached to broker and be persistent. when a message is persistent, the callback returned by sendAsync will complete. you can add listeners to the completable future and react to the result.
If you don’t need acknowledgments from sendAsync, you can just ignore the completeable future returned.
----
2018-06-08 09:37:29 UTC - Idan: i dont need to wait but I do need to handle if an error occours
----
2018-06-08 09:41:38 UTC - Sijie Guo: yeah. so just use #sendAsync and register a callback to the completable future returned by it. and you don’t need to distinguish batching or not, just let pulsar default values handle for you.
----
2018-06-08 09:43:22 UTC - Idan: thats cool! thanks
----
2018-06-08 09:57:50 UTC - Idan: @Sijie Guo is it possible also to consume batch messages from the broker? like get 10 messages in one request?
----
2018-06-08 09:58:09 UTC - Idan: was looking at docs. couldnt find references for such behavior
----
2018-06-08 10:04:25 UTC - Sijie Guo: @Idan oh pulsar does that transparently for you. what you can control is a “receiverQueueSize”, what does that mean - it is the max number of message that pulsar client received from brokers and keep in the consumer queue but haven’t consumed by the application.
----
2018-06-08 10:04:58 UTC - Idan: but does the API gives me the ability to get List of messages somehow?
----
2018-06-08 10:06:30 UTC - Sijie Guo: oh. the API doesn’t have a get List of messages api, pulsar dispatches individual message from the list via MessageListener or receive/receiveAsync.
----
2018-06-08 10:07:00 UTC - Sijie Guo: though fundementally pulsar is getting a list of messages, so it should be trivial to add this api if you need that.
----
2018-06-08 10:08:17 UTC - Idan: yes. well.. it’s trivial but not straight forward since we need to create internal-queue to consume messages in batch from your listener
----
2018-06-08 10:08:21 UTC - Idan: and flush it away to the consuming logic
----
2018-06-08 10:13:10 UTC - Sijie Guo: yeah. agreed with current api, it is not straightforward.
I was saying it is “trivial” to add a receive api or a listener api in pulsar to return a list of messages for you.
----
2018-06-08 10:18:28 UTC - Idan: ah well i think it’s essential when someone thinking about consume-batch most reasons coz he need list of messages to batch them later on into diff destinations (e.g sql)
----
2018-06-08 10:23:16 UTC - Sijie Guo: yeah agreed
----
2018-06-08 10:36:05 UTC - Idan: @Sijie Guo via pulsar docs i read that when not ACKING a message a re-try mechanism will re-send this message again. is there a possibility to extract on the message properties (after consuming it) the re-try counter? iam trying to create a DLQ forwarder logic. for that I need to understand how many times message was “re-tried”
----
2018-06-08 14:05:01 UTC - David Asher: Hey anyone luck getting the dashboard displaying information of a standalone pulsar instance? I can't see anything not even the "cluster" name
----
2018-06-08 14:29:58 UTC - Karthik Palanivelu: Hello there, I am using your zookeeper.yaml as sample to create statefulsets within my k8s. I updated only image of my own and deployed it. It got deployed and I modified the yaml to new image based on errors but when I delete it I am getting following message and if I redeploy it is always taking older image. Please help me here as I am new to k8s.
```Error from server (NotFound): error when stopping "/apache-pulsar/deploy/zk.yaml": the server could not find the requested resource```
----
2018-06-08 15:32:28 UTC - Karthik Palanivelu: @Matteo Merli @Sijie Guo I just found that yaml when I delete it, it is not deleting the statefulsets, can you please help?
----
2018-06-08 15:35:58 UTC - Sijie Guo: @Karthikeyan Palanivelu
> It got deployed and I modified the yaml to new image based on errors but when I delete it I am getting following message
so you deploy it first, then modify something in the yaml, and use that modified yaml to delete.
do i understand the sequence correctly?
----
2018-06-08 15:39:55 UTC - Karthik Palanivelu: No @Sijie Guo, I created it and deleted it with same yaml without modification and deletion of statefulsets did not happen. Let me double check
----
2018-06-08 15:42:24 UTC - Sijie Guo: no sure what are the modifications in the yaml. but I guess this might be related to you are using a different yaml (where you might change some names) to delete a cluster that was created before. so here is my suggestions:
- use the original yaml which you used to create the cluster, to delete.
- if it is difficult to find what has changed, you can
``` delete the statefulset by its name
kubectl delete statefulsets <statefulset-name>
```
and delete the corresponding headless service
``` delete the headless service
kubectl delete service <service-name>
```
you can use `kubectl get statefulsets` and `kubectl get service` to find them.
----
2018-06-08 15:44:46 UTC - Sijie Guo: @David Asher
> Hey anyone luck getting the dashboard displaying information of a standalone pulsar instance? I can’t see anything not even the “cluster” name
I haven’t tried the dashboard with standalone. let me try and give you some instructions.
+1 : David Asher
----
2018-06-08 15:45:08 UTC - Idan: @Sijie Guo i was testing the ack mechanisem. and I see some timing issues. is that expected behavior? My scenraio is simple. Iam producing message, consume is successfully but without acking it. my ack timeout is 5000ms. only after 10000ms i actually consume the message again.. that goes on on with constant delay (of 5000ms). shouldnt I get the message right after 5000ms ?
----
2018-06-08 15:46:02 UTC - Idan: if I decrease the timeout into 1000ms i get the message after 4000ms (delay of 3000ms). it never get right on time
----
2018-06-08 15:47:27 UTC - Sijie Guo: @Idan - what kind of settings are you using? e.g. subscription type and how many consumer instances?
----
2018-06-08 15:48:46 UTC - Idan: @Idan uploaded a file: <https://apache-pulsar.slack.com/files/UALJD8929/FB5B19XST/Untitled.pl|Untitled>
----
2018-06-08 15:50:39 UTC - Karthik Palanivelu: @Sijie Guo I did not modify any between creation and deletion but statefulsets are not getting deleted when I use the cmd -
```
kubectl delete -f /apache-pulsar/deploy/zk.yaml
configmap "zookeeper-config" deleted
poddisruptionbudget.policy "zk-budget" deleted
service "zookeeper" deleted
Error from server (NotFound): error when stopping "/apache-pulsar/deploy/zk.yaml": the server could not find the requested resource
```
```
$ kubectl get statefulsets
NAME DESIRED CURRENT AGE
zookeeper 3 3 6m
```
----
2018-06-08 15:50:53 UTC - Sijie Guo: @Idan and you are using 2.0.0 client, right?
----
2018-06-08 15:51:16 UTC - Idan: yes. ill show you my output logs. its constantly getting after 10 seconds (expecte 5 seconds)
----
2018-06-08 15:51:54 UTC - Sijie Guo: @Karthikeyan Palanivelu gotcha. can you try “kubectl delete statefulsets zookeeper” to delete the statefulsets.
----
2018-06-08 15:52:06 UTC - Idan: @Idan uploaded a file: <https://apache-pulsar.slack.com/files/UALJD8929/FB3JLCBRN/Untitled.pl|Untitled>
----
2018-06-08 15:52:43 UTC - Idan: look at the timestamps
----
2018-06-08 15:52:51 UTC - Idan: 10 seconds delay exactly everytime
----
2018-06-08 15:53:35 UTC - Idan: ofcourse that it’s on my local machine so i expect slight latency. but it’s constantly delaying within 5 seconds. abit suspicious
----
2018-06-08 15:54:04 UTC - Sijie Guo: oh i see. one second, let me check something
----
2018-06-08 15:54:06 UTC - Sijie Guo: very quickly
----
2018-06-08 15:54:16 UTC - Idan: sure. btw: i got also the pulsar logs if you need
----
2018-06-08 16:05:28 UTC - Sijie Guo: @Idan are you using MessageListener or calling #receiveAsyc ?
----
2018-06-08 16:05:39 UTC - Idan: i call receiveAsyc
----
2018-06-08 16:06:09 UTC - Idan: sorry
----
2018-06-08 16:06:13 UTC - Idan: this time i used: Message<String> msg = consumer.receive();
----
2018-06-08 16:06:34 UTC - Sijie Guo: is that in a simple loop?
----
2018-06-08 16:06:44 UTC - Idan: sec.. ill pass you the logic
----
2018-06-08 16:06:51 UTC - Sijie Guo: that would be great
----
2018-06-08 16:14:33 UTC - Idan: @Idan uploaded a file: <https://apache-pulsar.slack.com/files/UALJD8929/FB4A835DG/-.pl|Untitled>
----
2018-06-08 16:14:41 UTC - Idan: i tried to remove our internal stuff
----
2018-06-08 16:14:50 UTC - Idan: in the end it left pretty simple loop
----
2018-06-08 16:15:12 UTC - Idan: this is how i send the message
----
2018-06-08 16:15:18 UTC - Idan: @Idan uploaded a file: <https://apache-pulsar.slack.com/files/UALJD8929/FB571437G/-.pl|Untitled>
----
2018-06-08 16:15:59 UTC - Sijie Guo: gotcha
----
2018-06-08 16:17:08 UTC - Idan: now here is another fact: if I disconnect the consumer client and re-connect it
----
2018-06-08 16:17:17 UTC - Idan: it consumes all the non-acked messages right away
----
2018-06-08 16:17:22 UTC - Idan: on the spot
----
2018-06-08 16:18:25 UTC - Idan: i also printed the consumer thread name to make sure iam using only one: Thread[vert.x-sqs-consumer-thread-0,5,main]
----
2018-06-08 16:21:30 UTC - Idan: ok plesase ignore my last comment
----
2018-06-08 16:21:52 UTC - Idan: it doesnt matter. even if I re-connect the consumer client it still delays with 10 seconds btw one consuming to another
----
2018-06-08 16:23:08 UTC - Sijie Guo: I see - so the logic for ack timeout is pretty simple in pulsar, it basically trigger calling redeliver unacked message in the configured ack timeout interleaved. so if you configure 5 seconds timeout, it will ty to ask brokers to deliver unacked messages every 5 seconds. I am trying to see where the additional seconds would come from.
----
2018-06-08 16:23:12 UTC - Sijie Guo: give me a few more minutes
----
2018-06-08 16:23:50 UTC - Idan: no prob.
----
2018-06-08 16:24:01 UTC - Idan: wonder if you were able to simulate this on your side
----
2018-06-08 16:24:31 UTC - Sijie Guo: yeah. good point. let me reproduce it
----
2018-06-08 16:49:20 UTC - Byron: just noticed the 2.0.0-incubating (final) image is not up Docker hub, only rc1. Had anything changed beyond rc1?
----
2018-06-08 16:55:23 UTC - Matteo Merli: @Byron 2.0.0-rc1-incubating was the “official” release. the naming with “rc1" was discussed in community to reflect that, since a lot of changes went into the release, the likelihood of patch release was fairly high.
We have already a 2.0.1-incubating release that is being voted in the dev@ list: <https://lists.apache.org/thread.html/620813d66d43a43680d3afd587e72f0d6f9607f1566aa88ef73d359c@%3Cdev.pulsar.apache.org%3E>
If you have the chance, please take test it out and vote on the thread.
----
2018-06-08 16:55:59 UTC - Matteo Merli: the 2.0.1 release will fix the issues marked at It fixes the following issues: <https://github.com/apache/incubator-pulsar/milestone/14?closed=1>
----
2018-06-08 16:56:26 UTC - Byron: Understood thanks for the explanation
----
2018-06-08 17:00:21 UTC - Sijie Guo: @Idan I got the same behavior as you saw. it seems to be that the netty hashed wheel timer that pulsar client is using timeout a bit late than the configured 5 seconds interval. I am still figuring out why, but if @Matteo Merli you have some insights, please chime in.
----
2018-06-08 17:01:24 UTC - Matteo Merli: (reading the log above..)
----
2018-06-08 17:01:32 UTC - Idan: Alright
----
2018-06-08 17:01:40 UTC - Idan: Waiting for your insights
----
2018-06-08 17:02:10 UTC - Idan: @Sijie Guo btw: is it possible to know how many times specific message was re-delivered?
----
2018-06-08 17:04:28 UTC - Idan: I can do it on consumer side by mapping counters to pulsar's messageID but could be very useful if that state was populated on broke side
----
2018-06-08 17:16:03 UTC - Byron: I am getting a “Can not attach to current VM” error when trying to run standalone.. I’ve seen that before
----
2018-06-08 17:16:36 UTC - Byron: Unrelated but out of curiosity bookkeeper warns `Unable to link C library. Native methods will be disabled.`
----
2018-06-08 17:17:15 UTC - Idan: @Byron i see that warning aswell. But standalone works well
----
2018-06-08 17:18:36 UTC - Byron: Yes, I was just curious what it meant. I know its unrelated to the VM error
----
2018-06-08 17:20:52 UTC - Sijie Guo: bookkeeper has some optimizations on using linux systemcalls with JNA. if it feels to find the c library, it will fall back and disable those optimizations.
----
2018-06-08 17:21:11 UTC - Byron: I see thanks
----
2018-06-08 17:21:50 UTC - Idan: Guys any idea about my question regarding the retry counter?
----
2018-06-08 17:21:57 UTC - Matteo Merli: @Idan the Ack timeout is is not applied on a per message base, rather it gets applied in “bulk”
----
2018-06-08 17:22:43 UTC - Matteo Merli: If you set timeout to 10s, the message will be replayed between 10 to 20 sec
----
2018-06-08 17:23:31 UTC - Idan: Whats the motivation? What shall we do if need to quickly re-try as 1000ms would be 1000-5000 I assume
----
2018-06-08 17:23:55 UTC - Sijie Guo: but I thought netty has the tickDurations and ticksPerWheel to optimize that, no? @Matteo Merli
----
2018-06-08 17:24:35 UTC - Matteo Merli: still there is a lot of overhead in maintaining a timer per message
----
2018-06-08 17:25:49 UTC - Matteo Merli: `UnAckedMessageTracker` has the logic of maintaining 2 sets to catch all timed out messages: <https://github.com/apache/incubator-pulsar/blob/master/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java>
----
2018-06-08 17:39:20 UTC - Byron: I was not running Java 8.. although oddly it worked for me when evaluating 1.22 for the first time.
----
2018-06-08 17:54:54 UTC - Sijie Guo: @Matteo Merli: any suggestions to @Idan’s timeout value then? It doesn’t seem hashed wheel timer can really do finer granularity timeout?
----
2018-06-08 17:56:49 UTC - Matteo Merli: Not with current implementation. The main purpose of ack-timeout, as it was designed, was to act as a helper in case the application fail to process / ack a message. For that, a “precise” timeout was not important, rather just the assurance that message would be replayed, within a reasonable ballpark
----
2018-06-08 17:57:18 UTC - Byron: I am getting a very long stacktrace using the pulsar-client:
```
WARNING: HK2 service reification failed for [org.glassfish.jersey.jaxb.internal.XmlRootElementJaxbProvider$General] with an exception:
MultiException stack 1 of 2
java.lang.NoClassDefFoundError: javax/xml/bind/JAXBException
```
----
2018-06-08 17:57:40 UTC - Matteo Merli: there is an option to manually trigger the redelivery of all unacked messages as well, not sure whether that might be usefult in this context
----
2018-06-08 17:57:42 UTC - Idan: Reasonable within low latency systems is pretty much defined:)
----
2018-06-08 17:58:54 UTC - Idan: So in my context it means i got one shot.. ok ill see how we will be able to live with it. Thanks
----
2018-06-08 17:59:52 UTC - Idan: Could you please respond to that? We gotta know if we shall start implement
----
2018-06-08 18:00:24 UTC - Sijie Guo: @Idan: I don’t think that metric is exposed.
----
2018-06-08 18:00:34 UTC - Sijie Guo: what would expect it look like?
----
2018-06-08 18:01:57 UTC - Idan: My motivation is that: we need to create DLQ's
----
2018-06-08 18:02:19 UTC - Idan: For DLQ we need to define 'exhusted' messages
----
2018-06-08 18:02:38 UTC - Idan: Msg considered exhusted after X retries
----
2018-06-08 18:03:14 UTC - Idan: Therefore if I could get the retry counter on the messape response from the broker
----
2018-06-08 18:03:22 UTC - Idan: We can forward to DLQ
----
2018-06-08 18:03:38 UTC - Idan: It make sense it will be stored at the broker side
----
2018-06-08 18:04:03 UTC - Idan: Its not metric. More like property on the pulsar MessgeImpl
----
2018-06-08 18:04:16 UTC - Idan: That exposed to the app
----
2018-06-08 18:04:19 UTC - Idan: Dig it eh?
----
2018-06-08 18:06:03 UTC - Matteo Merli: I think that would need some support on broker side. It should be easy to add
----
2018-06-08 18:06:44 UTC - Matteo Merli: (at least a best-effort implementation that keeps the delivery counter in-memory in broker)
----
2018-06-08 18:07:17 UTC - Idan: Exactly. I mean the broker already got all the information about the messageId
----
2018-06-08 18:07:26 UTC - Idan: Need to add counter
----
2018-06-08 18:07:38 UTC - Idan: After delivered
----
2018-06-08 18:07:42 UTC - Idan: And inc
----
2018-06-08 18:08:03 UTC - Idan: Guess the key is the messageId eh
----
2018-06-08 18:08:30 UTC - Idan: If i get that the way from here to DLQ impl would be very easy
----
2018-06-08 18:08:51 UTC - Idan: Otherwise we need to maitain counters at all our consumers
----
2018-06-08 18:09:01 UTC - Idan: And in cluster it will be lots of headache
----
2018-06-08 18:09:02 UTC - Matteo Merli: yes, DLQ could be done by application (or inside client library) at that point
----
2018-06-08 18:11:51 UTC - Idan: You think those counters can be impl on pulsar in the near month?
----
2018-06-08 18:12:10 UTC - Matteo Merli: Yes, it should be easy to add
----
2018-06-08 18:26:24 UTC - Karthik Palanivelu: Team, I am using the similar zookeeper.yaml to deploy zk clusters in K8s as mentioned above. My zookeeper is not coming up based on similar config as in that file because of the error `Invalid Config, exiting abnormally`. Root cause is that the below lines as it could not find the other zks. DOMAIN parameter in the script(bin/generate-zookeeper-config.sh conf/zookeeper.conf) = `hostname -d` is not working so I hard coded the domain name as host to get it working but failing. Please help me here.
```
server.1=zk-0.<DOMAIN_NAME>:2888:3888
server.2=zk-1.<DOMAIN_NAME>:2888:3888
server.3=zk-2.<DOMAIN_NAME>:2888:3888
```
$kubectl get pods
```
zookeeper-0 0/1 CrashLoopBackOff 6 6m
zookeeper-1 0/1 Error 6 6m
zookeeper-2 0/1 CrashLoopBackOff 6 6m
```
----
2018-06-08 18:26:36 UTC - Idan: Cool. So we'll hold the dlq impl till we have this. Regarding the retry latency.. ill check the lowest latency we can get via the broker
----
2018-06-08 18:35:54 UTC - Sijie Guo: @Karthikeyan Palanivelu I think it should be something like:
server.1=zookeeper-0:2888:3888
server.2=zookeeper-1:2888:3888
server.3=zookeeper-2:2888:3888
----
2018-06-08 18:48:51 UTC - Karthik Palanivelu: ok, Let me try that. So no need to set the DOMAIN variable?
----
2018-06-08 18:51:06 UTC - Karthik Palanivelu: Another question is how does the MY_ID is generated, I am not finding in Yaml. Am I missing something here?
----
2018-06-08 18:57:09 UTC - Matteo Merli: There is a script included in Docker image: <https://github.com/apache/incubator-pulsar/blob/master/docker/pulsar/scripts/generate-zookeeper-config.sh>
----
2018-06-08 18:59:43 UTC - Karthik Palanivelu: Thanks @Sijie Guo for pointing out to my issue. That fixed it and I see my id is getting generated. Let me move on to Bookie
+1 : Sijie Guo
bananadance : Sijie Guo
----
2018-06-08 18:59:53 UTC - Karthik Palanivelu: Thanks @Matteo Merli
----
2018-06-08 19:00:02 UTC - Karthik Palanivelu: I am using that script
----
2018-06-08 19:00:24 UTC - Karthik Palanivelu: BTW, I removed the DOMAIN, Let me test and Report back
----
2018-06-08 21:16:16 UTC - Ali Ahmed: @Idan Here is a poc for an embedded pulsar test runner
<https://github.com/streamlio/embedded-pulsar>
----
2018-06-08 21:17:44 UTC - Idan: nice, ill put and try it in my unit-testings. thank you
----
2018-06-08 21:21:19 UTC - Ali Ahmed: we probably need randomized directories and ports with a builder pattern and cleaner shutdown method before we can submit to the pulsar repo
----
2018-06-09 00:06:22 UTC - Ali Ahmed: We are considering deploying <https://www.discourse.org/> for the apache pulsar community to help manage the interactions better. It’s available both as open source and hosted solutions.
----
2018-06-09 00:07:47 UTC - Ali Ahmed: How does the community feel about such a deployment ? slack will still be available but long running thread discussion will probably be guided to a discourse forum
----
2018-06-09 00:26:45 UTC - Byron: @Ali Ahmed I agree with that. Much better for documenting and finding recurring questions. Slack is way too chatty and loses history (and is hard to search it even with unlimited history).
----