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/11/08 09:11:02 UTC
Slack digest for #general - 2018-11-08
2018-11-07 10:55:05 UTC - Rodrigo Malacarne: Hello all, is there any difference between these two commands?
`./bin/pulsar-admin topics compact <persistent://my-tenant/my-namespace/my-topic>`
and
`./bin/pulsar compact-topic --topic <persistent://my-tenant/my-namespace/my-topic>`
----
2018-11-07 10:58:53 UTC - Rodrigo Malacarne: Another question... is there any way to automatically create a compacted topic? What I observed is, in order to create a compact topic, I have not only to set the compaction at the namespace level, but also run a manual compaction at least once. Is this procedure correct? Or is there any way, as said, to automatically create a compacted topic?
----
2018-11-07 11:34:25 UTC - jia zhai: Hi @Rodrigo Malacarne Hope this page contains all the information that you need: <http://pulsar.apache.org/docs/en/cookbooks-compaction>
> ./bin/pulsar-admin topics compact & ./bin/pulsar compact-topic
The pulsar-admin tool runs compaction via the Pulsar REST API. To run compaction in its own dedicated process, i.e. not through the REST API, you can use the pulsar compact-topic command.
Besides the manually trigger use your commands, You could also set the auto trigger in namespace level:
```
Tenant administrators can configure a policy for compaction at the namespace level. The policy specifies how large the topic backlog can grow before compaction is triggered.
For example, to trigger compaction when the backlog reaches 100MB:
$ bin/pulsar-admin namespaces set-compaction-threshold \
--threshold 100M my-tenant/my-namespace
Copy
Configuring the compaction threshold on a namespace will apply to all topics within that namespace.
```
----
2018-11-07 11:38:12 UTC - Rodrigo Malacarne: @jia zhai Thank you very mush... I have set the compaction at the namespace level, however the compaction somehow didn't work for the topics I've created inside the namespace. That's exactly my point ...
----
2018-11-07 11:39:37 UTC - jia zhai: Oh, what is the size in your setting
----
2018-11-07 11:39:47 UTC - Rodrigo Malacarne: 1K
----
2018-11-07 11:40:48 UTC - jia zhai: Would you please help provide more information about your whole steps?
----
2018-11-07 11:41:16 UTC - Rodrigo Malacarne: What I did was:
1) `./bin/pulsar-admin namespaces set-compaction-threshold --threshold 1K my-tenant/my-namespace`
After that, topics created inside this namespace were NOT automatically compacted...
Then I had to manually run
2) `./bin/pulsar-admin topics compact <persistent://my-tenant/my-namespace/my-topic>`
----
2018-11-07 11:41:54 UTC - Rodrigo Malacarne: I had to run step 2 for all the topics (currently around 70 topics)
----
2018-11-07 11:42:00 UTC - jia zhai: Pulsar consumers and readers need to be configured to read from compacted topics.
----
2018-11-07 11:42:08 UTC - jia zhai: Consumer<byte[]> compactedTopicConsumer = client.newConsumer()
.topic(“some-compacted-topic”)
.readCompacted(true)
.subscribe();
----
2018-11-07 11:42:22 UTC - Rodrigo Malacarne: All consumers and readers are configured for compaction ...
----
2018-11-07 11:43:39 UTC - Rodrigo Malacarne: After step 1, even though consumers and readers are correctly configured, the compaction doesn't happen somehow.
----
2018-11-07 11:44:05 UTC - Rodrigo Malacarne: Then I have to manually run step 2 for all topics...
----
2018-11-07 11:44:22 UTC - jia zhai: pulsar-admin namespaces get-compaction-threshold tenant/namespace
----
2018-11-07 11:44:33 UTC - jia zhai: Will this command get the right setting value?
----
2018-11-07 11:44:39 UTC - jia zhai: 1K
----
2018-11-07 11:44:44 UTC - jia zhai: as in your setting
----
2018-11-07 11:46:21 UTC - jia zhai: <https://github.com/apache/pulsar/blob/master/tests/integration/src/test/java/org/apache/pulsar/tests/integration/compaction/TestCompaction.java>
----
2018-11-07 11:46:59 UTC - jia zhai: Here is the UT for this feature, It is running in every PR, Seems recently it is not reporting error.
----
2018-11-07 11:47:35 UTC - jia zhai: line-305 contains a auto compaction test case.
----
2018-11-07 11:49:27 UTC - Rodrigo Malacarne: @jia zhai, yes:
----
2018-11-07 11:51:32 UTC - Rodrigo Malacarne: @jia zhai Thank you again ... I'll check the Java example.
----
2018-11-07 11:51:58 UTC - jia zhai: It is strange.:joy:
----
2018-11-07 11:52:39 UTC - Rodrigo Malacarne: @jia zhai I know.... :sleepy:
----
2018-11-07 11:55:21 UTC - Rodrigo Malacarne: My question is exactly that ... how to automatically create compacted topics. According to the docs (the compaction cookbook you linked) the only thing needed to trigger compaction is to set the compaction at the namespace level. And that's not what I've observed...
----
2018-11-07 12:00:02 UTC - jia zhai: seems the only different is you are using 1024, while in the tests, it uses 1.
----
2018-11-07 12:07:36 UTC - Rodrigo Malacarne: @jia zhai, You mean instead of 1024 it should return 1?
----
2018-11-07 12:08:18 UTC - jia zhai: ./bin/pulsar-admin namespaces set-compaction-threshold --threshold 1K my-tenant/my-namespace
----
2018-11-07 12:08:28 UTC - jia zhai: you are setting 1k = 1024
----
2018-11-07 12:08:36 UTC - Rodrigo Malacarne: Exactly ...
----
2018-11-07 12:08:40 UTC - Rodrigo Malacarne: I know that ..
----
2018-11-07 12:08:41 UTC - jia zhai: in the test, it sets 1
----
2018-11-07 12:09:06 UTC - Rodrigo Malacarne: And in this case 1 means "1K" ?
----
2018-11-07 12:09:17 UTC - jia zhai: 1 is easier to trigger than 1K
----
2018-11-07 12:09:37 UTC - jia zhai: 1k == 1024
----
2018-11-07 12:09:39 UTC - Rodrigo Malacarne: sure ...
----
2018-11-07 12:10:00 UTC - Rodrigo Malacarne: But my question is: if I set 1, then Pulsar knows it's one byte?
----
2018-11-07 12:10:12 UTC - Rodrigo Malacarne: Or should I use "1B" for instance?
----
2018-11-07 12:10:13 UTC - jia zhai: yes
----
2018-11-07 12:10:22 UTC - Rodrigo Malacarne: @jia zhai great!
----
2018-11-07 12:10:27 UTC - jia zhai: 1 == 1byte
----
2018-11-07 12:10:34 UTC - jia zhai: 1k == 1024 bytes
----
2018-11-07 12:10:39 UTC - Rodrigo Malacarne: Amazing!!!!
----
2018-11-07 12:10:43 UTC - Rodrigo Malacarne: Will check now !
----
2018-11-07 12:11:02 UTC - Rodrigo Malacarne: :+1:
----
2018-11-07 12:11:11 UTC - jia zhai: hope it will work
----
2018-11-07 12:11:14 UTC - jia zhai: :joy:
----
2018-11-07 12:11:44 UTC - Rodrigo Malacarne: :grin:
----
2018-11-07 12:42:10 UTC - David Tinker: Is it possible to limit the number of un-acknowledged messages Pulsar will deliver to a consumer i.e. stop delivery of new messages until some are acknowledged?
----
2018-11-07 12:57:50 UTC - David Tinker: I have tried setting maxUnackedMessagesPerConsumer=1000 in broker.conf and ConsumerBuilder.receiverQueueSize(1000) and neither seems to have any effect.
----
2018-11-07 13:37:55 UTC - jia zhai: > maxUnackedMessagesPerConsumer
This is the setting
----
2018-11-07 13:41:42 UTC - jia zhai: <https://github.com/apache/pulsar/blob/master/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java#L1058>
@David Tinker FYI. Here is a UT contains the steps to verify this. testConsumerBlockingWithUnAckedMessages
----
2018-11-08 06:18:23 UTC - David Tinker: It would be nice to be able to control message flow from the client but still use MessageListener to be notified when new messages arrive. I know I can use Consumer.receive(..) and receiverQueueSize instead but then I have to poll thousands of consumers at regular intervals. Not as nice as using MessageListener. What about adding Consumer.pause(boolean) to suspend sending flow permits to the broker? I think this could be easily implemented in ConsumerImpl.increaseAvailablePermits(). I could do a PR?
----
2018-11-08 06:20:21 UTC - Sijie Guo: Sounds an interesting idea. You are welcome to send a PR. We can discuss the approaches in a PR @David Tinker
----
2018-11-08 06:20:49 UTC - David Tinker: Will do. Tx.
----
2018-11-08 06:23:13 UTC - Matteo Merli: > It would be nice to be able to control message flow from the client but still use MessageListener to be notified when new messages arrive
The listener has the same underlying implementation used for receive(). The idea is that the flow control is tied to how fast application can call receive(). For listener, it’s based on how fast the thread is able to process messages.
----
2018-11-08 06:25:12 UTC - Matteo Merli: regarding pausing the listener for a specific consumer.. It was actually implemented in C++ client and not in Java :slightly_smiling_face: <http://pulsar.apache.org/api/cpp/classpulsar_1_1_consumer.html>
----
2018-11-08 06:25:32 UTC - Matteo Merli: `pauseMessageListener()` and `resumeMessageListener()`
----
2018-11-08 06:29:25 UTC - David Tinker: I see. So adding the same methods to Java Consumer would be the way to go? I don't want to block in MessageListener.receive(..) as that will impact other topics.
----
2018-11-08 06:30:51 UTC - Matteo Merli: Ok, then it would make sense to have the same option in Java lib as well
----