fix: wrap errors using %w to preserve context (#1321)

reugn · web-flow · commit 024e230d32fc · 2024-12-19T10:25:08.000+08:00
* fix: wrap errors using %w to preserve context

* move the consumer state check
diff --git a/oauth2/auth.go b/oauth2/auth.go
@@ -112,7 +112,7 @@ func ExtractUserName(token oauth2.Token) (string, error) {
 	p := jwt.Parser{}
 	claims := jwt.MapClaims{}
 	if _, _, err := p.ParseUnverified(token.AccessToken, claims); err != nil {
-		return "", fmt.Errorf("unable to decode the access token: %v", err)
+		return "", fmt.Errorf("unable to decode the access token: %w", err)
 	}
 	username, ok := claims[ClaimNameUserName]
 	if !ok {
diff --git a/oauth2/cache/cache.go b/oauth2/cache/cache.go
@@ -80,7 +80,7 @@ func (t *tokenCache) Token() (*xoauth2.Token, error) {
 	// load from the store and use the access token if it isn't expired
 	grant, err := t.store.LoadGrant(t.audience)
 	if err != nil {
-		return nil, fmt.Errorf("LoadGrant: %v", err)
+		return nil, fmt.Errorf("LoadGrant: %w", err)
 	}
 	t.token = grant.Token
 	if t.token != nil && t.validateAccessToken(*t.token) {
@@ -90,13 +90,13 @@ func (t *tokenCache) Token() (*xoauth2.Token, error) {
 	// obtain and cache a fresh access token
 	grant, err = t.refresher.Refresh(grant)
 	if err != nil {
-		return nil, fmt.Errorf("RefreshGrant: %v", err)
+		return nil, fmt.Errorf("RefreshGrant: %w", err)
 	}
 	t.token = grant.Token
 	err = t.store.SaveGrant(t.audience, *grant)
 	if err != nil {
 		// TODO log rather than throw
-		return nil, fmt.Errorf("SaveGrant: %v", err)
+		return nil, fmt.Errorf("SaveGrant: %w", err)
 	}
 
 	return t.token, nil
@@ -117,14 +117,14 @@ func (t *tokenCache) InvalidateToken() error {
 	}
 	grant, err := t.store.LoadGrant(t.audience)
 	if err != nil {
-		return fmt.Errorf("LoadGrant: %v", err)
+		return fmt.Errorf("LoadGrant: %w", err)
 	}
 	if grant.Token != nil && grant.Token.AccessToken == previous.AccessToken {
 		grant.Token.Expiry = time.Unix(0, 0).Add(expiryDelta)
 		err = t.store.SaveGrant(t.audience, *grant)
 		if err != nil {
 			// TODO log rather than throw
-			return fmt.Errorf("SaveGrant: %v", err)
+			return fmt.Errorf("SaveGrant: %w", err)
 		}
 	}
 	return nil
diff --git a/oauth2/store/keyring.go b/oauth2/store/keyring.go
@@ -20,6 +20,7 @@ package store
 import (
 	"crypto/sha1"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"sync"
 
@@ -92,7 +93,7 @@ func (f *KeyringStore) LoadGrant(audience string) (*oauth2.AuthorizationGrant, e
 
 	item, err := f.getItem(audience)
 	if err != nil {
-		if err == keyring.ErrKeyNotFound {
+		if errors.Is(err, keyring.ErrKeyNotFound) {
 			return nil, ErrNoAuthenticationData
 		}
 		return nil, err
@@ -119,10 +120,10 @@ func (f *KeyringStore) WhoAmI(audience string) (string, error) {
 	key := hashKeyringKey(audience)
 	authItem, err := f.kr.Get(key)
 	if err != nil {
-		if err == keyring.ErrKeyNotFound {
+		if errors.Is(err, keyring.ErrKeyNotFound) {
 			return "", ErrNoAuthenticationData
 		}
-		return "", fmt.Errorf("unable to get information from the keyring: %v", err)
+		return "", fmt.Errorf("unable to get information from the keyring: %w", err)
 	}
 	return authItem.Label, nil
 }
@@ -134,13 +135,13 @@ func (f *KeyringStore) Logout() error {
 	var err error
 	keys, err := f.kr.Keys()
 	if err != nil {
-		return fmt.Errorf("unable to get information from the keyring: %v", err)
+		return fmt.Errorf("unable to get information from the keyring: %w", err)
 	}
 	for _, key := range keys {
 		err = f.kr.Remove(key)
 	}
 	if err != nil {
-		return fmt.Errorf("unable to update the keyring: %v", err)
+		return fmt.Errorf("unable to update the keyring: %w", err)
 	}
 	return nil
 }
@@ -180,7 +181,7 @@ func (f *KeyringStore) setItem(item storedItem) error {
 	}
 	err = f.kr.Set(i)
 	if err != nil {
-		return fmt.Errorf("unable to update the keyring: %v", err)
+		return fmt.Errorf("unable to update the keyring: %w", err)
 	}
 	return nil
 }
diff --git a/oauth2/store/store.go b/oauth2/store/store.go
@@ -26,7 +26,7 @@ import (
 // ErrNoAuthenticationData indicates that stored authentication data is not available
 var ErrNoAuthenticationData = errors.New("authentication data is not available")
 
-// ErrUnsupportedAuthData ndicates that stored authentication data is unusable
+// ErrUnsupportedAuthData indicates that stored authentication data is unusable
 var ErrUnsupportedAuthData = errors.New("authentication data is not usable")
 
 // Store is responsible for persisting authorization grants
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
@@ -1158,21 +1158,21 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 	// error decrypting the payload
 	if err != nil {
 		// default crypto failure action
-		crypToFailureAction := crypto.ConsumerCryptoFailureActionFail
+		cryptoFailureAction := crypto.ConsumerCryptoFailureActionFail
 		if pc.options.decryption != nil {
-			crypToFailureAction = pc.options.decryption.ConsumerCryptoFailureAction
+			cryptoFailureAction = pc.options.decryption.ConsumerCryptoFailureAction
 		}
 
-		switch crypToFailureAction {
+		switch cryptoFailureAction {
 		case crypto.ConsumerCryptoFailureActionFail:
-			pc.log.Errorf("consuming message failed due to decryption err :%v", err)
+			pc.log.Errorf("consuming message failed due to decryption err: %v", err)
 			pc.NackID(newTrackingMessageID(int64(pbMsgID.GetLedgerId()), int64(pbMsgID.GetEntryId()), 0, 0, 0, nil))
 			return err
 		case crypto.ConsumerCryptoFailureActionDiscard:
 			pc.discardCorruptedMessage(pbMsgID, pb.CommandAck_DecryptionError)
-			return fmt.Errorf("discarding message on decryption error :%v", err)
+			return fmt.Errorf("discarding message on decryption error: %w", err)
 		case crypto.ConsumerCryptoFailureActionConsume:
-			pc.log.Warnf("consuming encrypted message due to error in decryption :%v", err)
+			pc.log.Warnf("consuming encrypted message due to error in decryption: %v", err)
 			messages := []*message{
 				{
 					publishTime:  timeFromUnixTimestampMillis(msgMeta.GetPublishTime()),
@@ -1767,16 +1767,16 @@ func (pc *partitionConsumer) runEventsLoop() {
 func (pc *partitionConsumer) internalClose(req *closeRequest) {
 	defer close(req.doneCh)
 	state := pc.getConsumerState()
-	if state != consumerReady {
-		// this might be redundant but to ensure nack tracker is closed
+	if state == consumerClosed || state == consumerClosing {
+		pc.log.WithField("state", state).Error("Consumer is closing or has closed")
 		if pc.nackTracker != nil {
 			pc.nackTracker.Close()
 		}
 		return
 	}
 
-	if state == consumerClosed || state == consumerClosing {
-		pc.log.WithField("state", state).Error("Consumer is closing or has closed")
+	if state != consumerReady {
+		// this might be redundant but to ensure nack tracker is closed
 		if pc.nackTracker != nil {
 			pc.nackTracker.Close()
 		}
diff --git a/pulsar/crypto/default_message_crypto.go b/pulsar/crypto/default_message_crypto.go
@@ -95,7 +95,7 @@ func (d *DefaultMessageCrypto) addPublicKeyCipher(keyName string, keyReader KeyR
 	d.cipherLock.Lock()
 	defer d.cipherLock.Unlock()
 	if keyName == "" || keyReader == nil {
-		return fmt.Errorf("keyname or keyreader is null")
+		return fmt.Errorf("keyname or keyreader is nil")
 	}
 
 	// read the public key and its info using keyReader
@@ -212,7 +212,7 @@ func (d *DefaultMessageCrypto) Encrypt(encKeys []string,
 func (d *DefaultMessageCrypto) Decrypt(msgMetadata MessageMetadataSupplier,
 	payload []byte,
 	keyReader KeyReader) ([]byte, error) {
-	// if data key is present, attempt to derypt using the existing key
+	// if data key is present, attempt to decrypt using the existing key
 	if d.dataKey != nil {
 		decryptedData, err := d.getKeyAndDecryptData(msgMetadata, payload)
 		if err != nil {
@@ -342,20 +342,20 @@ func (d *DefaultMessageCrypto) loadPrivateKey(key []byte) (gocrypto.PrivateKey,
 
 // read the public key into RSA key
 func (d *DefaultMessageCrypto) loadPublicKey(key []byte) (gocrypto.PublicKey, error) {
-	var publickKey gocrypto.PublicKey
+	var publicKey gocrypto.PublicKey
 
 	pubPem, _ := pem.Decode(key)
 	if pubPem == nil {
-		return publickKey, fmt.Errorf("failed to decode public key")
+		return publicKey, fmt.Errorf("failed to decode public key")
 	}
 
 	genericPublicKey, err := x509.ParsePKIXPublicKey(pubPem.Bytes)
 	if err != nil {
-		return publickKey, err
+		return publicKey, err
 	}
-	publickKey = genericPublicKey
+	publicKey = genericPublicKey
 
-	return publickKey, nil
+	return publicKey, nil
 }
 
 func generateDataKey() ([]byte, error) {
diff --git a/pulsar/internal/commands.go b/pulsar/internal/commands.go
@@ -272,7 +272,7 @@ func serializeMessage(wb Buffer,
 	encryptedPayload, err := encryptor.Encrypt(compressedPayload, msgMetadata)
 	if err != nil {
 		// error occurred while encrypting the payload, ProducerCryptoFailureAction is set to Fail
-		return fmt.Errorf("encryption of message failed, ProducerCryptoFailureAction is set to Fail. Error :%v", err)
+		return fmt.Errorf("encryption of message failed, ProducerCryptoFailureAction is set to Fail. Error: %w", err)
 	}
 
 	cmdSize := uint32(proto.Size(cmdSend))
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
@@ -947,7 +947,7 @@ func (c *connection) handleTopicMigrated(commandTopicMigrated *pb.CommandTopicMi
 	resourceID := commandTopicMigrated.GetResourceId()
 	migratedBrokerServiceURL := c.getMigratedBrokerServiceURL(commandTopicMigrated)
 	if migratedBrokerServiceURL == "" {
-		c.log.Warnf("Failed to find the migrated broker url for resource: %s, migratedBrokerUrl: %s, migratedBrokerUrlTls:%s",
+		c.log.Warnf("Failed to find the migrated broker url for resource: %d, migratedBrokerUrl: %s, migratedBrokerUrlTls:%s",
 			resourceID,
 			commandTopicMigrated.GetBrokerServiceUrl(),
 			commandTopicMigrated.GetBrokerServiceUrlTls())
diff --git a/pulsar/internal/crypto/producer_encryptor.go b/pulsar/internal/crypto/producer_encryptor.go
@@ -55,19 +55,19 @@ func (e *producerEncryptor) Encrypt(payload []byte, msgMetadata *pb.MessageMetad
 		crypto.NewMessageMetadataSupplier(msgMetadata),
 		payload)
 
-	// error encryping the payload
+	// error encrypting the payload
 	if err != nil {
 		// error occurred in encrypting the payload
 		// crypto ProducerCryptoFailureAction is set to send
-		// send unencrypted message
+		// unencrypted message
 		if e.producerCryptoFailureAction == crypto.ProducerCryptoFailureActionSend {
 			e.logger.
 				WithError(err).
 				Warnf("Encryption failed for payload sending unencrypted message ProducerCryptoFailureAction is set to send")
 			return payload, nil
 		}
 
-		return nil, fmt.Errorf("ProducerCryptoFailureAction is set to Fail and error occurred in encrypting payload :%v", err)
+		return nil, fmt.Errorf("ProducerCryptoFailureAction is set to Fail and error occurred in encrypting payload: %w", err)
 	}
 	return encryptedPayload, nil
 }
diff --git a/pulsar/primitiveSerDe.go b/pulsar/primitiveSerDe.go
@@ -100,14 +100,14 @@ func (b BinaryFreeList) Uint64(r io.Reader, byteOrder binary.ByteOrder) (uint64,
 
 func (b BinaryFreeList) Float64(buf []byte) (float64, error) {
 	if len(buf) < 8 {
-		return 0, fmt.Errorf("cannot decode binary double: %s", io.ErrShortBuffer)
+		return 0, fmt.Errorf("cannot decode binary double: %w", io.ErrShortBuffer)
 	}
 	return math.Float64frombits(binary.BigEndian.Uint64(buf[:8])), nil
 }
 
 func (b BinaryFreeList) Float32(buf []byte) (float32, error) {
 	if len(buf) < 4 {
-		return 0, fmt.Errorf("cannot decode binary float: %s", io.ErrShortBuffer)
+		return 0, fmt.Errorf("cannot decode binary float: %w", io.ErrShortBuffer)
 	}
 	return math.Float32frombits(binary.BigEndian.Uint32(buf[:4])), nil
 }
diff --git a/pulsar/producer_impl.go b/pulsar/producer_impl.go
@@ -143,7 +143,7 @@ func newProducer(client *client, options *ProducerOptions) (*producer, error) {
 				true,
 				client.log.SubLogger(log.Fields{"topic": p.topic}))
 			if err != nil {
-				return nil, fmt.Errorf("unable to get MessageCrypto instance. Producer creation is abandoned. %v", err)
+				return nil, fmt.Errorf("unable to get MessageCrypto instance. Producer creation is abandoned. %w", err)
 			}
 			p.options.Encryption.MessageCrypto = messageCrypto
 		}
diff --git a/pulsar/table_view_impl.go b/pulsar/table_view_impl.go
@@ -41,8 +41,8 @@ type TableViewImpl struct {
 	dataMu sync.Mutex
 	data   map[string]interface{}
 
-	readersMu    sync.Mutex
-	cancelRaders map[string]cancelReader
+	readersMu     sync.Mutex
+	cancelReaders map[string]cancelReader
 
 	listenersMu sync.Mutex
 	listeners   []func(string, interface{}) error
@@ -73,12 +73,12 @@ func newTableView(client *client, options TableViewOptions) (TableView, error) {
 	}
 
 	tv := TableViewImpl{
-		client:       client,
-		options:      options,
-		data:         make(map[string]interface{}),
-		cancelRaders: make(map[string]cancelReader),
-		logger:       logger,
-		closedCh:     make(chan struct{}),
+		client:        client,
+		options:       options,
+		data:          make(map[string]interface{}),
+		cancelReaders: make(map[string]cancelReader),
+		logger:        logger,
+		closedCh:      make(chan struct{}),
 	}
 
 	// Do an initial round of partition update check to make sure we can populate the partition readers
@@ -104,16 +104,16 @@ func (tv *TableViewImpl) partitionUpdateCheck() error {
 	tv.readersMu.Lock()
 	defer tv.readersMu.Unlock()
 
-	for partition, cancelReader := range tv.cancelRaders {
+	for partition, cancelReader := range tv.cancelReaders {
 		if _, ok := partitions[partition]; !ok {
 			cancelReader.cancelFunc()
 			cancelReader.reader.Close()
-			delete(tv.cancelRaders, partition)
+			delete(tv.cancelReaders, partition)
 		}
 	}
 
 	for partition := range partitions {
-		if _, ok := tv.cancelRaders[partition]; !ok {
+		if _, ok := tv.cancelReaders[partition]; !ok {
 			reader, err := newReader(tv.client, ReaderOptions{
 				Topic:          partition,
 				StartMessageID: EarliestMessageID(),
@@ -127,14 +127,14 @@ func (tv *TableViewImpl) partitionUpdateCheck() error {
 			for reader.HasNext() {
 				msg, err := reader.Next(context.Background())
 				if err != nil {
-					tv.logger.Errorf("read next message failed for %s: %w", partition, err)
+					tv.logger.Errorf("read next message failed for %s: %v", partition, err)
 				}
 				if msg != nil {
 					tv.handleMessage(msg)
 				}
 			}
 			ctx, cancelFunc := context.WithCancel(context.Background())
-			tv.cancelRaders[partition] = cancelReader{
+			tv.cancelReaders[partition] = cancelReader{
 				reader:     reader,
 				cancelFunc: cancelFunc,
 			}
@@ -148,7 +148,7 @@ func (tv *TableViewImpl) partitionUpdateCheck() error {
 func (tv *TableViewImpl) periodicPartitionUpdateCheck() {
 	for {
 		if err := tv.partitionUpdateCheck(); err != nil {
-			tv.logger.Errorf("failed to check for changes in number of partitions: %w", err)
+			tv.logger.Errorf("failed to check for changes in number of partitions: %v", err)
 		}
 		select {
 		case <-tv.closedCh:
@@ -236,7 +236,7 @@ func (tv *TableViewImpl) Close() {
 
 	if !tv.closed {
 		tv.closed = true
-		for _, cancelReader := range tv.cancelRaders {
+		for _, cancelReader := range tv.cancelReaders {
 			cancelReader.reader.Close()
 		}
 		close(tv.closedCh)
@@ -259,7 +259,7 @@ func (tv *TableViewImpl) handleMessage(msg Message) {
 
 	for _, listener := range tv.listeners {
 		if err := listener(msg.Key(), reflect.Indirect(payload).Interface()); err != nil {
-			tv.logger.Errorf("table view listener failed for %v: %w", msg, err)
+			tv.logger.Errorf("table view listener failed for %v: %v", msg, err)
 		}
 	}
 }
@@ -268,7 +268,7 @@ func (tv *TableViewImpl) watchReaderForNewMessages(ctx context.Context, reader R
 	for {
 		msg, err := reader.Next(ctx)
 		if err != nil {
-			tv.logger.Errorf("read next message failed for %s: %w", reader.Topic(), err)
+			tv.logger.Errorf("read next message failed for %s: %v", reader.Topic(), err)
 		}
 		var e *Error
 		if (errors.As(err, &e) && e.Result() == ConsumerClosed) || errors.Is(err, context.Canceled) {

Original file line number	Diff line number	Diff line change
`@@ -112,7 +112,7 @@ func ExtractUserName(token oauth2.Token) (string, error) {`
`112`	`112`	`p := jwt.Parser{}`
`113`	`113`	`claims := jwt.MapClaims{}`
`114`	`114`	`if _, _, err := p.ParseUnverified(token.AccessToken, claims); err != nil {`
`115`		`- return "", fmt.Errorf("unable to decode the access token: %v", err)`
	`115`	`+ return "", fmt.Errorf("unable to decode the access token: %w", err)`
`116`	`116`	`}`
`117`	`117`	`username, ok := claims[ClaimNameUserName]`
`118`	`118`	`if !ok {`
Original file line number	Diff line number	Diff line change
`@@ -80,7 +80,7 @@ func (t tokenCache) Token() (xoauth2.Token, error) {`
`80`	`80`	`// load from the store and use the access token if it isn't expired`
`81`	`81`	`grant, err := t.store.LoadGrant(t.audience)`
`82`	`82`	`if err != nil {`
`83`		`- return nil, fmt.Errorf("LoadGrant: %v", err)`
	`83`	`+ return nil, fmt.Errorf("LoadGrant: %w", err)`
`84`	`84`	`}`
`85`	`85`	`t.token = grant.Token`
`86`	`86`	`if t.token != nil && t.validateAccessToken(*t.token) {`
`@@ -90,13 +90,13 @@ func (t tokenCache) Token() (xoauth2.Token, error) {`
`90`	`90`	`// obtain and cache a fresh access token`
`91`	`91`	`grant, err = t.refresher.Refresh(grant)`
`92`	`92`	`if err != nil {`
`93`		`- return nil, fmt.Errorf("RefreshGrant: %v", err)`
	`93`	`+ return nil, fmt.Errorf("RefreshGrant: %w", err)`
`94`	`94`	`}`
`95`	`95`	`t.token = grant.Token`
`96`	`96`	`err = t.store.SaveGrant(t.audience, *grant)`
`97`	`97`	`if err != nil {`
`98`	`98`	`// TODO log rather than throw`
`99`		`- return nil, fmt.Errorf("SaveGrant: %v", err)`
	`99`	`+ return nil, fmt.Errorf("SaveGrant: %w", err)`
`100`	`100`	`}`
`101`	`101`
`102`	`102`	`return t.token, nil`
`@@ -117,14 +117,14 @@ func (t *tokenCache) InvalidateToken() error {`
`117`	`117`	`}`
`118`	`118`	`grant, err := t.store.LoadGrant(t.audience)`
`119`	`119`	`if err != nil {`
`120`		`- return fmt.Errorf("LoadGrant: %v", err)`
	`120`	`+ return fmt.Errorf("LoadGrant: %w", err)`
`121`	`121`	`}`
`122`	`122`	`if grant.Token != nil && grant.Token.AccessToken == previous.AccessToken {`
`123`	`123`	`grant.Token.Expiry = time.Unix(0, 0).Add(expiryDelta)`
`124`	`124`	`err = t.store.SaveGrant(t.audience, *grant)`
`125`	`125`	`if err != nil {`
`126`	`126`	`// TODO log rather than throw`
`127`		`- return fmt.Errorf("SaveGrant: %v", err)`
	`127`	`+ return fmt.Errorf("SaveGrant: %w", err)`
`128`	`128`	`}`
`129`	`129`	`}`
`130`	`130`	`return nil`
Original file line number	Diff line number	Diff line change
`@@ -20,6 +20,7 @@ package store`
`20`	`20`	`import (`
`21`	`21`	`"crypto/sha1"`
`22`	`22`	`"encoding/json"`
	`23`	`+ "errors"`
`23`	`24`	`"fmt"`
`24`	`25`	`"sync"`
`25`	`26`
`@@ -92,7 +93,7 @@ func (f KeyringStore) LoadGrant(audience string) (oauth2.AuthorizationGrant, e`
`92`	`93`
`93`	`94`	`item, err := f.getItem(audience)`
`94`	`95`	`if err != nil {`
`95`		`- if err == keyring.ErrKeyNotFound {`
	`96`	`+ if errors.Is(err, keyring.ErrKeyNotFound) {`
`96`	`97`	`return nil, ErrNoAuthenticationData`
`97`	`98`	`}`
`98`	`99`	`return nil, err`
`@@ -119,10 +120,10 @@ func (f *KeyringStore) WhoAmI(audience string) (string, error) {`
`119`	`120`	`key := hashKeyringKey(audience)`
`120`	`121`	`authItem, err := f.kr.Get(key)`
`121`	`122`	`if err != nil {`
`122`		`- if err == keyring.ErrKeyNotFound {`
	`123`	`+ if errors.Is(err, keyring.ErrKeyNotFound) {`
`123`	`124`	`return "", ErrNoAuthenticationData`
`124`	`125`	`}`
`125`		`- return "", fmt.Errorf("unable to get information from the keyring: %v", err)`
	`126`	`+ return "", fmt.Errorf("unable to get information from the keyring: %w", err)`
`126`	`127`	`}`
`127`	`128`	`return authItem.Label, nil`
`128`	`129`	`}`
`@@ -134,13 +135,13 @@ func (f *KeyringStore) Logout() error {`
`134`	`135`	`var err error`
`135`	`136`	`keys, err := f.kr.Keys()`
`136`	`137`	`if err != nil {`
`137`		`- return fmt.Errorf("unable to get information from the keyring: %v", err)`
	`138`	`+ return fmt.Errorf("unable to get information from the keyring: %w", err)`
`138`	`139`	`}`
`139`	`140`	`for _, key := range keys {`
`140`	`141`	`err = f.kr.Remove(key)`
`141`	`142`	`}`
`142`	`143`	`if err != nil {`
`143`		`- return fmt.Errorf("unable to update the keyring: %v", err)`
	`144`	`+ return fmt.Errorf("unable to update the keyring: %w", err)`
`144`	`145`	`}`
`145`	`146`	`return nil`
`146`	`147`	`}`
`@@ -180,7 +181,7 @@ func (f *KeyringStore) setItem(item storedItem) error {`
`180`	`181`	`}`
`181`	`182`	`err = f.kr.Set(i)`
`182`	`183`	`if err != nil {`
`183`		`- return fmt.Errorf("unable to update the keyring: %v", err)`
	`184`	`+ return fmt.Errorf("unable to update the keyring: %w", err)`
`184`	`185`	`}`
`185`	`186`	`return nil`
`186`	`187`	`}`
Original file line number	Diff line number	Diff line change
`@@ -272,7 +272,7 @@ func serializeMessage(wb Buffer,`
`272`	`272`	`encryptedPayload, err := encryptor.Encrypt(compressedPayload, msgMetadata)`
`273`	`273`	`if err != nil {`
`274`	`274`	`// error occurred while encrypting the payload, ProducerCryptoFailureAction is set to Fail`
`275`		`- return fmt.Errorf("encryption of message failed, ProducerCryptoFailureAction is set to Fail. Error :%v", err)`
	`275`	`+ return fmt.Errorf("encryption of message failed, ProducerCryptoFailureAction is set to Fail. Error: %w", err)`
`276`	`276`	`}`
`277`	`277`
`278`	`278`	`cmdSize := uint32(proto.Size(cmdSend))`
Original file line number	Diff line number	Diff line change
`@@ -100,14 +100,14 @@ func (b BinaryFreeList) Uint64(r io.Reader, byteOrder binary.ByteOrder) (uint64,`
`100`	`100`
`101`	`101`	`func (b BinaryFreeList) Float64(buf []byte) (float64, error) {`
`102`	`102`	`if len(buf) < 8 {`
`103`		`- return 0, fmt.Errorf("cannot decode binary double: %s", io.ErrShortBuffer)`
	`103`	`+ return 0, fmt.Errorf("cannot decode binary double: %w", io.ErrShortBuffer)`
`104`	`104`	`}`
`105`	`105`	`return math.Float64frombits(binary.BigEndian.Uint64(buf[:8])), nil`
`106`	`106`	`}`
`107`	`107`
`108`	`108`	`func (b BinaryFreeList) Float32(buf []byte) (float32, error) {`
`109`	`109`	`if len(buf) < 4 {`
`110`		`- return 0, fmt.Errorf("cannot decode binary float: %s", io.ErrShortBuffer)`
	`110`	`+ return 0, fmt.Errorf("cannot decode binary float: %w", io.ErrShortBuffer)`
`111`	`111`	`}`
`112`	`112`	`return math.Float32frombits(binary.BigEndian.Uint32(buf[:4])), nil`
`113`	`113`	`}`