Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

test(pubsub): dedupe by message data in ordering keys json test #8526

Merged
merged 6 commits into from
Sep 11, 2023
Merged

test(pubsub): dedupe by message data in ordering keys json test #8526

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
bbc2680
test(pubsub): dedupe by message data in ordering keys json test
hongalex Sep 5, 2023
e3ecd0a
switch cmp.Diff order so the error logging makes sense
hongalex Sep 5, 2023
93dc88e
Merge branch 'main' into pubsub-test-ordered-keys-flake
hongalex Sep 7, 2023
cf070d6
switch to using t.Logf
hongalex Sep 11, 2023
86cdab4
merge
hongalex Sep 11, 2023
933d9ad
Merge branch 'main' into pubsub-test-ordered-keys-flake
hongalex Sep 11, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 14 additions & 6 deletions pubsub/integration_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1263,11 +1263,17 @@ func TestIntegration_OrderedKeys_JSON(t *testing.T) {
key := parts[0]
msg := parts[1]
publishData = append(publishData, testutil2.OrderedKeyMsg{Key: key, Data: msg})
topic.Publish(ctx, &Message{
ID: msg,
r := topic.Publish(ctx, &Message{
Data: []byte(msg),
OrderingKey: key,
})
go func() {
_, err := r.Get(ctx)
if err != nil {
// Can't fail inside goroutine, so just log the error.
t.Logf("publish error for message(%s): %v", msg, err)
}
}()
wg.Add(1)
}
if err := scanner.Err(); err != nil {
Expand All @@ -1276,15 +1282,17 @@ func TestIntegration_OrderedKeys_JSON(t *testing.T) {

go func() {
if err := sub.Receive(ctx, func(ctx context.Context, msg *Message) {
defer msg.Ack()
mu.Lock()
defer mu.Unlock()
if _, ok := receiveSet[msg.ID]; ok {
// Messages are deduped using the data field, since in this case all
// messages are unique.
if _, ok := receiveSet[string(msg.Data)]; ok {
return
}
receiveSet[msg.ID] = struct{}{}
receiveSet[string(msg.Data)] = struct{}{}
receiveData = append(receiveData, testutil2.OrderedKeyMsg{Key: msg.OrderingKey, Data: string(msg.Data)})
wg.Done()
msg.Ack()
}); err != nil {
if c := status.Code(err); c != codes.Canceled {
t.Error(err)
Expand All @@ -1307,7 +1315,7 @@ func TestIntegration_OrderedKeys_JSON(t *testing.T) {
mu.Lock()
defer mu.Unlock()
if err := testutil2.VerifyKeyOrdering(publishData, receiveData); err != nil {
t.Fatalf("CreateTopic error: %v", err)
t.Fatalf("VerifyKeyOrdering error: %v", err)
}
}

Expand Down
2 changes: 1 addition & 1 deletion pubsub/internal/testutil/verifier.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ func VerifyKeyOrdering(publishData, receiveData []OrderedKeyMsg) error {
return fmt.Errorf("saw key %s, but we never published this key", k)
}

if diff := cmp.Diff(pb, rd); diff != "" {
if diff := cmp.Diff(rd, pb); diff != "" {
return fmt.Errorf("%s: got -, want +\n\t%s", k, diff)
}
}
Expand Down