Skip to content

Commit 7bf4904

Browse files
hongalexhongalex
authored
feat(pubsub): use Streaming Pull response for ordering check (#9682)
* feat(pubsub): remove subscription config check with ordering * remove old ordering check * put back comment, save properties variable
1 parent 53a1a75 commit 7bf4904

3 files changed

Lines changed: 32 additions & 26 deletions

File tree

‎pubsub/integration_test.go‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1434,9 +1434,6 @@ func TestIntegration_OrderedKeys_SubscriptionOrdering(t *testing.T) {
14341434
msg.Ack()
14351435
atomic.AddInt32(&numAcked, 1)
14361436
})
1437-
if sub.enableOrdering != enableMessageOrdering {
1438-
t.Fatalf("enableOrdering mismatch: got: %v, want: %v", sub.enableOrdering, enableMessageOrdering)
1439-
}
14401437
// If the messages were received on a subscription with the EnableMessageOrdering=true,
14411438
// total processing would exceed the timeout and only one message would be processed.
14421439
if numAcked < 2 {

‎pubsub/iterator.go‎

Lines changed: 27 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,12 @@ type messageIterator struct {
9696
eoMu sync.RWMutex
9797
enableExactlyOnceDelivery bool
9898
sendNewAckDeadline bool
99+
100+
orderingMu sync.RWMutex
101+
// enableOrdering determines if messages should be processed in order. This is populated
102+
// by the response in StreamingPull and can change mid Receive. Must be accessed
103+
// with the lock held.
104+
enableOrdering bool
99105
}
100106

101107
// newMessageIterator starts and returns a new messageIterator.
@@ -352,12 +358,30 @@ func (it *messageIterator) recvMessages() ([]*pb.ReceivedMessage, error) {
352358
if err != nil {
353359
return nil, err
354360
}
355-
it.eoMu.Lock()
356-
if got := res.GetSubscriptionProperties().GetExactlyOnceDeliveryEnabled(); got != it.enableExactlyOnceDelivery {
361+
362+
// If the new exactly once settings are different than the current settings, update it.
363+
it.eoMu.RLock()
364+
enableEOD := it.enableExactlyOnceDelivery
365+
it.eoMu.RUnlock()
366+
367+
subProp := res.GetSubscriptionProperties()
368+
if got := subProp.GetExactlyOnceDeliveryEnabled(); got != enableEOD {
369+
it.eoMu.Lock()
357370
it.sendNewAckDeadline = true
358371
it.enableExactlyOnceDelivery = got
372+
it.eoMu.Unlock()
373+
}
374+
375+
// Also update the subscriber's ordering setting if stale.
376+
it.orderingMu.RLock()
377+
enableOrdering := it.enableOrdering
378+
it.orderingMu.RUnlock()
379+
380+
if got := subProp.GetMessageOrderingEnabled(); got != enableOrdering {
381+
it.orderingMu.Lock()
382+
it.enableOrdering = got
383+
it.orderingMu.Unlock()
359384
}
360-
it.eoMu.Unlock()
361385
return res.ReceivedMessages, nil
362386
}
363387

‎pubsub/subscription.go‎

Lines changed: 5 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -50,8 +50,6 @@ type Subscription struct {
5050

5151
mu sync.Mutex
5252
receiveActive bool
53-
54-
enableOrdering bool
5553
}
5654

5755
// Subscription creates a reference to a subscription.
@@ -1238,8 +1236,6 @@ func (s *Subscription) Receive(ctx context.Context, f func(context.Context, *Mes
12381236
s.mu.Unlock()
12391237
defer func() { s.mu.Lock(); s.receiveActive = false; s.mu.Unlock() }()
12401238

1241-
s.checkOrdering(ctx)
1242-
12431239
// TODO(hongalex): move settings check to a helper function to make it more testable
12441240
maxCount := s.ReceiveSettings.MaxOutstandingMessages
12451241
if maxCount == 0 {
@@ -1392,11 +1388,14 @@ func (s *Subscription) Receive(ctx context.Context, f func(context.Context, *Mes
13921388
iter.eoMu.RUnlock()
13931389

13941390
wg.Add(1)
1395-
// Make sure the subscription has ordering enabled before adding to scheduler.
1391+
// Only schedule messages in order if an ordering key is present and the subscriber client
1392+
// received the ordering flag from a Streaming Pull response.
13961393
var key string
1397-
if s.enableOrdering {
1394+
iter.orderingMu.RLock()
1395+
if iter.enableOrdering {
13981396
key = msg.OrderingKey
13991397
}
1398+
iter.orderingMu.RUnlock()
14001399
msgLen := len(msg.Data)
14011400
if err := sched.Add(key, msg, func(msg interface{}) {
14021401
defer wg.Done()
@@ -1436,20 +1435,6 @@ func (s *Subscription) Receive(ctx context.Context, f func(context.Context, *Mes
14361435
return group.Wait()
14371436
}
14381437

1439-
// checkOrdering calls Config to check theEnableMessageOrdering field.
1440-
// If this call fails (e.g. because the service account doesn't have
1441-
// the roles/viewer or roles/pubsub.viewer role) we will assume
1442-
// EnableMessageOrdering to be true.
1443-
// See: https://github.com/googleapis/google-cloud-go/issues/3884
1444-
func (s *Subscription) checkOrdering(ctx context.Context) {
1445-
cfg, err := s.Config(ctx)
1446-
if err != nil {
1447-
s.enableOrdering = true
1448-
} else {
1449-
s.enableOrdering = cfg.EnableMessageOrdering
1450-
}
1451-
}
1452-
14531438
type pullOptions struct {
14541439
maxExtension time.Duration // the maximum time to extend a message's ack deadline in total
14551440
maxExtensionPeriod time.Duration // the maximum time to extend a message's ack deadline per modack rpc

0 commit comments

Comments
 (0)