2018-10-31 11:40:07 UTC - Ganga Lakshmanasamy: Is it possible to create producers dynamically in streamlio? We have a program which need to create producers when a user adds up an account. Is it safe to have multiple asynchronous threads running in the producer? We are trying to create it using java 8 ---- 2018-10-31 13:40:18 UTC - David Tinker: How can I change the --zookeeper and --configuration-store previously set with ./pulsar/bin/pulsar initialize-cluster-metadata? My zookeeper is now binding on a different network interface. Tx. ---- 2018-10-31 15:32:11 UTC - Elena Poughia: @Elena Poughia has joined the channel ---- 2018-10-31 16:07:16 UTC - Matteo Merli: @David Tinker When passing the zk and config store `pulsar initialize-cluster-metadata` these are only used to connect to ZK and initialize some metadata. That metadata doesn’t contain the ZK connection string.
So if you move you ZK cluster, you just need to update the connection string in `broker.conf` and `bookkeeper.conf` slightly_smiling_face : David Tinker ---- 2018-10-31 16:08:43 UTC - Matteo Merli: @Ganga Lakshmanasamy Yes, in Pulsar (:wink:) you can add as many producers as you want to a topic. All the objects in the API (client, producers, consumers) are thread safe. ---- 2018-10-31 16:27:55 UTC - Ganga Lakshmanasamy: ok great!! Thanks for the info ---- 2018-10-31 16:28:13 UTC - Ganga Lakshmanasamy: @Matteo Merli Is there a GUI available to configure topic? ---- 2018-10-31 16:57:56 UTC - Ryan Samo: Thanks for the tip @Sanjeev Kulkarni . So how would you go about configuring the bookkeepers to be rack and data center aware so that when a client rights to a broker, the data is replicated on both sides and separate racks? Do you have to use geo-replication to pull this off or can you do a stretch cluster and also pull this off? Thanks! ---- 2018-10-31 17:06:41 UTC - Sanjeev Kulkarni: Have you browsed these? <https://bookkeeper.apache.org/distributedlog/docs/0.5.0/user_guide/implementation/storage.html#rackaware-and-regionaware> ---- 2018-10-31 17:07:05 UTC - Sanjeev Kulkarni: you dont need geo-replication, since this is effectively one cluster ---- 2018-10-31 17:36:38 UTC - Nick Marchessault: As a consumer, what does it mean to acknowledge a message after synchronously receiving one? If you dont acknowledge a message and it goes "unacked", what does the broker do with it? Will it be re-consumed? I have a java consumer that is sychronously receiving messages from 13 different topics, and there are over 800 unacknowledge messages in my subscription. I tried looking through the documentation to see what the ack does but couldnt find anything that goes into much detail. ---- 2018-10-31 17:39:14 UTC - Grant Wu: Does <https://pulsar.apache.org/docs/en/concepts-messaging/#acknowledgement> help? ---- 2018-10-31 17:39:26 UTC - Grant Wu: <https://pulsar.apache.org/docs/en/concepts-architecture-overview/#persistent-storage> and this ---- 2018-10-31 17:39:37 UTC - Grant Wu: I guess it doesn’t really go into “what happens when messages go unacked” ---- 2018-10-31 17:39:43 UTC - Grant Wu: I believe the behavior is that they get redelivered eventually ---- 2018-10-31 17:40:40 UTC - Ryan Samo: I have not, let me look further to understand how it works. Thanks for the info! ---- 2018-10-31 17:49:29 UTC - Sanjeev Kulkarni: If a message doesn’t get acked within a certain time limit, it will get redelivered by the broker ---- 2018-10-31 17:52:30 UTC - Nick Marchessault: Ok thanks! ---- 2018-10-31 18:13:28 UTC - Alex Mault: Hey all! I notice as part of the protocol, I have the option to ack messages with something like a response code. (Wether it be an error, or other) ---- 2018-10-31 18:14:05 UTC - Alex Mault: Is anyone currently using this functionality to mark events as attempted, but failed, and then have a system to re-queue them? ---- 2018-10-31 18:16:28 UTC - Alex Mault: Looking at the proto, I’m talking about `commandAck`’s `enum validationError` ---- 2018-10-31 18:17:06 UTC - Alex Mault: If my client library fails in one of the ways that is listed as a validation error, do I ack the message with the error and then pulsar tries to retransmit? ---- 2018-10-31 18:17:33 UTC - Matteo Merli: > If a message doesn’t get acked within a certain time limit, it will get redelivered by the broker @Nick Marchessault that’s not the default behavior though. By default messages are only replayed when a consumer gets disconnected. To enable redelivery on timeout: <http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ConsumerBuilder.html#ackTimeout-long-java.util.concurrent.TimeUnit-> ---- 2018-10-31 18:18:20 UTC - Grant Wu: Did you mean to @ sanjeev? ---- 2018-10-31 18:18:30 UTC - Matteo Merli: @Alex Mault Is used to track messages that are failing the CRC or decompression ---- 2018-10-31 18:20:03 UTC - Alex Mault: @Matteo Merli Yea, sorry about that - I realize I wasn’t clear on the fact that I am helping write a pulsar client in rust (communicating with the binary protocol). I am trying to figure out the proper way to handle packets that do fail CRC, decompression or decryption. Should they be ack’d with the validation error, or should they simply be ignored and the broker will resend them? party-parrot : Grant Wu, Sijie Guo clap : Grant Wu, Sijie Guo pray : Grant Wu, Sijie Guo, Karthik Ramasamy fire : Grant Wu, Sijie Guo, Ali Ahmed, Karthik Ramasamy ---- 2018-10-31 18:21:27 UTC - Nick Marchessault: So is there a default ack time out? or are you saying you have to set one, and then if a message doesnt get acknowledged, it will then get replayed? am I understanding that correctly? ---- 2018-10-31 18:23:57 UTC - Matteo Merli: Oh nice! ---- 2018-10-31 18:24:49 UTC - Matteo Merli: Yes, the behavior is to log locally in client and ack with error to broker. If messages cannot be processed for these reasons there’s no way we can recover from there ---- 2018-10-31 18:25:54 UTC - Alex Mault: On a seperate topic, we noticed an error in the `Development` -> `binary protocol` docs page in the `Consumer` diagram. It shows a `subscribe(...)` followed by a `Flow(...)`, two messages sent and ack’d, then connection close. This image doesn’t make it clear that you must `Flow(...)` again to tell the broker to send more messages. That is, the client, not the broker is responsible for keeping the client queue full. (We initially thought that `Flow(1000)` means the broker would send up to 1000, then send one additional for each additional ack) The text below does describe it well `A Flow command gives additional permits to send messages...` +1 : dba ---- 2018-10-31 18:27:15 UTC - Grant Wu: I am curious, who is “we” here :stuck_out_tongue: ---- 2018-10-31 18:28:36 UTC - Matteo Merli: I see, yes it’s just a small example in the diagram ---- 2018-10-31 18:29:38 UTC - Matteo Merli: The purpose of flow mechanism is to have broker to dispatch as fast as a consumer application calls `consumer.receive()` ---- 2018-10-31 18:30:07 UTC - Matteo Merli: And the flow is decoupled from the acknowledgments ---- 2018-10-31 18:30:47 UTC - Matteo Merli: Are you say the diagram should be expanded with a more realistic interaction? ---- 2018-10-31 19:00:11 UTC - Alex Mault: @Matteo Merli yes, I was hoping to change it to an example of `Flow(2)`, msg, msg, ack, ack, `Flow(2)` msg…. close ---- 2018-10-31 19:00:24 UTC - Alex Mault: or even `Flow(1)` ---- 2018-10-31 19:00:36 UTC - Alex Mault: just to more clearly show the binary protocol ---- 2018-10-31 19:01:51 UTC - Alex Mault: @Grant Wu “We” is Wyyerd Group. +1 : Ali Ahmed ---- 2018-10-31 19:02:05 UTC - Alex Mault: You can take a look at the very beginnings of the library at <https://github.com/wyyerd/pulsar-rs> ---- 2018-10-31 19:02:36 UTC - Grant Wu: Out of curiosity, did you consider the “wrap the C++ library” route? ---- 2018-10-31 19:02:42 UTC - Grant Wu: And then maybe add a Futures layer on top? ---- 2018-10-31 19:02:50 UTC - Grant Wu: Note: I don’t actually know Rust. ---- 2018-10-31 19:03:08 UTC - Alex Mault: It currently does simple producer and subscriber multiplexed into a client. ---- 2018-10-31 19:04:59 UTC - Alex Mault: @Grant Wu We did consider the FFI method at first and chose to do it rust-native for build simplicity and to control the performance tightly. Also code safety is critical to us and something we can nearly guarantee with rust. (baring logic errors ofc :slightly_smiling_face: ) ---- 2018-10-31 19:05:24 UTC - Ryan Samo: Cool so the distributed log portion supports the functionality, that’s perfect. I guess from here, I am having trouble trying to relate this config to the actual Pulsar config. Would these flags be set in the bookkeeper.conf? I’m trying to translate how to make Pulsar behave in that way. ---- 2018-10-31 19:05:37 UTC - Alex Mault: @Colin Stearns Is the Wyyerd engineer that built this first round library in a couple days as a proof of concept :slightly_smiling_face: tada : Matteo Merli, Sijie Guo, Karthik Ramasamy ---- 2018-10-31 19:12:55 UTC - Ali Ahmed: @Ganga Lakshmanasamy the pulsar admin gui can monitor things but not configure. What are you trying to do ? ---- 2018-11-01 01:16:50 UTC - Karthik Ramasamy: @Karthikeyan Palanivelu - who has experience in setting this. Please describe how you setup the geo replication cluster? ---- 2018-11-01 01:17:26 UTC - Sanjeev Kulkarni: @Matteo Merli might also add more color ---- 2018-11-01 03:36:40 UTC - pulsar: @pulsar has joined the channel ---- 2018-11-01 04:29:15 UTC - jnstommy: @jnstommy has joined the channel ---- 2018-11-01 07:19:53 UTC - dba: @Matteo Merli Just wanted to make sure you saw this. ---- 2018-11-01 07:25:57 UTC - hj: In PulsarDecoder.java, I have some questions <https://github.com/apache/pulsar/blob/master/pulsar-common/src/main/java/org/apache/pulsar/common/api/PulsarDecoder.java> ```code // Q1: cmdSize ? int cmdSize = (int) buffer.readUnsignedInt(); int writerIndex = buffer.writerIndex(); // Q2: why change writerIndex here and plus cmdSize buffer.writerIndex(buffer.readerIndex() + cmdSize); ByteBufCodedInputStream cmdInputStream = ByteBufCodedInputStream.get(buffer); cmdBuilder = BaseCommand.newBuilder(); cmd = cmdBuilder.mergeFrom(cmdInputStream, null).build(); // Q3: why reset writerIndex here ? buffer.writerIndex(writerIndex); ``` ---- 2018-11-01 07:33:45 UTC - Ganga Lakshmanasamy: @Ali Ahmed Heard that the messages that are put into topics can run through different rules like duplication. We basically need duplicate values to be removed before the consumer picks up the message for further processing. Also we need to query and check the messages that are available in topic. ---- 2018-11-01 07:50:01 UTC - Ali Ahmed: @Ganga Lakshmanasamy pulsar can guarantee no message duplication if used appropriately so there is nothing to change there. In terms of querying data, pulsar can be queried through sql. <https://pulsar.apache.org/docs/fr/next/sql-overview/> ----
