feat: enable compression for publishing (#1360)

* feat: enable compression for publishing

This is enabled by default and should save on network usage for clients

* feat: enable compression for publishing

This is enabled by default and should save on network usage for clients

* 🦉 Updates from OwlBot post-processor

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

---------

Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>

Author: dpcollins-google

google-cloud-pubsublite/clirr-ignored-differences.xml

@@ -28,6 +28,12 @@
     <method>*</method>
     <to>**</to>
   </difference>
+  <difference>
+    <differenceType>7006</differenceType>
+    <className>com/google/cloud/pubsublite/internal/**</className>
+    <method>*</method>
+    <to>**</to>
+  </difference>
   <difference>
     <differenceType>7009</differenceType>
     <className>com/google/cloud/pubsublite/internal/**</className>

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/cloudpubsub/PublisherSettings.java

@@ -22,7 +22,7 @@
 
 import com.google.api.gax.batching.BatchingSettings;
 import com.google.api.gax.core.CredentialsProvider;
-import com.google.api.gax.rpc.ApiCallContext;
+import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.rpc.ApiException;
 import com.google.api.gax.rpc.StatusCode.Code;
 import com.google.auto.value.AutoValue;
@@ -50,6 +50,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.protobuf.ByteString;
 import com.google.pubsub.v1.PubsubMessage;
+import io.grpc.CallOptions;
 import java.util.Optional;
 import org.threeten.bp.Duration;
 
@@ -82,11 +83,14 @@ public abstract class PublisherSettings {
   abstract BatchingSettings batchingSettings();
 
   /**
-   * Whether to enable publish idempotence, where the server will ensure that unique messages within
-   * a single publisher session are stored only once. Default true.
+   * Whether idempotence is enabled, where the server will ensure that unique messages within a
+   * single publisher session are stored only once. Default true.
    */
   abstract boolean enableIdempotence();
 
+  /** Whether request compression is enabled. Default true. */
+  abstract boolean enableCompression();
+
   /** A provider for credentials. */
   abstract CredentialsProvider credentialsProvider();
 
@@ -115,6 +119,7 @@ public static Builder newBuilder() {
             PublisherServiceSettings.defaultCredentialsProviderBuilder().build())
         .setBatchingSettings(DEFAULT_BATCHING_SETTINGS)
         .setEnableIdempotence(true)
+        .setEnableCompression(true)
         .setUnderlyingBuilder(SinglePartitionPublisherBuilder.newBuilder());
   }
 
@@ -137,11 +142,14 @@ public abstract Builder setMessageTransformer(
     public abstract Builder setBatchingSettings(BatchingSettings batchingSettings);
 
     /**
-     * Whether to enable publish idempotence, where the server will ensure that unique messages
-     * within a single publisher session are stored only once. Default true.
+     * Whether idempotence is enabled, where the server will ensure that unique messages within a
+     * single publisher session are stored only once. Default true.
      */
     public abstract Builder setEnableIdempotence(boolean enableIdempotence);
 
+    /** Whether request compression is enabled. Default true. */
+    public abstract Builder setEnableCompression(boolean enableCompression);
+
     /** A provider for credentials. */
     public abstract Builder setCredentialsProvider(CredentialsProvider credentialsProvider);
 
@@ -190,10 +198,14 @@ public com.google.cloud.pubsublite.internal.Publisher<MessageMetadata> newPublis
                 .setPartition(partition)
                 .setStreamFactory(
                     responseStream -> {
-                      ApiCallContext context =
+                      GrpcCallContext context =
                           getCallContext(
                               PubsubContext.of(framework()),
                               RoutingMetadata.of(topicPath(), partition));
+                      if (enableCompression()) {
+                        context =
+                            context.withCallOptions(CallOptions.DEFAULT.withCompression("gzip"));
+                      }
                       return client.publishCallable().splitCall(responseStream, context);
                     });
         if (enableIdempotence()) {

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/wire/ServiceClients.java

@@ -22,7 +22,6 @@
 import com.google.api.gax.grpc.ChannelPoolSettings;
 import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
-import com.google.api.gax.rpc.ApiCallContext;
 import com.google.api.gax.rpc.ApiException;
 import com.google.api.gax.rpc.ClientSettings;
 import com.google.api.gax.rpc.TransportChannelProvider;
@@ -70,7 +69,7 @@ Settings addDefaultSettings(CloudRegion target, Builder builder) throws ApiExcep
     }
   }
 
-  public static ApiCallContext getCallContext(
+  public static GrpcCallContext getCallContext(
       PubsubContext context, RoutingMetadata routingMetadata) {
     return GrpcCallContext.createDefault()
         .withExtraHeaders(