You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/08/16 15:19:09 UTC
[GitHub] [pulsar-client-go] severinson opened a new pull request, #829: Enable seeking individual topic partitions
severinson opened a new pull request, #829:
URL: https://github.com/apache/pulsar-client-go/pull/829
Currently, seeking on a consumer with a KeyShared subscription fails. This PR removes an unnecessary check to seek the underlying partitionConsumer responsible for the topic.
--
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] wolfstudy commented on a diff in pull request #829: Enable seeking individual topic partitions
Posted by GitBox <gi...@apache.org>.
wolfstudy commented on code in PR #829:
URL: https://github.com/apache/pulsar-client-go/pull/829#discussion_r951165206
##########
pulsar/consumer_impl.go:
##########
@@ -575,15 +575,21 @@ func (c *consumer) Seek(msgID MessageID) error {
c.Lock()
defer c.Unlock()
- if len(c.consumers) > 1 {
- return newError(SeekFailed, "for partition topic, seek command should perform on the individual partitions")
- }
-
mid, ok := c.messageID(msgID)
if !ok {
return nil
}
+ if mid.partitionIdx < 0 {
+ return newError(SeekFailed, "partitionIdx is negative")
+ }
Review Comment:
In Go SDK, the partition index less than 0 is not allowed.
```
// did we receive a valid partition index?
if partition < 0 || partition >= len(c.consumers) {
c.log.Warnf("invalid partition index %d expected a partition between [0-%d]",
partition, len(c.consumers))
return trackingMessageID{}, false
}
```
And why do we need to modify the current logic, where the number of consumers is equivalent to the number of partitions, and each partition corresponds to a unique consumer.
The Go SDK currently does not support seek operations on partitioned topics, and this logic is still being implemented.
--
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] wolfstudy commented on pull request #829: Enable seeking individual topic partitions
Posted by GitBox <gi...@apache.org>.
wolfstudy commented on PR #829:
URL: https://github.com/apache/pulsar-client-go/pull/829#issuecomment-1222048942
If you want to directly perform the seek operation on a partitioned topic, you can refer to the behavior here. https://github.com/apache/pulsar-client-go/pull/782
--
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] severinson commented on a diff in pull request #829: Enable seeking individual topic partitions
Posted by GitBox <gi...@apache.org>.
severinson commented on code in PR #829:
URL: https://github.com/apache/pulsar-client-go/pull/829#discussion_r948075136
##########
pulsar/consumer_impl.go:
##########
@@ -575,15 +575,21 @@ func (c *consumer) Seek(msgID MessageID) error {
c.Lock()
defer c.Unlock()
- if len(c.consumers) > 1 {
- return newError(SeekFailed, "for partition topic, seek command should perform on the individual partitions")
- }
-
mid, ok := c.messageID(msgID)
if !ok {
return nil
}
+ if mid.partitionIdx < 0 {
+ return newError(SeekFailed, "partitionIdx is negative")
+ }
Review Comment:
Is that really true? The pre-existing code would do the following:
```
return c.consumers[mid.partitionIdx].Seek(mid)
```
This suggests that the partition index is 0 for non-partitioned topics. Otherwise seek() on a non-partitioned topic would've always resulted in an index out of bounds panic.
--
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] codelipenghui commented on pull request #829: Enable seeking individual topic partitions
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on PR #829:
URL: https://github.com/apache/pulsar-client-go/pull/829#issuecomment-1222067853
@severinson After checking more details about this part. Before 2.8.0, the Java client only supported seek with the earliest/latest position. https://github.com/apache/pulsar/pull/10033 has introduced a new API https://github.com/apache/pulsar/pull/10033/files#diff-6010f94fead60a5a3ed8aa58a37fe96fdf2cd9c5c1573989cdf89abcb9f9c256R613 which able to provide seek message ID for each partition or topic(For multi-topic/regex consumer)
I think the go client should also follow the same way.
--
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] codelipenghui commented on a diff in pull request #829: Enable seeking individual topic partitions
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on code in PR #829:
URL: https://github.com/apache/pulsar-client-go/pull/829#discussion_r948035080
##########
pulsar/consumer_impl.go:
##########
@@ -575,15 +575,21 @@ func (c *consumer) Seek(msgID MessageID) error {
c.Lock()
defer c.Unlock()
- if len(c.consumers) > 1 {
- return newError(SeekFailed, "for partition topic, seek command should perform on the individual partitions")
- }
-
mid, ok := c.messageID(msgID)
if !ok {
return nil
}
+ if mid.partitionIdx < 0 {
+ return newError(SeekFailed, "partitionIdx is negative")
+ }
Review Comment:
Will this break the non-partitioned topic message acknowledgment?
The partition index -1 means the message from a non-partitioned topic.
@freeznet @wolfstudy Please help confirm
--
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] severinson commented on a diff in pull request #829: Enable seeking individual topic partitions
Posted by GitBox <gi...@apache.org>.
severinson commented on code in PR #829:
URL: https://github.com/apache/pulsar-client-go/pull/829#discussion_r948075136
##########
pulsar/consumer_impl.go:
##########
@@ -575,15 +575,21 @@ func (c *consumer) Seek(msgID MessageID) error {
c.Lock()
defer c.Unlock()
- if len(c.consumers) > 1 {
- return newError(SeekFailed, "for partition topic, seek command should perform on the individual partitions")
- }
-
mid, ok := c.messageID(msgID)
if !ok {
return nil
}
+ if mid.partitionIdx < 0 {
+ return newError(SeekFailed, "partitionIdx is negative")
+ }
Review Comment:
Is that really true? The pre-existing code would do the following:
```
return c.consumers[mid.partitionIdx].Seek(mid)
```
This suggests that the partition index is 0 for non-partitioned topics. Otherwise seek() would've always resulted in an index out of bounds panic.
--
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] severinson commented on a diff in pull request #829: Enable seeking individual topic partitions
Posted by GitBox <gi...@apache.org>.
severinson commented on code in PR #829:
URL: https://github.com/apache/pulsar-client-go/pull/829#discussion_r952235283
##########
pulsar/consumer_impl.go:
##########
@@ -575,15 +575,21 @@ func (c *consumer) Seek(msgID MessageID) error {
c.Lock()
defer c.Unlock()
- if len(c.consumers) > 1 {
- return newError(SeekFailed, "for partition topic, seek command should perform on the individual partitions")
- }
-
mid, ok := c.messageID(msgID)
if !ok {
return nil
}
+ if mid.partitionIdx < 0 {
+ return newError(SeekFailed, "partitionIdx is negative")
+ }
Review Comment:
Thanks for the comments. I have a few questions.
> And why do we need to modify the current logic, where the number of consumers is equivalent to the number of partitions, and each partition corresponds to a unique consumer.
I don't understand what you mean. I'm not changing the current logic. I'm just passing through the message id to the partitionConsumer responsible for the partition.
> The Go SDK currently does not support seek operations on partitioned topics, and this logic is still being implemented.
All right. Do you know what work still remains to have the Go client support seeking by partition?
##########
pulsar/consumer_impl.go:
##########
@@ -575,15 +575,21 @@ func (c *consumer) Seek(msgID MessageID) error {
c.Lock()
defer c.Unlock()
- if len(c.consumers) > 1 {
- return newError(SeekFailed, "for partition topic, seek command should perform on the individual partitions")
- }
-
mid, ok := c.messageID(msgID)
if !ok {
return nil
}
+ if mid.partitionIdx < 0 {
+ return newError(SeekFailed, "partitionIdx is negative")
+ }
Review Comment:
Thanks for the comments. I have a few questions.
> And why do we need to modify the current logic, where the number of consumers is equivalent to the number of partitions, and each partition corresponds to a unique consumer.
I don't understand what you mean. I'm not changing the current logic. I'm just passing through the message id to the partitionConsumer responsible for the partition.
> The Go SDK currently does not support seek operations on partitioned topics, and this logic is still being implemented.
All right. Do you know what work still remains to have the Go client support seeking by partition?
--
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] severinson commented on pull request #829: Enable seeking individual topic partitions
Posted by GitBox <gi...@apache.org>.
severinson commented on PR #829:
URL: https://github.com/apache/pulsar-client-go/pull/829#issuecomment-1217712214
@codelipenghui I've added tests.
--
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