fix: set x-goog-request-params for streaming pull request (#1522)

kamalaboulhosn · gcf-owl-bot[bot] · web-flow · commit 52e23d6640ac · 2023-03-16T14:15:45.000-04:00
* samples: schema evolution * samples: schema evolution * Format fixes * Fix documentation for field. * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Add back in working asserts * Formatting fixes * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Version/delete fixes * samples: schema evolution * samples: schema evolution * Format fixes * Fix documentation for field. * Add back in working asserts * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Formatting fixes * Version/delete fixes * samples: Schema evolution (#1499) * samples: schema evolution * samples: schema evolution * Format fixes * Fix documentation for field. * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Add back in working asserts * Formatting fixes * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Version/delete fixes * samples: schema evolution * samples: schema evolution * Format fixes * Fix documentation for field. * Add back in working asserts * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Formatting fixes * Version/delete fixes --------- Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com> * Minor fixes for comments * samples: Schema evolution (#1499) * samples: schema evolution * samples: schema evolution * Format fixes * Fix documentation for field. * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Add back in working asserts * Formatting fixes * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Version/delete fixes * samples: schema evolution * samples: schema evolution * Format fixes * Fix documentation for field. * Add back in working asserts * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Formatting fixes * Version/delete fixes --------- Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com> * Fix rollback example * Formatting * Formatting and wording fixes * Add new schemas to test directory * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Samples: Fix exception handling * fix: Set x-goog-request-params for streaming pull request * Revert "fix: Set x-goog-request-params for streaming pull request" This reverts commit 3185a3e. * Revert "Revert "fix: Set x-goog-request-params for streaming pull request"" This reverts commit 3b1f4d9. --------- Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
@@ -36,6 +36,8 @@
 import com.google.api.gax.rpc.StreamController;
 import com.google.cloud.pubsub.v1.MessageDispatcher.AckProcessor;
 import com.google.cloud.pubsub.v1.stub.SubscriberStub;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.protobuf.Any;
@@ -86,6 +88,8 @@ final class StreamingSubscriberConnection extends AbstractApiService implements
 
   private Duration inititalStreamAckDeadline;
 
+  private final Map<String, List<String>> streamMetadata;
+
   private final SubscriberStub subscriberStub;
   private final int channelAffinity;
   private final String subscription;
@@ -134,6 +138,9 @@ private StreamingSubscriberConnection(Builder builder) {
       inititalStreamAckDeadline = builder.maxDurationPerAckExtension;
     }
 
+    streamMetadata =
+        ImmutableMap.of("x-goog-request-params", ImmutableList.of("subscription=" + subscription));
+
     subscriberStub = builder.subscriberStub;
     channelAffinity = builder.channelAffinity;
 
@@ -273,7 +280,9 @@ private void initialize() {
             .streamingPullCallable()
             .splitCall(
                 responseObserver,
-                GrpcCallContext.createDefault().withChannelAffinity(channelAffinity));
+                GrpcCallContext.createDefault()
+                    .withChannelAffinity(channelAffinity)
+                    .withExtraHeaders(streamMetadata));
 
     logger.log(Level.FINER, "Initializing stream to subscription {0}", subscription);
     // We need to set streaming ack deadline, but it's not useful since we'll modack to send receipt
