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 2020/06/17 21:16:36 UTC
[GitHub] [pulsar-client-go] merlimat opened a new pull request #284: Introduced lifecycle for compression providers
merlimat opened a new pull request #284:
URL: https://github.com/apache/pulsar-client-go/pull/284
### Motivation
Use individual instances of compression providers for each producer/consumer, with a defined lifecycle, instead of a static singleton instance.
Some compression implementation can use private state to improve the performance. Typically these need to be used by a single go-routine. For this, it's better to have private instances.
----------------------------------------------------------------
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] merlimat commented on a change in pull request #284: Introduced lifecycle for compression providers
Posted by GitBox <gi...@apache.org>.
merlimat commented on a change in pull request #284:
URL: https://github.com/apache/pulsar-client-go/pull/284#discussion_r442422926
##########
File path: pulsar/consumer_partition.go
##########
@@ -846,11 +844,15 @@ func getPreviousMessage(mid *messageID) *messageID {
}
func (pc *partitionConsumer) Decompress(msgMeta *pb.MessageMetadata, payload internal.Buffer) (internal.Buffer, error) {
- provider, ok := compressionProviders[msgMeta.GetCompression()]
+ provider, ok := pc.compressionProviders[msgMeta.GetCompression()]
if !ok {
- err := fmt.Errorf("unsupported compression type: %v", msgMeta.GetCompression())
- pc.log.WithError(err).Error("Failed to decompress message.")
- return nil, err
+ var err error
+ if provider, err = pc.initializeCompressionProvider(msgMeta.GetCompression()); err != nil {
+ pc.log.WithError(err).Error("Failed to decompress message.")
+ return nil, err
+ }
+
+ pc.compressionProviders[msgMeta.GetCompression()] = provider
Review comment:
Yes, it's done at line 683 above
----------------------------------------------------------------
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 #284: Introduced lifecycle for compression providers
Posted by GitBox <gi...@apache.org>.
cckellogg commented on a change in pull request #284:
URL: https://github.com/apache/pulsar-client-go/pull/284#discussion_r442418002
##########
File path: pulsar/consumer_partition.go
##########
@@ -846,11 +844,15 @@ func getPreviousMessage(mid *messageID) *messageID {
}
func (pc *partitionConsumer) Decompress(msgMeta *pb.MessageMetadata, payload internal.Buffer) (internal.Buffer, error) {
- provider, ok := compressionProviders[msgMeta.GetCompression()]
+ provider, ok := pc.compressionProviders[msgMeta.GetCompression()]
if !ok {
- err := fmt.Errorf("unsupported compression type: %v", msgMeta.GetCompression())
- pc.log.WithError(err).Error("Failed to decompress message.")
- return nil, err
+ var err error
+ if provider, err = pc.initializeCompressionProvider(msgMeta.GetCompression()); err != nil {
+ pc.log.WithError(err).Error("Failed to decompress message.")
+ return nil, err
+ }
+
+ pc.compressionProviders[msgMeta.GetCompression()] = provider
Review comment:
Do we need to close these providers when we the consumer is closed?
----------------------------------------------------------------
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] merlimat merged pull request #284: Introduced lifecycle for compression providers
Posted by GitBox <gi...@apache.org>.
merlimat merged pull request #284:
URL: https://github.com/apache/pulsar-client-go/pull/284
----------------------------------------------------------------
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