spring-cloud-gcp: Inconsistent Message Ordering with High-Frequency Publishing Interval

Issue

When publishing multiple events/message having the same ordering key on a interval of 1 to 500 ms, the message ordering is not maintained consistently.

Despite taking all the necessary steps to ensure proper ordering, (mentioned below) the consumed events are received out of order.

• publishing messages to a regional end-point.
• enabling message ordering flag at subscription level and
• enabling message ordering flag at the publisher level.

However, if a sleep delay of approximately 1 second is introduced before publishing the individual events, the messages are received in the expected order.

Steps to reproduce

1. Set up a publisher-subscriber system with proper configuration for message ordering.
2. Publish a sequence of events with the same ordering key, each separated by a 1 to 500 ms interval. (Don’t bring up the consumer yet)
3. Start the consumer that subscribes to a subscription which has message ordering flag enabled.
4. Observe the received order of the consumed events.

Code example

Configurations

  @Bean
  public CredentialsProvider credentialsProvider() {
    return () ->
        ServiceAccountCredentials.fromStream(Files.newInputStream(Paths.get(credentialsFilePath)))
            .createScoped(PublisherStubSettings.getDefaultServiceScopes());
  }

  @Bean
  public PubSubPublisherTemplate pubSubPublisherTemplate(
      CredentialsProvider credentialsProvider, PubSubMessageConverter pubSubMessageConverter) {

    DefaultPublisherFactory factory = new DefaultPublisherFactory(() -> projectId);
    factory.setEnableMessageOrdering(true);
    factory.setCredentialsProvider(credentialsProvider);
    factory.setEndpoint("us-east1-pubsub.googleapis.com:443");

    PubSubPublisherTemplate template = new PubSubPublisherTemplate(factory);
    template.setMessageConverter(pubSubMessageConverter);
    return template;
  }

  @Bean
  public PubSubSubscriberTemplate subscriberTemplate(
      CredentialsProvider credentialsProvider, PubSubMessageConverter pubSubMessageConverter) {

    GcpPubSubProperties gcpPubSubProperties = new GcpPubSubProperties();
    gcpPubSubProperties.setKeepAliveIntervalMinutes(1);
    gcpPubSubProperties.initialize(projectId);

    DefaultSubscriberFactory factory =
        new DefaultSubscriberFactory(() -> projectId, gcpPubSubProperties);
    factory.setCredentialsProvider(credentialsProvider);

    PubSubSubscriberTemplate template = new PubSubSubscriberTemplate(factory);
    template.setMessageConverter(pubSubMessageConverter);
    return template;
  }

  @Bean
  public PubSubTemplate pubSubTemplateBean(
      PubSubPublisherTemplate pubSubPublisherTemplate,
      PubSubSubscriberTemplate subscriberTemplate) {
    return new PubSubTemplate(pubSubPublisherTemplate, subscriberTemplate);
  }

Publisher and Subscriber

  private void publishWithOrdering(List<Item> items) {
    items.forEach(
        item -> {
          log.info(
              "Publishing item (with ordering key) : [{}] with event description : [{}] ",
              item.getItemNumber(),
              item.getItemDescription());

          pubSubTemplate.publish(
              topic(), item, Collections.singletonMap(GcpPubSubHeaders.ORDERING_KEY, "1234"));

          try {
            Thread.sleep(100);
          } catch (InterruptedException e) {
            throw new RuntimeException(e);
          }
        });
  }

@Configuration
public class PubSubListener {
  @Bean
  public Subscriber itemListener(PubSubTemplate pubSubTemplate) {

    return pubSubTemplate.subscribeAndConvert(getSubscription(), this::handleMessage, Item.class);
  }

  public void handleMessage(ConvertedBasicAcknowledgeablePubsubMessage<Item> message) {
    Item payload = message.getPayload();
    log.info(
        "Item received with item ID : [{}] and event description : [{}] with ordering key [{}]",
        payload.getItemNumber(),
        payload.getItemDescription(),
        message.getPubsubMessage().getOrderingKey());

    message.ack();
  }

  public String getSubscription() {
    return "item-topic-subscription";
  }
}

Dependencies

<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>com.google.cloud</groupId>
      <artifactId>spring-cloud-gcp-dependencies</artifactId>
      <version>4.5.1</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>

<dependency>
  <groupId>com.google.cloud</groupId>
  <artifactId>spring-cloud-gcp-starter-pubsub</artifactId>
</dependency>

<dependency>
  <groupId>com.google.cloud</groupId>
  <artifactId>spring-cloud-gcp-starter</artifactId>
</dependency>