Pub/Sub message ordering support

[skymeyer]

Fixes #723

Message ordering may require producing the events using an explicit regional endpoint. Not sure if we should add a helper to produce a pubsub client using a regional service endpoint or leave this as an exercise to the integrator.

[n3wscott]

oh awesome. taking a look, back from kubecon

[n3wscott]

Looks like you got tripped up on a data race in the unit test, could you take a look @skymeyer ? overall this code looks good, it might be a simple fix for just the test.

WARNING: DATA RACE
Write at 0x00c00028b9f0 by goroutine 91:
  github.com/cloudevents/sdk-go/protocol/pubsub/v2/internal.(*Connection).getOrCreateTopicInfo()
      /home/runner/work/sdk-go/sdk-go/protocol/pubsub/v2/internal/connection.go:143 +0x43a
  github.com/cloudevents/sdk-go/protocol/pubsub/v2/internal.(*Connection).getOrCreateTopic()
      /home/runner/work/sdk-go/sdk-go/protocol/pubsub/v2/internal/connection.go:149 +0x74
  github.com/cloudevents/sdk-go/protocol/pubsub/v2/internal.(*Connection).Publish()
      /home/runner/work/sdk-go/sdk-go/protocol/pubsub/v2/internal/connection.go:300 +0x6d
  github.com/cloudevents/sdk-go/protocol/pubsub/v2/internal.TestPublishReceiveRoundtripWithOrderingKey()
      /home/runner/work/sdk-go/sdk-go/protocol/pubsub/v2/internal/connection_test.go:790 +0x946
  testing.tRunner()
      /opt/hostedtoolcache/go/1.14.15/x64/src/testing/testing.go:1050 +0x1eb

Previous write at 0x00c00028b9f0 by goroutine 162:
  github.com/cloudevents/sdk-go/protocol/pubsub/v2/internal.(*Connection).getOrCreateTopicInfo()
      /home/runner/work/sdk-go/sdk-go/protocol/pubsub/v2/internal/connection.go:143 +0x43a
  github.com/cloudevents/sdk-go/protocol/pubsub/v2/internal.(*Connection).getOrCreateTopic()
      /home/runner/work/sdk-go/sdk-go/protocol/pubsub/v2/internal/connection.go:149 +0x74
  github.com/cloudevents/sdk-go/protocol/pubsub/v2/internal.(*Connection).getOrCreateSubscriptionInfo.func1()
      /home/runner/work/sdk-go/sdk-go/protocol/pubsub/v2/internal/connection.go:224 +0x643
  sync.(*Once).doSlow()
      /opt/hostedtoolcache/go/1.14.15/x64/src/sync/once.go:66 +0x103
  sync.(*Once).Do()
      /opt/hostedtoolcache/go/1.14.15/x64/src/sync/once.go:57 +0x68
  github.com/cloudevents/sdk-go/protocol/pubsub/v2/internal.(*Connection).getOrCreateSubscriptionInfo()
      /home/runner/work/sdk-go/sdk-go/protocol/pubsub/v2/internal/connection.go:207 +0x202
  github.com/cloudevents/sdk-go/protocol/pubsub/v2/internal.(*Connection).getOrCreateSubscription()
      /home/runner/work/sdk-go/sdk-go/protocol/pubsub/v2/internal/connection.go:266 +0x74
  github.com/cloudevents/sdk-go/protocol/pubsub/v2/internal.(*Connection).Receive()
      /home/runner/work/sdk-go/sdk-go/protocol/pubsub/v2/internal/connection.go:313 +0x79

Goroutine 91 (running) created at:
  testing.(*T).Run()
      /opt/hostedtoolcache/go/1.14.15/x64/src/testing/testing.go:1095 +0x537
  testing.runTests.func1()
      /opt/hostedtoolcache/go/1.14.15/x64/src/testing/testing.go:1339 +0xa6
  testing.tRunner()
      /opt/hostedtoolcache/go/1.14.15/x64/src/testing/testing.go:1050 +0x1eb
  testing.runTests()
      /opt/hostedtoolcache/go/1.14.15/x64/src/testing/testing.go:1337 +0x594
  testing.(*M).Run()
      /opt/hostedtoolcache/go/1.14.15/x64/src/testing/testing.go:1252 +0x2ff
  main.main()
      _testmain.go:134 +0x347

Goroutine 162 (running) created at:
  github.com/cloudevents/sdk-go/protocol/pubsub/v2/internal.TestPublishReceiveRoundtripWithOrderingKey()
      /home/runner/work/sdk-go/sdk-go/protocol/pubsub/v2/internal/connection_test.go:772 +0x5bd
  testing.tRunner()
      /opt/hostedtoolcache/go/1.14.15/x64/src/testing/testing.go:1050 +0x1eb
==================
    testing.go:965: race detected during execution of test
--- FAIL: TestPublishReceiveRoundtripWithOrderingKey (1.13s)

[skymeyer]

Thanks @n3wscott - will take a look, prob need to add some lock.

[skymeyer]

@n3wscott I was setting ti.topic.EnableMessageOrdering at an incorrect spot, should be addressed now.

[rriolobos]

I tested pubsub ordering key with this library but I found an issue. When you send 3 or more messages, with same ordering key, receiver only waits for first one, second and third messages are consumed concurrently. I added a 5 second sleep in message handler in order to check that the total amount of time was 15 seconds as expected, but it was 10 second. It can also be checked with logs. When I try the same directly invoking pubsub receiver, it worked as expected.

func receive(ctx context.Context, event event.Event) error {
	fmt.Printf("Event Context: %+v\n", event.Context)

	fmt.Printf("Protocol Context: %+v\n", pscontext.ProtocolContextFrom(ctx))

	data := &Example{}
	if err := event.DataAs(data); err != nil {
		fmt.Printf("Got Data Error: %s\n", err.Error())
	}

	time.Sleep(8 * time.Second)

	fmt.Printf("Data processed: %+v\n", data)

	fmt.Printf("----------------------------\n")

	return nil
}

func main() {
	ctx := context.Background()

	var env envConfig
	if err := envconfig.Process("", &env); err != nil {
		log.Printf("[ERROR] Failed to process env var: %s", err)
		os.Exit(1)
	}

	t, err := cepubsub.New(context.Background(),
		cepubsub.WithProjectID(env.ProjectID),
		cepubsub.WithTopicID(env.TopicID),
		cepubsub.WithReceiveSettings(&pubsub.ReceiveSettings{}),
		cepubsub.WithSubscriptionID(env.SubscriptionID))
	if err != nil {
		log.Fatalf("failed to create pubsub protocol, %s", err.Error())
	}

	t.ReceiveSettings.NumGoroutines = 1

	c, err := cloudevents.NewClient(t)

	if err != nil {
		log.Fatalf("failed to create client, %s", err.Error())
	}

	log.Println("Created client, listening...")

	if err := c.StartReceiver(ctx, receive); err != nil {
		log.Fatalf("failed to start pubsub receiver, %s", err.Error())
	}
}

[skymeyer]

Question from @rriolobos is tracked under #753.