You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "Gleiphir2769 (via GitHub)" <gi...@apache.org> on 2023/03/17 09:39:18 UTC
[GitHub] [pulsar-client-go] Gleiphir2769 opened a new pull request, #991: [feat] Support consumer client memory limit
Gleiphir2769 opened a new pull request, #991:
URL: https://github.com/apache/pulsar-client-go/pull/991
Master Issue: #927
### Motivation
This is part of the work for [PIP 74](https://github.com/apache/pulsar/wiki/PIP-74%3A-Pulsar-client-memory-limits) in go client.
More details in https://github.com/apache/pulsar-client-go/issues/927 .
### Modifications
- Consumer adds memory limit relate logic.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
### Documentation
- Does this pull request introduce a new feature? (**yes** / no)
- If yes, how is the feature documented? (not applicable / docs / **GoDocs** / not documented)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] Gleiphir2769 commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "Gleiphir2769 (via GitHub)" <gi...@apache.org>.
Gleiphir2769 commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1145775394
##########
pulsar/internal/memory_limit_controller.go:
##########
@@ -31,18 +31,41 @@ type MemoryLimitController interface {
CurrentUsage() int64
CurrentUsagePercent() float64
IsMemoryLimited() bool
+ RegisterTrigger(trigger func())
}
type memoryLimitController struct {
limit int64
chCond *chCond
currentUsage int64
+
+ triggers []*thresholdTrigger
+ // valid range is (0, 1.0)
+ triggerThreshold float64
+}
+
+type thresholdTrigger struct {
+ triggerFunc func()
+ triggerRunning int32
+}
+
+func (t *thresholdTrigger) canTryRunning() bool {
+ return atomic.CompareAndSwapInt32(&t.triggerRunning, 0, 1)
+}
+
+func (t *thresholdTrigger) setRunning(isRunning bool) {
+ if isRunning {
+ atomic.StoreInt32(&t.triggerRunning, 1)
+ }
+ atomic.StoreInt32(&t.triggerRunning, 0)
Review Comment:
Sorry for missing it.
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] Gleiphir2769 commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "Gleiphir2769 (via GitHub)" <gi...@apache.org>.
Gleiphir2769 commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1145845933
##########
pulsar/consumer_test.go:
##########
@@ -4113,3 +4113,204 @@ func TestConsumerBatchIndexAckDisabled(t *testing.T) {
assert.Nil(t, err)
assert.Equal(t, []byte("done"), message.Payload())
}
+
+func TestConsumerMemoryLimit(t *testing.T) {
+ // Create client 1 without memory limit
+ cli1, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ })
+
+ assert.Nil(t, err)
+ defer cli1.Close()
+
+ // Create client 1 with memory limit
+ cli2, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ MemoryLimitBytes: 10 * 1024,
+ })
+
+ assert.Nil(t, err)
+ defer cli2.Close()
+
+ topic := newTopicName()
+
+ // Use client 1 to create producer p1
+ p1, err := cli1.CreateProducer(ProducerOptions{
+ Topic: topic,
+ DisableBatching: false,
+ })
+ assert.Nil(t, err)
+ defer p1.Close()
+
+ // Use mem-limited client 2 to create consumer c1
+ c1, err := cli2.Subscribe(ConsumerOptions{
+ Topic: topic,
+ SubscriptionName: "my-sub-1",
+ Type: Exclusive,
+ EnableAutoScaledReceiverQueueSize: true,
+ })
+ assert.Nil(t, err)
+ defer c1.Close()
+ pc1 := c1.(*consumer).consumers[0]
+
+ // Fill up the messageCh of c1
+ for i := 0; i < 10; i++ {
+ p1.SendAsync(
+ context.Background(),
+ &ProducerMessage{Payload: createTestMessagePayload(1)},
+ func(id MessageID, producerMessage *ProducerMessage, err error) {
+ },
+ )
+ }
+
+ retryAssert(t, 5, 200, func() {}, func(t assert.TestingT) bool {
+ return assert.Equal(t, 10, len(pc1.messageCh))
+ })
+
+ // Get current receiver queue size of c1
+ prevQueueSize := pc1.currentQueueSize.Load()
+
+ // Make the client 1 exceed the memory limit
+ _, err = p1.Send(context.Background(), &ProducerMessage{
+ Payload: createTestMessagePayload(10*1024 + 1),
+ })
+ assert.NoError(t, err)
+
+ // c1 should shrink it's receiver queue size
+ retryAssert(t, 5, 200, func() {}, func(t assert.TestingT) bool {
+ return assert.Equal(t, prevQueueSize/2, pc1.currentQueueSize.Load())
+ })
+
+ // Use mem-limited client 2 to create consumer c2
+ c2, err := cli2.Subscribe(ConsumerOptions{
+ Topic: topic,
+ SubscriptionName: "my-sub-2",
+ Type: Exclusive,
+ SubscriptionInitialPosition: SubscriptionPositionEarliest,
+ EnableAutoScaledReceiverQueueSize: true,
+ })
+ assert.Nil(t, err)
+ defer c2.Close()
+ pc2 := c2.(*consumer).consumers[0]
+
+ // Try to induce c2 receiver queue size expansion
+ for i := 0; i < 10; i++ {
+ p1.SendAsync(
+ context.Background(),
+ &ProducerMessage{Payload: createTestMessagePayload(1)},
+ func(id MessageID, producerMessage *ProducerMessage, err error) {
+ },
+ )
+ }
+
+ retryAssert(t, 5, 200, func() {}, func(t assert.TestingT) bool {
+ return assert.Equal(t, 10, len(pc1.messageCh))
+ })
+
+ // c2 receiver queue size should not expansion because client 1 has exceeded the memory limit
+ assert.Equal(t, 1, int(pc2.currentQueueSize.Load()))
+
+ // Use mem-limited client 2 to create producer p2
+ p2, err := cli2.CreateProducer(ProducerOptions{
+ Topic: topic,
+ DisableBatching: false,
+ DisableBlockIfQueueFull: true,
+ })
+ assert.Nil(t, err)
+ defer p2.Close()
+
+ _, err = p2.Send(context.Background(), &ProducerMessage{
+ Payload: createTestMessagePayload(1),
+ })
+ // Producer can't send message
+ assert.Equal(t, true, errors.Is(err, errMemoryBufferIsFull))
+}
+
+func TestMultiConsumerMemoryLimit(t *testing.T) {
+ // Create client 1 without memory limit
+ cli1, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ })
+
+ assert.Nil(t, err)
+ defer cli1.Close()
+
+ // Create client 1 with memory limit
+ cli2, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ MemoryLimitBytes: 10 * 1024,
+ })
+
+ assert.Nil(t, err)
+ defer cli2.Close()
+
+ topic := newTopicName()
+
+ // Use client 1 to create producer p1
+ p1, err := cli1.CreateProducer(ProducerOptions{
+ Topic: topic,
+ DisableBatching: false,
+ })
+ assert.Nil(t, err)
+ defer p1.Close()
+
+ // Use mem-limited client 2 to create consumer c1
+ c1, err := cli2.Subscribe(ConsumerOptions{
+ Topic: topic,
+ SubscriptionName: "my-sub-1",
+ Type: Exclusive,
+ EnableAutoScaledReceiverQueueSize: true,
+ })
+ assert.Nil(t, err)
+ defer c1.Close()
+ pc1 := c1.(*consumer).consumers[0]
+
+ // Use mem-limited client 2 to create consumer c1
+ c2, err := cli2.Subscribe(ConsumerOptions{
+ Topic: topic,
+ SubscriptionName: "my-sub-2",
+ Type: Exclusive,
+ EnableAutoScaledReceiverQueueSize: true,
+ })
+ assert.Nil(t, err)
+ defer c2.Close()
+ pc2 := c1.(*consumer).consumers[0]
Review Comment:
Done.
```
pc2 := c2.(*consumer).consumers[0]
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] shibd merged pull request #991: [feat] Support consumer client memory limit
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd merged PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] shibd commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1145591286
##########
pulsar/consumer_partition.go:
##########
@@ -1760,6 +1770,21 @@ func (pc *partitionConsumer) markScaleIfNeed() {
}
}
+func (pc *partitionConsumer) shrinkReceiverQueueSize() {
+ if !pc.options.autoReceiverQueueSize {
+ return
+ }
+
+ oldSize := pc.currentQueueSize.Load()
Review Comment:
oh, I see. Already protected in `memory_limit_controller`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] shibd commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1145767777
##########
pulsar/internal/memory_limit_controller.go:
##########
@@ -31,18 +31,41 @@ type MemoryLimitController interface {
CurrentUsage() int64
CurrentUsagePercent() float64
IsMemoryLimited() bool
+ RegisterTrigger(trigger func())
}
type memoryLimitController struct {
limit int64
chCond *chCond
currentUsage int64
+
+ triggers []*thresholdTrigger
+ // valid range is (0, 1.0)
+ triggerThreshold float64
+}
+
+type thresholdTrigger struct {
+ triggerFunc func()
+ triggerRunning int32
+}
+
+func (t *thresholdTrigger) canTryRunning() bool {
+ return atomic.CompareAndSwapInt32(&t.triggerRunning, 0, 1)
+}
+
+func (t *thresholdTrigger) setRunning(isRunning bool) {
+ if isRunning {
+ atomic.StoreInt32(&t.triggerRunning, 1)
+ }
+ atomic.StoreInt32(&t.triggerRunning, 0)
Review Comment:
```suggestion
} else {
atomic.StoreInt32(&t.triggerRunning, 0)
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] shibd commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1145592045
##########
pulsar/consumer_partition.go:
##########
@@ -333,6 +335,7 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
}
if pc.options.autoReceiverQueueSize {
pc.currentQueueSize.Store(initialReceiverQueueSize)
+ pc.client.memLimit.RegisterTrigger(receiverQueueShrinkMemThreshold, pc.shrinkReceiverQueueSize)
Review Comment:
This threshold should be managed by the `memory_limit_controller`, and we don't need to set it when we register trigger.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] Gleiphir2769 commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "Gleiphir2769 (via GitHub)" <gi...@apache.org>.
Gleiphir2769 commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1141799619
##########
pulsar/consumer_partition.go:
##########
@@ -1752,6 +1762,37 @@ func (pc *partitionConsumer) markScaleIfNeed() {
}
}
+func (pc *partitionConsumer) reserveMemory(size int64) {
+ pc.client.memLimit.ForceReserveMemory(size)
+ if pc.client.memLimit.CurrentUsagePercent() >= receiverQueueShrinkMemThreshold {
Review Comment:
The reasons here are not the same as those implemented by the Java client:
1. I don’t think it matters whether this is triggered by `memory limiter` as long as the `shrink` is not triggered repeated.
3. If we want to implement it like the Java client, we need to modify the `memory limiter` to allow it to support a trigger, but I feel like this in itself is an unnecessary code coupling.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] Anonymitaet commented on pull request #991: [feat] Support consumer client memory limit
Posted by "Anonymitaet (via GitHub)" <gi...@apache.org>.
Anonymitaet commented on PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#issuecomment-1512391159
Hi @shibd, thanks for introducing this great feature!
========================================
As we discussed just now, for the doc side:
### ✅ Pulsar feature matrix
I've ticked the boxes of ["Memory Limits" for Go](https://docs.google.com/spreadsheets/d/1YHYTkIXR8-Ql103u-IMI18TXLlGStK8uJjDsOOA0T20/edit#gid=1784579914&range=22:22).
### ✅ Pulsar client docs
#### 🌈 Collobration way
To add docs for it, feel free to choose your preferred method (either way works for me):
- You provide tech inputs to me. I'll write docs based on them and request your review. Or
- You submit a PR to the pulsar-site repo directly and request my review.
#### 🌈 TO-DO tasks
1. Create a tab for Go and a topic named "Set memory limits" in [work with clients](https://pulsar.apache.org/docs/next/client-libraries-clients/). It should explain:
- What is it?
- How to use it? (e.g., any considerations? Sample code, etc)
<img width="1809" alt="image" src="https://user-images.githubusercontent.com/50226895/232664030-f39d313a-ea80-4479-b912-4c52dadb2e65.png">
2. This doc is only added to `NEXT` (submit a PR [here](https://github.com/apache/pulsar-site/edit/main/docs/client-libraries-clients.md)) since this feature is server-independent.
========================================
Feel free to reach out to me if you have questions. TIA!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] RobertIndie commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1145831375
##########
pulsar/consumer_test.go:
##########
@@ -4113,3 +4113,204 @@ func TestConsumerBatchIndexAckDisabled(t *testing.T) {
assert.Nil(t, err)
assert.Equal(t, []byte("done"), message.Payload())
}
+
+func TestConsumerMemoryLimit(t *testing.T) {
+ // Create client 1 without memory limit
+ cli1, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ })
+
+ assert.Nil(t, err)
+ defer cli1.Close()
+
+ // Create client 1 with memory limit
+ cli2, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ MemoryLimitBytes: 10 * 1024,
+ })
+
+ assert.Nil(t, err)
+ defer cli2.Close()
+
+ topic := newTopicName()
+
+ // Use client 1 to create producer p1
+ p1, err := cli1.CreateProducer(ProducerOptions{
+ Topic: topic,
+ DisableBatching: false,
+ })
+ assert.Nil(t, err)
+ defer p1.Close()
+
+ // Use mem-limited client 2 to create consumer c1
+ c1, err := cli2.Subscribe(ConsumerOptions{
+ Topic: topic,
+ SubscriptionName: "my-sub-1",
+ Type: Exclusive,
+ EnableAutoScaledReceiverQueueSize: true,
+ })
+ assert.Nil(t, err)
+ defer c1.Close()
+ pc1 := c1.(*consumer).consumers[0]
+
+ // Fill up the messageCh of c1
+ for i := 0; i < 10; i++ {
+ p1.SendAsync(
+ context.Background(),
+ &ProducerMessage{Payload: createTestMessagePayload(1)},
+ func(id MessageID, producerMessage *ProducerMessage, err error) {
+ },
+ )
+ }
+
+ retryAssert(t, 5, 200, func() {}, func(t assert.TestingT) bool {
+ return assert.Equal(t, 10, len(pc1.messageCh))
+ })
+
+ // Get current receiver queue size of c1
+ prevQueueSize := pc1.currentQueueSize.Load()
+
+ // Make the client 1 exceed the memory limit
+ _, err = p1.Send(context.Background(), &ProducerMessage{
+ Payload: createTestMessagePayload(10*1024 + 1),
+ })
+ assert.NoError(t, err)
+
+ // c1 should shrink it's receiver queue size
+ retryAssert(t, 5, 200, func() {}, func(t assert.TestingT) bool {
+ return assert.Equal(t, prevQueueSize/2, pc1.currentQueueSize.Load())
+ })
+
+ // Use mem-limited client 2 to create consumer c2
+ c2, err := cli2.Subscribe(ConsumerOptions{
+ Topic: topic,
+ SubscriptionName: "my-sub-2",
+ Type: Exclusive,
+ SubscriptionInitialPosition: SubscriptionPositionEarliest,
+ EnableAutoScaledReceiverQueueSize: true,
+ })
+ assert.Nil(t, err)
+ defer c2.Close()
+ pc2 := c2.(*consumer).consumers[0]
+
+ // Try to induce c2 receiver queue size expansion
+ for i := 0; i < 10; i++ {
+ p1.SendAsync(
+ context.Background(),
+ &ProducerMessage{Payload: createTestMessagePayload(1)},
+ func(id MessageID, producerMessage *ProducerMessage, err error) {
+ },
+ )
+ }
+
+ retryAssert(t, 5, 200, func() {}, func(t assert.TestingT) bool {
+ return assert.Equal(t, 10, len(pc1.messageCh))
+ })
+
+ // c2 receiver queue size should not expansion because client 1 has exceeded the memory limit
+ assert.Equal(t, 1, int(pc2.currentQueueSize.Load()))
+
+ // Use mem-limited client 2 to create producer p2
+ p2, err := cli2.CreateProducer(ProducerOptions{
+ Topic: topic,
+ DisableBatching: false,
+ DisableBlockIfQueueFull: true,
+ })
+ assert.Nil(t, err)
+ defer p2.Close()
+
+ _, err = p2.Send(context.Background(), &ProducerMessage{
+ Payload: createTestMessagePayload(1),
+ })
+ // Producer can't send message
+ assert.Equal(t, true, errors.Is(err, errMemoryBufferIsFull))
+}
+
+func TestMultiConsumerMemoryLimit(t *testing.T) {
+ // Create client 1 without memory limit
+ cli1, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ })
+
+ assert.Nil(t, err)
+ defer cli1.Close()
+
+ // Create client 1 with memory limit
+ cli2, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ MemoryLimitBytes: 10 * 1024,
+ })
+
+ assert.Nil(t, err)
+ defer cli2.Close()
+
+ topic := newTopicName()
+
+ // Use client 1 to create producer p1
+ p1, err := cli1.CreateProducer(ProducerOptions{
+ Topic: topic,
+ DisableBatching: false,
+ })
+ assert.Nil(t, err)
+ defer p1.Close()
+
+ // Use mem-limited client 2 to create consumer c1
+ c1, err := cli2.Subscribe(ConsumerOptions{
+ Topic: topic,
+ SubscriptionName: "my-sub-1",
+ Type: Exclusive,
+ EnableAutoScaledReceiverQueueSize: true,
+ })
+ assert.Nil(t, err)
+ defer c1.Close()
+ pc1 := c1.(*consumer).consumers[0]
+
+ // Use mem-limited client 2 to create consumer c1
+ c2, err := cli2.Subscribe(ConsumerOptions{
+ Topic: topic,
+ SubscriptionName: "my-sub-2",
+ Type: Exclusive,
+ EnableAutoScaledReceiverQueueSize: true,
+ })
+ assert.Nil(t, err)
+ defer c2.Close()
+ pc2 := c1.(*consumer).consumers[0]
Review Comment:
```suggestion
pc2 := c1.(*consumer).consumers[1]
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] shibd commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1141767660
##########
pulsar/consumer_partition.go:
##########
@@ -1752,6 +1762,37 @@ func (pc *partitionConsumer) markScaleIfNeed() {
}
}
+func (pc *partitionConsumer) reserveMemory(size int64) {
+ pc.client.memLimit.ForceReserveMemory(size)
+ if pc.client.memLimit.CurrentUsagePercent() >= receiverQueueShrinkMemThreshold {
Review Comment:
This shrinking strategy is different from java client implementation.
The java client implementation:
Will try to trigger shrinking memory after each `ReserveMemory`, this means that the producer will also trigger the consumer to reduce the queue when sending data.
https://github.com/shibd/pulsar/blob/82237d3684fe506bcb6426b3b23f413422e6e4fb/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MemoryLimitController.java#L65
Your implementation will cause the following scenario where producers cannot produce data.
- One client instance has `1` producer instance and `3` consumer.
- If `3` consumers run out of memory, the producer's production data will block.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] shibd commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1142204396
##########
pulsar/consumer_partition.go:
##########
@@ -1752,6 +1762,37 @@ func (pc *partitionConsumer) markScaleIfNeed() {
}
}
+func (pc *partitionConsumer) reserveMemory(size int64) {
+ pc.client.memLimit.ForceReserveMemory(size)
+ if pc.client.memLimit.CurrentUsagePercent() >= receiverQueueShrinkMemThreshold {
Review Comment:
> OK, I see. But even in Java client, reduceReceiverQueue() can not free memory immediately. It just controls next flow to prefetch fewer messages.
> It just controls next flow to prefetch fewer messages.
Yes, We need these inactive consumers to reduce the size of the pull flow so that it can give memory to more active consumers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] shibd commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1145586634
##########
pulsar/consumer_partition.go:
##########
@@ -1760,6 +1770,21 @@ func (pc *partitionConsumer) markScaleIfNeed() {
}
}
+func (pc *partitionConsumer) shrinkReceiverQueueSize() {
+ if !pc.options.autoReceiverQueueSize {
+ return
+ }
+
+ oldSize := pc.currentQueueSize.Load()
Review Comment:
We still need `receiverQueueShrinking ` to protect can only be shrinking once at a time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] shibd commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1145631152
##########
pulsar/consumer_partition.go:
##########
@@ -333,6 +335,7 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
}
if pc.options.autoReceiverQueueSize {
pc.currentQueueSize.Store(initialReceiverQueueSize)
+ pc.client.memLimit.RegisterTrigger(receiverQueueShrinkMemThreshold, pc.shrinkReceiverQueueSize)
Review Comment:
I think it can be passed in via the constructor(`NewMemoryLimitController`).
In the short, We don't have to set a threshold for each trigger, it should be unique in `memory_limit_controller`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] shibd commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1141872335
##########
pulsar/consumer_partition.go:
##########
@@ -1752,6 +1762,37 @@ func (pc *partitionConsumer) markScaleIfNeed() {
}
}
+func (pc *partitionConsumer) reserveMemory(size int64) {
+ pc.client.memLimit.ForceReserveMemory(size)
+ if pc.client.memLimit.CurrentUsagePercent() >= receiverQueueShrinkMemThreshold {
Review Comment:
> By the way, is it necessary for the producer to trigger the consumer shrink? If the memory usage exceeds the threshold, the consumer will always shrink it's receiver queue, right?
Sorry, the previous example is not quite accurate.
I think the key difference is whether one `consumer` or `producer` needs to trigger all consumers to reduce the queue size.
- The java client implementation:
When a producer or consumer call the `ReserveMemory` option, If the memory exceeds the threshold, it triggers all consumers to shrink the queue size
- Your implementation:
When a consumer checks that memory exceeds a threshold, it will only try to reduce its own queue size.
I think there is a memory usage starvation issue here:
- One client instance has `1` producer instance and `3` consumer.
- If the queue of these three consumers is already relatively large, if the user of `consumer-1` and `consumer 2` is not actively called (there are many messages piled up in queurCh), then if `consumer-3` checks and reduces its queue size after receiving a message, although the memory limit is reduced, but `consumer-1` and `consumer-2` still occupy more memory (a lot of pending consumption),
- The same way, if the producer wants to send a message, Since it cannot trigger other consumers to reduce the queue size, it will also send more slowly.
In short, in this case, Memory is wasted on inactive consumers(consumer-1, consumer-2)
If it is in implementing the Java client, the problem does not exist.
Each time, all the consumers will reduce the queue size, and then let them re-promote the queue size according to the activity of the consumers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] Gleiphir2769 commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "Gleiphir2769 (via GitHub)" <gi...@apache.org>.
Gleiphir2769 commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1145595299
##########
pulsar/consumer_partition.go:
##########
@@ -333,6 +335,7 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
}
if pc.options.autoReceiverQueueSize {
pc.currentQueueSize.Store(initialReceiverQueueSize)
+ pc.client.memLimit.RegisterTrigger(receiverQueueShrinkMemThreshold, pc.shrinkReceiverQueueSize)
Review Comment:
But `memory_limit_controller` is created by client. If threshold is managed by `memory_limit_controller`, will the threshold become a magic number?
I'm not sure if my thinking is correct.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] shibd commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1145592045
##########
pulsar/consumer_partition.go:
##########
@@ -333,6 +335,7 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
}
if pc.options.autoReceiverQueueSize {
pc.currentQueueSize.Store(initialReceiverQueueSize)
+ pc.client.memLimit.RegisterTrigger(receiverQueueShrinkMemThreshold, pc.shrinkReceiverQueueSize)
Review Comment:
This threshold should be managed by the ·memory_limit_controller·, and we don't need to set it when we register trigger.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] Gleiphir2769 commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "Gleiphir2769 (via GitHub)" <gi...@apache.org>.
Gleiphir2769 commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1145749318
##########
pulsar/consumer_partition.go:
##########
@@ -333,6 +335,7 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
}
if pc.options.autoReceiverQueueSize {
pc.currentQueueSize.Store(initialReceiverQueueSize)
+ pc.client.memLimit.RegisterTrigger(receiverQueueShrinkMemThreshold, pc.shrinkReceiverQueueSize)
Review Comment:
>
I see. I will modify it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] Gleiphir2769 commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "Gleiphir2769 (via GitHub)" <gi...@apache.org>.
Gleiphir2769 commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1141897797
##########
pulsar/consumer_partition.go:
##########
@@ -1752,6 +1762,37 @@ func (pc *partitionConsumer) markScaleIfNeed() {
}
}
+func (pc *partitionConsumer) reserveMemory(size int64) {
+ pc.client.memLimit.ForceReserveMemory(size)
+ if pc.client.memLimit.CurrentUsagePercent() >= receiverQueueShrinkMemThreshold {
Review Comment:
> In short, in this case, Memory is wasted on inactive consumers(consumer-1, consumer-2)
OK, I see. But even in Java client, `reduceReceiverQueue()` can not free memory immediately. It just controls next flow to prefetch fewer messages. IMO, the memory limit on the consumer side is lazy. The memory will only be freed when pending messages are consumed.
So I think this question can be summed up like this.
- Even `memory limiter` can trigger all the consumers to reduce it's queue size. The actual memory release still depends on the consumer's consumption.
- If the consumer is not active, even reducing the queue size will not reduce the memory.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] Gleiphir2769 commented on a diff in pull request #991: [feat] Support consumer client memory limit
Posted by "Gleiphir2769 (via GitHub)" <gi...@apache.org>.
Gleiphir2769 commented on code in PR #991:
URL: https://github.com/apache/pulsar-client-go/pull/991#discussion_r1141794538
##########
pulsar/consumer_partition.go:
##########
@@ -1752,6 +1762,37 @@ func (pc *partitionConsumer) markScaleIfNeed() {
}
}
+func (pc *partitionConsumer) reserveMemory(size int64) {
+ pc.client.memLimit.ForceReserveMemory(size)
+ if pc.client.memLimit.CurrentUsagePercent() >= receiverQueueShrinkMemThreshold {
Review Comment:
I think `shrink` revoked by `memory limiter` and revoked by `consumer` are equivalent.
> One client instance has 1 producer instance and 3 consumer.
> If 3 consumers run out of memory, the producer's production data will block.
In this scenario, the Java client will also block. Because `consumer` use the `forceReserve`.
By the way, is it necessary for the producer to trigger the consumer `shrink`? If the memory usage exceeds the threshold, the consumer will always shrink it's receiver queue, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org