Repository: apache/pulsar-client-go
Pull request #1342 (2025-03-06 12:35:43)

Improve message decompress error
### Motivation

Improve the error for the message decompress to include more useful information.
The error log would be :
```
level=error msg="handle message Id: ledgerId:5  entryId:0  partition:-1" consumerID=1 error="failed to decompress message: some error, compression type: 1, payload size: 4, uncompressed size: 4" local_addr="127.0.0.1:49301" remote_addr="pulsar://localhost:6650"

```


### Does this pull request potentially affect one of the following parts:

*If `yes` was chosen, please highlight the changes*

  - Dependencies (does it add or upgrade a dependency): (yes / no)
  - The public API: (yes / no)
  - The schema: (yes / no / don't know)
  - The default values of configurations: (yes / no)
  - The wire protocol: (yes / no)

### Documentation

  - Does this pull request introduce a new feature? (yes / no)
  - If yes, how is the feature documented? (not applicable / docs / GoDocs / not documented)
  - If a feature is not applicable for documentation, explain why?
  - If a feature is not documented yet in this PR, please create a followup issue for adding the documentation



Patch:
diff --git a/go.mod b/go.mod
index 7cf845a..c69d1fd 100644
--- a/go.mod
+++ b/go.mod
@@ -17,7 +17,7 @@ require (
 	github.com/onsi/ginkgo v1.16.5
 	github.com/onsi/gomega v1.19.0
 	github.com/opentracing/opentracing-go v1.2.0
-	github.com/pierrec/lz4 v2.0.5+incompatible
+	github.com/pierrec/lz4/v4 v4.1.22
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.14.0
 	github.com/sirupsen/logrus v1.9.3
diff --git a/go.sum b/go.sum
index e633932..f8c8db1 100644
--- a/go.sum
+++ b/go.sum
@@ -298,8 +298,8 @@ github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQ
 github.com/opencontainers/image-spec v1.1.0/go.mod h1:W4s4sFTMaBeK1BQLXbG4AdM2szdn85PY75RI83NrTrM=
 github.com/opentracing/opentracing-go v1.2.0 h1:uEJPy/1a5RIPAJ0Ov+OIO8OxWu77jEv+1B0VhjKrZUs=
 github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYrxe9dPLANfrWvHYVTgc=
-github.com/pierrec/lz4 v2.0.5+incompatible h1:2xWsjqPFWcplujydGg4WmhC/6fZqK42wMM8aXeqhl0I=
-github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
+github.com/pierrec/lz4/v4 v4.1.22 h1:cKFw6uJDK+/gfw5BcDL0JL5aBsAFdsIT18eRtLj7VIU=
+github.com/pierrec/lz4/v4 v4.1.22/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
@@ -405,8 +405,6 @@ go.opentelemetry.io/otel/trace v1.24.0 h1:CsKnnL4dUAr/0llH9FKuc698G04IrpWV0MQA/Y
 go.opentelemetry.io/otel/trace v1.24.0/go.mod h1:HPc3Xr/cOApsBI154IU0OI0HJexz+aw5uPdbs3UCjNU=
 go.opentelemetry.io/proto/otlp v1.0.0 h1:T0TX0tmXU8a3CbNXzEKGeU5mIVOdf0oykP+u2lIVU/I=
 go.opentelemetry.io/proto/otlp v1.0.0/go.mod h1:Sy6pihPLfYHkr3NkUbEhGHFhINUSI/v80hjKIs5JXpM=
-go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
-go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
 go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 13fa9b4..aa922af 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -2147,7 +2147,12 @@ func (pc *partitionConsumer) Decompress(msgMeta *pb.MessageMetadata, payload int
 
 	uncompressed, err := provider.Decompress(nil, payload.ReadableSlice(), int(msgMeta.GetUncompressedSize()))
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to decompress message: %w, compression type: %d, payload size: %d, "+
+			"uncompressed size: %d",
+			err,
+			msgMeta.GetCompression(),
+			len(payload.ReadableSlice()),
+			msgMeta.GetUncompressedSize())
 	}
 
 	return internal.NewBufferWrapper(uncompressed), nil
diff --git a/pulsar/internal/compression/lz4.go b/pulsar/internal/compression/lz4.go
index 745ea83..a5b7ee7 100644
--- a/pulsar/internal/compression/lz4.go
+++ b/pulsar/internal/compression/lz4.go
@@ -18,7 +18,7 @@
 package compression
 
 import (
-	"github.com/pierrec/lz4"
+	"github.com/pierrec/lz4/v4"
 )
 
 const (
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index 1faccc9..a04fa2a 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -129,11 +129,6 @@ type request struct {
 	callback func(command *pb.BaseCommand, err error)
 }
 
-type incomingCmd struct {
-	cmd               *pb.BaseCommand
-	headersAndPayload Buffer
-}
-
 type connection struct {
 	started           int32
 	connectionTimeout time.Duration
@@ -160,7 +155,6 @@ type connection struct {
 
 	incomingRequestsWG sync.WaitGroup
 	incomingRequestsCh chan *request
-	incomingCmdCh      chan *incomingCmd
 	closeCh            chan struct{}
 	readyCh            chan struct{}
 	writeRequestsCh    chan Buffer
@@ -209,7 +203,6 @@ func newConnection(opts connectionOptions) *connection {
 		closeCh:            make(chan struct{}),
 		readyCh:            make(chan struct{}),
 		incomingRequestsCh: make(chan *request, 10),
-		incomingCmdCh:      make(chan *incomingCmd, 10),
 
 		// This channel is used to pass data from producers to the connection
 		// go routine. It can become contended or blocking if we have multiple
@@ -438,9 +431,6 @@ func (c *connection) run() {
 		select {
 		case <-c.closeCh:
 			return
-
-		case cmd := <-c.incomingCmdCh:
-			c.internalReceivedCommand(cmd.cmd, cmd.headersAndPayload)
 		case data := <-c.writeRequestsCh:
 			if data == nil {
 				return
@@ -534,10 +524,6 @@ func (c *connection) writeCommand(cmd *pb.BaseCommand) {
 }
 
 func (c *connection) receivedCommand(cmd *pb.BaseCommand, headersAndPayload Buffer) {
-	c.incomingCmdCh <- &incomingCmd{cmd, headersAndPayload}
-}
-
-func (c *connection) internalReceivedCommand(cmd *pb.BaseCommand, headersAndPayload Buffer) {
 	c.log.Debugf("Received command: %s -- payload: %v", cmd, headersAndPayload)
 	c.setLastDataReceived(time.Now())
 


Test patch:
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index 33e4d05..edef440 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -49,7 +49,7 @@ import (
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
 	plog "github.com/apache/pulsar-client-go/pulsar/log"
 	"github.com/google/uuid"
-	"github.com/pierrec/lz4"
+	"github.com/pierrec/lz4/v4"
 	"github.com/stretchr/testify/assert"
 	"google.golang.org/protobuf/proto"
 )