Internal change

COPYBARA_INTEGRATE_REVIEW=https://github.com/grpc/grpc/pull/35043 from gtcooke94:deprecate_old_crl_apis 003057a93c
PiperOrigin-RevId: 585744149

include/grpc/grpc_security.h

@@ -894,7 +894,10 @@ GRPCAPI void grpc_tls_credentials_options_set_identity_cert_name(
 GRPCAPI void grpc_tls_credentials_options_set_cert_request_type(
     grpc_tls_credentials_options* options,
     grpc_ssl_client_certificate_request_type type);
-/**
+
+/** Deprecated in favor of grpc_tls_credentials_options_set_crl_provider. The
+ * crl provider interface provides a significantly more flexible approach to
+ * using CRLs. See gRFC A69 for details.
  * EXPERIMENTAL API - Subject to change
  *
  * If set, gRPC will read all hashed x.509 CRL files in the directory and

src/core/lib/event_engine/posix_engine/tcp_socket_utils.cc

@@ -75,6 +75,7 @@ int AdjustValue(int default_value, int min_value, int max_value,
   return *actual_value;
 }
 
+#ifdef GRPC_POSIX_SOCKET_UTILS_COMMON
 // The default values for TCP_USER_TIMEOUT are currently configured to be in
 // line with the default values of KEEPALIVE_TIMEOUT as proposed in
 // https://github.com/grpc/proposal/blob/master/A18-tcp-user-timeout.md */
@@ -83,8 +84,6 @@ int kDefaultServerUserTimeoutMs = 20000;
 bool kDefaultClientUserTimeoutEnabled = false;
 bool kDefaultServerUserTimeoutEnabled = true;
 
-#ifdef GRPC_POSIX_SOCKET_UTILS_COMMON
-
 absl::Status ErrorForFd(
     int fd, const experimental::EventEngine::ResolvedAddress& addr) {
   if (fd >= 0) return absl::OkStatus();

src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi

@@ -470,6 +470,7 @@ async def _finish_handler_with_stream_responses(RPCState rpc_state,
     """
     cdef object async_response_generator
     cdef object response_message
+    install_context_from_request_call_event_aio(rpc_state)
 
     if inspect.iscoroutinefunction(stream_handler):
         # Case 1: Coroutine async handler - using reader-writer API
@@ -523,6 +524,7 @@ async def _finish_handler_with_stream_responses(RPCState rpc_state,
     rpc_state.metadata_sent = True
     rpc_state.status_sent = True
     await execute_batch(rpc_state, finish_ops, loop)
+    uninstall_context()
 
 
 async def _handle_unary_unary_rpc(object method_handler,

test/cpp/interop/xds_stats_watcher.cc

@@ -51,6 +51,19 @@ std::unordered_set<std::string> ToLowerCase(
   return result;
 }
 
+bool HasNonEmptyMetadata(
+    const std::map<std::string, LoadBalancerStatsResponse::MetadataByPeer>&
+        metadata_by_peer) {
+  for (const auto& entry : metadata_by_peer) {
+    for (const auto& rpc_metadata : entry.second.rpc_metadata()) {
+      if (rpc_metadata.metadata_size() > 0) {
+        return true;
+      }
+    }
+  }
+  return false;
+}
+
 }  // namespace
 
 XdsStatsWatcher::XdsStatsWatcher(int start_id, int end_id,
@@ -113,8 +126,12 @@ LoadBalancerStatsResponse XdsStatsWatcher::WaitForRpcStatsResponse(
                [this] { return rpcs_needed_ == 0; });
   response.mutable_rpcs_by_peer()->insert(rpcs_by_peer_.begin(),
                                           rpcs_by_peer_.end());
+  // Return metadata if at least one RPC had relevant metadata. Note that empty
+  // entries would be returned for RCPs with no relevant metadata in this case.
+  if (HasNonEmptyMetadata(metadata_by_peer_)) {
     response.mutable_metadatas_by_peer()->insert(metadata_by_peer_.begin(),
                                                  metadata_by_peer_.end());
+  }
   auto& response_rpcs_by_method = *response.mutable_rpcs_by_method();
   for (const auto& rpc_by_type : rpcs_by_type_) {
     std::string method_name;

test/cpp/interop/xds_stats_watcher_test.cc

@@ -152,7 +152,7 @@ TEST(XdsStatsWatcherTest, WaitForRpcStatsResponseReturnsAll) {
             watcher.WaitForRpcStatsResponse(0).DebugString());
 }
 
-TEST(XdsStatsWatcherTest, WaitForRpcStatsResponseIgnoresMetadata) {
+TEST(XdsStatsWatcherTest, WaitForRpcStatsResponseExcludesMetadata) {
   XdsStatsWatcher watcher(0, 3, {});
   // RPC had metadata - but watcher should ignore it
   watcher.RpcCompleted(BuildCallResult(0), "peer1",
@@ -163,11 +163,6 @@ TEST(XdsStatsWatcherTest, WaitForRpcStatsResponseIgnoresMetadata) {
                        {{"k1", "v5"}, {"k2", "v6"}, {"k3", "v7"}});
   LoadBalancerStatsResponse expected;
   expected.mutable_rpcs_by_peer()->insert({{"peer1", 2}, {"peer2", 1}});
-  // There will still be an empty metadata collection for each RPC
-  expected.mutable_metadatas_by_peer()->insert({
-      {"peer1", BuildMetadatas({{}, {}})},
-      {"peer2", BuildMetadatas({{}})},
-  });
   (*expected.mutable_rpcs_by_method())["UnaryCall"]
       .mutable_rpcs_by_peer()
       ->insert({{"peer1", 2}, {"peer2", 1}});