fix: set maxInboundMetadataSize in pubsub (#3157)

* fix: set maxInboundMetadataSize

* add comment

* add doc

* format doc

docs/src/main/asciidoc/pubsub.adoc

@@ -127,6 +127,9 @@ The behavior when the specified limits are exceeded. | No | Block
 NOTE: By default, subscription-specific threads are named after fully-qualified subscription name, ex: `gcp-pubsub-subscriber-projects/project-id/subscriptions/subscription-name`.
 This can be customized, by registering a `SelectiveSchedulerThreadNameProvider` bean.
 
+NOTE: By default, subscription-specific `maxInboundMetadataSize` is set to 4MB.
+This can be customized, by registering a `TransportChannelProvider` bean.
+
 ==== GRPC Connection Settings
 
 The Pub/Sub API uses the https://cloud.google.com/pubsub/docs/reference/service_apis_overview#grpc_api[GRPC] protocol to send API requests to the Pub/Sub service.

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

@@ -436,6 +436,9 @@ public SubscriptionAdminClient subscriptionAdminClient(
   @ConditionalOnMissingBean(name = "subscriberTransportChannelProvider")
   public TransportChannelProvider subscriberTransportChannelProvider() {
     return SubscriberStubSettings.defaultGrpcTransportProviderBuilder()
+        // default value specified by pubsub client library,
+        // see https://github.com/googleapis/java-pubsub/blob/main/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java#L487.
+        .setMaxInboundMetadataSize(4 * 1024 * 1024)
         .setKeepAliveTime(
             Duration.ofMinutes(this.gcpPubSubProperties.getKeepAliveIntervalMinutes()))
         .build();

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

@@ -133,6 +133,17 @@ void maxInboundMessageSize_default() {
         });
   }
 
+  @Test
+  void maxInboundMetadataSize_default() {
+    contextRunner.run(
+        ctx -> {
+          TransportChannelProvider subscriberTcp =
+              ctx.getBean("subscriberTransportChannelProvider", TransportChannelProvider.class);
+          assertThat(FieldUtils.readField(subscriberTcp, "maxInboundMetadataSize", true))
+              .isEqualTo(4 * 1024 * 1024);
+        });
+  }
+
   @Test
   void retryableCodes_default() {
     contextRunner.run(