You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "geniusjoe (via GitHub)" <gi...@apache.org> on 2025/08/18 04:32:20 UTC
[PR] Support Snappy CompressionType [pulsar-client-go]
geniusjoe opened a new pull request, #1406:
URL: https://github.com/apache/pulsar-client-go/pull/1406
Master Issue: https://github.com/apache/pulsar-client-go/issues/936
### Motivation
As above issue says. In go client, Snappy compression is not yet supported. When using go client to consume Snappy compressed messge, it will throw a `DecompressionError` exception
### Modifications
- Realize SnappyProvider in `pulsar/internal/compression/snappy.go` based on `github.com/klauspost/compress/snappy` package
- Support Snappy provider in producer `pulsar/internal/batch_builder#goGetCompressionProvider` and consumer `pulsar/consumer_partition.go#initializeCompressionProvider`
### Verifying this change
- [x] Make sure that the change passes the CI checks.
This change is already covered by existing tests, such as
`pulsar/internal/compression/compression_test.go#TestCompression`
`pulsar/internal/compression/compression_test.go#TestCompressionNoBuffers`
`pulsar/producer_test.go#TestProducerCompression`
This change added tests and can be verified as follows:
`pulsar/consumer_test.go#TestConsumerMultiCompressions`
### Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API: (no)
- The schema: (no)
- The default values of configurations: (no)
- The wire protocol: (no)
### Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (docs)
- If a feature is not documented yet in this PR, please create a followup issue for adding the documentation
I'll propose a pr to add this feature to the Client Feature Matrix later.
### Benchmark
- Compression speed: golang-snappy ≈ 1.5x klauspost-compress-snappy
- Decompression speed: klauspost-compress-snappy ≈ 1.5x golang-snappy
- Compression speed: klauspost-compress-snappy > lz4 > zstd-pure-go-default > zlib
- Decompression speed: lz4 > klauspost-compress-snappy > zstd-pure-go-default > zlib
```
pulsar/internal/compression/compression_bench_test.go#BenchmarkCompression
goos: linux
goarch: amd64
pkg: github.com/apache/pulsar-client-go/pulsar/internal/compression
cpu: AMD EPYC 7K62 48-Core Processor
BenchmarkCompression
BenchmarkCompression/zlib
BenchmarkCompression/zlib-8 224 5216098 ns/op 19.29 MB/s
BenchmarkCompression/lz4
BenchmarkCompression/lz4-8 1970 606331 ns/op 165.93 MB/s
BenchmarkCompression/zstd-pure-go-fastest
BenchmarkCompression/zstd-pure-go-fastest-8 1226 939331 ns/op 107.10 MB/s
BenchmarkCompression/zstd-pure-go-default
BenchmarkCompression/zstd-pure-go-default-8 735 1521550 ns/op 66.12 MB/s
BenchmarkCompression/zstd-pure-go-best
BenchmarkCompression/zstd-pure-go-best-8 385 3033820 ns/op 33.16 MB/s
BenchmarkCompression/zstd-cgo-level-fastest
BenchmarkCompression/zstd-cgo-level-fastest-8 2497 489938 ns/op 205.34 MB/s
BenchmarkCompression/zstd-cgo-level-default
BenchmarkCompression/zstd-cgo-level-default-8 886 1334805 ns/op 75.37 MB/s
BenchmarkCompression/zstd-cgo-level-best
BenchmarkCompression/zstd-cgo-level-best-8 360 3253330 ns/op 30.92 MB/s
BenchmarkCompression/klauspost-compress-snappy
BenchmarkCompression/klauspost-compress-snappy-8 2307 515983 ns/op 194.98 MB/s
BenchmarkCompression/golang-snappy
BenchmarkCompression/golang-snappy-8 3483 348149 ns/op 288.97 MB/s
```
```
pulsar/internal/compression/compression_bench_test.go#BenchmarkDecompression
goos: linux
goarch: amd64
pkg: github.com/apache/pulsar-client-go/pulsar/internal/compression
cpu: AMD EPYC 7K62 48-Core Processor
BenchmarkDecompression
BenchmarkDecompression/zlib
BenchmarkDecompression/zlib-8 1184 898020 ns/op 112.03 MB/s
BenchmarkDecompression/lz4
BenchmarkDecompression/lz4-8 39049 30671 ns/op 3280.15 MB/s
BenchmarkDecompression/zstd-pure-go-fastest
BenchmarkDecompression/zstd-pure-go-fastest-8 4549 255393 ns/op 393.93 MB/s
BenchmarkDecompression/zstd-pure-go-default
BenchmarkDecompression/zstd-pure-go-default-8 4567 251818 ns/op 399.52 MB/s
BenchmarkDecompression/zstd-pure-go-best
BenchmarkDecompression/zstd-pure-go-best-8 4579 256995 ns/op 391.47 MB/s
BenchmarkDecompression/zstd-cgo-level-fastest
BenchmarkDecompression/zstd-cgo-level-fastest-8 7965 135827 ns/op 740.69 MB/s
BenchmarkDecompression/zstd-cgo-level-default
BenchmarkDecompression/zstd-cgo-level-default-8 7032 172644 ns/op 582.74 MB/s
BenchmarkDecompression/zstd-cgo-level-best
BenchmarkDecompression/zstd-cgo-level-best-8 8055 154114 ns/op 652.80 MB/s
BenchmarkDecompression/klauspost-compress-snappy
BenchmarkDecompression/klauspost-compress-snappy-8 10000 111700 ns/op 900.68 MB/s
BenchmarkDecompression/golang-snappy
BenchmarkDecompression/golang-snappy-8 7765 156358 ns/op 643.44 MB/s
```
```
pulsar/internal/compression/compression_bench_test.go#BenchmarkCompressionParallel
goos: linux
goarch: amd64
pkg: github.com/apache/pulsar-client-go/pulsar/internal/compression
cpu: AMD EPYC 7K62 48-Core Processor
BenchmarkCompressionParallel
BenchmarkCompressionParallel/zlib
BenchmarkCompressionParallel/zlib-8 1219 939369 ns/op 107.10 MB/s
BenchmarkCompressionParallel/lz4
BenchmarkCompressionParallel/lz4-8 9488 110573 ns/op 909.86 MB/s
BenchmarkCompressionParallel/zstd-pure-go-fastest
BenchmarkCompressionParallel/zstd-pure-go-fastest-8 3723 314296 ns/op 320.10 MB/s
BenchmarkCompressionParallel/zstd-pure-go-default
BenchmarkCompressionParallel/zstd-pure-go-default-8 3666 315090 ns/op 319.29 MB/s
BenchmarkCompressionParallel/zstd-pure-go-best
BenchmarkCompressionParallel/zstd-pure-go-best-8 2583 394209 ns/op 255.21 MB/s
BenchmarkCompressionParallel/zstd-cgo-level-fastest
BenchmarkCompressionParallel/zstd-cgo-level-fastest-8 3166 336781 ns/op 298.73 MB/s
BenchmarkCompressionParallel/zstd-cgo-level-default
BenchmarkCompressionParallel/zstd-cgo-level-default-8 3261 327483 ns/op 307.21 MB/s
BenchmarkCompressionParallel/zstd-cgo-level-best
BenchmarkCompressionParallel/zstd-cgo-level-best-8 4357 264377 ns/op 380.54 MB/s
BenchmarkCompressionParallel/klauspost-compress-snappy
BenchmarkCompressionParallel/klauspost-compress-snappy-8 12405 95724 ns/op 1051.01 MB/s
BenchmarkCompressionParallel/golang-snappy
BenchmarkCompressionParallel/golang-snappy-8 16046 68610 ns/op 1466.36 MB/s
```
--
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
Re: [PR] Support Snappy CompressionType [pulsar-client-go]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower merged PR #1406:
URL: https://github.com/apache/pulsar-client-go/pull/1406
--
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
Re: [PR] Support Snappy CompressionType [pulsar-client-go]
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #1406:
URL: https://github.com/apache/pulsar-client-go/pull/1406#discussion_r2284865460
##########
pulsar/consumer_test.go:
##########
@@ -1293,6 +1293,91 @@ func TestConsumerCompression(t *testing.T) {
}
}
+func TestConsumerMultiCompressions(t *testing.T) {
+ type testProvider struct {
+ name string
+ compressionType CompressionType
+ }
+
+ providers := []testProvider{
+ {"zlib", ZLib},
+ {"lz4", LZ4},
+ {"zstd", ZSTD},
+ {"snappy", SNAPPY},
+ }
+
+ for _, provider := range providers {
+ p := provider
+ t.Run(p.name, func(t *testing.T) {
+ client, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ })
+
+ assert.Nil(t, err)
+ defer client.Close()
+
+ topicName := newTopicName()
+ ctx := context.Background()
+
+ // enable batching
+ p1, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ CompressionType: p.compressionType,
+ })
+ assert.Nil(t, err)
+ defer p1.Close()
+
+ // disable batching
+ p2, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ CompressionType: p.compressionType,
+ DisableBatching: true,
+ })
+ assert.Nil(t, err)
+ defer p2.Close()
+
+ consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "sub-1",
+ })
+ assert.Nil(t, err)
+ defer consumer.Close()
+
+ const N = 100
+
+ for i := 0; i < N; i++ {
+ if _, err := p1.Send(ctx, &ProducerMessage{
+ Payload: []byte(fmt.Sprintf("msg-content-%d-batching-enabled", i)),
+ }); err != nil {
+ t.Fatal(err)
+ }
Review Comment:
If we want to test the batching producer, we should use the SendAsync. Otherwise, the batching won't take effect.
--
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
Re: [PR] Support Snappy CompressionType [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1406:
URL: https://github.com/apache/pulsar-client-go/pull/1406#discussion_r2285070904
##########
pulsar/consumer_test.go:
##########
@@ -1293,6 +1293,91 @@ func TestConsumerCompression(t *testing.T) {
}
}
+func TestConsumerMultiCompressions(t *testing.T) {
+ type testProvider struct {
+ name string
+ compressionType CompressionType
+ }
+
+ providers := []testProvider{
+ {"zlib", ZLib},
+ {"lz4", LZ4},
+ {"zstd", ZSTD},
+ {"snappy", SNAPPY},
+ }
+
+ for _, provider := range providers {
+ p := provider
+ t.Run(p.name, func(t *testing.T) {
+ client, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ })
+
+ assert.Nil(t, err)
+ defer client.Close()
+
+ topicName := newTopicName()
+ ctx := context.Background()
+
+ // enable batching
+ p1, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ CompressionType: p.compressionType,
+ })
+ assert.Nil(t, err)
+ defer p1.Close()
+
+ // disable batching
+ p2, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ CompressionType: p.compressionType,
+ DisableBatching: true,
+ })
+ assert.Nil(t, err)
+ defer p2.Close()
+
+ consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "sub-1",
+ })
+ assert.Nil(t, err)
+ defer consumer.Close()
+
+ const N = 100
+
+ for i := 0; i < N; i++ {
+ if _, err := p1.Send(ctx, &ProducerMessage{
+ Payload: []byte(fmt.Sprintf("msg-content-%d-batching-enabled", i)),
+ }); err != nil {
+ t.Fatal(err)
+ }
Review Comment:
Fixed
--
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