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 2021/10/22 16:43:17 UTC
[GitHub] [pulsar-client-go] taiyang-li opened a new pull request #303: Close msg channel when consumer close
taiyang-li opened a new pull request #303:
URL: https://github.com/apache/pulsar-client-go/pull/303
## Motivation
When consumer close, message channel is not closed. It will cause the loop receiving messages by consumer.Chan() can't be break. Just like the code below. I fix it in this PR.
``` go
func main() {
client, err := pulsar.NewClient(pulsar.ClientOptions{
URL: "pulsar://localhost:6650",
})
if err != nil {
log.Fatalln(err)
}
defer client.Close()
consumer, err := client.Subscribe(pulsar.ConsumerOptions{
Topic: "mytopic",
SubscriptionName: "mysub",
Type: pulsar.Shared,
})
if err != nil {
log.Fatalln(err)
}
defer consumer.Close()
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
// The loop will never be break even if you execute consumer.Close(),
// unless user make it break using some flag, but he can't judge whether there are some msgs unreceived in the channel.
for msg := range consumer.Chan() {
consumer.Ack(msg)
}
}()
time.Sleep(10 * time.Second)
consumer.Unsubscribe()
consumer.Close()
fmt.Println("closed")
wg.Wait()
fmt.Println("finish wait")
}
```
## Modifications
Add close message channel in consumer.Close()
--
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 #303: Close msg channel when consumer close
Posted by GitBox <gi...@apache.org>.
wolfstudy commented on pull request #303:
URL: https://github.com/apache/pulsar-client-go/pull/303#issuecomment-654000882
@taiyang-li Can you merge master code to track the CI again?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] taiyang-li closed pull request #303: Close msg channel when consumer close
Posted by GitBox <gi...@apache.org>.
taiyang-li closed pull request #303:
URL: https://github.com/apache/pulsar-client-go/pull/303
--
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] cckellogg commented on a change in pull request #303: Close msg channel when consumer close
Posted by GitBox <gi...@apache.org>.
cckellogg commented on a change in pull request #303:
URL: https://github.com/apache/pulsar-client-go/pull/303#discussion_r450617663
##########
File path: pulsar/consumer_impl.go
##########
@@ -374,6 +374,7 @@ func (c *consumer) Close() {
}
wg.Wait()
close(c.closeCh)
+ close(c.messageCh)
Review comment:
If a users passes a message channel in the client should not close that channel.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] taiyang-li commented on a change in pull request #303: Close msg channel when consumer close
Posted by GitBox <gi...@apache.org>.
taiyang-li commented on a change in pull request #303:
URL: https://github.com/apache/pulsar-client-go/pull/303#discussion_r450586026
##########
File path: pulsar/consumer_impl.go
##########
@@ -374,6 +374,7 @@ func (c *consumer) Close() {
}
wg.Wait()
close(c.closeCh)
+ close(c.messageCh)
Review comment:
Maybe I should close message channel in multiTopicConsumer.Close() and regexConsumer.Close(), and remove it in internal consumer.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] cckellogg commented on pull request #303: Close msg channel when consumer close
Posted by GitBox <gi...@apache.org>.
cckellogg commented on pull request #303:
URL: https://github.com/apache/pulsar-client-go/pull/303#issuecomment-654613386
Looking at the code example reading messages from the channel after the consumer has unsubscribed or closed is kind of useless since you can't acknowledge those messages and they will be resent when the client connects again. I personally think it's better to have your own close flag to know when you should stop processing messages.
Closing the message channel in the client is a little tricky since multiple go routines could be sending on it and the multi topic and regex consumer are built on the base consumer. There would probably need to be some sort of flag/lock to know when and who should close the message channel.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] taiyang-li commented on a change in pull request #303: Close msg channel when consumer close
Posted by GitBox <gi...@apache.org>.
taiyang-li commented on a change in pull request #303:
URL: https://github.com/apache/pulsar-client-go/pull/303#discussion_r450581019
##########
File path: pulsar/consumer_impl.go
##########
@@ -374,6 +374,7 @@ func (c *consumer) Close() {
}
wg.Wait()
close(c.closeCh)
+ close(c.messageCh)
Review comment:
Yes. I already found this problem. It is hard to decide where to close the message channel.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] taiyang-li commented on pull request #303: Close msg channel when consumer close
Posted by GitBox <gi...@apache.org>.
taiyang-li commented on pull request #303:
URL: https://github.com/apache/pulsar-client-go/pull/303#issuecomment-654626423
> Looking at the code example reading messages from the channel after the consumer has unsubscribed or closed is kind of useless since you can't acknowledge those messages and they will be resent when the client connects again. I personally think it's better to have your own close flag to know when you should stop processing messages.
>
> Closing the message channel in the client is a little tricky since multiple go routines could be sending on it and the multi topic and regex consumer are built on the base consumer. There would probably need to be some sort of flag/lock to know when and who should close the message channel.
Indeed this is a workaround of the problem I mentioned above. But new users must remember that he must break the consume loop himself, instead of client closing message channel. At least this make some confusion for me when I first use this library
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] taiyang-li commented on a change in pull request #303: Close msg channel when consumer close
Posted by GitBox <gi...@apache.org>.
taiyang-li commented on a change in pull request #303:
URL: https://github.com/apache/pulsar-client-go/pull/303#discussion_r452074274
##########
File path: pulsar/consumer_singletopic.go
##########
@@ -0,0 +1,41 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pulsar
+
+type singleTopicConsumer struct {
Review comment:
Usually, it is better channel closed by producer, not consumer. When application new a message channel and passes it to singleTopicConsumer/multiTopicConsumer/regexConsumer, we can still close the channel in interface Close() exactly once.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] taiyang-li commented on a change in pull request #303: Close msg channel when consumer close
Posted by GitBox <gi...@apache.org>.
taiyang-li commented on a change in pull request #303:
URL: https://github.com/apache/pulsar-client-go/pull/303#discussion_r451241499
##########
File path: pulsar/consumer_singletopic.go
##########
@@ -0,0 +1,41 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pulsar
+
+type singleTopicConsumer struct {
Review comment:
since I can't close message channel directly in internal consumer. So I wrap internal consumer for single topic subscription scene.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] cckellogg commented on a change in pull request #303: Close msg channel when consumer close
Posted by GitBox <gi...@apache.org>.
cckellogg commented on a change in pull request #303:
URL: https://github.com/apache/pulsar-client-go/pull/303#discussion_r450616616
##########
File path: pulsar/consumer_singletopic.go
##########
@@ -0,0 +1,41 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pulsar
+
+type singleTopicConsumer struct {
Review comment:
What is the purpose of this?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] taiyang-li commented on a change in pull request #303: Close msg channel when consumer close
Posted by GitBox <gi...@apache.org>.
taiyang-li commented on a change in pull request #303:
URL: https://github.com/apache/pulsar-client-go/pull/303#discussion_r450592165
##########
File path: pulsar/consumer_impl.go
##########
@@ -374,6 +374,7 @@ func (c *consumer) Close() {
}
wg.Wait()
close(c.closeCh)
+ close(c.messageCh)
Review comment:
> Maybe I should close message channel in multiTopicConsumer.Close() and regexConsumer.Close(), and remove it in internal consumer.
Done. Commit again
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] cckellogg commented on a change in pull request #303: Close msg channel when consumer close
Posted by GitBox <gi...@apache.org>.
cckellogg commented on a change in pull request #303:
URL: https://github.com/apache/pulsar-client-go/pull/303#discussion_r450446421
##########
File path: pulsar/consumer_impl.go
##########
@@ -374,6 +374,7 @@ func (c *consumer) Close() {
}
wg.Wait()
close(c.closeCh)
+ close(c.messageCh)
Review comment:
If the user passes in a channel and the channel is shared close could be called multiple times and there will be a 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] taiyang-li commented on pull request #303: Close msg channel when consumer close
Posted by GitBox <gi...@apache.org>.
taiyang-li commented on pull request #303:
URL: https://github.com/apache/pulsar-client-go/pull/303#issuecomment-656445206
> @taiyang-li Can you merge master code to track the CI again?
Already merged.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] cckellogg commented on a change in pull request #303: Close msg channel when consumer close
Posted by GitBox <gi...@apache.org>.
cckellogg commented on a change in pull request #303:
URL: https://github.com/apache/pulsar-client-go/pull/303#discussion_r451622413
##########
File path: pulsar/consumer_singletopic.go
##########
@@ -0,0 +1,41 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pulsar
+
+type singleTopicConsumer struct {
Review comment:
What happens if this is shared message channel passed in by the application?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-go] flowchartsman commented on a change in pull request #303: Close msg channel when consumer close
Posted by GitBox <gi...@apache.org>.
flowchartsman commented on a change in pull request #303:
URL: https://github.com/apache/pulsar-client-go/pull/303#discussion_r693023004
##########
File path: pulsar/consumer_impl.go
##########
@@ -374,6 +374,7 @@ func (c *consumer) Close() {
}
wg.Wait()
close(c.closeCh)
+ close(c.messageCh)
Review comment:
I just came across this issue, and allowing the user to pass in a message channel seems like a mistake to me, because it inverts control. If the user wants to multiplex messages from different consumers onto a channel they control, it should be up to them to handle that with a select clause, [runtime.Select](https://pkg.go.dev/reflect#Select), or a combination of concurrency primitives in their code.
--
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