You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2025/02/20 12:58:53 UTC
(pulsar-client-go) branch master updated: Chore: deprecated usages (#1329)
This is an automated email from the ASF dual-hosted git repository.
zixuan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-client-go.git
The following commit(s) were added to refs/heads/master by this push:
new 020b55c1 Chore: deprecated usages (#1329)
020b55c1 is described below
commit 020b55c1dce4475fe8781f4fb6f858de903844b8
Author: Collignon-Ducret Rémi <mi...@users.noreply.github.com>
AuthorDate: Thu Feb 20 13:58:48 2025 +0100
Chore: deprecated usages (#1329)
* chore: move from CAS to CompareAndSwap
because deprecated
* chore: since go1.20 initialize a random seed is deprecated
---
.gitignore | 2 ++
go.mod | 2 +-
go.sum | 2 ++
pulsar/consumer_partition.go | 8 ++++----
pulsar/message_chunking_test.go | 11 -----------
pulsar/producer_partition.go | 9 +++++----
pulsar/producer_test.go | 2 +-
pulsar/transaction_coordinator_client.go | 2 +-
8 files changed, 16 insertions(+), 22 deletions(-)
diff --git a/.gitignore b/.gitignore
index 3db3c8c7..bfa04f03 100644
--- a/.gitignore
+++ b/.gitignore
@@ -14,3 +14,5 @@ coverage.html
perf/perf
pulsar-perf
bin
+
+vendor/
diff --git a/go.mod b/go.mod
index 6f09ee18..7cf845ab 100644
--- a/go.mod
+++ b/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
diff --git a/go.sum b/go.sum
index e8237bb9..e6339326 100644
--- a/go.sum
+++ b/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=
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 24ffa401..13fa9b4c 100644
--- a/pulsar/consumer_partition.go
+++ b/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)
}
diff --git a/pulsar/message_chunking_test.go b/pulsar/message_chunking_test.go
index 59fdb5ec..c2e1113d 100644
--- a/pulsar/message_chunking_test.go
+++ b/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,
})
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index 4b950dff..448f780c 100755
--- a/pulsar/producer_partition.go
+++ b/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() {
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index 3af4a89c..b8279299 100644
--- a/pulsar/producer_test.go
+++ b/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{
diff --git a/pulsar/transaction_coordinator_client.go b/pulsar/transaction_coordinator_client.go
index 9382da6e..ec29b032 100644
--- a/pulsar/transaction_coordinator_client.go
+++ b/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)