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 PubSubMessage
s, 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());