deprecate(pubsub/v2): remove and rename existing settings (#11375)

* deprecate(pubsub/v2): deprecate and rename user settings

* remove more code related to synchronous and byte limit

Author: hongalex

pubsub/loadtest/loadtest.go

@@ -66,10 +66,9 @@ func (l *PubServer) Start(ctx context.Context, req *pb.StartRequest) (*pb.StartR
 func (l *PubServer) init(c *pubsub.Client, topicName string, msgSize, batchSize int32, batchDur time.Duration, ordered bool) {
 	topic := c.Topic(topicName)
 	topic.PublishSettings = pubsub.PublishSettings{
-		DelayThreshold:    batchDur,
-		CountThreshold:    950,
-		ByteThreshold:     9500000,
-		BufferedByteLimit: 2e9,
+		DelayThreshold: batchDur,
+		CountThreshold: 950,
+		ByteThreshold:  9500000,
 	}
 	topic.EnableMessageOrdering = ordered
 

pubsub/v2/integration_test.go

@@ -98,14 +98,12 @@ func TestIntegration_PublishReceive(t *testing.T) {
 	ctx := context.Background()
 	client := integrationTestClient(ctx, t)
 
-	for _, sync := range []bool{false, true} {
-		for _, maxMsgs := range []int{0, 3, -1} { // MaxOutstandingMessages = default, 3, unlimited
-			testPublishAndReceive(t, client, maxMsgs, sync, false, 10, 0)
-		}
-
-		// Tests for large messages (larger than the 4MB gRPC limit).
-		testPublishAndReceive(t, client, 0, sync, false, 1, 5*1024*1024)
+	for _, maxMsgs := range []int{0, 3, -1} { // MaxOutstandingMessages = default, 3, unlimited
+		testPublishAndReceive(t, client, maxMsgs, false, 10, 0)
 	}
+
+	// Tests for large messages (larger than the 4MB gRPC limit).
+	testPublishAndReceive(t, client, 0, false, 1, 5*1024*1024)
 }
 
 // withGoogleClientInfo sets the name and version of the application in
@@ -126,8 +124,8 @@ func withGoogleClientInfo(ctx context.Context) context.Context {
 	return metadata.NewOutgoingContext(ctx, metadata.Join(allMDs...))
 }
 
-func testPublishAndReceive(t *testing.T, client *Client, maxMsgs int, synchronous, exactlyOnceDelivery bool, numMsgs, extraBytes int) {
-	t.Run(fmt.Sprintf("maxMsgs:%d,synchronous:%t,exactlyOnceDelivery:%t,numMsgs:%d", maxMsgs, synchronous, exactlyOnceDelivery, numMsgs), func(t *testing.T) {
+func testPublishAndReceive(t *testing.T, client *Client, maxMsgs int, exactlyOnceDelivery bool, numMsgs, extraBytes int) {
+	t.Run(fmt.Sprintf("maxMsgs:%d,exactlyOnceDelivery:%t,numMsgs:%d", maxMsgs, exactlyOnceDelivery, numMsgs), func(t *testing.T) {
 		t.Parallel()
 		testutil.Retry(t, 3, 10*time.Second, func(r *testutil.R) {
 			ctx := context.Background()
@@ -193,7 +191,6 @@ func testPublishAndReceive(t *testing.T, client *Client, maxMsgs int, synchronou
 			}
 
 			sub.ReceiveSettings.MaxOutstandingMessages = maxMsgs
-			sub.ReceiveSettings.Synchronous = synchronous
 
 			// Use a timeout to ensure that Pull does not block indefinitely if there are
 			// unexpectedly few messages available.
@@ -219,8 +216,8 @@ func testPublishAndReceive(t *testing.T, client *Client, maxMsgs int, synchronou
 				got[md.ID] = md
 			}
 			if !testutil.Equal(got, want) {
-				r.Errorf("MaxOutstandingMessages=%d, Synchronous=%t, messages got: %+v, messages want: %+v",
-					maxMsgs, synchronous, got, want)
+				r.Errorf("MaxOutstandingMessages=%d, messages got: %+v, messages want: %+v",
+					maxMsgs, got, want)
 			}
 		})
 	})
@@ -601,7 +598,6 @@ func TestIntegration_OrderedKeys_ResumePublish(t *testing.T) {
 		t.Fatalf("topic %v should exist, but it doesn't", topic)
 	}
 
-	topic.PublishSettings.BufferedByteLimit = 100
 	topic.EnableMessageOrdering = true
 
 	orderingKey := "some-ordering-key2"
@@ -792,7 +788,7 @@ func TestIntegration_ExactlyOnceDelivery_PublishReceive(t *testing.T) {
 	client := integrationTestClient(ctx, t)
 
 	for _, maxMsgs := range []int{0, 3, -1} { // MaxOutstandingMessages = default, 3, unlimited
-		testPublishAndReceive(t, client, maxMsgs, false, true, 10, 0)
+		testPublishAndReceive(t, client, maxMsgs, true, 10, 0)
 	}
 }
 

pubsub/v2/iterator.go

@@ -27,13 +27,11 @@ import (
 	"cloud.google.com/go/pubsub/internal/distribution"
 	vkit "cloud.google.com/go/pubsub/v2/apiv1"
 	pb "cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
-	gax "github.com/googleapis/gax-go/v2"
 	"github.com/googleapis/gax-go/v2/apierror"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/propagation"
 	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
 	"go.opentelemetry.io/otel/trace"
-	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/encoding/protowire"
@@ -126,16 +124,10 @@ type messageIterator struct {
 // Stop must be called on the messageIterator when it is no longer needed.
 // The iterator always uses the background context for acking messages and extending message deadlines.
 func newMessageIterator(subc *vkit.SubscriptionAdminClient, subName string, po *pullOptions) *messageIterator {
-	var ps *pullStream
-	if !po.synchronous {
-		maxMessages := po.maxOutstandingMessages
-		maxBytes := po.maxOutstandingBytes
-		if po.useLegacyFlowControl {
-			maxMessages = 0
-			maxBytes = 0
-		}
-		ps = newPullStream(context.Background(), subc.StreamingPull, subName, po.clientID, maxMessages, maxBytes, po.maxExtensionPeriod)
-	}
+	maxMessages := po.maxOutstandingMessages
+	maxBytes := po.maxOutstandingBytes
+
+	ps := newPullStream(context.Background(), subc.StreamingPull, subName, po.clientID, maxMessages, maxBytes, po.maxExtensionPeriod)
 	// The period will update each tick based on the distribution of acks. We'll start by arbitrarily sending
 	// the first keepAlive halfway towards the minimum ack deadline.
 	keepAlivePeriod := minDurationPerLeaseExtension / 2
@@ -270,21 +262,17 @@ func (it *messageIterator) receive(maxToPull int32) ([]*Message, error) {
 
 	var rmsgs []*pb.ReceivedMessage
 	var err error
-	if it.po.synchronous {
-		rmsgs, err = it.pullMessages(maxToPull)
-	} else {
-		rmsgs, err = it.recvMessages()
-		// If stopping the iterator results in the grpc stream getting shut down and
-		// returning an error here, treat the same as above and return EOF.
-		// If the cancellation comes from the underlying grpc client getting closed,
-		// do propagate the cancellation error.
-		// See https://github.com/googleapis/google-cloud-go/pull/10153#discussion_r1600814775
-		if err != nil && errors.Is(it.ps.ctx.Err(), context.Canceled) {
+	rmsgs, err = it.recvMessages()
+	// If stopping the iterator results in the grpc stream getting shut down and
+	// returning an error here, treat the same as above and return EOF.
+	// If the cancellation comes from the underlying grpc client getting closed,
+	// do propagate the cancellation error.
+	// See https://github.com/googleapis/google-cloud-go/pull/10153#discussion_r1600814775
+	if err != nil {
+		if errors.Is(it.ps.ctx.Err(), context.Canceled) {
 			err = io.EOF
 		}
-	}
-	// Any error here is fatal.
-	if err != nil {
+		// Any error here is fatal.
 		return nil, it.fail(err)
 	}
 
@@ -396,27 +384,6 @@ func (it *messageIterator) receive(maxToPull int32) ([]*Message, error) {
 	return nil, nil
 }
 
-// Get messages using the Pull RPC.
-// This may block indefinitely. It may also return zero messages, after some time waiting.
-func (it *messageIterator) pullMessages(maxToPull int32) ([]*pb.ReceivedMessage, error) {
-	// Use it.ctx as the RPC context, so that if the iterator is stopped, the call
-	// will return immediately.
-	res, err := it.subc.Pull(it.ctx, &pb.PullRequest{
-		Subscription: it.subName,
-		MaxMessages:  maxToPull,
-	}, gax.WithGRPCOptions(grpc.MaxCallRecvMsgSize(maxSendRecvBytes)))
-	switch {
-	case errors.Is(err, context.Canceled):
-		return nil, nil
-	case status.Code(err) == codes.Canceled:
-		return nil, nil
-	case err != nil:
-		return nil, err
-	default:
-		return res.ReceivedMessages, nil
-	}
-}
-
 func (it *messageIterator) recvMessages() ([]*pb.ReceivedMessage, error) {
 	res, err := it.ps.Recv()
 	if err != nil {
@@ -510,7 +477,7 @@ func (it *messageIterator) sender() {
 		case <-it.pingTicker.C:
 			it.mu.Lock()
 			// Ping only if we are processing messages via streaming.
-			sendPing = !it.po.synchronous
+			sendPing = true
 		case <-it.receiptTicker.C:
 			it.mu.Lock()
 			sendReceipt = (len(it.pendingReceipts) > 0)

pubsub/v2/iterator_test.go

@@ -402,27 +402,6 @@ func TestIterator_ModifyAckContextDeadline(t *testing.T) {
 	}
 }
 
-func TestIterator_SynchronousPullCancel(t *testing.T) {
-	srv := pstest.NewServer()
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	srv.Publish(fullyQualifiedTopicName, []byte("creating a topic"), nil)
-
-	_, client, err := initConn(ctx, srv.Addr)
-	if err != nil {
-		t.Fatal(err)
-	}
-	iter := newMessageIterator(client.subc, fullyQualifiedTopicName, &pullOptions{})
-
-	// Cancelling the iterator and pulling should not result in any errors.
-	iter.cancel()
-
-	if _, err := iter.pullMessages(100); err != nil {
-		t.Fatalf("Got error in pullMessages: %v", err)
-	}
-}
-
 func TestIterator_BoundedDuration(t *testing.T) {
 	// Use exported fields for time.Duration fields so they
 	// print nicely. Otherwise, they will print as integers.
@@ -529,7 +508,6 @@ func TestIterator_StreamingPullExactlyOnce(t *testing.T) {
 	srv.Publish(fullyQualifiedTopicName, []byte("msg"), nil)
 
 	iter := newMessageIterator(client.subc, fullyQualifiedSubName, &pullOptions{
-		synchronous:            false,
 		maxOutstandingMessages: 100,
 		maxOutstandingBytes:    1e6,
 		maxPrefetch:            30,