Exposes publisher endpoint for Pub/Sub message ordering

docs/src/main/asciidoc/pubsub.adoc

@@ -79,6 +79,10 @@ After this amount of time has elapsed (counting from the first element added), t
 Enables batching. | No | false
 | `spring.cloud.gcp.pubsub.publisher.enable-message-ordering`|
 Enables message ordering. | No | false
+| `spring.cloud.gcp.pubsub.publisher.endpoint`|
+The publisher endpoint.
+Example: `"us-east1-pubsub.googleapis.com:443"`.
+This is useful in conjunction with enabling message ordering because sending messages to the same region ensures they are received in order even when multiple publishers are used. | No | pubsub.googleapis.com:443
 |===
 
 ==== GRPC Connection Settings
@@ -182,6 +186,7 @@ By default, the `SimplePubSubMessageConverter` is used to convert payloads of ty
 
 If you are relying on message converters and would like to provide an ordering key, use the `GcpPubSubHeaders.ORDERING_KEY` header.
 You will also need to make sure to enable message ordering on the publisher via the `spring.cloud.gcp.pubsub.publisher.enable-message-ordering` property.
+Additionally, if you are using multiple publishers, you will want to set the `spring.cloud.gcp.pubsub.publisher.endpoint` to a regional endpoint such as `"us-east1-pubsub.googleapis.com:443"` so that messages are sent to the same region and received in order.
 
 [source,java,indent=0]
 ----

spring-cloud-gcp-autoconfigure/src/main/java/com/google/cloud/spring/autoconfigure/pubsub/GcpPubSubAutoConfiguration.java

@@ -310,6 +310,7 @@ public PublisherFactory defaultPublisherFactory(
 		retrySettings.ifAvailable(factory::setRetrySettings);
 		batchingSettings.ifAvailable(factory::setBatchingSettings);
 		factory.setEnableMessageOrdering(gcpPubSubProperties.getPublisher().getEnableMessageOrdering());
+		factory.setEndpoint(gcpPubSubProperties.getPublisher().getEndpoint());
 		return factory;
 	}
 

spring-cloud-gcp-autoconfigure/src/main/java/com/google/cloud/spring/autoconfigure/pubsub/GcpPubSubProperties.java

@@ -131,6 +131,11 @@ public static class Publisher {
 		 */
 		private Boolean enableMessageOrdering;
 
+		/**
+		 * Set publisher endpoint. Example: "us-east1-pubsub.googleapis.com:443".
+		 */
+		private String endpoint;
+
 		public Batching getBatching() {
 			return this.batching;
 		}
@@ -154,6 +159,14 @@ public Boolean getEnableMessageOrdering() {
 		public void setEnableMessageOrdering(Boolean enableMessageOrdering) {
 			this.enableMessageOrdering = enableMessageOrdering;
 		}
+
+		public String getEndpoint() {
+			return endpoint;
+		}
+
+		public void setEndpoint(String endpoint) {
+			this.endpoint = endpoint;
+		}
 	}
 
 	/**

spring-cloud-gcp-autoconfigure/src/test/java/com/google/cloud/spring/autoconfigure/pubsub/it/PubSubTemplateDocumentationIntegrationTests.java

@@ -67,7 +67,8 @@ public class PubSubTemplateDocumentationIntegrationTests {
 
 	private ApplicationContextRunner contextRunner = new ApplicationContextRunner()
 			.withPropertyValues("spring.cloud.gcp.pubsub.subscriber.max-ack-extension-period=0",
-					"spring.cloud.gcp.pubsub.publisher.enable-message-ordering=true")
+					"spring.cloud.gcp.pubsub.publisher.enable-message-ordering=true",
+					"spring.cloud.gcp.pubsub.publisher.endpoint=us-east1-pubsub.googleapis.com:443")
 			.withConfiguration(AutoConfigurations.of(GcpContextAutoConfiguration.class,
 					GcpPubSubAutoConfiguration.class));
 

spring-cloud-gcp-pubsub/src/main/java/com/google/cloud/spring/pubsub/support/DefaultPublisherFactory.java

@@ -63,6 +63,8 @@ public class DefaultPublisherFactory implements PublisherFactory {
 
 	private Boolean enableMessageOrdering;
 
+	private String endpoint;
+
 	/**
 	 * Create {@link DefaultPublisherFactory} instance based on the provided {@link GcpProjectIdProvider}.
 	 * <p>The {@link GcpProjectIdProvider} must not be null, neither provide an empty {@code projectId}.
@@ -133,6 +135,17 @@ public void setEnableMessageOrdering(Boolean enableMessageOrdering) {
 		this.enableMessageOrdering = enableMessageOrdering;
 	}
 
+	/**
+	 * Set the publisher endpoint. Example: "us-east1-pubsub.googleapis.com:443".
+	 * This is useful in conjunction with enabling message ordering because
+	 * sending messages to the same region ensures they are received in order
+	 * even when multiple publishers are used.
+	 * @param endpoint publisher endpoint
+	 */
+	public void setEndpoint(String endpoint) {
+		this.endpoint = endpoint;
+	}
+
 	@Override
 	public Publisher createPublisher(String topic) {
 		return this.publishers.computeIfAbsent(topic, key -> {
@@ -179,6 +192,10 @@ void applyPublisherSettings(Publisher.Builder publisherBuilder) {
 		if (this.enableMessageOrdering != null) {
 			publisherBuilder.setEnableMessageOrdering(this.enableMessageOrdering);
 		}
+
+		if (this.endpoint != null) {
+			publisherBuilder.setEndpoint(this.endpoint);
+		}
 	}
 
 	Map<String, Publisher> getCache() {