4

Our spring boot app uses Messaging Gateway to publish and consume message from GCP pubsub. The existing topics and their subscriptions which we use does not have ordering enabled. Now we want to add a topic and subscription with ordering of messages enabled. I have tested this using a simple python script and works exactly as we want. Now I am implementing the publisher in our Spring boot app. The message gets published correctly but the ordering key which i am sending as the message header is not reflecting and my consumer also is not able to fetch it.

Here is the MessageGateway

@MessagingGateway(defaultRequestChannel = "OutboundChannel")
public interface FiFoGateway {
  void execute(T message);
}

Publisher code

Message<MyPubSubMessage> orderedMessage = MessageBuilder
                .withPayload(pubSubMessage)
                .setHeader("orderingKey", "key1")
                .build();
fiFoGateway.execute(orderedMessage);

My publisher config

  @Bean("fIFOPublisher")
  @ServiceActivator(inputChannel = "OutboundChannel")
  public PubSubMessageHandler fiFOPublisher(PubSubTemplate pubSubTemplate) {
    return new PubSubMessageHandler(pubSubTemplate, "myTopic");
  }

My consumer also not consuming it.

It isappearing in the GCP page but we can see the ordering key colum in empty. enter image description here

Not sure what am I missing ? Any suggestions or pointers ?

4
  • Did you try to purge your subscription and to test again? It could be old messages first. Commented Nov 4, 2023 at 7:35
  • @guillaumeblaquiere I purged the messages and tried again. The ordering key is not coming in the new messages.
    – Tiny Rick
    Commented Nov 4, 2023 at 10:53
  • @TinyRick did you recreate subscription with ordering enabled or you are using old subscription? Commented Nov 7, 2023 at 16:37
  • @ChandanPasunoori I created a fresh topic and created a fresh subscription with ordering enabled. Still no luck. The spring boot version i am using is 2.3.4 and spring cloud gcp version is 1.2.8.RELEASE. I suspect there is no support for ordering in this version
    – Tiny Rick
    Commented Nov 9, 2023 at 4:46

1 Answer 1

1

All the Pub/Sub client APIs offer a flag that allows you to enable message ordering.

Using Spring Cloud GCP, which internally uses the Pub/Sub Java client library, the easiest way to set this flag is setting the configuration property spring.cloud.gcp.pubsub.publisher.enable-message-ordering to true, it is false by default.

Please, consider review the Spring GCP documentation, especially when they describe ordering messages when using PubSubTemplate and mapping headers with Spring integration.

Please, be aware that in order to use the mentioned configuration property you need at least the 2.0.2 version of the Spring Cloud GCP Pub/Sub related libraries: the change was addressed in this issue and first included in the aforementioned library version.

If you cannot upgrade your libraries, you may try the following.

First, define a custom PubSubMessageConverter, the class in charge of performing the actual creation of the PubSubMessages, to deal with message ordering, something like:

public class OrderingAwarePubSubNessageConverter extends SimplePubSubMessageConverter {

  private final Charset charset;

  public OrderingAwarePubSubNessageConverter() {
    this(Charset.defaultCharset());
  }

  public OrderingAwarePubSubNessageConverter(Charset charset) {
    super(charset);

    this.charset = charset;
  }

  // Override the toPubSubMessage conversion method including the logic for ordering messages
  @Override
  public PubsubMessage toPubSubMessage(Object payload, Map<String, String> headers) {
    ByteString convertedPayload;
    if (payload instanceof ByteString) {
      convertedPayload = (ByteString)payload;
    } else if (payload instanceof String) {
      convertedPayload = ByteString.copyFrom(((String)payload).getBytes(this.charset));
    } else if (payload instanceof ByteBuffer) {
      convertedPayload = ByteString.copyFrom((ByteBuffer)payload);
    } else {
      if (!(payload instanceof byte[])) {
        throw new PubSubMessageConversionException("Unable to convert payload of type " + payload.getClass().getName() + " to byte[] for sending to Pub/Sub.");
      }

      convertedPayload = ByteString.copyFrom((byte[])payload);
    }

    PubsubMessage.Builder pubsubMessageBuilder = PubsubMessage.newBuilder().setData(convertedPayload);
    if (headers != null) {
      pubsubMessageBuilder.putAllAttributes(headers);

      // This code is the one required for instructing PubsubMessage.Builder to order messages
      if (headers.containsKey("yourOrderingKeyHeader")) {
        pubsubMessageBuilder.removeAttributes("yourOrderingKeyHeader");
        pubsubMessageBuilder.setOrderingKey(headers.get("yourOrderingKeyHeader"));
      }
    }

    return pubsubMessageBuilder.build();
  }
}

Please, note the code related to ordering messages.

Then, configure your PubSubTemplate for using an instance of this new converter. Please, define in your code:

pubSubTemplate.setMessageConverter(new OrderingAwarePubSubNessageConverter());
3
  • I actually followed this link databasesandlife.com/spring-pubsub-ordering The link mentions the property enabling. However the spring version the link mentions is different from mine. The spring boot version i am using is 2.3.4 and spring cloud gcp version is 1.2.8.RELEASE. I suspect there is no support for ordering in this version
    – Tiny Rick
    Commented Nov 9, 2023 at 4:48
  • You are right @TinyRick. Please, see the updated answer. Please, try upgrading the version of your Spring Cloud GCP library, I think it may be possible. Commented Nov 9, 2023 at 10:26
  • @TinyRick I updated the answer with a possible solution based on the library version you are currently using. I hope it helps. Commented Nov 10, 2023 at 23:08

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.