Forward porting changes from 2.0.4.patch branch

CHANGELOG.adoc

@@ -5,9 +5,13 @@ https://spring.io/projects/spring-cloud-gcp[Spring Cloud GCP] is a set of integr
 This document provides a high-level overview of the changes introduced in Spring Cloud GCP by release.
 For a detailed view of what has changed, refer to the https://github.com/GoogleCloudPlatform/spring-cloud-gcp/commits/main[commit history] on GitHub.
 
-== 2.0.5.BUILD-SNAPSHOT
+== 2.0.5 (2021-10-25)
 
-== 2.0.4
+=== Pub/Sub
+
+* Fixed: Allow overriding Pub/Sub retryableCodes in pull settings (https://github.com/GoogleCloudPlatform/spring-cloud-gcp/pull/670[#670]).
+
+== 2.0.4 (2021-08-11)
 
 === General
 

docs/src/main/asciidoc/pubsub.adoc

@@ -97,6 +97,7 @@ They do *not* control message redelivery; only message acknowledgement deadline
 |===
 | Name | Description | Required | Default value
 | `spring.cloud.gcp.pubsub.keepAliveIntervalMinutes` | Determines frequency of keepalive gRPC ping | No | `5 minutes`
+| `spring.cloud.gcp.pubsub.subscriber.retryableCodes` | RPC status codes that should be retried when pulling messages. | No | UNKNOWN,ABORTED,UNAVAILABLE
 | `spring.cloud.gcp.pubsub.[subscriber,publisher].retry.total-timeout-seconds`|
 TotalTimeout has ultimate control over how long the logic should keep trying the remote call until it gives up completely.
 The higher the total timeout, the more retries can be attempted. | No | 0

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

@@ -235,6 +235,9 @@ public SubscriberFactory defaultSubscriberFactory(
 					"The subscriberRetrySettings bean is being deprecated. Please use application.properties to configure properties");
 			factory.setSubscriberStubRetrySettings(retrySettings.getIfAvailable());
 		}
+		if (this.gcpPubSubProperties.getSubscriber().getRetryableCodes() != null) {
+			factory.setRetryableCodes(gcpPubSubProperties.getSubscriber().getRetryableCodes());
+		}
 		return factory;
 	}
 

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

@@ -22,6 +22,7 @@
 import com.google.api.gax.core.ExecutorProvider;
 import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
 import com.google.api.gax.retrying.RetrySettings;
+import com.google.api.gax.rpc.StatusCode.Code;
 import com.google.api.gax.rpc.TransportChannelProvider;
 import com.google.auth.Credentials;
 import com.google.cloud.spring.core.GcpProjectIdProvider;
@@ -109,6 +110,65 @@ public void maxInboundMessageSize_default() {
 		});
 	}
 
+	@Test
+	public void retryableCodes_default() {
+		ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+				.withConfiguration(AutoConfigurations.of(GcpPubSubAutoConfiguration.class))
+				.withUserConfiguration(TestConfig.class);
+
+		contextRunner.run(ctx -> {
+
+			DefaultSubscriberFactory defaultSubscriberFactory = ctx.getBean("defaultSubscriberFactory", DefaultSubscriberFactory.class);
+			assertThat(FieldUtils.readField(defaultSubscriberFactory, "retryableCodes", true))
+					.isNull();
+		});
+	}
+
+	@Test
+	public void retryableCodes_empty() {
+		ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+				.withConfiguration(AutoConfigurations.of(GcpPubSubAutoConfiguration.class))
+				.withUserConfiguration(TestConfig.class)
+				.withPropertyValues("spring.cloud.gcp.pubsub.subscriber.retryableCodes=");
+
+		contextRunner.run(ctx -> {
+
+			DefaultSubscriberFactory defaultSubscriberFactory = ctx.getBean("defaultSubscriberFactory", DefaultSubscriberFactory.class);
+			assertThat(FieldUtils.readField(defaultSubscriberFactory, "retryableCodes", true))
+					.isEqualTo(new Code[] {});
+		});
+	}
+
+	@Test
+	public void retryableCodes_INTERNAL() {
+		ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+				.withConfiguration(AutoConfigurations.of(GcpPubSubAutoConfiguration.class))
+				.withUserConfiguration(TestConfig.class)
+				.withPropertyValues("spring.cloud.gcp.pubsub.subscriber.retryableCodes=INTERNAL");
+
+		contextRunner.run(ctx -> {
+
+			DefaultSubscriberFactory defaultSubscriberFactory = ctx.getBean("defaultSubscriberFactory", DefaultSubscriberFactory.class);
+			assertThat(FieldUtils.readField(defaultSubscriberFactory, "retryableCodes", true))
+					.isEqualTo(new Code[] { Code.INTERNAL });
+		});
+	}
+
+	@Test
+	public void retryableCodes_many() {
+		ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+				.withConfiguration(AutoConfigurations.of(GcpPubSubAutoConfiguration.class))
+				.withUserConfiguration(TestConfig.class)
+				.withPropertyValues("spring.cloud.gcp.pubsub.subscriber.retryableCodes=UNKNOWN,ABORTED,UNAVAILABLE,INTERNAL");
+
+		contextRunner.run(ctx -> {
+
+			DefaultSubscriberFactory defaultSubscriberFactory = ctx.getBean("defaultSubscriberFactory", DefaultSubscriberFactory.class);
+			assertThat(FieldUtils.readField(defaultSubscriberFactory, "retryableCodes", true))
+					.isEqualTo(new Code[] { Code.UNKNOWN, Code.ABORTED, Code.UNAVAILABLE, Code.INTERNAL });
+		});
+	}
+
 	@Test
 	public void customExecutorProviderUsedWhenProvided() {
 		ExecutorProvider executorProvider = mock(ExecutorProvider.class);

spring-cloud-gcp-pubsub/src/main/java/com/google/cloud/spring/pubsub/core/PubSubConfiguration.java

@@ -20,6 +20,7 @@
 import java.util.concurrent.ConcurrentMap;
 
 import com.google.api.gax.batching.FlowController.LimitExceededBehavior;
+import com.google.api.gax.rpc.StatusCode.Code;
 import com.google.cloud.spring.pubsub.support.PubSubSubscriptionUtils;
 import com.google.pubsub.v1.ProjectSubscriptionName;
 
@@ -302,6 +303,10 @@ public static class Subscriber {
 		 * Flow control settings for subscriber factory.
 		 */
 		private final FlowControl flowControl = new FlowControl();
+		/**
+		 * RPC status codes that should be retried when pulling messages.
+		 */
+		private Code[] retryableCodes = null;
 
 		public boolean isGlobal() {
 			return global;
@@ -311,6 +316,14 @@ public Retry getRetry() {
 			return this.retry;
 		}
 
+		public Code[] getRetryableCodes() {
+			return retryableCodes;
+		}
+
+		public void setRetryableCodes(Code[] retryableCodes) {
+			this.retryableCodes = retryableCodes;
+		}
+
 		public FlowControl getFlowControl() {
 			return this.flowControl;
 		}

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

@@ -29,6 +29,7 @@
 import com.google.api.gax.core.FixedExecutorProvider;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.rpc.HeaderProvider;
+import com.google.api.gax.rpc.StatusCode.Code;
 import com.google.api.gax.rpc.TransportChannelProvider;
 import com.google.cloud.pubsub.v1.MessageReceiver;
 import com.google.cloud.pubsub.v1.Subscriber;
@@ -88,6 +89,8 @@ public class DefaultSubscriberFactory implements SubscriberFactory {
 
 	private ExecutorProvider defaultExecutorProvider;
 
+	private Code[] retryableCodes;
+
 	/**
 	 * Default {@link DefaultSubscriberFactory} constructor.
 	 * @param projectIdProvider provides the default GCP project ID for selecting the
@@ -209,6 +212,14 @@ public void setSubscriberStubRetrySettings(RetrySettings subscriberStubRetrySett
 		this.subscriberStubRetrySettings = subscriberStubRetrySettings;
 	}
 
+	/**
+	 * Set the retryable codes for subscriber pull settings.
+	 * @param retryableCodes pull RPC response codes that should be retried.
+	 */
+	public void setRetryableCodes(Code[] retryableCodes) {
+		this.retryableCodes = retryableCodes;
+	}
+
 	@Override
 	public Subscriber createSubscriber(String subscriptionName, MessageReceiver receiver) {
 		Subscriber.Builder subscriberBuilder = Subscriber.newBuilder(
@@ -314,6 +325,11 @@ public SubscriberStub createSubscriberStub(String subscriptionName) {
 			subscriberStubSettings.pullSettings().setRetrySettings(retrySettings);
 		}
 
+		if (this.retryableCodes != null) {
+			subscriberStubSettings.pullSettings().setRetryableCodes(
+					this.retryableCodes);
+		}
+
 		try {
 			return GrpcSubscriberStub.create(subscriberStubSettings.build());
 		}