diff --git a/BUILD b/BUILD
index 47a0c46d429..3dd6e4f8d62 100644
--- a/BUILD
+++ b/BUILD
@@ -1531,9 +1531,9 @@ grpc_cc_library(
         "//src/core:arena_promise",
         "//src/core:atomic_utils",
         "//src/core:bitset",
+        "//src/core:call_factory",
         "//src/core:call_filters",
         "//src/core:call_final_info",
-        "//src/core:call_size_estimator",
         "//src/core:call_spine",
         "//src/core:cancel_callback",
         "//src/core:channel_args",
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 020ff143c2e..12959352064 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -2500,6 +2500,7 @@ add_library(grpc
   src/core/lib/surface/version.cc
   src/core/lib/transport/batch_builder.cc
   src/core/lib/transport/bdp_estimator.cc
+  src/core/lib/transport/call_factory.cc
   src/core/lib/transport/call_filters.cc
   src/core/lib/transport/call_final_info.cc
   src/core/lib/transport/call_size_estimator.cc
@@ -3223,6 +3224,7 @@ add_library(grpc_unsecure
   src/core/lib/surface/version.cc
   src/core/lib/transport/batch_builder.cc
   src/core/lib/transport/bdp_estimator.cc
+  src/core/lib/transport/call_factory.cc
   src/core/lib/transport/call_filters.cc
   src/core/lib/transport/call_final_info.cc
   src/core/lib/transport/call_size_estimator.cc
@@ -5330,6 +5332,7 @@ add_library(grpc_authorization_provider
   src/core/lib/surface/validate_metadata.cc
   src/core/lib/surface/version.cc
   src/core/lib/transport/batch_builder.cc
+  src/core/lib/transport/call_factory.cc
   src/core/lib/transport/call_filters.cc
   src/core/lib/transport/call_final_info.cc
   src/core/lib/transport/call_size_estimator.cc
diff --git a/Makefile b/Makefile
index 62801ec68c8..d72aa44ba90 100644
--- a/Makefile
+++ b/Makefile
@@ -1682,6 +1682,7 @@ LIBGRPC_SRC = \
     src/core/lib/surface/version.cc \
     src/core/lib/transport/batch_builder.cc \
     src/core/lib/transport/bdp_estimator.cc \
+    src/core/lib/transport/call_factory.cc \
     src/core/lib/transport/call_filters.cc \
     src/core/lib/transport/call_final_info.cc \
     src/core/lib/transport/call_size_estimator.cc \
@@ -2239,6 +2240,7 @@ LIBGRPC_UNSECURE_SRC = \
     src/core/lib/surface/version.cc \
     src/core/lib/transport/batch_builder.cc \
     src/core/lib/transport/bdp_estimator.cc \
+    src/core/lib/transport/call_factory.cc \
     src/core/lib/transport/call_filters.cc \
     src/core/lib/transport/call_final_info.cc \
     src/core/lib/transport/call_size_estimator.cc \
diff --git a/Package.swift b/Package.swift
index 77c64ae043f..39976a00980 100644
--- a/Package.swift
+++ b/Package.swift
@@ -1848,6 +1848,8 @@ let package = Package(
         "src/core/lib/transport/batch_builder.h",
         "src/core/lib/transport/bdp_estimator.cc",
         "src/core/lib/transport/bdp_estimator.h",
+        "src/core/lib/transport/call_factory.cc",
+        "src/core/lib/transport/call_factory.h",
         "src/core/lib/transport/call_filters.cc",
         "src/core/lib/transport/call_filters.h",
         "src/core/lib/transport/call_final_info.cc",
diff --git a/bazel/experiments.bzl b/bazel/experiments.bzl
index a1b87eca1d6..ca5894b6010 100644
--- a/bazel/experiments.bzl
+++ b/bazel/experiments.bzl
@@ -18,6 +18,7 @@
 
 EXPERIMENT_ENABLES = {
     "call_status_override_on_cancellation": "call_status_override_on_cancellation",
+    "call_v3": "call_v3",
     "canary_client_privacy": "canary_client_privacy",
     "client_idleness": "client_idleness",
     "client_privacy": "client_privacy",
@@ -37,7 +38,6 @@ EXPERIMENT_ENABLES = {
     "promise_based_server_call": "promise_based_server_call",
     "registered_method_lookup_in_transport": "registered_method_lookup_in_transport",
     "promise_based_inproc_transport": "promise_based_client_call,promise_based_inproc_transport,promise_based_server_call,registered_method_lookup_in_transport",
-    "registered_methods_map": "registered_methods_map",
     "rfc_max_concurrent_streams": "rfc_max_concurrent_streams",
     "round_robin_delegate_to_pick_first": "round_robin_delegate_to_pick_first",
     "rstpit": "rstpit",
diff --git a/build_autogenerated.yaml b/build_autogenerated.yaml
index 27a146c8193..1964edf6d8a 100644
--- a/build_autogenerated.yaml
+++ b/build_autogenerated.yaml
@@ -1152,6 +1152,7 @@ libs:
   - src/core/lib/surface/wait_for_cq_end_op.h
   - src/core/lib/transport/batch_builder.h
   - src/core/lib/transport/bdp_estimator.h
+  - src/core/lib/transport/call_factory.h
   - src/core/lib/transport/call_filters.h
   - src/core/lib/transport/call_final_info.h
   - src/core/lib/transport/call_size_estimator.h
@@ -1956,6 +1957,7 @@ libs:
   - src/core/lib/surface/version.cc
   - src/core/lib/transport/batch_builder.cc
   - src/core/lib/transport/bdp_estimator.cc
+  - src/core/lib/transport/call_factory.cc
   - src/core/lib/transport/call_filters.cc
   - src/core/lib/transport/call_final_info.cc
   - src/core/lib/transport/call_size_estimator.cc
@@ -2608,6 +2610,7 @@ libs:
   - src/core/lib/surface/wait_for_cq_end_op.h
   - src/core/lib/transport/batch_builder.h
   - src/core/lib/transport/bdp_estimator.h
+  - src/core/lib/transport/call_factory.h
   - src/core/lib/transport/call_filters.h
   - src/core/lib/transport/call_final_info.h
   - src/core/lib/transport/call_size_estimator.h
@@ -3031,6 +3034,7 @@ libs:
   - src/core/lib/surface/version.cc
   - src/core/lib/transport/batch_builder.cc
   - src/core/lib/transport/bdp_estimator.cc
+  - src/core/lib/transport/call_factory.cc
   - src/core/lib/transport/call_filters.cc
   - src/core/lib/transport/call_final_info.cc
   - src/core/lib/transport/call_size_estimator.cc
@@ -4670,6 +4674,7 @@ libs:
   - src/core/lib/surface/validate_metadata.h
   - src/core/lib/surface/wait_for_cq_end_op.h
   - src/core/lib/transport/batch_builder.h
+  - src/core/lib/transport/call_factory.h
   - src/core/lib/transport/call_filters.h
   - src/core/lib/transport/call_final_info.h
   - src/core/lib/transport/call_size_estimator.h
@@ -4971,6 +4976,7 @@ libs:
   - src/core/lib/surface/validate_metadata.cc
   - src/core/lib/surface/version.cc
   - src/core/lib/transport/batch_builder.cc
+  - src/core/lib/transport/call_factory.cc
   - src/core/lib/transport/call_filters.cc
   - src/core/lib/transport/call_final_info.cc
   - src/core/lib/transport/call_size_estimator.cc
diff --git a/config.m4 b/config.m4
index fd7cc94eb7a..30bbb2dcf5c 100644
--- a/config.m4
+++ b/config.m4
@@ -810,6 +810,7 @@ if test "$PHP_GRPC" != "no"; then
     src/core/lib/surface/version.cc \
     src/core/lib/transport/batch_builder.cc \
     src/core/lib/transport/bdp_estimator.cc \
+    src/core/lib/transport/call_factory.cc \
     src/core/lib/transport/call_filters.cc \
     src/core/lib/transport/call_final_info.cc \
     src/core/lib/transport/call_size_estimator.cc \
diff --git a/config.w32 b/config.w32
index 4f3f9167d47..b316bb9bbcb 100644
--- a/config.w32
+++ b/config.w32
@@ -775,6 +775,7 @@ if (PHP_GRPC != "no") {
     "src\\core\\lib\\surface\\version.cc " +
     "src\\core\\lib\\transport\\batch_builder.cc " +
     "src\\core\\lib\\transport\\bdp_estimator.cc " +
+    "src\\core\\lib\\transport\\call_factory.cc " +
     "src\\core\\lib\\transport\\call_filters.cc " +
     "src\\core\\lib\\transport\\call_final_info.cc " +
     "src\\core\\lib\\transport\\call_size_estimator.cc " +
diff --git a/examples/cpp/csm/Dockerfile.client b/examples/cpp/csm/Dockerfile.client
index 8773240b916..69f01f5248a 100644
--- a/examples/cpp/csm/Dockerfile.client
+++ b/examples/cpp/csm/Dockerfile.client
@@ -22,7 +22,7 @@ RUN ln -s /usr/bin/python3 /usr/bin/python
 RUN mkdir /artifacts
 
 COPY . .
-RUN OVERRIDE_BAZEL_VERSION=5.4.0 tools/bazel build //examples/cpp/csm:csm_greeter_client
+RUN tools/bazel build //examples/cpp/csm:csm_greeter_client
 RUN cp -rL /workdir/bazel-bin/examples/cpp/csm/csm_greeter_client /artifacts/
 
 FROM python:3.9-slim-bookworm
diff --git a/examples/cpp/csm/README.md b/examples/cpp/csm/README.md
index 5774d2b33b8..db766e42f1a 100644
--- a/examples/cpp/csm/README.md
+++ b/examples/cpp/csm/README.md
@@ -6,6 +6,30 @@ This CSM example builds on the [Hello World Example](https://github.com/grpc/grp
 
 The client takes the following command-line arguments -
 * target - By default, the client tries to connect to the xDS "xds:///helloworld:50051" and gRPC would use xDS to resolve this target and connect to the server backend. This can be overriden to change the target.
+* cookie_name - session affinity cookie name. Defaults to "GSSA"
+* delay_s - delay (in seconds) between the RPCs. Default value is 5
+
 
 The server takes the following command-line arguments -
 * port - Port on which the Hello World service is run. Defaults to 50051.
+
+## Building
+
+From the gRPC workspace folder:
+
+Client:
+```
+docker build -f examples/cpp/csm/Dockerfile.client
+```
+Server:
+```
+docker build -f examples/cpp/csm/Dockerfile.server
+```
+
+To push to a registry, add a tag to the image either by adding a `-t` flag to `docker build` command above or run:
+
+```
+docker image tag ${sha from build command above} ${tag}
+```
+
+And then push the tagged image using `docker push`
\ No newline at end of file
diff --git a/examples/cpp/csm/csm_greeter_client.cc b/examples/cpp/csm/csm_greeter_client.cc
index 6ad635f5a63..3aba7058562 100644
--- a/examples/cpp/csm/csm_greeter_client.cc
+++ b/examples/cpp/csm/csm_greeter_client.cc
@@ -16,16 +16,20 @@
  *
  */
 
+#include <sys/types.h>
+
+#include <chrono>
 #include <condition_variable>
 #include <iostream>
 #include <memory>
-#include <mutex>
+#include <optional>
 #include <string>
 
 #include "absl/flags/flag.h"
 #include "absl/flags/parse.h"
 #include "absl/strings/str_join.h"
 #include "absl/strings/str_split.h"
+#include "absl/types/optional.h"
 #include "opentelemetry/exporters/prometheus/exporter_factory.h"
 #include "opentelemetry/exporters/prometheus/exporter_options.h"
 #include "opentelemetry/sdk/metrics/meter_provider.h"
@@ -41,6 +45,8 @@
 #endif
 
 ABSL_FLAG(std::string, target, "xds:///helloworld:50051", "Target string");
+ABSL_FLAG(std::string, cookie_name, "GSSA", "Cookie name");
+ABSL_FLAG(uint, delay_s, 5, "Delay between requests");
 
 using grpc::Channel;
 using grpc::ClientContext;
@@ -49,15 +55,13 @@ using helloworld::Greeter;
 using helloworld::HelloReply;
 using helloworld::HelloRequest;
 
+namespace {
+
 struct Cookie {
   std::string name;
   std::string value;
   std::set<std::string> attributes;
 
-  std::pair<std::string, std::string> Header() const {
-    return std::make_pair("cookie", absl::StrFormat("%s=%s", name, value));
-  }
-
   template <typename Sink>
   friend void AbslStringify(Sink& sink, const Cookie& cookie) {
     absl::Format(&sink, "(Cookie: %s, value: %s, attributes: {%s})",
@@ -66,49 +70,45 @@ struct Cookie {
   }
 };
 
-class GreeterClient {
- protected:
-  static Cookie ParseCookie(absl::string_view header) {
-    Cookie cookie;
-    std::pair<absl::string_view, absl::string_view> name_value =
-        absl::StrSplit(header, absl::MaxSplits('=', 1));
-    cookie.name = std::string(name_value.first);
-    std::pair<absl::string_view, absl::string_view> value_attrs =
-        absl::StrSplit(name_value.second, absl::MaxSplits(';', 1));
-    cookie.value = std::string(value_attrs.first);
-    for (absl::string_view segment : absl::StrSplit(value_attrs.second, ';')) {
-      cookie.attributes.emplace(absl::StripAsciiWhitespace(segment));
-    }
-    return cookie;
+Cookie ParseCookie(absl::string_view header) {
+  Cookie cookie;
+  std::pair<absl::string_view, absl::string_view> name_value =
+      absl::StrSplit(header, absl::MaxSplits('=', 1));
+  cookie.name = std::string(name_value.first);
+  std::pair<absl::string_view, absl::string_view> value_attrs =
+      absl::StrSplit(name_value.second, absl::MaxSplits(';', 1));
+  cookie.value = std::string(value_attrs.first);
+  for (absl::string_view segment : absl::StrSplit(value_attrs.second, ';')) {
+    cookie.attributes.emplace(absl::StripAsciiWhitespace(segment));
   }
+  return cookie;
+}
 
-  static std::vector<Cookie> GetCookies(
-      const std::multimap<grpc::string_ref, grpc::string_ref>&
-          server_initial_metadata,
-      absl::string_view cookie_name) {
-    std::vector<Cookie> values;
-    auto pair = server_initial_metadata.equal_range("set-cookie");
-    for (auto it = pair.first; it != pair.second; ++it) {
-      gpr_log(GPR_INFO, "set-cookie header: %s", it->second.data());
-      const auto cookie = ParseCookie(it->second.data());
-      if (cookie.name == cookie_name) {
-        values.emplace_back(cookie);
-      }
+std::vector<Cookie> GetCookies(
+    const std::multimap<grpc::string_ref, grpc::string_ref>& initial_metadata,
+    absl::string_view cookie_name) {
+  std::vector<Cookie> values;
+  auto pair = initial_metadata.equal_range("set-cookie");
+  for (auto it = pair.first; it != pair.second; ++it) {
+    const auto cookie = ParseCookie(it->second.data());
+    if (cookie.name == cookie_name) {
+      values.emplace_back(std::move(cookie));
     }
-    return values;
   }
+  return values;
+}
 
+class GreeterClient {
  public:
-  GreeterClient(std::shared_ptr<Channel> channel)
-      : stub_(Greeter::NewStub(channel)) {}
+  GreeterClient(std::shared_ptr<Channel> channel, absl::string_view cookie_name)
+      : stub_(Greeter::NewStub(channel)), cookie_name_(cookie_name) {}
 
   // Assembles the client's payload, sends it and presents the response back
   // from the server.
-  std::string SayHello(const std::string& user, Cookie* cookieFromServer,
-                       const Cookie* cookieToServer) {
+  void SayHello() {
     // Data we are sending to the server.
     HelloRequest request;
-    request.set_name(user);
+    request.set_name("world");
 
     // Container for the data we expect from the server.
     HelloReply reply;
@@ -120,59 +120,45 @@ class GreeterClient {
     // The actual RPC.
     std::mutex mu;
     std::condition_variable cv;
-    bool done = false;
-    Status status;
-    if (cookieToServer != NULL) {
-      std::pair<std::string, std::string> cookieHeader =
-          cookieToServer->Header();
-      context.AddMetadata(cookieHeader.first, cookieHeader.second);
+    absl::optional<Status> status;
+    // Set the cookie header if we already got a cookie from the server
+    if (cookie_from_server_.has_value()) {
+      context.AddMetadata("cookie",
+                          absl::StrFormat("%s=%s", cookie_from_server_->name,
+                                          cookie_from_server_->value));
     }
-    stub_->async()->SayHello(&context, &request, &reply,
-                             [&mu, &cv, &done, &status](Status s) {
-                               status = std::move(s);
-                               std::lock_guard<std::mutex> lock(mu);
-                               done = true;
-                               cv.notify_one();
-                             });
-
     std::unique_lock<std::mutex> lock(mu);
-    while (!done) {
+    stub_->async()->SayHello(&context, &request, &reply, [&](Status s) {
+      std::lock_guard<std::mutex> lock(mu);
+      status = std::move(s);
+      cv.notify_one();
+    });
+    while (!status.has_value()) {
       cv.wait(lock);
     }
-
-    // Act upon its status.
-    if (status.ok()) {
-      if (cookieFromServer != NULL) {
-        const std::multimap<grpc::string_ref, grpc::string_ref>&
-            server_initial_metadata = context.GetServerInitialMetadata();
-        std::vector<Cookie> cookies =
-            GetCookies(server_initial_metadata, "GSSA");
-        if (!cookies.empty()) {
-          *cookieFromServer = cookies.front();
-        }
-      }
-      return reply.message();
-    } else {
-      std::cout << status.error_code() << ": " << status.error_message()
-                << std::endl;
-      return "RPC failed";
+    if (!status->ok()) {
+      std::cout << "RPC failed" << status->error_code() << ": "
+                << status->error_message() << std::endl;
+      return;
     }
+    const std::multimap<grpc::string_ref, grpc::string_ref>&
+        server_initial_metadata = context.GetServerInitialMetadata();
+    // Update a cookie after a successful request
+    std::vector<Cookie> cookies =
+        GetCookies(server_initial_metadata, cookie_name_);
+    if (!cookies.empty()) {
+      cookie_from_server_.emplace(std::move(cookies.front()));
+    }
+    std::cout << "Greeter received: " << reply.message() << std::endl;
   }
 
  private:
   std::unique_ptr<Greeter::Stub> stub_;
+  std::string cookie_name_;
+  absl::optional<Cookie> cookie_from_server_;
 };
 
-static void sayHello(GreeterClient& greeter, Cookie* cookieFromServer,
-                     const Cookie* cookieToServer) {
-  std::string user("world");
-  std::string reply = greeter.SayHello(user, cookieFromServer, cookieToServer);
-  std::cout << "Greeter received: " << reply << std::endl;
-  std::this_thread::sleep_for(std::chrono::seconds(5));
-}
-
-int main(int argc, char** argv) {
-  absl::ParseCommandLine(argc, argv);
+absl::StatusOr<grpc::experimental::CsmObservability> InitializeObservability() {
   opentelemetry::exporter::metrics::PrometheusExporterOptions opts;
   // default was "localhost:9464" which causes connection issue across GKE pods
   opts.url = "0.0.0.0:9464";
@@ -181,21 +167,29 @@ int main(int argc, char** argv) {
   auto meter_provider =
       std::make_shared<opentelemetry::sdk::metrics::MeterProvider>();
   meter_provider->AddMetricReader(std::move(prometheus_exporter));
-  auto observability = grpc::experimental::CsmObservabilityBuilder()
-                           .SetMeterProvider(std::move(meter_provider))
-                           .BuildAndRegister();
+  return grpc::experimental::CsmObservabilityBuilder()
+      .SetMeterProvider(std::move(meter_provider))
+      .BuildAndRegister();
+}
+
+}  // namespace
+
+int main(int argc, char** argv) {
+  absl::ParseCommandLine(argc, argv);
+  // Setup the observability
+  auto observability = InitializeObservability();
   if (!observability.ok()) {
     std::cerr << "CsmObservability::Init() failed: "
               << observability.status().ToString() << std::endl;
     return static_cast<int>(observability.status().code());
   }
-  GreeterClient greeter(grpc::CreateChannel(
-      absl::GetFlag(FLAGS_target), grpc::InsecureChannelCredentials()));
-
-  Cookie session_cookie;
-  sayHello(greeter, &session_cookie, NULL);
+  GreeterClient greeter(grpc::CreateChannel(absl::GetFlag(FLAGS_target),
+                                            grpc::InsecureChannelCredentials()),
+                        absl::GetFlag(FLAGS_cookie_name));
   while (true) {
-    sayHello(greeter, NULL, &session_cookie);
+    greeter.SayHello();
+    std::this_thread::sleep_for(
+        std::chrono::seconds(absl::GetFlag(FLAGS_delay_s)));
   }
   return 0;
 }
diff --git a/gRPC-C++.podspec b/gRPC-C++.podspec
index 400bb4f4430..1b5a250be1b 100644
--- a/gRPC-C++.podspec
+++ b/gRPC-C++.podspec
@@ -1256,6 +1256,7 @@ Pod::Spec.new do |s|
                       'src/core/lib/surface/wait_for_cq_end_op.h',
                       'src/core/lib/transport/batch_builder.h',
                       'src/core/lib/transport/bdp_estimator.h',
+                      'src/core/lib/transport/call_factory.h',
                       'src/core/lib/transport/call_filters.h',
                       'src/core/lib/transport/call_final_info.h',
                       'src/core/lib/transport/call_size_estimator.h',
@@ -2512,6 +2513,7 @@ Pod::Spec.new do |s|
                               'src/core/lib/surface/wait_for_cq_end_op.h',
                               'src/core/lib/transport/batch_builder.h',
                               'src/core/lib/transport/bdp_estimator.h',
+                              'src/core/lib/transport/call_factory.h',
                               'src/core/lib/transport/call_filters.h',
                               'src/core/lib/transport/call_final_info.h',
                               'src/core/lib/transport/call_size_estimator.h',
diff --git a/gRPC-Core.podspec b/gRPC-Core.podspec
index 53440b2b239..19436806e23 100644
--- a/gRPC-Core.podspec
+++ b/gRPC-Core.podspec
@@ -1957,6 +1957,8 @@ Pod::Spec.new do |s|
                       'src/core/lib/transport/batch_builder.h',
                       'src/core/lib/transport/bdp_estimator.cc',
                       'src/core/lib/transport/bdp_estimator.h',
+                      'src/core/lib/transport/call_factory.cc',
+                      'src/core/lib/transport/call_factory.h',
                       'src/core/lib/transport/call_filters.cc',
                       'src/core/lib/transport/call_filters.h',
                       'src/core/lib/transport/call_final_info.cc',
@@ -3291,6 +3293,7 @@ Pod::Spec.new do |s|
                               'src/core/lib/surface/wait_for_cq_end_op.h',
                               'src/core/lib/transport/batch_builder.h',
                               'src/core/lib/transport/bdp_estimator.h',
+                              'src/core/lib/transport/call_factory.h',
                               'src/core/lib/transport/call_filters.h',
                               'src/core/lib/transport/call_final_info.h',
                               'src/core/lib/transport/call_size_estimator.h',
diff --git a/grpc.gemspec b/grpc.gemspec
index fce5c38066b..b78139acfa3 100644
--- a/grpc.gemspec
+++ b/grpc.gemspec
@@ -1850,6 +1850,8 @@ Gem::Specification.new do |s|
   s.files += %w( src/core/lib/transport/batch_builder.h )
   s.files += %w( src/core/lib/transport/bdp_estimator.cc )
   s.files += %w( src/core/lib/transport/bdp_estimator.h )
+  s.files += %w( src/core/lib/transport/call_factory.cc )
+  s.files += %w( src/core/lib/transport/call_factory.h )
   s.files += %w( src/core/lib/transport/call_filters.cc )
   s.files += %w( src/core/lib/transport/call_filters.h )
   s.files += %w( src/core/lib/transport/call_final_info.cc )
diff --git a/grpc.gyp b/grpc.gyp
index a7d33960d3c..b43595e2b0d 100644
--- a/grpc.gyp
+++ b/grpc.gyp
@@ -996,6 +996,7 @@
         'src/core/lib/surface/version.cc',
         'src/core/lib/transport/batch_builder.cc',
         'src/core/lib/transport/bdp_estimator.cc',
+        'src/core/lib/transport/call_factory.cc',
         'src/core/lib/transport/call_filters.cc',
         'src/core/lib/transport/call_final_info.cc',
         'src/core/lib/transport/call_size_estimator.cc',
@@ -1493,6 +1494,7 @@
         'src/core/lib/surface/version.cc',
         'src/core/lib/transport/batch_builder.cc',
         'src/core/lib/transport/bdp_estimator.cc',
+        'src/core/lib/transport/call_factory.cc',
         'src/core/lib/transport/call_filters.cc',
         'src/core/lib/transport/call_final_info.cc',
         'src/core/lib/transport/call_size_estimator.cc',
@@ -2282,6 +2284,7 @@
         'src/core/lib/surface/validate_metadata.cc',
         'src/core/lib/surface/version.cc',
         'src/core/lib/transport/batch_builder.cc',
+        'src/core/lib/transport/call_factory.cc',
         'src/core/lib/transport/call_filters.cc',
         'src/core/lib/transport/call_final_info.cc',
         'src/core/lib/transport/call_size_estimator.cc',
diff --git a/package.xml b/package.xml
index e85ce015ec0..2248624d4d4 100644
--- a/package.xml
+++ b/package.xml
@@ -1832,6 +1832,8 @@
     <file baseinstalldir="/" name="src/core/lib/transport/batch_builder.h" role="src" />
     <file baseinstalldir="/" name="src/core/lib/transport/bdp_estimator.cc" role="src" />
     <file baseinstalldir="/" name="src/core/lib/transport/bdp_estimator.h" role="src" />
+    <file baseinstalldir="/" name="src/core/lib/transport/call_factory.cc" role="src" />
+    <file baseinstalldir="/" name="src/core/lib/transport/call_factory.h" role="src" />
     <file baseinstalldir="/" name="src/core/lib/transport/call_filters.cc" role="src" />
     <file baseinstalldir="/" name="src/core/lib/transport/call_filters.h" role="src" />
     <file baseinstalldir="/" name="src/core/lib/transport/call_final_info.cc" role="src" />
diff --git a/src/core/BUILD b/src/core/BUILD
index 8380b8d715b..6ee421b2c23 100644
--- a/src/core/BUILD
+++ b/src/core/BUILD
@@ -6734,6 +6734,38 @@ grpc_cc_library(
     ],
 )
 
+grpc_cc_library(
+    name = "call_factory",
+    srcs = [
+        "lib/transport/call_factory.cc",
+    ],
+    hdrs = [
+        "lib/transport/call_factory.h",
+    ],
+    deps = [
+        "arena",
+        "call_size_estimator",
+        "call_spine",
+        "channel_args",
+        "ref_counted",
+        "resource_quota",
+        "//:gpr_platform",
+        "//:stats",
+    ],
+)
+
+grpc_cc_library(
+    name = "call_destination",
+    hdrs = [
+        "lib/transport/call_destination.h",
+    ],
+    deps = [
+        "call_spine",
+        "//:gpr_platform",
+        "//:orphanable",
+    ],
+)
+
 grpc_cc_library(
     name = "parsed_metadata",
     srcs = [
diff --git a/src/core/lib/event_engine/forkable.cc b/src/core/lib/event_engine/forkable.cc
index 98e77d8babf..a51f1176506 100644
--- a/src/core/lib/event_engine/forkable.cc
+++ b/src/core/lib/event_engine/forkable.cc
@@ -43,13 +43,15 @@ bool IsForkEnabled() {
 void ObjectGroupForkHandler::RegisterForkable(
     std::shared_ptr<Forkable> forkable, GRPC_UNUSED void (*prepare)(void),
     GRPC_UNUSED void (*parent)(void), GRPC_UNUSED void (*child)(void)) {
-  GPR_ASSERT(!is_forking_);
-  forkables_.emplace_back(forkable);
+  if (IsForkEnabled()) {
+    GPR_ASSERT(!is_forking_);
+    forkables_.emplace_back(forkable);
 #ifdef GRPC_POSIX_FORK_ALLOW_PTHREAD_ATFORK
-  if (!std::exchange(registered_, true)) {
-    pthread_atfork(prepare, parent, child);
-  }
+    if (!std::exchange(registered_, true)) {
+      pthread_atfork(prepare, parent, child);
+    }
 #endif  // GRPC_POSIX_FORK_ALLOW_PTHREAD_ATFORK
+  }
 }
 
 void ObjectGroupForkHandler::Prefork() {
diff --git a/src/core/lib/experiments/experiments.cc b/src/core/lib/experiments/experiments.cc
index 23fd0d91640..38eaf479058 100644
--- a/src/core/lib/experiments/experiments.cc
+++ b/src/core/lib/experiments/experiments.cc
@@ -29,6 +29,8 @@ const char* const description_call_status_override_on_cancellation =
     "with cancellation.";
 const char* const additional_constraints_call_status_override_on_cancellation =
     "{}";
+const char* const description_call_v3 = "Promise-based call version 3.";
+const char* const additional_constraints_call_v3 = "{}";
 const char* const description_canary_client_privacy =
     "If set, canary client privacy";
 const char* const additional_constraints_canary_client_privacy = "{}";
@@ -105,9 +107,6 @@ const uint8_t required_experiments_promise_based_inproc_transport[] = {
     static_cast<uint8_t>(grpc_core::kExperimentIdPromiseBasedServerCall),
     static_cast<uint8_t>(
         grpc_core::kExperimentIdRegisteredMethodLookupInTransport)};
-const char* const description_registered_methods_map =
-    "Use absl::flat_hash_map for registered methods.";
-const char* const additional_constraints_registered_methods_map = "{}";
 const char* const description_rfc_max_concurrent_streams =
     "If set, enable rfc-compliant behavior (cancellation) in the advent that "
     "max concurrent streams are exceeded in chttp2. See "
@@ -191,6 +190,8 @@ const ExperimentMetadata g_experiment_metadata[] = {
      description_call_status_override_on_cancellation,
      additional_constraints_call_status_override_on_cancellation, nullptr, 0,
      kDefaultForDebugOnly, true},
+    {"call_v3", description_call_v3, additional_constraints_call_v3, nullptr, 0,
+     false, true},
     {"canary_client_privacy", description_canary_client_privacy,
      additional_constraints_canary_client_privacy, nullptr, 0, false, false},
     {"client_idleness", description_client_idleness,
@@ -234,8 +235,6 @@ const ExperimentMetadata g_experiment_metadata[] = {
      description_promise_based_inproc_transport,
      additional_constraints_promise_based_inproc_transport,
      required_experiments_promise_based_inproc_transport, 3, false, false},
-    {"registered_methods_map", description_registered_methods_map,
-     additional_constraints_registered_methods_map, nullptr, 0, false, true},
     {"rfc_max_concurrent_streams", description_rfc_max_concurrent_streams,
      additional_constraints_rfc_max_concurrent_streams, nullptr, 0, false,
      true},
@@ -293,6 +292,8 @@ const char* const description_call_status_override_on_cancellation =
     "with cancellation.";
 const char* const additional_constraints_call_status_override_on_cancellation =
     "{}";
+const char* const description_call_v3 = "Promise-based call version 3.";
+const char* const additional_constraints_call_v3 = "{}";
 const char* const description_canary_client_privacy =
     "If set, canary client privacy";
 const char* const additional_constraints_canary_client_privacy = "{}";
@@ -369,9 +370,6 @@ const uint8_t required_experiments_promise_based_inproc_transport[] = {
     static_cast<uint8_t>(grpc_core::kExperimentIdPromiseBasedServerCall),
     static_cast<uint8_t>(
         grpc_core::kExperimentIdRegisteredMethodLookupInTransport)};
-const char* const description_registered_methods_map =
-    "Use absl::flat_hash_map for registered methods.";
-const char* const additional_constraints_registered_methods_map = "{}";
 const char* const description_rfc_max_concurrent_streams =
     "If set, enable rfc-compliant behavior (cancellation) in the advent that "
     "max concurrent streams are exceeded in chttp2. See "
@@ -455,6 +453,8 @@ const ExperimentMetadata g_experiment_metadata[] = {
      description_call_status_override_on_cancellation,
      additional_constraints_call_status_override_on_cancellation, nullptr, 0,
      kDefaultForDebugOnly, true},
+    {"call_v3", description_call_v3, additional_constraints_call_v3, nullptr, 0,
+     false, true},
     {"canary_client_privacy", description_canary_client_privacy,
      additional_constraints_canary_client_privacy, nullptr, 0, false, false},
     {"client_idleness", description_client_idleness,
@@ -498,8 +498,6 @@ const ExperimentMetadata g_experiment_metadata[] = {
      description_promise_based_inproc_transport,
      additional_constraints_promise_based_inproc_transport,
      required_experiments_promise_based_inproc_transport, 3, false, false},
-    {"registered_methods_map", description_registered_methods_map,
-     additional_constraints_registered_methods_map, nullptr, 0, false, true},
     {"rfc_max_concurrent_streams", description_rfc_max_concurrent_streams,
      additional_constraints_rfc_max_concurrent_streams, nullptr, 0, false,
      true},
@@ -557,6 +555,8 @@ const char* const description_call_status_override_on_cancellation =
     "with cancellation.";
 const char* const additional_constraints_call_status_override_on_cancellation =
     "{}";
+const char* const description_call_v3 = "Promise-based call version 3.";
+const char* const additional_constraints_call_v3 = "{}";
 const char* const description_canary_client_privacy =
     "If set, canary client privacy";
 const char* const additional_constraints_canary_client_privacy = "{}";
@@ -633,9 +633,6 @@ const uint8_t required_experiments_promise_based_inproc_transport[] = {
     static_cast<uint8_t>(grpc_core::kExperimentIdPromiseBasedServerCall),
     static_cast<uint8_t>(
         grpc_core::kExperimentIdRegisteredMethodLookupInTransport)};
-const char* const description_registered_methods_map =
-    "Use absl::flat_hash_map for registered methods.";
-const char* const additional_constraints_registered_methods_map = "{}";
 const char* const description_rfc_max_concurrent_streams =
     "If set, enable rfc-compliant behavior (cancellation) in the advent that "
     "max concurrent streams are exceeded in chttp2. See "
@@ -719,6 +716,8 @@ const ExperimentMetadata g_experiment_metadata[] = {
      description_call_status_override_on_cancellation,
      additional_constraints_call_status_override_on_cancellation, nullptr, 0,
      kDefaultForDebugOnly, true},
+    {"call_v3", description_call_v3, additional_constraints_call_v3, nullptr, 0,
+     false, true},
     {"canary_client_privacy", description_canary_client_privacy,
      additional_constraints_canary_client_privacy, nullptr, 0, false, false},
     {"client_idleness", description_client_idleness,
@@ -762,8 +761,6 @@ const ExperimentMetadata g_experiment_metadata[] = {
      description_promise_based_inproc_transport,
      additional_constraints_promise_based_inproc_transport,
      required_experiments_promise_based_inproc_transport, 3, false, false},
-    {"registered_methods_map", description_registered_methods_map,
-     additional_constraints_registered_methods_map, nullptr, 0, false, true},
     {"rfc_max_concurrent_streams", description_rfc_max_concurrent_streams,
      additional_constraints_rfc_max_concurrent_streams, nullptr, 0, false,
      true},
diff --git a/src/core/lib/experiments/experiments.h b/src/core/lib/experiments/experiments.h
index b9a6ecca096..f7d214fa8e9 100644
--- a/src/core/lib/experiments/experiments.h
+++ b/src/core/lib/experiments/experiments.h
@@ -67,6 +67,7 @@ inline bool IsCallStatusOverrideOnCancellationEnabled() {
   return true;
 #endif
 }
+inline bool IsCallV3Enabled() { return false; }
 inline bool IsCanaryClientPrivacyEnabled() { return false; }
 #define GRPC_EXPERIMENT_IS_INCLUDED_CLIENT_IDLENESS
 inline bool IsClientIdlenessEnabled() { return true; }
@@ -92,7 +93,6 @@ inline bool IsPromiseBasedServerCallEnabled() { return false; }
 #define GRPC_EXPERIMENT_IS_INCLUDED_REGISTERED_METHOD_LOOKUP_IN_TRANSPORT
 inline bool IsRegisteredMethodLookupInTransportEnabled() { return true; }
 inline bool IsPromiseBasedInprocTransportEnabled() { return false; }
-inline bool IsRegisteredMethodsMapEnabled() { return false; }
 inline bool IsRfcMaxConcurrentStreamsEnabled() { return false; }
 #define GRPC_EXPERIMENT_IS_INCLUDED_ROUND_ROBIN_DELEGATE_TO_PICK_FIRST
 inline bool IsRoundRobinDelegateToPickFirstEnabled() { return true; }
@@ -128,6 +128,7 @@ inline bool IsCallStatusOverrideOnCancellationEnabled() {
   return true;
 #endif
 }
+inline bool IsCallV3Enabled() { return false; }
 inline bool IsCanaryClientPrivacyEnabled() { return false; }
 #define GRPC_EXPERIMENT_IS_INCLUDED_CLIENT_IDLENESS
 inline bool IsClientIdlenessEnabled() { return true; }
@@ -154,7 +155,6 @@ inline bool IsPromiseBasedServerCallEnabled() { return false; }
 #define GRPC_EXPERIMENT_IS_INCLUDED_REGISTERED_METHOD_LOOKUP_IN_TRANSPORT
 inline bool IsRegisteredMethodLookupInTransportEnabled() { return true; }
 inline bool IsPromiseBasedInprocTransportEnabled() { return false; }
-inline bool IsRegisteredMethodsMapEnabled() { return false; }
 inline bool IsRfcMaxConcurrentStreamsEnabled() { return false; }
 #define GRPC_EXPERIMENT_IS_INCLUDED_ROUND_ROBIN_DELEGATE_TO_PICK_FIRST
 inline bool IsRoundRobinDelegateToPickFirstEnabled() { return true; }
@@ -190,6 +190,7 @@ inline bool IsCallStatusOverrideOnCancellationEnabled() {
   return true;
 #endif
 }
+inline bool IsCallV3Enabled() { return false; }
 inline bool IsCanaryClientPrivacyEnabled() { return false; }
 #define GRPC_EXPERIMENT_IS_INCLUDED_CLIENT_IDLENESS
 inline bool IsClientIdlenessEnabled() { return true; }
@@ -217,7 +218,6 @@ inline bool IsPromiseBasedServerCallEnabled() { return false; }
 #define GRPC_EXPERIMENT_IS_INCLUDED_REGISTERED_METHOD_LOOKUP_IN_TRANSPORT
 inline bool IsRegisteredMethodLookupInTransportEnabled() { return true; }
 inline bool IsPromiseBasedInprocTransportEnabled() { return false; }
-inline bool IsRegisteredMethodsMapEnabled() { return false; }
 inline bool IsRfcMaxConcurrentStreamsEnabled() { return false; }
 #define GRPC_EXPERIMENT_IS_INCLUDED_ROUND_ROBIN_DELEGATE_TO_PICK_FIRST
 inline bool IsRoundRobinDelegateToPickFirstEnabled() { return true; }
@@ -246,6 +246,7 @@ inline bool IsWrrDelegateToPickFirstEnabled() { return true; }
 #else
 enum ExperimentIds {
   kExperimentIdCallStatusOverrideOnCancellation,
+  kExperimentIdCallV3,
   kExperimentIdCanaryClientPrivacy,
   kExperimentIdClientIdleness,
   kExperimentIdClientPrivacy,
@@ -265,7 +266,6 @@ enum ExperimentIds {
   kExperimentIdPromiseBasedServerCall,
   kExperimentIdRegisteredMethodLookupInTransport,
   kExperimentIdPromiseBasedInprocTransport,
-  kExperimentIdRegisteredMethodsMap,
   kExperimentIdRfcMaxConcurrentStreams,
   kExperimentIdRoundRobinDelegateToPickFirst,
   kExperimentIdRstpit,
@@ -290,6 +290,10 @@ enum ExperimentIds {
 inline bool IsCallStatusOverrideOnCancellationEnabled() {
   return IsExperimentEnabled(kExperimentIdCallStatusOverrideOnCancellation);
 }
+#define GRPC_EXPERIMENT_IS_INCLUDED_CALL_V3
+inline bool IsCallV3Enabled() {
+  return IsExperimentEnabled(kExperimentIdCallV3);
+}
 #define GRPC_EXPERIMENT_IS_INCLUDED_CANARY_CLIENT_PRIVACY
 inline bool IsCanaryClientPrivacyEnabled() {
   return IsExperimentEnabled(kExperimentIdCanaryClientPrivacy);
@@ -366,10 +370,6 @@ inline bool IsRegisteredMethodLookupInTransportEnabled() {
 inline bool IsPromiseBasedInprocTransportEnabled() {
   return IsExperimentEnabled(kExperimentIdPromiseBasedInprocTransport);
 }
-#define GRPC_EXPERIMENT_IS_INCLUDED_REGISTERED_METHODS_MAP
-inline bool IsRegisteredMethodsMapEnabled() {
-  return IsExperimentEnabled(kExperimentIdRegisteredMethodsMap);
-}
 #define GRPC_EXPERIMENT_IS_INCLUDED_RFC_MAX_CONCURRENT_STREAMS
 inline bool IsRfcMaxConcurrentStreamsEnabled() {
   return IsExperimentEnabled(kExperimentIdRfcMaxConcurrentStreams);
diff --git a/src/core/lib/experiments/experiments.yaml b/src/core/lib/experiments/experiments.yaml
index a008eb40304..cfcb561ed3a 100644
--- a/src/core/lib/experiments/experiments.yaml
+++ b/src/core/lib/experiments/experiments.yaml
@@ -47,6 +47,11 @@
   expiry: 2024/04/01
   owner: vigneshbabu@google.com
   test_tags: []
+- name: call_v3
+  description: Promise-based call version 3.
+  expiry: 2024/06/01
+  owner: ctiller@google.com
+  test_tags: []
 - name: canary_client_privacy
   description:
     If set, canary client privacy
@@ -178,12 +183,6 @@
   expiry: 2024/03/31
   owner: yashkt@google.com
   test_tags: ["surface_registered_method_lookup"]
-- name: registered_methods_map
-  description:
-    Use absl::flat_hash_map for registered methods.
-  expiry: 2024/01/31
-  owner: alishananda@google.com
-  test_tags: []
 - name: rfc_max_concurrent_streams
   description:
     If set, enable rfc-compliant behavior (cancellation) in the advent that
diff --git a/src/core/lib/experiments/rollouts.yaml b/src/core/lib/experiments/rollouts.yaml
index 8f895e2fbb2..8c9e77b36d3 100644
--- a/src/core/lib/experiments/rollouts.yaml
+++ b/src/core/lib/experiments/rollouts.yaml
@@ -42,6 +42,8 @@
 
 - name: call_status_override_on_cancellation
   default: debug
+- name: call_v3
+  default: false
 - name: canary_client_privacy
   default: false
 - name: client_idleness
@@ -92,8 +94,6 @@
   default: false
 - name: registered_method_lookup_in_transport
   default: true
-- name: registered_methods_map
-  default: false
 - name: rfc_max_concurrent_streams
   default: false
 - name: round_robin_delegate_to_pick_first
diff --git a/src/core/lib/surface/channel.cc b/src/core/lib/surface/channel.cc
index e2015e3acdf..7886baba05e 100644
--- a/src/core/lib/surface/channel.cc
+++ b/src/core/lib/surface/channel.cc
@@ -58,12 +58,25 @@
 #include "src/core/lib/surface/channel_init.h"
 #include "src/core/lib/surface/channel_stack_type.h"
 #include "src/core/lib/surface/init_internally.h"
+#include "src/core/lib/transport/call_factory.h"
 #include "src/core/lib/transport/transport.h"
 
 // IWYU pragma: no_include <type_traits>
 
 namespace grpc_core {
 
+namespace {
+
+class NotReallyACallFactory final : public CallFactory {
+ public:
+  using CallFactory::CallFactory;
+  CallInitiator CreateCall(ClientMetadataHandle, Arena*) override {
+    Crash("NotReallyACallFactory::CreateCall should never be called");
+  }
+};
+
+}  // namespace
+
 Channel::Channel(bool is_client, bool is_promising, std::string target,
                  const ChannelArgs& channel_args,
                  grpc_compression_options compression_options,
@@ -71,14 +84,10 @@ Channel::Channel(bool is_client, bool is_promising, std::string target,
     : is_client_(is_client),
       is_promising_(is_promising),
       compression_options_(compression_options),
-      call_size_estimator_(channel_stack->call_stack_size +
-                           grpc_call_get_initial_size_estimate()),
       channelz_node_(channel_args.GetObjectRef<channelz::ChannelNode>()),
-      allocator_(channel_args.GetObject<ResourceQuota>()
-                     ->memory_quota()
-                     ->CreateMemoryOwner()),
       target_(std::move(target)),
-      channel_stack_(std::move(channel_stack)) {
+      channel_stack_(std::move(channel_stack)),
+      call_factory_(MakeRefCounted<NotReallyACallFactory>(channel_args)) {
   // We need to make sure that grpc_shutdown() does not shut things down
   // until after the channel is destroyed.  However, the channel may not
   // actually be destroyed by the time grpc_channel_destroy() returns,
@@ -107,17 +116,6 @@ Channel::Channel(bool is_client, bool is_promising, std::string target,
   };
 }
 
-Arena* Channel::CreateArena() {
-  const size_t initial_size = call_size_estimator_.CallSizeEstimate();
-  global_stats().IncrementCallInitialSize(initial_size);
-  return Arena::Create(initial_size, &allocator_);
-}
-
-void Channel::DestroyArena(Arena* arena) {
-  call_size_estimator_.UpdateCallSizeEstimate(arena->TotalUsedBytes());
-  arena->Destroy();
-}
-
 absl::StatusOr<RefCountedPtr<Channel>> Channel::CreateWithBuilder(
     ChannelStackBuilder* builder) {
   auto channel_args = builder->channel_args();
diff --git a/src/core/lib/surface/channel.h b/src/core/lib/surface/channel.h
index d7a6e1f5acc..4796331893e 100644
--- a/src/core/lib/surface/channel.h
+++ b/src/core/lib/surface/channel.h
@@ -52,10 +52,9 @@
 #include "src/core/lib/gprpp/sync.h"
 #include "src/core/lib/gprpp/time.h"
 #include "src/core/lib/iomgr/iomgr_fwd.h"
-#include "src/core/lib/resource_quota/memory_quota.h"
 #include "src/core/lib/slice/slice.h"
 #include "src/core/lib/surface/channel_stack_type.h"
-#include "src/core/lib/transport/call_size_estimator.h"
+#include "src/core/lib/transport/call_factory.h"
 #include "src/core/lib/transport/transport.h"
 
 /// The same as grpc_channel_destroy, but doesn't create an ExecCtx, and so
@@ -122,8 +121,8 @@ class Channel : public RefCounted<Channel>,
 
   channelz::ChannelNode* channelz_node() const { return channelz_node_.get(); }
 
-  Arena* CreateArena();
-  void DestroyArena(Arena* arena);
+  Arena* CreateArena() { return call_factory_->CreateArena(); }
+  void DestroyArena(Arena* arena) { return call_factory_->DestroyArena(arena); }
 
   absl::string_view target() const { return target_; }
   bool is_client() const { return is_client_; }
@@ -148,12 +147,11 @@ class Channel : public RefCounted<Channel>,
   const bool is_client_;
   const bool is_promising_;
   const grpc_compression_options compression_options_;
-  CallSizeEstimator call_size_estimator_;
   CallRegistrationTable registration_table_;
   RefCountedPtr<channelz::ChannelNode> channelz_node_;
-  MemoryAllocator allocator_;
   std::string target_;
   const RefCountedPtr<grpc_channel_stack> channel_stack_;
+  const RefCountedPtr<CallFactory> call_factory_;
 };
 
 }  // namespace grpc_core
diff --git a/src/core/lib/transport/call_destination.h b/src/core/lib/transport/call_destination.h
new file mode 100644
index 00000000000..e291bd703cb
--- /dev/null
+++ b/src/core/lib/transport/call_destination.h
@@ -0,0 +1,35 @@
+// Copyright 2024 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef GRPC_SRC_CORE_LIB_TRANSPORT_CALL_DESTINATION_H
+#define GRPC_SRC_CORE_LIB_TRANSPORT_CALL_DESTINATION_H
+
+#include <grpc/support/port_platform.h>
+
+#include "src/core/lib/gprpp/orphanable.h"
+#include "src/core/lib/transport/call_spine.h"
+
+namespace grpc_core {
+
+// CallDestination is responsible for the processing of a CallHandler.
+// It might be a transport, the server API, or a subchannel on the client (for
+// instance).
+class CallDestination : public Orphanable {
+ public:
+  virtual void StartCall(CallHandler call_handler) = 0;
+};
+
+}  // namespace grpc_core
+
+#endif  // GRPC_SRC_CORE_LIB_TRANSPORT_CALL_DESTINATION_H
diff --git a/src/core/lib/transport/call_factory.cc b/src/core/lib/transport/call_factory.cc
new file mode 100644
index 00000000000..bcc01cc87ff
--- /dev/null
+++ b/src/core/lib/transport/call_factory.cc
@@ -0,0 +1,41 @@
+// Copyright 2024 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <grpc/support/port_platform.h>
+
+#include "src/core/lib/transport/call_factory.h"
+
+#include "src/core/lib/debug/stats.h"
+#include "src/core/lib/resource_quota/resource_quota.h"
+
+namespace grpc_core {
+
+CallFactory::CallFactory(const ChannelArgs& args)
+    : call_size_estimator_(1024),
+      allocator_(args.GetObject<ResourceQuota>()
+                     ->memory_quota()
+                     ->CreateMemoryOwner()) {}
+
+Arena* CallFactory::CreateArena() {
+  const size_t initial_size = call_size_estimator_.CallSizeEstimate();
+  global_stats().IncrementCallInitialSize(initial_size);
+  return Arena::Create(initial_size, &allocator_);
+}
+
+void CallFactory::DestroyArena(Arena* arena) {
+  call_size_estimator_.UpdateCallSizeEstimate(arena->TotalUsedBytes());
+  arena->Destroy();
+}
+
+}  // namespace grpc_core
diff --git a/src/core/lib/transport/call_factory.h b/src/core/lib/transport/call_factory.h
new file mode 100644
index 00000000000..99bb1768650
--- /dev/null
+++ b/src/core/lib/transport/call_factory.h
@@ -0,0 +1,56 @@
+// Copyright 2024 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef GRPC_SRC_CORE_LIB_TRANSPORT_CALL_FACTORY_H
+#define GRPC_SRC_CORE_LIB_TRANSPORT_CALL_FACTORY_H
+
+#include <grpc/support/port_platform.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gprpp/ref_counted.h"
+#include "src/core/lib/resource_quota/arena.h"
+#include "src/core/lib/transport/call_size_estimator.h"
+#include "src/core/lib/transport/call_spine.h"
+
+namespace grpc_core {
+
+// CallFactory creates calls.
+class CallFactory : public RefCounted<CallFactory> {
+ public:
+  explicit CallFactory(const ChannelArgs& args);
+
+  // Create an arena for a call.
+  // We do this as a separate step so that servers can create arenas without
+  // creating the call into it - in the case that we have a HTTP/2 rapid reset
+  // like attack this saves a lot of cpu time.
+  Arena* CreateArena();
+  // Destroy an arena created by CreateArena.
+  // Updates the call size estimator so that we always create arenas of about
+  // the right size.
+  void DestroyArena(Arena* arena);
+
+  // Create a call. The call will be created in the given arena.
+  // It is the CallFactory's responsibility to ensure that the CallHandler
+  // associated with the call is eventually handled by something (typically a
+  // CallDestination, but this is not strictly required).
+  virtual CallInitiator CreateCall(ClientMetadataHandle md, Arena* arena) = 0;
+
+ private:
+  CallSizeEstimator call_size_estimator_;
+  MemoryAllocator allocator_;
+};
+
+}  // namespace grpc_core
+
+#endif  // GRPC_SRC_CORE_LIB_TRANSPORT_CALL_FACTORY_H
diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py
index 89da4fae139..9d766455be7 100644
--- a/src/python/grpcio/grpc_core_dependencies.py
+++ b/src/python/grpcio/grpc_core_dependencies.py
@@ -784,6 +784,7 @@ CORE_SOURCE_FILES = [
     'src/core/lib/surface/version.cc',
     'src/core/lib/transport/batch_builder.cc',
     'src/core/lib/transport/bdp_estimator.cc',
+    'src/core/lib/transport/call_factory.cc',
     'src/core/lib/transport/call_filters.cc',
     'src/core/lib/transport/call_final_info.cc',
     'src/core/lib/transport/call_size_estimator.cc',
diff --git a/templates/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_0/Dockerfile.template b/templates/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_0/Dockerfile.template
new file mode 100644
index 00000000000..2bff360192c
--- /dev/null
+++ b/templates/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_0/Dockerfile.template
@@ -0,0 +1,23 @@
+%YAML 1.2
+--- |
+  # Copyright 2024 gRPC authors.
+  #
+  # Licensed under the Apache License, Version 2.0 (the "License");
+  # you may not use this file except in compliance with the License.
+  # You may obtain a copy of the License at
+  #
+  #     http://www.apache.org/licenses/LICENSE-2.0
+  #
+  # Unless required by applicable law or agreed to in writing, software
+  # distributed under the License is distributed on an "AS IS" BASIS,
+  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  # See the License for the specific language governing permissions and
+  # limitations under the License.
+
+  FROM debian:10
+
+  <%include file="../../apt_get_basic.include"/>
+  <%include file="../../ruby_3_0_deps.include"/>
+
+  # Define the default command.
+  CMD ["bash"]
diff --git a/templates/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_1/Dockerfile.template b/templates/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_1/Dockerfile.template
new file mode 100644
index 00000000000..a0f67ff7d89
--- /dev/null
+++ b/templates/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_1/Dockerfile.template
@@ -0,0 +1,23 @@
+%YAML 1.2
+--- |
+  # Copyright 2024 gRPC authors.
+  #
+  # Licensed under the Apache License, Version 2.0 (the "License");
+  # you may not use this file except in compliance with the License.
+  # You may obtain a copy of the License at
+  #
+  #     http://www.apache.org/licenses/LICENSE-2.0
+  #
+  # Unless required by applicable law or agreed to in writing, software
+  # distributed under the License is distributed on an "AS IS" BASIS,
+  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  # See the License for the specific language governing permissions and
+  # limitations under the License.
+
+  FROM debian:10
+
+  <%include file="../../apt_get_basic.include"/>
+  <%include file="../../ruby_3_1_deps.include"/>
+
+  # Define the default command.
+  CMD ["bash"]
diff --git a/templates/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_2/Dockerfile.template b/templates/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_2/Dockerfile.template
new file mode 100644
index 00000000000..a19720ff436
--- /dev/null
+++ b/templates/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_2/Dockerfile.template
@@ -0,0 +1,23 @@
+%YAML 1.2
+--- |
+  # Copyright 2024 gRPC authors.
+  #
+  # Licensed under the Apache License, Version 2.0 (the "License");
+  # you may not use this file except in compliance with the License.
+  # You may obtain a copy of the License at
+  #
+  #     http://www.apache.org/licenses/LICENSE-2.0
+  #
+  # Unless required by applicable law or agreed to in writing, software
+  # distributed under the License is distributed on an "AS IS" BASIS,
+  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  # See the License for the specific language governing permissions and
+  # limitations under the License.
+
+  FROM debian:10
+
+  <%include file="../../apt_get_basic.include"/>
+  <%include file="../../ruby_3_2_deps.include"/>
+
+  # Define the default command.
+  CMD ["bash"]
diff --git a/templates/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_3/Dockerfile.template b/templates/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_3/Dockerfile.template
new file mode 100644
index 00000000000..8fc44ef1a80
--- /dev/null
+++ b/templates/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_3/Dockerfile.template
@@ -0,0 +1,23 @@
+%YAML 1.2
+--- |
+  # Copyright 2024 gRPC authors.
+  #
+  # Licensed under the Apache License, Version 2.0 (the "License");
+  # you may not use this file except in compliance with the License.
+  # You may obtain a copy of the License at
+  #
+  #     http://www.apache.org/licenses/LICENSE-2.0
+  #
+  # Unless required by applicable law or agreed to in writing, software
+  # distributed under the License is distributed on an "AS IS" BASIS,
+  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  # See the License for the specific language governing permissions and
+  # limitations under the License.
+
+  FROM debian:10
+
+  <%include file="../../apt_get_basic.include"/>
+  <%include file="../../ruby_3_3_deps.include"/>
+
+  # Define the default command.
+  CMD ["bash"]
diff --git a/templates/tools/dockerfile/ruby_3_0_deps.include b/templates/tools/dockerfile/ruby_3_0_deps.include
new file mode 100644
index 00000000000..7ca23e21325
--- /dev/null
+++ b/templates/tools/dockerfile/ruby_3_0_deps.include
@@ -0,0 +1,16 @@
+#==================
+# Ruby dependencies
+
+# Install rvm
+RUN apt-get update && apt-get install -y gnupg2 && apt-get clean
+RUN gpg2 --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3 7D2BAF1CF37B13E2069D6956105BD0E739499BDB
+RUN \curl -sSL https://get.rvm.io | bash -s stable
+
+# Install Ruby 3.0
+RUN apt-get update && apt-get install -y procps && apt-get clean
+RUN /bin/bash -l -c "rvm install ruby-3.0.5"
+RUN /bin/bash -l -c "rvm use --default ruby-3.0.5"
+RUN /bin/bash -l -c "echo 'gem: --no-document' > ~/.gemrc"
+RUN /bin/bash -l -c "echo 'export PATH=/usr/local/rvm/bin:$PATH' >> ~/.bashrc"
+RUN /bin/bash -l -c "echo 'rvm --default use ruby-3.0.5' >> ~/.bashrc"
+RUN /bin/bash -l -c "gem install bundler --no-document"
diff --git a/templates/tools/dockerfile/ruby_3_1_deps.include b/templates/tools/dockerfile/ruby_3_1_deps.include
new file mode 100644
index 00000000000..f5c46ef84f1
--- /dev/null
+++ b/templates/tools/dockerfile/ruby_3_1_deps.include
@@ -0,0 +1,16 @@
+#==================
+# Ruby dependencies
+
+# Install rvm
+RUN apt-get update && apt-get install -y gnupg2 && apt-get clean
+RUN gpg2 --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3 7D2BAF1CF37B13E2069D6956105BD0E739499BDB
+RUN \curl -sSL https://get.rvm.io | bash -s stable
+
+# Install Ruby 3.1
+RUN apt-get update && apt-get install -y procps && apt-get clean
+RUN /bin/bash -l -c "rvm install ruby-3.1.3"
+RUN /bin/bash -l -c "rvm use --default ruby-3.1.3"
+RUN /bin/bash -l -c "echo 'gem: --no-document' > ~/.gemrc"
+RUN /bin/bash -l -c "echo 'export PATH=/usr/local/rvm/bin:$PATH' >> ~/.bashrc"
+RUN /bin/bash -l -c "echo 'rvm --default use ruby-3.1.3' >> ~/.bashrc"
+RUN /bin/bash -l -c "gem install bundler --no-document"
diff --git a/templates/tools/dockerfile/ruby_3_2_deps.include b/templates/tools/dockerfile/ruby_3_2_deps.include
new file mode 100644
index 00000000000..36f6711e946
--- /dev/null
+++ b/templates/tools/dockerfile/ruby_3_2_deps.include
@@ -0,0 +1,16 @@
+#==================
+# Ruby dependencies
+
+# Install rvm
+RUN apt-get update && apt-get install -y gnupg2 && apt-get clean
+RUN gpg2 --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3 7D2BAF1CF37B13E2069D6956105BD0E739499BDB
+RUN \curl -sSL https://get.rvm.io | bash -s stable
+
+# Install Ruby 3.2
+RUN apt-get update && apt-get install -y procps && apt-get clean
+RUN /bin/bash -l -c "rvm install ruby-3.2.0"
+RUN /bin/bash -l -c "rvm use --default ruby-3.2.0"
+RUN /bin/bash -l -c "echo 'gem: --no-document' > ~/.gemrc"
+RUN /bin/bash -l -c "echo 'export PATH=/usr/local/rvm/bin:$PATH' >> ~/.bashrc"
+RUN /bin/bash -l -c "echo 'rvm --default use ruby-3.2.0' >> ~/.bashrc"
+RUN /bin/bash -l -c "gem install bundler --no-document"
diff --git a/templates/tools/dockerfile/ruby_3_3_deps.include b/templates/tools/dockerfile/ruby_3_3_deps.include
new file mode 100644
index 00000000000..ae3fb964949
--- /dev/null
+++ b/templates/tools/dockerfile/ruby_3_3_deps.include
@@ -0,0 +1,16 @@
+#==================
+# Ruby dependencies
+
+# Install rvm
+RUN apt-get update && apt-get install -y gnupg2 && apt-get clean
+RUN gpg2 --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3 7D2BAF1CF37B13E2069D6956105BD0E739499BDB
+RUN \curl -sSL https://get.rvm.io | bash -s stable
+
+# Install Ruby 3.3
+RUN apt-get update && apt-get install -y procps && apt-get clean
+RUN /bin/bash -l -c "rvm install ruby-3.3.0"
+RUN /bin/bash -l -c "rvm use --default ruby-3.3.0"
+RUN /bin/bash -l -c "echo 'gem: --no-document' > ~/.gemrc"
+RUN /bin/bash -l -c "echo 'export PATH=/usr/local/rvm/bin:$PATH' >> ~/.bashrc"
+RUN /bin/bash -l -c "echo 'rvm --default use ruby-3.3.0' >> ~/.bashrc"
+RUN /bin/bash -l -c "gem install bundler --no-document"
diff --git a/third_party/utf8_range/.github/workflows/bazel_tests.yml b/third_party/utf8_range/.github/workflows/bazel_tests.yml
index ae3e970419c..adf711859aa 100644
--- a/third_party/utf8_range/.github/workflows/bazel_tests.yml
+++ b/third_party/utf8_range/.github/workflows/bazel_tests.yml
@@ -29,8 +29,15 @@ jobs:
 
     steps:
       - uses: actions/checkout@v2
+      - name: Setup Bazel
+        run: |
+          echo "BAZEL=bazel" >> $GITHUB_ENV
+          echo "USE_BAZEL_VERSION=5.1.1" >> $GITHUB_ENV
+      - name: Setup Bazel (macOS)
+        if: runner.os == 'macOS'
+        run: echo "BAZEL=bazelisk" >> $GITHUB_ENV
       - name: Install dependencies
         run: sudo apt update && sudo apt install -y ${{ matrix.install }}
         if: matrix.install != ''
       - name: Run tests
-        run: cd ${{ github.workspace }} && CC=${{ matrix.CC }} bazel test --test_output=errors ... ${{ matrix.flags }}
+        run: cd ${{ github.workspace }} && CC=${{ matrix.CC }} ${{ env.BAZEL }} test --test_output=errors ... ${{ matrix.flags }}
diff --git a/third_party/utf8_range/BUILD.bazel b/third_party/utf8_range/BUILD.bazel
index 8582f1bdc48..b9ba04e17c6 100644
--- a/third_party/utf8_range/BUILD.bazel
+++ b/third_party/utf8_range/BUILD.bazel
@@ -4,6 +4,12 @@
 # license that can be found in the LICENSE file or at
 # https://opensource.org/licenses/MIT.
 
+DEFAULT_VISIBILITY = [
+    "//third_party/protobuf:__subpackages__",
+    "//:__subpackages__",
+    "//util/utf8/internal:__subpackages__",
+]
+
 package(default_visibility = ["//visibility:public"])
 
 licenses(["notice"])
@@ -21,6 +27,7 @@ filegroup(
     ],
     visibility = [
         "@com_google_protobuf//:__subpackages__",
+        "@upb//:__subpackages__",
     ],
 )
 
diff --git a/third_party/utf8_range/CMakeLists.txt b/third_party/utf8_range/CMakeLists.txt
index 69a4d899077..344952d38cf 100644
--- a/third_party/utf8_range/CMakeLists.txt
+++ b/third_party/utf8_range/CMakeLists.txt
@@ -73,6 +73,11 @@ if (utf8_range_ENABLE_INSTALL)
   install(FILES "${PROJECT_BINARY_DIR}/${PROJECT_NAME}-config.cmake"
     DESTINATION "${CMAKE_INSTALL_LIBDIR}/cmake/${PROJECT_NAME}"
   )
+  configure_file(${CMAKE_CURRENT_SOURCE_DIR}/cmake/utf8_range.pc.cmake
+                 ${CMAKE_CURRENT_BINARY_DIR}/utf8_range.pc @ONLY)
+  install(
+    FILES ${CMAKE_CURRENT_BINARY_DIR}/utf8_range.pc
+    DESTINATION "${CMAKE_INSTALL_LIBDIR}/pkgconfig")
 
   # Install public headers explicitly.
   install(FILES utf8_range.h utf8_validity.h
diff --git a/third_party/utf8_range/cmake/utf8_range.pc.cmake b/third_party/utf8_range/cmake/utf8_range.pc.cmake
new file mode 100644
index 00000000000..64b4b770b29
--- /dev/null
+++ b/third_party/utf8_range/cmake/utf8_range.pc.cmake
@@ -0,0 +1,11 @@
+prefix=@CMAKE_INSTALL_PREFIX@
+exec_prefix=@CMAKE_INSTALL_PREFIX@
+libdir=@CMAKE_INSTALL_FULL_LIBDIR@
+includedir=@CMAKE_INSTALL_FULL_INCLUDEDIR@
+
+Name: UTF8 Range
+Description: Google's UTF8 Library
+Version: 1.0
+Requires: absl_strings
+Libs: -L${libdir} -lutf8_validity -lutf8_range @CMAKE_THREAD_LIBS_INIT@
+Cflags: -I${includedir}
diff --git a/third_party/utf8_range/utf8_validity.cc b/third_party/utf8_range/utf8_validity.cc
index db811993f81..9e945766732 100644
--- a/third_party/utf8_range/utf8_validity.cc
+++ b/third_party/utf8_range/utf8_validity.cc
@@ -133,6 +133,7 @@ size_t ValidUTF8Span(const char* data, const char* end) {
   return err_pos + (1 - ReturnPosition);
 }
 
+#ifdef __SSE4_1__
 /* Returns the number of bytes needed to skip backwards to get to the first
    byte of codepoint.
  */
@@ -148,6 +149,7 @@ inline int CodepointSkipBackwards(int32_t codepoint_word) {
   }
   return 0;
 }
+#endif  // __SSE4_1__
 
 /* Skipping over ASCII as much as possible, per 8 bytes. It is intentional
    as most strings to check for validity consist only of 1 byte codepoints.
diff --git a/tools/bazelify_tests/dockerimage_current_versions.bzl b/tools/bazelify_tests/dockerimage_current_versions.bzl
index 1d4696194d1..638fdcaada4 100644
--- a/tools/bazelify_tests/dockerimage_current_versions.bzl
+++ b/tools/bazelify_tests/dockerimage_current_versions.bzl
@@ -57,7 +57,10 @@ DOCKERIMAGE_CURRENT_VERSIONS = {
     "tools/dockerfile/distribtest/ruby_centos7_x64.current_version": "docker://us-docker.pkg.dev/grpc-testing/testing-images-public/ruby_centos7_x64@sha256:4d529b984b78ca179086f7f9b416605e2d9a96ca0a28a71f4421bb5ffdc18f96",
     "tools/dockerfile/distribtest/ruby_debian10_x64.current_version": "docker://us-docker.pkg.dev/grpc-testing/testing-images-public/ruby_debian10_x64@sha256:1298c39c950b2a48261555b6cff1ae66230a5020f100d3b381759285f0caf84e",
     "tools/dockerfile/distribtest/ruby_debian10_x64_ruby_2_7.current_version": "docker://us-docker.pkg.dev/grpc-testing/testing-images-public/ruby_debian10_x64_ruby_2_7@sha256:5ee26ad3abe2683c9a8ee03987ab0ae63f50793c3d3f5e4be6e6cbacb4556fcf",
-    "tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_0.current_version": "docker://us-docker.pkg.dev/grpc-testing/testing-images-public/ruby_debian10_x64_ruby_3_0@sha256:9190da90a2a95eca1370cef64dcba7ddee9f59cc7487093da6711c1280a0b0f9",
+    "tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_0.current_version": "docker://us-docker.pkg.dev/grpc-testing/testing-images-public/ruby_debian10_x64_ruby_3_0@sha256:be9dfa852af7fe5c700747bbad5b93d15c3f030ac980a107e98fc9b70fee6077",
+    "tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_1.current_version": "docker://us-docker.pkg.dev/grpc-testing/testing-images-public/ruby_debian10_x64_ruby_3_1@sha256:282a052275fa9e0378301d759f5febcc9f1ff0ebb702e536690cd4fed1a43904",
+    "tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_2.current_version": "docker://us-docker.pkg.dev/grpc-testing/testing-images-public/ruby_debian10_x64_ruby_3_2@sha256:f1ba9030d7d7496a6ed42c0e1a9a2fe3204208e94983d7d4b6b4c7752042ed28",
+    "tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_3.current_version": "docker://us-docker.pkg.dev/grpc-testing/testing-images-public/ruby_debian10_x64_ruby_3_3@sha256:a26d6bfe8f799ae817db0bd0280283c2cc9ea0f02174fe9a2cb0cd10d71d81f8",
     "tools/dockerfile/distribtest/ruby_ubuntu2004_x64.current_version": "docker://us-docker.pkg.dev/grpc-testing/testing-images-public/ruby_ubuntu2004_x64@sha256:426cbf625df0c0e7451b9716041996dc6a35a3788c1a24d68891256f84733d8e",
     "tools/dockerfile/distribtest/ruby_ubuntu2204_x64.current_version": "docker://us-docker.pkg.dev/grpc-testing/testing-images-public/ruby_ubuntu2204_x64@sha256:1c74c312f8a4ab37e629732a35daa3056e12136b90f37540cdf9fa11303c7eb8",
     "tools/dockerfile/grpc_artifact_centos6_x64.current_version": "docker://us-docker.pkg.dev/grpc-testing/testing-images-public/grpc_artifact_centos6_x64@sha256:3285047265ea2b7c5d4df4c769b2d05f56288d947c75e16d27ae2dee693f791b",
diff --git a/tools/buildgen/extract_metadata_from_bazel_xml.py b/tools/buildgen/extract_metadata_from_bazel_xml.py
index 2a289c5e2bb..23c41909ff4 100755
--- a/tools/buildgen/extract_metadata_from_bazel_xml.py
+++ b/tools/buildgen/extract_metadata_from_bazel_xml.py
@@ -99,11 +99,13 @@ EXTERNAL_PROTO_LIBRARIES = {
 # For that we need mapping from external repo name to a corresponding
 # path to a git submodule.
 EXTERNAL_SOURCE_PREFIXES = {
-    "@utf8_range": "third_party/utf8_range",
-    "@com_googlesource_code_re2": "third_party/re2",
-    "@com_google_googletest": "third_party/googletest",
-    "@com_google_protobuf": "third_party/upb",
-    "@zlib": "third_party/zlib",
+    # TODO(veblush): Remove @utf8_range// item once protobuf is upgraded to 26.x
+    "@utf8_range//": "third_party/utf8_range",
+    "@com_googlesource_code_re2//": "third_party/re2",
+    "@com_google_googletest//": "third_party/googletest",
+    "@com_google_protobuf//upb": "third_party/upb/upb",
+    "@com_google_protobuf//third_party/utf8_range": "third_party/utf8_range",
+    "@zlib//": "third_party/zlib",
 }
 
 
@@ -209,9 +211,9 @@ def _try_extract_source_file_path(label: str) -> str:
         # This is an external source file. We are only interested in sources
         # for some of the external libraries.
         for lib_name, prefix in EXTERNAL_SOURCE_PREFIXES.items():
-            if label.startswith(lib_name + "//"):
+            if label.startswith(lib_name):
                 return (
-                    label.replace("%s//" % lib_name, prefix + "/")
+                    label.replace("%s" % lib_name, prefix)
                     .replace(":", "/")
                     .replace("//", "/")
                 )
@@ -1100,7 +1102,7 @@ _BUILD_EXTRA_METADATA = {
         "build": "all",
         "_RENAME": "upb_textformat_lib",
     },
-    "@utf8_range//:utf8_range": {
+    "@com_google_protobuf//third_party/utf8_range:utf8_range": {
         "language": "c",
         "build": "all",
         # rename to utf8_range_lib is necessary for now to avoid clash with utf8_range target in protobuf's cmake
@@ -1400,6 +1402,15 @@ tests = _exclude_unwanted_cc_tests(_extract_cc_tests(bazel_rules))
 # only very little "extra metadata" would be needed and/or it would be trivial
 # to generate it automatically.
 all_extra_metadata = {}
+# TODO(veblush): Remove this workaround once protobuf is upgraded to 26.x
+if "@com_google_protobuf//third_party/utf8_range:utf8_range" not in bazel_rules:
+    md = _BUILD_EXTRA_METADATA[
+        "@com_google_protobuf//third_party/utf8_range:utf8_range"
+    ]
+    del _BUILD_EXTRA_METADATA[
+        "@com_google_protobuf//third_party/utf8_range:utf8_range"
+    ]
+    _BUILD_EXTRA_METADATA["@utf8_range//:utf8_range"] = md
 all_extra_metadata.update(_BUILD_EXTRA_METADATA)
 all_extra_metadata.update(
     _generate_build_extra_metadata_for_tests(tests, bazel_rules)
diff --git a/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_0.current_version b/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_0.current_version
index 1f83abbda4d..709c65daebb 100644
--- a/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_0.current_version
+++ b/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_0.current_version
@@ -1 +1 @@
-us-docker.pkg.dev/grpc-testing/testing-images-public/ruby_debian10_x64_ruby_3_0:c277dbca7dbd28ff6112cbc842ba3144c49266bd@sha256:9190da90a2a95eca1370cef64dcba7ddee9f59cc7487093da6711c1280a0b0f9
\ No newline at end of file
+us-docker.pkg.dev/grpc-testing/testing-images-public/ruby_debian10_x64_ruby_3_0:dce91a018316b35b962e5fa576fded4775457077@sha256:be9dfa852af7fe5c700747bbad5b93d15c3f030ac980a107e98fc9b70fee6077
\ No newline at end of file
diff --git a/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_0/Dockerfile b/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_0/Dockerfile
index 1ca6d03190d..639bd90158c 100644
--- a/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_0/Dockerfile
+++ b/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_0/Dockerfile
@@ -1,4 +1,4 @@
-# Copyright 2020 gRPC authors.
+# Copyright 2024 gRPC authors.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,29 +14,71 @@
 
 FROM debian:10
 
-# Install Git and basic packages.
+#=================
+# Basic C core dependencies
+
+# C/C++ dependencies according to https://github.com/grpc/grpc/blob/master/BUILDING.md
+RUN apt-get update && apt-get install -y \
+  build-essential \
+  autoconf \
+  libtool \
+  pkg-config \
+  && apt-get clean
+
+# GCC
+RUN apt-get update && apt-get install -y \
+  gcc \
+  g++ \
+  && apt-get clean
+
+# libc6
 RUN apt-get update && apt-get install -y \
-  gnupg2 \
-  procps \
+  libc6 \
+  libc6-dbg \
+  libc6-dev \
+  && apt-get clean
+
+# Tools
+RUN apt-get update && apt-get install -y \
+  bzip2 \
   curl \
-  gcc && apt-get clean
+  dnsutils \
+  git \
+  lcov \
+  make \
+  strace \
+  time \
+  unzip \
+  wget \
+  zip \
+  && apt-get clean
+
+#=================
+# Setup git to access working directory across docker boundary.
+# This avoids the "fatal: detected dubious ownership in repository XYZ"
+# git error.
+
+RUN git config --global --add safe.directory '*'
+RUN git config --global protocol.file.allow always
+
 
 #==================
 # Ruby dependencies
 
 # Install rvm
-RUN gpg2 --keyserver hkp://keyserver.ubuntu.com --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3 7D2BAF1CF37B13E2069D6956105BD0E739499BDB
+RUN apt-get update && apt-get install -y gnupg2 && apt-get clean
+RUN gpg2 --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3 7D2BAF1CF37B13E2069D6956105BD0E739499BDB
 RUN \curl -sSL https://get.rvm.io | bash -s stable
 
 # Install Ruby 3.0
-RUN /bin/bash -l -c "rvm install ruby-3.0.0"
-RUN /bin/bash -l -c "rvm use --default ruby-3.0.0"
+RUN apt-get update && apt-get install -y procps && apt-get clean
+RUN /bin/bash -l -c "rvm install ruby-3.0.5"
+RUN /bin/bash -l -c "rvm use --default ruby-3.0.5"
 RUN /bin/bash -l -c "echo 'gem: --no-document' > ~/.gemrc"
 RUN /bin/bash -l -c "echo 'export PATH=/usr/local/rvm/bin:$PATH' >> ~/.bashrc"
-RUN /bin/bash -l -c "echo 'rvm --default use ruby-3.0.0' >> ~/.bashrc"
+RUN /bin/bash -l -c "echo 'rvm --default use ruby-3.0.5' >> ~/.bashrc"
 RUN /bin/bash -l -c "gem install bundler --no-document"
 
-RUN mkdir /var/local/jenkins
 
 # Define the default command.
 CMD ["bash"]
diff --git a/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_1.current_version b/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_1.current_version
new file mode 100644
index 00000000000..4995cf042c1
--- /dev/null
+++ b/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_1.current_version
@@ -0,0 +1 @@
+us-docker.pkg.dev/grpc-testing/testing-images-public/ruby_debian10_x64_ruby_3_1:348a4b975f4d19772bb9a1a95fc03720f89edbc0@sha256:282a052275fa9e0378301d759f5febcc9f1ff0ebb702e536690cd4fed1a43904
\ No newline at end of file
diff --git a/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_1/Dockerfile b/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_1/Dockerfile
new file mode 100644
index 00000000000..314b0d889fb
--- /dev/null
+++ b/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_1/Dockerfile
@@ -0,0 +1,84 @@
+# Copyright 2024 gRPC authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+FROM debian:10
+
+#=================
+# Basic C core dependencies
+
+# C/C++ dependencies according to https://github.com/grpc/grpc/blob/master/BUILDING.md
+RUN apt-get update && apt-get install -y \
+  build-essential \
+  autoconf \
+  libtool \
+  pkg-config \
+  && apt-get clean
+
+# GCC
+RUN apt-get update && apt-get install -y \
+  gcc \
+  g++ \
+  && apt-get clean
+
+# libc6
+RUN apt-get update && apt-get install -y \
+  libc6 \
+  libc6-dbg \
+  libc6-dev \
+  && apt-get clean
+
+# Tools
+RUN apt-get update && apt-get install -y \
+  bzip2 \
+  curl \
+  dnsutils \
+  git \
+  lcov \
+  make \
+  strace \
+  time \
+  unzip \
+  wget \
+  zip \
+  && apt-get clean
+
+#=================
+# Setup git to access working directory across docker boundary.
+# This avoids the "fatal: detected dubious ownership in repository XYZ"
+# git error.
+
+RUN git config --global --add safe.directory '*'
+RUN git config --global protocol.file.allow always
+
+
+#==================
+# Ruby dependencies
+
+# Install rvm
+RUN apt-get update && apt-get install -y gnupg2 && apt-get clean
+RUN gpg2 --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3 7D2BAF1CF37B13E2069D6956105BD0E739499BDB
+RUN \curl -sSL https://get.rvm.io | bash -s stable
+
+# Install Ruby 3.1
+RUN apt-get update && apt-get install -y procps && apt-get clean
+RUN /bin/bash -l -c "rvm install ruby-3.1.3"
+RUN /bin/bash -l -c "rvm use --default ruby-3.1.3"
+RUN /bin/bash -l -c "echo 'gem: --no-document' > ~/.gemrc"
+RUN /bin/bash -l -c "echo 'export PATH=/usr/local/rvm/bin:$PATH' >> ~/.bashrc"
+RUN /bin/bash -l -c "echo 'rvm --default use ruby-3.1.3' >> ~/.bashrc"
+RUN /bin/bash -l -c "gem install bundler --no-document"
+
+
+# Define the default command.
+CMD ["bash"]
diff --git a/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_2.current_version b/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_2.current_version
new file mode 100644
index 00000000000..f17c0cea7c5
--- /dev/null
+++ b/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_2.current_version
@@ -0,0 +1 @@
+us-docker.pkg.dev/grpc-testing/testing-images-public/ruby_debian10_x64_ruby_3_2:e9f8640c6bb80a029c65ed6ebdfddd9574a041e4@sha256:f1ba9030d7d7496a6ed42c0e1a9a2fe3204208e94983d7d4b6b4c7752042ed28
\ No newline at end of file
diff --git a/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_2/Dockerfile b/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_2/Dockerfile
new file mode 100644
index 00000000000..546acc6d0d3
--- /dev/null
+++ b/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_2/Dockerfile
@@ -0,0 +1,84 @@
+# Copyright 2024 gRPC authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+FROM debian:10
+
+#=================
+# Basic C core dependencies
+
+# C/C++ dependencies according to https://github.com/grpc/grpc/blob/master/BUILDING.md
+RUN apt-get update && apt-get install -y \
+  build-essential \
+  autoconf \
+  libtool \
+  pkg-config \
+  && apt-get clean
+
+# GCC
+RUN apt-get update && apt-get install -y \
+  gcc \
+  g++ \
+  && apt-get clean
+
+# libc6
+RUN apt-get update && apt-get install -y \
+  libc6 \
+  libc6-dbg \
+  libc6-dev \
+  && apt-get clean
+
+# Tools
+RUN apt-get update && apt-get install -y \
+  bzip2 \
+  curl \
+  dnsutils \
+  git \
+  lcov \
+  make \
+  strace \
+  time \
+  unzip \
+  wget \
+  zip \
+  && apt-get clean
+
+#=================
+# Setup git to access working directory across docker boundary.
+# This avoids the "fatal: detected dubious ownership in repository XYZ"
+# git error.
+
+RUN git config --global --add safe.directory '*'
+RUN git config --global protocol.file.allow always
+
+
+#==================
+# Ruby dependencies
+
+# Install rvm
+RUN apt-get update && apt-get install -y gnupg2 && apt-get clean
+RUN gpg2 --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3 7D2BAF1CF37B13E2069D6956105BD0E739499BDB
+RUN \curl -sSL https://get.rvm.io | bash -s stable
+
+# Install Ruby 3.2
+RUN apt-get update && apt-get install -y procps && apt-get clean
+RUN /bin/bash -l -c "rvm install ruby-3.2.0"
+RUN /bin/bash -l -c "rvm use --default ruby-3.2.0"
+RUN /bin/bash -l -c "echo 'gem: --no-document' > ~/.gemrc"
+RUN /bin/bash -l -c "echo 'export PATH=/usr/local/rvm/bin:$PATH' >> ~/.bashrc"
+RUN /bin/bash -l -c "echo 'rvm --default use ruby-3.2.0' >> ~/.bashrc"
+RUN /bin/bash -l -c "gem install bundler --no-document"
+
+
+# Define the default command.
+CMD ["bash"]
diff --git a/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_3.current_version b/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_3.current_version
new file mode 100644
index 00000000000..bbe14a70714
--- /dev/null
+++ b/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_3.current_version
@@ -0,0 +1 @@
+us-docker.pkg.dev/grpc-testing/testing-images-public/ruby_debian10_x64_ruby_3_3:c063f13c0987355593ce6bcf2123bb9717eeba35@sha256:a26d6bfe8f799ae817db0bd0280283c2cc9ea0f02174fe9a2cb0cd10d71d81f8
\ No newline at end of file
diff --git a/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_3/Dockerfile b/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_3/Dockerfile
new file mode 100644
index 00000000000..f7f27b649f7
--- /dev/null
+++ b/tools/dockerfile/distribtest/ruby_debian10_x64_ruby_3_3/Dockerfile
@@ -0,0 +1,84 @@
+# Copyright 2024 gRPC authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+FROM debian:10
+
+#=================
+# Basic C core dependencies
+
+# C/C++ dependencies according to https://github.com/grpc/grpc/blob/master/BUILDING.md
+RUN apt-get update && apt-get install -y \
+  build-essential \
+  autoconf \
+  libtool \
+  pkg-config \
+  && apt-get clean
+
+# GCC
+RUN apt-get update && apt-get install -y \
+  gcc \
+  g++ \
+  && apt-get clean
+
+# libc6
+RUN apt-get update && apt-get install -y \
+  libc6 \
+  libc6-dbg \
+  libc6-dev \
+  && apt-get clean
+
+# Tools
+RUN apt-get update && apt-get install -y \
+  bzip2 \
+  curl \
+  dnsutils \
+  git \
+  lcov \
+  make \
+  strace \
+  time \
+  unzip \
+  wget \
+  zip \
+  && apt-get clean
+
+#=================
+# Setup git to access working directory across docker boundary.
+# This avoids the "fatal: detected dubious ownership in repository XYZ"
+# git error.
+
+RUN git config --global --add safe.directory '*'
+RUN git config --global protocol.file.allow always
+
+
+#==================
+# Ruby dependencies
+
+# Install rvm
+RUN apt-get update && apt-get install -y gnupg2 && apt-get clean
+RUN gpg2 --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3 7D2BAF1CF37B13E2069D6956105BD0E739499BDB
+RUN \curl -sSL https://get.rvm.io | bash -s stable
+
+# Install Ruby 3.3
+RUN apt-get update && apt-get install -y procps && apt-get clean
+RUN /bin/bash -l -c "rvm install ruby-3.3.0"
+RUN /bin/bash -l -c "rvm use --default ruby-3.3.0"
+RUN /bin/bash -l -c "echo 'gem: --no-document' > ~/.gemrc"
+RUN /bin/bash -l -c "echo 'export PATH=/usr/local/rvm/bin:$PATH' >> ~/.bashrc"
+RUN /bin/bash -l -c "echo 'rvm --default use ruby-3.3.0' >> ~/.bashrc"
+RUN /bin/bash -l -c "gem install bundler --no-document"
+
+
+# Define the default command.
+CMD ["bash"]
diff --git a/tools/dockerfile/grpc_clang_format/clang_format_all_the_things.sh b/tools/dockerfile/grpc_clang_format/clang_format_all_the_things.sh
index 96fb2237f1c..ef654434917 100755
--- a/tools/dockerfile/grpc_clang_format/clang_format_all_the_things.sh
+++ b/tools/dockerfile/grpc_clang_format/clang_format_all_the_things.sh
@@ -52,6 +52,7 @@ do
     -and -not -name grpc_tls_credentials_options_comparator_test.cc \
     -and -not -path '*/cmake/build/*' \
     -and -not -path '*/third_party/*' \
+    -and -not -path '*/src/python/grpcio_observability/grpc_root/*' \
     `"
   done
 done
diff --git a/tools/doxygen/Doxyfile.c++.internal b/tools/doxygen/Doxyfile.c++.internal
index a175782598a..70c8d9c1a39 100644
--- a/tools/doxygen/Doxyfile.c++.internal
+++ b/tools/doxygen/Doxyfile.c++.internal
@@ -2849,6 +2849,8 @@ src/core/lib/transport/batch_builder.cc \
 src/core/lib/transport/batch_builder.h \
 src/core/lib/transport/bdp_estimator.cc \
 src/core/lib/transport/bdp_estimator.h \
+src/core/lib/transport/call_factory.cc \
+src/core/lib/transport/call_factory.h \
 src/core/lib/transport/call_filters.cc \
 src/core/lib/transport/call_filters.h \
 src/core/lib/transport/call_final_info.cc \
diff --git a/tools/doxygen/Doxyfile.core.internal b/tools/doxygen/Doxyfile.core.internal
index ccbeed1cb31..022958293f7 100644
--- a/tools/doxygen/Doxyfile.core.internal
+++ b/tools/doxygen/Doxyfile.core.internal
@@ -2630,6 +2630,8 @@ src/core/lib/transport/batch_builder.cc \
 src/core/lib/transport/batch_builder.h \
 src/core/lib/transport/bdp_estimator.cc \
 src/core/lib/transport/bdp_estimator.h \
+src/core/lib/transport/call_factory.cc \
+src/core/lib/transport/call_factory.h \
 src/core/lib/transport/call_filters.cc \
 src/core/lib/transport/call_filters.h \
 src/core/lib/transport/call_final_info.cc \
diff --git a/tools/interop_matrix/client_matrix.py b/tools/interop_matrix/client_matrix.py
index 1eaaf59c751..48caa2560a0 100644
--- a/tools/interop_matrix/client_matrix.py
+++ b/tools/interop_matrix/client_matrix.py
@@ -130,6 +130,7 @@ LANG_RELEASE_MATRIX = {
             ("v1.57.0", ReleaseInfo()),
             ("v1.58.0", ReleaseInfo()),
             ("v1.60.0", ReleaseInfo()),
+            ("v1.61.0", ReleaseInfo()),
         ]
     ),
     "go": OrderedDict(
@@ -764,6 +765,12 @@ LANG_RELEASE_MATRIX = {
                     runtimes=["python"], testcases_file="python__master"
                 ),
             ),
+            (
+                "v1.61.0",
+                ReleaseInfo(
+                    runtimes=["python"], testcases_file="python__master"
+                ),
+            ),
         ]
     ),
     "node": OrderedDict(
@@ -858,6 +865,7 @@ LANG_RELEASE_MATRIX = {
             ("v1.57.0", ReleaseInfo()),
             ("v1.58.0", ReleaseInfo()),
             ("v1.60.0", ReleaseInfo()),
+            ("v1.61.0", ReleaseInfo()),
         ]
     ),
     "php": OrderedDict(
@@ -916,6 +924,7 @@ LANG_RELEASE_MATRIX = {
             ("v1.57.0", ReleaseInfo()),
             ("v1.58.0", ReleaseInfo()),
             ("v1.60.0", ReleaseInfo()),
+            ("v1.61.0", ReleaseInfo()),
         ]
     ),
     "csharp": OrderedDict(
diff --git a/tools/run_tests/artifacts/distribtest_targets.py b/tools/run_tests/artifacts/distribtest_targets.py
index b611a1560a8..cff61987d04 100644
--- a/tools/run_tests/artifacts/distribtest_targets.py
+++ b/tools/run_tests/artifacts/distribtest_targets.py
@@ -476,6 +476,15 @@ def targets():
         RubyDistribTest(
             "linux", "x64", "debian10", ruby_version="ruby_3_0", presubmit=True
         ),
+        RubyDistribTest(
+            "linux", "x64", "debian10", ruby_version="ruby_3_1", presubmit=True
+        ),
+        RubyDistribTest(
+            "linux", "x64", "debian10", ruby_version="ruby_3_2", presubmit=True
+        ),
+        RubyDistribTest(
+            "linux", "x64", "debian10", ruby_version="ruby_3_3", presubmit=True
+        ),
         RubyDistribTest("linux", "x64", "centos7"),
         RubyDistribTest("linux", "x64", "ubuntu2004"),
         RubyDistribTest("linux", "x64", "ubuntu2204", presubmit=True),