Chore: deprecated usages (#1329)

* chore: move from CAS to CompareAndSwap

because deprecated

* chore: since go1.20 initialize a random seed is deprecated

Author: miton18

.gitignore

@@ -14,3 +14,5 @@ coverage.html
 perf/perf
 pulsar-perf
 bin
+
+vendor/

go.mod

@@ -25,7 +25,7 @@ require (
 	github.com/spf13/cobra v1.6.1
 	github.com/stretchr/testify v1.9.0
 	github.com/testcontainers/testcontainers-go v0.32.0
-	go.uber.org/atomic v1.7.0
+	go.uber.org/atomic v1.11.0
 	golang.org/x/mod v0.20.0
 	golang.org/x/oauth2 v0.11.0
 	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8

go.sum

@@ -407,6 +407,8 @@ go.opentelemetry.io/proto/otlp v1.0.0 h1:T0TX0tmXU8a3CbNXzEKGeU5mIVOdf0oykP+u2lI
 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=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=

pulsar/consumer_partition.go

@@ -257,7 +257,7 @@ func (p *availablePermits) flowIfNeed() {
 		availablePermits := current
 		requestedPermits := current
 		// check if permits changed
-		if !p.permits.CAS(current, 0) {
+		if !p.permits.CompareAndSwap(current, 0) {
 			return
 		}
 
@@ -2084,13 +2084,13 @@ func (pc *partitionConsumer) expectMoreIncomingMessages() {
 	if !pc.options.autoReceiverQueueSize {
 		return
 	}
-	if pc.scaleReceiverQueueHint.CAS(true, false) {
+	if pc.scaleReceiverQueueHint.CompareAndSwap(true, false) {
 		oldSize := pc.currentQueueSize.Load()
 		maxSize := int32(pc.options.receiverQueueSize)
 		newSize := int32(math.Min(float64(maxSize), float64(oldSize*2)))
 		usagePercent := pc.client.memLimit.CurrentUsagePercent()
 		if usagePercent < receiverQueueExpansionMemThreshold && newSize > oldSize {
-			pc.currentQueueSize.CAS(oldSize, newSize)
+			pc.currentQueueSize.CompareAndSwap(oldSize, newSize)
 			pc.availablePermits.add(newSize - oldSize)
 			pc.log.Debugf("update currentQueueSize from %d -> %d", oldSize, newSize)
 		}
@@ -2116,7 +2116,7 @@ func (pc *partitionConsumer) shrinkReceiverQueueSize() {
 	minSize := int32(math.Min(float64(initialReceiverQueueSize), float64(pc.options.receiverQueueSize)))
 	newSize := int32(math.Max(float64(minSize), float64(oldSize/2)))
 	if newSize < oldSize {
-		pc.currentQueueSize.CAS(oldSize, newSize)
+		pc.currentQueueSize.CompareAndSwap(oldSize, newSize)
 		pc.availablePermits.add(newSize - oldSize)
 		pc.log.Debugf("update currentQueueSize from %d -> %d", oldSize, newSize)
 	}

pulsar/message_chunking_test.go

@@ -57,8 +57,6 @@ func TestInvalidChunkingConfig(t *testing.T) {
 }
 
 func TestLargeMessage(t *testing.T) {
-	rand.Seed(time.Now().Unix())
-
 	client, err := NewClient(ClientOptions{
 		URL: lookupURL,
 	})
@@ -208,7 +206,6 @@ func TestMaxPendingChunkMessages(t *testing.T) {
 }
 
 func TestExpireIncompleteChunks(t *testing.T) {
-	rand.Seed(time.Now().Unix())
 	client, err := NewClient(ClientOptions{
 		URL: lookupURL,
 	})
@@ -240,8 +237,6 @@ func TestExpireIncompleteChunks(t *testing.T) {
 }
 
 func TestChunksEnqueueFailed(t *testing.T) {
-	rand.Seed(time.Now().Unix())
-
 	client, err := NewClient(ClientOptions{
 		URL: lookupURL,
 	})
@@ -278,8 +273,6 @@ func TestChunksEnqueueFailed(t *testing.T) {
 }
 
 func TestSeekChunkMessages(t *testing.T) {
-	rand.Seed(time.Now().Unix())
-
 	client, err := NewClient(ClientOptions{
 		URL: lookupURL,
 	})
@@ -343,8 +336,6 @@ func TestSeekChunkMessages(t *testing.T) {
 }
 
 func TestChunkAckAndNAck(t *testing.T) {
-	rand.Seed(time.Now().Unix())
-
 	client, err := NewClient(ClientOptions{
 		URL: lookupURL,
 	})
@@ -400,8 +391,6 @@ func TestChunkAckAndNAck(t *testing.T) {
 }
 
 func TestChunkSize(t *testing.T) {
-	rand.Seed(time.Now().Unix())
-
 	client, err := NewClient(ClientOptions{
 		URL: lookupURL,
 	})

pulsar/producer_partition.go

@@ -1089,7 +1089,7 @@ func (p *partitionProducer) Send(ctx context.Context, msg *ProducerMessage) (Mes
 	doneCh := make(chan struct{})
 
 	p.internalSendAsync(ctx, msg, func(ID MessageID, _ *ProducerMessage, e error) {
-		if isDone.CAS(false, true) {
+		if isDone.CompareAndSwap(false, true) {
 			err = e
 			msgID = ID
 			close(doneCh)
@@ -1394,7 +1394,8 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 		)
 
 		if sr.totalChunks > 1 {
-			if sr.chunkID == 0 {
+			switch sr.chunkID {
+			case 0:
 				sr.chunkRecorder.setFirstChunkID(
 					&messageID{
 						int64(response.MessageId.GetLedgerId()),
@@ -1403,7 +1404,7 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 						p.partitionIdx,
 						0,
 					})
-			} else if sr.chunkID == sr.totalChunks-1 {
+			case sr.totalChunks - 1:
 				sr.chunkRecorder.setLastChunkID(
 					&messageID{
 						int64(response.MessageId.GetLedgerId()),
@@ -1546,7 +1547,7 @@ func (p *partitionProducer) setProducerState(state producerState) {
 // set a new producerState and return the last state
 // returns bool if the new state has been set or not
 func (p *partitionProducer) casProducerState(oldState, newState producerState) bool {
-	return p.state.CAS(int32(oldState), int32(newState))
+	return p.state.CompareAndSwap(int32(oldState), int32(newState))
 }
 
 func (p *partitionProducer) Close() {

pulsar/producer_test.go

@@ -1977,8 +1977,8 @@ func TestWaitForExclusiveProducer(t *testing.T) {
 			Topic:              topicName,
 			ProducerAccessMode: ProducerAccessModeWaitForExclusive,
 		})
-		defer producer2.Close()
 		assert.NoError(t, err)
+		defer producer2.Close()
 		assert.NotNil(t, producer2)
 
 		id, err := producer2.Send(context.Background(), &ProducerMessage{

pulsar/transaction_coordinator_client.go

@@ -319,7 +319,7 @@ func (t *transactionHandler) endTxn(op *endTxnOp) {
 }
 
 func (t *transactionHandler) close() {
-	if !t.state.CAS(txnHandlerReady, txnHandlerClosed) {
+	if !t.state.CompareAndSwap(txnHandlerReady, txnHandlerClosed) {
 		return
 	}
 	close(t.closeCh)