diff --git a/gax-grpc/src/main/java/com/google/api/gax/grpc/GrpcCallContext.java b/gax-grpc/src/main/java/com/google/api/gax/grpc/GrpcCallContext.java
index fdcdd9588..d3fdd990e 100644
--- a/gax-grpc/src/main/java/com/google/api/gax/grpc/GrpcCallContext.java
+++ b/gax-grpc/src/main/java/com/google/api/gax/grpc/GrpcCallContext.java
@@ -34,6 +34,7 @@
 import com.google.api.gax.rpc.ApiCallContext;
 import com.google.api.gax.rpc.StatusCode;
 import com.google.api.gax.rpc.TransportChannel;
+import com.google.api.gax.rpc.internal.ApiCallContextOptions;
 import com.google.api.gax.rpc.internal.Headers;
 import com.google.api.gax.tracing.ApiTracer;
 import com.google.api.gax.tracing.BaseApiTracer;
@@ -43,7 +44,6 @@
 import com.google.common.collect.ImmutableSet;
 import io.grpc.CallCredentials;
 import io.grpc.CallOptions;
-import io.grpc.CallOptions.Key;
 import io.grpc.Channel;
 import io.grpc.Deadline;
 import io.grpc.Metadata;
@@ -66,7 +66,7 @@
  */
 @BetaApi("Reference ApiCallContext instead - this class is likely to experience breaking changes")
 public final class GrpcCallContext implements ApiCallContext {
-  static final CallOptions.Key<ApiTracer> TRACER_KEY = Key.create("gax.tracer");
+  static final CallOptions.Key<ApiTracer> TRACER_KEY = CallOptions.Key.create("gax.tracer");
 
   private final Channel channel;
   private final CallOptions callOptions;
@@ -77,6 +77,7 @@ public final class GrpcCallContext implements ApiCallContext {
   @Nullable private final RetrySettings retrySettings;
   @Nullable private final ImmutableSet<StatusCode.Code> retryableCodes;
   private final ImmutableMap<String, List<String>> extraHeaders;
+  private final ApiCallContextOptions options;
 
   /** Returns an empty instance with a null channel and default {@link CallOptions}. */
   public static GrpcCallContext createDefault() {
@@ -88,6 +89,7 @@ public static GrpcCallContext createDefault() {
         null,
         null,
         ImmutableMap.<String, List<String>>of(),
+        ApiCallContextOptions.getDefaultOptions(),
         null,
         null);
   }
@@ -102,6 +104,7 @@ public static GrpcCallContext of(Channel channel, CallOptions callOptions) {
         null,
         null,
         ImmutableMap.<String, List<String>>of(),
+        ApiCallContextOptions.getDefaultOptions(),
         null,
         null);
   }
@@ -114,6 +117,7 @@ private GrpcCallContext(
       @Nullable Duration streamIdleTimeout,
       @Nullable Integer channelAffinity,
       ImmutableMap<String, List<String>> extraHeaders,
+      ApiCallContextOptions options,
       @Nullable RetrySettings retrySettings,
       @Nullable Set<StatusCode.Code> retryableCodes) {
     this.channel = channel;
@@ -123,6 +127,7 @@ private GrpcCallContext(
     this.streamIdleTimeout = streamIdleTimeout;
     this.channelAffinity = channelAffinity;
     this.extraHeaders = Preconditions.checkNotNull(extraHeaders);
+    this.options = Preconditions.checkNotNull(options);
     this.retrySettings = retrySettings;
     this.retryableCodes = retryableCodes == null ? null : ImmutableSet.copyOf(retryableCodes);
   }
@@ -187,6 +192,7 @@ public GrpcCallContext withTimeout(@Nullable Duration timeout) {
         this.streamIdleTimeout,
         this.channelAffinity,
         this.extraHeaders,
+        this.options,
         this.retrySettings,
         this.retryableCodes);
   }
@@ -212,6 +218,7 @@ public GrpcCallContext withStreamWaitTimeout(@Nullable Duration streamWaitTimeou
         this.streamIdleTimeout,
         this.channelAffinity,
         this.extraHeaders,
+        this.options,
         this.retrySettings,
         this.retryableCodes);
   }
@@ -231,6 +238,7 @@ public GrpcCallContext withStreamIdleTimeout(@Nullable Duration streamIdleTimeou
         streamIdleTimeout,
         this.channelAffinity,
         this.extraHeaders,
+        this.options,
         this.retrySettings,
         this.retryableCodes);
   }
@@ -245,6 +253,7 @@ public GrpcCallContext withChannelAffinity(@Nullable Integer affinity) {
         this.streamIdleTimeout,
         affinity,
         this.extraHeaders,
+        this.options,
         this.retrySettings,
         this.retryableCodes);
   }
@@ -263,6 +272,7 @@ public GrpcCallContext withExtraHeaders(Map<String, List<String>> extraHeaders)
         this.streamIdleTimeout,
         this.channelAffinity,
         newExtraHeaders,
+        this.options,
         this.retrySettings,
         this.retryableCodes);
   }
@@ -282,6 +292,7 @@ public GrpcCallContext withRetrySettings(RetrySettings retrySettings) {
         this.streamIdleTimeout,
         this.channelAffinity,
         this.extraHeaders,
+        this.options,
         retrySettings,
         this.retryableCodes);
   }
@@ -301,6 +312,7 @@ public GrpcCallContext withRetryableCodes(Set<StatusCode.Code> retryableCodes) {
         this.streamIdleTimeout,
         this.channelAffinity,
         this.extraHeaders,
+        this.options,
         this.retrySettings,
         retryableCodes);
   }
@@ -370,6 +382,8 @@ public ApiCallContext merge(ApiCallContext inputCallContext) {
     ImmutableMap<String, List<String>> newExtraHeaders =
         Headers.mergeHeaders(this.extraHeaders, grpcCallContext.extraHeaders);
 
+    ApiCallContextOptions newOptions = options.merge(grpcCallContext.options);
+
     CallOptions newCallOptions =
         grpcCallContext
             .callOptions
@@ -388,6 +402,7 @@ public ApiCallContext merge(ApiCallContext inputCallContext) {
         newStreamIdleTimeout,
         newChannelAffinity,
         newExtraHeaders,
+        newOptions,
         newRetrySettings,
         newRetryableCodes);
   }
@@ -448,6 +463,7 @@ public GrpcCallContext withChannel(Channel newChannel) {
         this.streamIdleTimeout,
         this.channelAffinity,
         this.extraHeaders,
+        this.options,
         this.retrySettings,
         this.retryableCodes);
   }
@@ -462,6 +478,7 @@ public GrpcCallContext withCallOptions(CallOptions newCallOptions) {
         this.streamIdleTimeout,
         this.channelAffinity,
         this.extraHeaders,
+        this.options,
         this.retrySettings,
         this.retryableCodes);
   }
@@ -491,6 +508,29 @@ public GrpcCallContext withTracer(@Nonnull ApiTracer tracer) {
     return withCallOptions(callOptions.withOption(TRACER_KEY, tracer));
   }
 
+  /** {@inheritDoc} */
+  @Override
+  public <T> GrpcCallContext withOption(Key<T> key, T value) {
+    ApiCallContextOptions newOptions = options.withOption(key, value);
+    return new GrpcCallContext(
+        this.channel,
+        this.callOptions,
+        this.timeout,
+        this.streamWaitTimeout,
+        this.streamIdleTimeout,
+        this.channelAffinity,
+        this.extraHeaders,
+        newOptions,
+        this.retrySettings,
+        this.retryableCodes);
+  }
+
+  /** {@inheritDoc} */
+  @Override
+  public <T> T getOption(Key<T> key) {
+    return options.getOption(key);
+  }
+
   @Override
   public int hashCode() {
     return Objects.hash(
@@ -501,6 +541,7 @@ public int hashCode() {
         streamIdleTimeout,
         channelAffinity,
         extraHeaders,
+        options,
         retrySettings,
         retryableCodes);
   }
@@ -522,6 +563,7 @@ public boolean equals(Object o) {
         && Objects.equals(this.streamIdleTimeout, that.streamIdleTimeout)
         && Objects.equals(this.channelAffinity, that.channelAffinity)
         && Objects.equals(this.extraHeaders, that.extraHeaders)
+        && Objects.equals(this.options, that.options)
         && Objects.equals(this.retrySettings, that.retrySettings)
         && Objects.equals(this.retryableCodes, that.retryableCodes);
   }
diff --git a/gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcCallContextTest.java b/gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcCallContextTest.java
index 1b9b5c187..4e563225e 100644
--- a/gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcCallContextTest.java
+++ b/gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcCallContextTest.java
@@ -354,6 +354,48 @@ public void testWithRetryableCodes() {
     assertNotNull(context.getRetryableCodes());
   }
 
+  @Test
+  public void testWithOptions() {
+    GrpcCallContext emptyCallContext = GrpcCallContext.createDefault();
+    ApiCallContext.Key<String> contextKey1 = ApiCallContext.Key.create("testKey1");
+    ApiCallContext.Key<String> contextKey2 = ApiCallContext.Key.create("testKey2");
+    String testContext1 = "test1";
+    String testContext2 = "test2";
+    String testContextOverwrite = "test1Overwrite";
+    GrpcCallContext context =
+        emptyCallContext
+            .withOption(contextKey1, testContext1)
+            .withOption(contextKey2, testContext2);
+    assertEquals(testContext1, context.getOption(contextKey1));
+    assertEquals(testContext2, context.getOption(contextKey2));
+    GrpcCallContext newContext = context.withOption(contextKey1, testContextOverwrite);
+    assertEquals(testContextOverwrite, newContext.getOption(contextKey1));
+  }
+
+  @Test
+  public void testMergeOptions() {
+    GrpcCallContext emptyCallContext = GrpcCallContext.createDefault();
+    ApiCallContext.Key<String> contextKey1 = ApiCallContext.Key.create("testKey1");
+    ApiCallContext.Key<String> contextKey2 = ApiCallContext.Key.create("testKey2");
+    ApiCallContext.Key<String> contextKey3 = ApiCallContext.Key.create("testKey3");
+    String testContext1 = "test1";
+    String testContext2 = "test2";
+    String testContext3 = "test3";
+    String testContextOverwrite = "test1Overwrite";
+    GrpcCallContext context1 =
+        emptyCallContext
+            .withOption(contextKey1, testContext1)
+            .withOption(contextKey2, testContext2);
+    GrpcCallContext context2 =
+        emptyCallContext
+            .withOption(contextKey1, testContextOverwrite)
+            .withOption(contextKey3, testContext3);
+    ApiCallContext mergedContext = context1.merge(context2);
+    assertEquals(testContextOverwrite, mergedContext.getOption(contextKey1));
+    assertEquals(testContext2, mergedContext.getOption(contextKey2));
+    assertEquals(testContext3, mergedContext.getOption(contextKey3));
+  }
+
   private static Map<String, List<String>> createTestExtraHeaders(String... keyValues) {
     Map<String, List<String>> extraHeaders = new HashMap<>();
     for (int i = 0; i < keyValues.length; i += 2) {
diff --git a/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallContext.java b/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallContext.java
index 700716217..0d3b00898 100644
--- a/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallContext.java
+++ b/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallContext.java
@@ -34,6 +34,7 @@
 import com.google.api.gax.rpc.ApiCallContext;
 import com.google.api.gax.rpc.StatusCode;
 import com.google.api.gax.rpc.TransportChannel;
+import com.google.api.gax.rpc.internal.ApiCallContextOptions;
 import com.google.api.gax.rpc.internal.Headers;
 import com.google.api.gax.tracing.ApiTracer;
 import com.google.api.gax.tracing.BaseApiTracer;
@@ -65,6 +66,7 @@ public final class HttpJsonCallContext implements ApiCallContext {
   private final Instant deadline;
   private final Credentials credentials;
   private final ImmutableMap<String, List<String>> extraHeaders;
+  private final ApiCallContextOptions options;
   private final ApiTracer tracer;
   private final RetrySettings retrySettings;
   private final ImmutableSet<StatusCode.Code> retryableCodes;
@@ -72,7 +74,15 @@ public final class HttpJsonCallContext implements ApiCallContext {
   /** Returns an empty instance. */
   public static HttpJsonCallContext createDefault() {
     return new HttpJsonCallContext(
-        null, null, null, null, ImmutableMap.<String, List<String>>of(), null, null, null);
+        null,
+        null,
+        null,
+        null,
+        ImmutableMap.<String, List<String>>of(),
+        ApiCallContextOptions.getDefaultOptions(),
+        null,
+        null,
+        null);
   }
 
   private HttpJsonCallContext(
@@ -81,6 +91,7 @@ private HttpJsonCallContext(
       Instant deadline,
       Credentials credentials,
       ImmutableMap<String, List<String>> extraHeaders,
+      ApiCallContextOptions options,
       ApiTracer tracer,
       RetrySettings defaultRetrySettings,
       Set<StatusCode.Code> defaultRetryableCodes) {
@@ -89,6 +100,7 @@ private HttpJsonCallContext(
     this.deadline = deadline;
     this.credentials = credentials;
     this.extraHeaders = extraHeaders;
+    this.options = options;
     this.tracer = tracer;
     this.retrySettings = defaultRetrySettings;
     this.retryableCodes =
@@ -152,6 +164,8 @@ public HttpJsonCallContext merge(ApiCallContext inputCallContext) {
     ImmutableMap<String, List<String>> newExtraHeaders =
         Headers.mergeHeaders(extraHeaders, httpJsonCallContext.extraHeaders);
 
+    ApiCallContextOptions newOptions = options.merge(httpJsonCallContext.options);
+
     ApiTracer newTracer = httpJsonCallContext.tracer;
     if (newTracer == null) {
       newTracer = this.tracer;
@@ -173,6 +187,7 @@ public HttpJsonCallContext merge(ApiCallContext inputCallContext) {
         newDeadline,
         newCredentials,
         newExtraHeaders,
+        newOptions,
         newTracer,
         newRetrySettings,
         newRetryableCodes);
@@ -186,6 +201,7 @@ public HttpJsonCallContext withCredentials(Credentials newCredentials) {
         this.deadline,
         newCredentials,
         this.extraHeaders,
+        this.options,
         this.tracer,
         this.retrySettings,
         this.retryableCodes);
@@ -220,6 +236,7 @@ public HttpJsonCallContext withTimeout(Duration timeout) {
         this.deadline,
         this.credentials,
         this.extraHeaders,
+        this.options,
         this.tracer,
         this.retrySettings,
         this.retryableCodes);
@@ -265,6 +282,7 @@ public ApiCallContext withExtraHeaders(Map<String, List<String>> extraHeaders) {
         this.deadline,
         this.credentials,
         newExtraHeaders,
+        this.options,
         this.tracer,
         this.retrySettings,
         this.retryableCodes);
@@ -276,6 +294,28 @@ public Map<String, List<String>> getExtraHeaders() {
     return extraHeaders;
   }
 
+  /** {@inheritDoc} */
+  @Override
+  public <T> ApiCallContext withOption(Key<T> key, T value) {
+    ApiCallContextOptions newOptions = options.withOption(key, value);
+    return new HttpJsonCallContext(
+        this.channel,
+        this.timeout,
+        this.deadline,
+        this.credentials,
+        this.extraHeaders,
+        newOptions,
+        this.tracer,
+        this.retrySettings,
+        this.retryableCodes);
+  }
+
+  /** {@inheritDoc} */
+  @Override
+  public <T> T getOption(Key<T> key) {
+    return options.getOption(key);
+  }
+
   public HttpJsonChannel getChannel() {
     return channel;
   }
@@ -301,6 +341,7 @@ public HttpJsonCallContext withRetrySettings(RetrySettings retrySettings) {
         this.deadline,
         this.credentials,
         this.extraHeaders,
+        this.options,
         this.tracer,
         retrySettings,
         this.retryableCodes);
@@ -319,6 +360,7 @@ public HttpJsonCallContext withRetryableCodes(Set<StatusCode.Code> retryableCode
         this.deadline,
         this.credentials,
         this.extraHeaders,
+        this.options,
         this.tracer,
         this.retrySettings,
         retryableCodes);
@@ -331,6 +373,7 @@ public HttpJsonCallContext withChannel(HttpJsonChannel newChannel) {
         this.deadline,
         this.credentials,
         this.extraHeaders,
+        this.options,
         this.tracer,
         this.retrySettings,
         this.retryableCodes);
@@ -343,6 +386,7 @@ public HttpJsonCallContext withDeadline(Instant newDeadline) {
         newDeadline,
         this.credentials,
         this.extraHeaders,
+        this.options,
         this.tracer,
         this.retrySettings,
         this.retryableCodes);
@@ -368,6 +412,7 @@ public HttpJsonCallContext withTracer(@Nonnull ApiTracer newTracer) {
         this.deadline,
         this.credentials,
         this.extraHeaders,
+        this.options,
         newTracer,
         this.retrySettings,
         this.retryableCodes);
@@ -387,6 +432,7 @@ public boolean equals(Object o) {
         && Objects.equals(this.deadline, that.deadline)
         && Objects.equals(this.credentials, that.credentials)
         && Objects.equals(this.extraHeaders, that.extraHeaders)
+        && Objects.equals(this.options, that.options)
         && Objects.equals(this.tracer, that.tracer)
         && Objects.equals(this.retrySettings, that.retrySettings)
         && Objects.equals(this.retryableCodes, that.retryableCodes);
@@ -400,6 +446,7 @@ public int hashCode() {
         deadline,
         credentials,
         extraHeaders,
+        options,
         tracer,
         retrySettings,
         retryableCodes);
diff --git a/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonCallContextTest.java b/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonCallContextTest.java
index 245ace65c..fc872d352 100644
--- a/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonCallContextTest.java
+++ b/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonCallContextTest.java
@@ -229,4 +229,46 @@ public void testWithExtraHeaders() {
     ApiCallContext context = emptyContext.withExtraHeaders(headers);
     assertEquals(headers, context.getExtraHeaders());
   }
+
+  @Test
+  public void testWithOptions() {
+    ApiCallContext emptyCallContext = HttpJsonCallContext.createDefault();
+    ApiCallContext.Key<String> contextKey1 = ApiCallContext.Key.create("testKey1");
+    ApiCallContext.Key<String> contextKey2 = ApiCallContext.Key.create("testKey2");
+    String testContext1 = "test1";
+    String testContext2 = "test2";
+    String testContextOverwrite = "test1Overwrite";
+    ApiCallContext context =
+        emptyCallContext
+            .withOption(contextKey1, testContext1)
+            .withOption(contextKey2, testContext2);
+    assertEquals(testContext1, context.getOption(contextKey1));
+    assertEquals(testContext2, context.getOption(contextKey2));
+    ApiCallContext newContext = context.withOption(contextKey1, testContextOverwrite);
+    assertEquals(testContextOverwrite, newContext.getOption(contextKey1));
+  }
+
+  @Test
+  public void testMergeOptions() {
+    ApiCallContext emptyCallContext = HttpJsonCallContext.createDefault();
+    ApiCallContext.Key<String> contextKey1 = ApiCallContext.Key.create("testKey1");
+    ApiCallContext.Key<String> contextKey2 = ApiCallContext.Key.create("testKey2");
+    ApiCallContext.Key<String> contextKey3 = ApiCallContext.Key.create("testKey3");
+    String testContext1 = "test1";
+    String testContext2 = "test2";
+    String testContext3 = "test3";
+    String testContextOverwrite = "test1Overwrite";
+    ApiCallContext context1 =
+        emptyCallContext
+            .withOption(contextKey1, testContext1)
+            .withOption(contextKey2, testContext2);
+    ApiCallContext context2 =
+        emptyCallContext
+            .withOption(contextKey1, testContextOverwrite)
+            .withOption(contextKey3, testContext3);
+    ApiCallContext mergedContext = context1.merge(context2);
+    assertEquals(testContextOverwrite, mergedContext.getOption(contextKey1));
+    assertEquals(testContext2, mergedContext.getOption(contextKey2));
+    assertEquals(testContext3, mergedContext.getOption(contextKey3));
+  }
 }
diff --git a/gax/src/main/java/com/google/api/gax/rpc/ApiCallContext.java b/gax/src/main/java/com/google/api/gax/rpc/ApiCallContext.java
index a9366fbc5..79e4ef0e7 100644
--- a/gax/src/main/java/com/google/api/gax/rpc/ApiCallContext.java
+++ b/gax/src/main/java/com/google/api/gax/rpc/ApiCallContext.java
@@ -36,6 +36,7 @@
 import com.google.api.gax.rpc.StatusCode.Code;
 import com.google.api.gax.tracing.ApiTracer;
 import com.google.auth.Credentials;
+import com.google.common.base.Preconditions;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -242,4 +243,31 @@ public interface ApiCallContext extends RetryingContext {
   /** Return the extra headers set for this context. */
   @BetaApi("The surface for extra headers is not stable yet and may change in the future.")
   Map<String, List<String>> getExtraHeaders();
+
+  /**
+   * Return a new ApiCallContext with additional option merged into the present instance. Any
+   * existing value of the key is overwritten.
+   */
+  @BetaApi("The surface for call context options is not stable yet and may change in the future.")
+  <T> ApiCallContext withOption(Key<T> key, T value);
+
+  /** Return the api call context option set for this context. */
+  @SuppressWarnings("unchecked")
+  @BetaApi("The surface for call context options is not stable yet and may change in the future.")
+  <T> T getOption(Key<T> key);
+
+  /** Key for api call context options key-value pair. */
+  final class Key<T> {
+    private final String name;
+
+    private Key(String name) {
+      this.name = name;
+    }
+
+    /** Factory method for creating instances of {@link Key}. */
+    public static <T> Key<T> create(String name) {
+      Preconditions.checkNotNull(name, "Key name cannot be null.");
+      return new Key<>(name);
+    }
+  }
 }
diff --git a/gax/src/main/java/com/google/api/gax/rpc/internal/ApiCallContextOptions.java b/gax/src/main/java/com/google/api/gax/rpc/internal/ApiCallContextOptions.java
new file mode 100644
index 000000000..daf64679c
--- /dev/null
+++ b/gax/src/main/java/com/google/api/gax/rpc/internal/ApiCallContextOptions.java
@@ -0,0 +1,91 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google LLC nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.google.api.gax.rpc.internal;
+
+import com.google.api.core.InternalApi;
+import com.google.api.gax.rpc.ApiCallContext;
+import com.google.api.gax.rpc.ApiCallContext.Key;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+
+/**
+ * ApiCallContextOptions encapsulates additional call options to pass in a {@link ApiCallContext}
+ */
+@InternalApi
+public final class ApiCallContextOptions {
+
+  private final ImmutableMap<Key, Object> options;
+  private static final ApiCallContextOptions DEFAULT_OPTIONS =
+      new ApiCallContextOptions(ImmutableMap.<Key, Object>of());
+
+  private ApiCallContextOptions(ImmutableMap<Key, Object> options) {
+    this.options = Preconditions.checkNotNull(options);
+  }
+
+  public static ApiCallContextOptions getDefaultOptions() {
+    return DEFAULT_OPTIONS;
+  }
+
+  /** Add an option. Any existing value of the key is overwritten. */
+  public <T> ApiCallContextOptions withOption(Key<T> key, T value) {
+    Preconditions.checkNotNull(key);
+    Preconditions.checkNotNull(value);
+    ImmutableMap.Builder builder = ImmutableMap.<Key, Object>builder();
+    if (!options.containsKey(key)) {
+      builder.putAll(options).put(key, value);
+    } else {
+      builder.put(key, value);
+      for (Key oldKey : options.keySet()) {
+        if (!oldKey.equals(key)) {
+          builder.put(oldKey, options.get(oldKey));
+        }
+      }
+    }
+    return new ApiCallContextOptions(builder.build());
+  }
+
+  /** Get an option. */
+  public <T> T getOption(Key<T> key) {
+    Preconditions.checkNotNull(key);
+    return (T) options.get(key);
+  }
+
+  /** Merge new options into existing ones. Any existing values of the keys are overwritten. */
+  public ApiCallContextOptions merge(ApiCallContextOptions newOptions) {
+    Preconditions.checkNotNull(newOptions);
+    ImmutableMap.Builder builder = ImmutableMap.<Key, Object>builder().putAll(newOptions.options);
+    for (Key key : options.keySet()) {
+      if (!newOptions.options.containsKey(key)) {
+        builder.put(key, options.get(key));
+      }
+    }
+    return new ApiCallContextOptions(builder.build());
+  }
+}
diff --git a/gax/src/test/java/com/google/api/gax/rpc/internal/ApiCallContextOptionsTest.java b/gax/src/test/java/com/google/api/gax/rpc/internal/ApiCallContextOptionsTest.java
new file mode 100644
index 000000000..cf4f785bb
--- /dev/null
+++ b/gax/src/test/java/com/google/api/gax/rpc/internal/ApiCallContextOptionsTest.java
@@ -0,0 +1,110 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google LLC nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.google.api.gax.rpc.internal;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.api.gax.rpc.ApiCallContext;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class ApiCallContextOptionsTest {
+
+  private static final ApiCallContext.Key<Integer> INTEGER_KEY = ApiCallContext.Key.create("key1");
+  private static final ApiCallContext.Key<String> STRING_KEY = ApiCallContext.Key.create("key2");
+  private static final ApiCallContext.Key<Double> DOUBLE_KEY = ApiCallContext.Key.create("key3");
+
+  @Test
+  public void testWithOption() {
+    ApiCallContextOptions options =
+        ApiCallContextOptions.getDefaultOptions()
+            .withOption(INTEGER_KEY, 1)
+            .withOption(STRING_KEY, "test")
+            .withOption(DOUBLE_KEY, 2.0);
+    assertThat(options.getOption(INTEGER_KEY)).isEqualTo(1);
+    assertThat(options.getOption(STRING_KEY)).isEqualTo("test");
+    assertThat(options.getOption(DOUBLE_KEY)).isEqualTo(2.0);
+
+    ApiCallContextOptions updatedOptions = options.withOption(INTEGER_KEY, 10);
+    assertThat(updatedOptions.getOption(INTEGER_KEY)).isEqualTo(10);
+    assertThat(options.getOption(STRING_KEY)).isEqualTo("test");
+    assertThat(options.getOption(DOUBLE_KEY)).isEqualTo(2.0);
+  }
+
+  @Test
+  public void testMergeEmpty() {
+    ApiCallContextOptions options =
+        ApiCallContextOptions.getDefaultOptions()
+            .withOption(INTEGER_KEY, 1)
+            .withOption(STRING_KEY, "test");
+
+    ApiCallContextOptions defaultOptions = ApiCallContextOptions.getDefaultOptions();
+
+    ApiCallContextOptions mergedOptions1 = options.merge(defaultOptions);
+    assertThat(mergedOptions1.getOption(INTEGER_KEY)).isEqualTo(1);
+    assertThat(mergedOptions1.getOption(STRING_KEY)).isEqualTo("test");
+
+    ApiCallContextOptions mergedOptions2 = defaultOptions.merge(options);
+    assertThat(mergedOptions2.getOption(INTEGER_KEY)).isEqualTo(1);
+    assertThat(mergedOptions2.getOption(STRING_KEY)).isEqualTo("test");
+  }
+
+  @Test
+  public void testMergeDifferentKeys() {
+    ApiCallContextOptions options1 =
+        ApiCallContextOptions.getDefaultOptions()
+            .withOption(INTEGER_KEY, 1)
+            .withOption(STRING_KEY, "test");
+    ApiCallContextOptions options2 =
+        ApiCallContextOptions.getDefaultOptions().withOption(DOUBLE_KEY, 5.0);
+    ApiCallContextOptions mergedOptions = options1.merge(options2);
+    assertThat(mergedOptions.getOption(INTEGER_KEY)).isEqualTo(1);
+    assertThat(mergedOptions.getOption(STRING_KEY)).isEqualTo("test");
+    assertThat(mergedOptions.getOption(DOUBLE_KEY)).isEqualTo(5.0);
+  }
+
+  @Test
+  public void testMergeDuplicateKeys() {
+    ApiCallContextOptions options1 =
+        ApiCallContextOptions.getDefaultOptions()
+            .withOption(INTEGER_KEY, 1)
+            .withOption(STRING_KEY, "test");
+    ApiCallContextOptions options2 =
+        ApiCallContextOptions.getDefaultOptions()
+            .withOption(STRING_KEY, "newTest")
+            .withOption(DOUBLE_KEY, 5.0);
+    ApiCallContextOptions mergedOptions = options1.merge(options2);
+    assertThat(mergedOptions.getOption(INTEGER_KEY)).isEqualTo(1);
+    assertThat(mergedOptions.getOption(STRING_KEY)).isEqualTo("newTest");
+    assertThat(mergedOptions.getOption(DOUBLE_KEY)).isEqualTo(5.0);
+  }
+}
diff --git a/gax/src/test/java/com/google/api/gax/rpc/testing/FakeCallContext.java b/gax/src/test/java/com/google/api/gax/rpc/testing/FakeCallContext.java
index 6d16048d1..84d40e11c 100644
--- a/gax/src/test/java/com/google/api/gax/rpc/testing/FakeCallContext.java
+++ b/gax/src/test/java/com/google/api/gax/rpc/testing/FakeCallContext.java
@@ -35,6 +35,7 @@
 import com.google.api.gax.rpc.ClientContext;
 import com.google.api.gax.rpc.StatusCode;
 import com.google.api.gax.rpc.TransportChannel;
+import com.google.api.gax.rpc.internal.ApiCallContextOptions;
 import com.google.api.gax.rpc.internal.Headers;
 import com.google.api.gax.tracing.ApiTracer;
 import com.google.api.gax.tracing.BaseApiTracer;
@@ -57,6 +58,7 @@ public class FakeCallContext implements ApiCallContext {
   private final Duration streamWaitTimeout;
   private final Duration streamIdleTimeout;
   private final ImmutableMap<String, List<String>> extraHeaders;
+  private final ApiCallContextOptions options;
   private final ApiTracer tracer;
   private final RetrySettings retrySettings;
   private final ImmutableSet<StatusCode.Code> retryableCodes;
@@ -68,6 +70,7 @@ private FakeCallContext(
       Duration streamWaitTimeout,
       Duration streamIdleTimeout,
       ImmutableMap<String, List<String>> extraHeaders,
+      ApiCallContextOptions options,
       ApiTracer tracer,
       RetrySettings retrySettings,
       Set<StatusCode.Code> retryableCodes) {
@@ -77,6 +80,7 @@ private FakeCallContext(
     this.streamWaitTimeout = streamWaitTimeout;
     this.streamIdleTimeout = streamIdleTimeout;
     this.extraHeaders = extraHeaders;
+    this.options = options;
     this.tracer = tracer;
     this.retrySettings = retrySettings;
     this.retryableCodes = retryableCodes == null ? null : ImmutableSet.copyOf(retryableCodes);
@@ -84,7 +88,16 @@ private FakeCallContext(
 
   public static FakeCallContext createDefault() {
     return new FakeCallContext(
-        null, null, null, null, null, ImmutableMap.<String, List<String>>of(), null, null, null);
+        null,
+        null,
+        null,
+        null,
+        null,
+        ImmutableMap.<String, List<String>>of(),
+        ApiCallContextOptions.getDefaultOptions(),
+        null,
+        null,
+        null);
   }
 
   @Override
@@ -157,6 +170,9 @@ public ApiCallContext merge(ApiCallContext inputCallContext) {
 
     ImmutableMap<String, List<String>> newExtraHeaders =
         Headers.mergeHeaders(extraHeaders, fakeCallContext.extraHeaders);
+
+    ApiCallContextOptions newOptions = options.merge(fakeCallContext.options);
+
     return new FakeCallContext(
         newCallCredentials,
         newChannel,
@@ -164,6 +180,7 @@ public ApiCallContext merge(ApiCallContext inputCallContext) {
         newStreamWaitTimeout,
         newStreamIdleTimeout,
         newExtraHeaders,
+        newOptions,
         newTracer,
         newRetrySettings,
         newRetryableCodes);
@@ -181,6 +198,7 @@ public FakeCallContext withRetrySettings(RetrySettings retrySettings) {
         this.streamWaitTimeout,
         this.streamIdleTimeout,
         this.extraHeaders,
+        this.options,
         this.tracer,
         retrySettings,
         this.retryableCodes);
@@ -198,6 +216,7 @@ public FakeCallContext withRetryableCodes(Set<StatusCode.Code> retryableCodes) {
         this.streamWaitTimeout,
         this.streamIdleTimeout,
         this.extraHeaders,
+        this.options,
         this.tracer,
         this.retrySettings,
         retryableCodes);
@@ -237,6 +256,7 @@ public FakeCallContext withCredentials(Credentials credentials) {
         this.streamWaitTimeout,
         this.streamIdleTimeout,
         this.extraHeaders,
+        this.options,
         this.tracer,
         this.retrySettings,
         this.retryableCodes);
@@ -261,6 +281,7 @@ public FakeCallContext withChannel(FakeChannel channel) {
         this.streamWaitTimeout,
         this.streamIdleTimeout,
         this.extraHeaders,
+        this.options,
         this.tracer,
         this.retrySettings,
         this.retryableCodes);
@@ -285,6 +306,7 @@ public FakeCallContext withTimeout(Duration timeout) {
         this.streamWaitTimeout,
         this.streamIdleTimeout,
         this.extraHeaders,
+        this.options,
         this.tracer,
         this.retrySettings,
         this.retryableCodes);
@@ -299,6 +321,7 @@ public ApiCallContext withStreamWaitTimeout(@Nullable Duration streamWaitTimeout
         streamWaitTimeout,
         this.streamIdleTimeout,
         this.extraHeaders,
+        this.options,
         this.tracer,
         this.retrySettings,
         this.retryableCodes);
@@ -314,6 +337,7 @@ public ApiCallContext withStreamIdleTimeout(@Nullable Duration streamIdleTimeout
         this.streamWaitTimeout,
         streamIdleTimeout,
         this.extraHeaders,
+        this.options,
         this.tracer,
         this.retrySettings,
         this.retryableCodes);
@@ -331,6 +355,7 @@ public ApiCallContext withExtraHeaders(Map<String, List<String>> extraHeaders) {
         streamWaitTimeout,
         streamIdleTimeout,
         newExtraHeaders,
+        this.options,
         this.tracer,
         this.retrySettings,
         this.retryableCodes);
@@ -341,6 +366,29 @@ public Map<String, List<String>> getExtraHeaders() {
     return this.extraHeaders;
   }
 
+  @Override
+  public <T> ApiCallContext withOption(Key<T> key, T value) {
+    Preconditions.checkNotNull(key);
+    ApiCallContextOptions newOptions = options.withOption(key, value);
+    return new FakeCallContext(
+        credentials,
+        channel,
+        timeout,
+        streamWaitTimeout,
+        streamIdleTimeout,
+        extraHeaders,
+        newOptions,
+        tracer,
+        retrySettings,
+        retryableCodes);
+  }
+
+  @Override
+  public <T> T getOption(Key<T> key) {
+    Preconditions.checkNotNull(key);
+    return options.getOption(key);
+  }
+
   /** {@inheritDoc} */
   @Override
   @Nonnull
@@ -363,6 +411,7 @@ public ApiCallContext withTracer(@Nonnull ApiTracer tracer) {
         this.streamWaitTimeout,
         this.streamIdleTimeout,
         this.extraHeaders,
+        this.options,
         tracer,
         this.retrySettings,
         this.retryableCodes);