Merge pull request #19584 from bigfacebear/add_IsValidTarget_api_to_ResolverRegistry

add IsValidTarget API to ResolverRegistry
diff --git a/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc b/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc
index 5ecdf4e..c5a27f1 100644
--- a/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc
+++ b/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc
@@ -89,9 +89,15 @@
     : Resolver(args.combiner, std::move(args.result_handler)) {
   GRPC_CLOSURE_INIT(&reresolution_closure_, ReturnReresolutionResult, this,
                     grpc_combiner_scheduler(combiner()));
-  channel_args_ = grpc_channel_args_copy(args.args);
   FakeResolverResponseGenerator* response_generator =
       FakeResolverResponseGenerator::GetFromArgs(args.args);
+  // Channels sharing the same subchannels may have different resolver response
+  // generators. If we don't remove this arg, subchannel pool will create new
+  // subchannels for the same address instead of reusing existing ones because
+  // of different values of this channel arg.
+  const char* args_to_remove[] = {GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR};
+  channel_args_ = grpc_channel_args_copy_and_remove(
+      args.args, args_to_remove, GPR_ARRAY_SIZE(args_to_remove));
   if (response_generator != nullptr) {
     response_generator->resolver_ = this;
     if (response_generator->has_result_) {
diff --git a/src/core/lib/slice/slice_buffer.cc b/src/core/lib/slice/slice_buffer.cc
index 2b69f57..7eedad3 100644
--- a/src/core/lib/slice/slice_buffer.cc
+++ b/src/core/lib/slice/slice_buffer.cc
@@ -262,9 +262,9 @@
   src->length = 0;
 }
 
+template <bool incref>
 static void slice_buffer_move_first_maybe_ref(grpc_slice_buffer* src, size_t n,
-                                              grpc_slice_buffer* dst,
-                                              bool incref) {
+                                              grpc_slice_buffer* dst) {
   GPR_ASSERT(src->length >= n);
   if (src->length == n) {
     grpc_slice_buffer_move_into(src, dst);
@@ -304,12 +304,12 @@
 
 void grpc_slice_buffer_move_first(grpc_slice_buffer* src, size_t n,
                                   grpc_slice_buffer* dst) {
-  slice_buffer_move_first_maybe_ref(src, n, dst, true);
+  slice_buffer_move_first_maybe_ref<true>(src, n, dst);
 }
 
 void grpc_slice_buffer_move_first_no_ref(grpc_slice_buffer* src, size_t n,
                                          grpc_slice_buffer* dst) {
-  slice_buffer_move_first_maybe_ref(src, n, dst, false);
+  slice_buffer_move_first_maybe_ref<false>(src, n, dst);
 }
 
 void grpc_slice_buffer_move_first_into_buffer(grpc_slice_buffer* src, size_t n,
diff --git a/src/python/grpcio/grpc/_channel.py b/src/python/grpcio/grpc/_channel.py
index 0bf8e03..24f928e 100644
--- a/src/python/grpcio/grpc/_channel.py
+++ b/src/python/grpcio/grpc/_channel.py
@@ -13,7 +13,6 @@
 # limitations under the License.
 """Invocation-side implementation of gRPC Python."""
 
-import functools
 import logging
 import sys
 import threading
@@ -82,6 +81,17 @@
         unknown_cygrpc_code, details)
 
 
+def _wait_once_until(condition, until):
+    if until is None:
+        condition.wait()
+    else:
+        remaining = until - time.time()
+        if remaining < 0:
+            raise grpc.FutureTimeoutError()
+        else:
+            condition.wait(timeout=remaining)
+
+
 class _RPCState(object):
 
     def __init__(self, due, initial_metadata, trailing_metadata, code, details):
@@ -168,11 +178,12 @@
 #pylint: disable=too-many-statements
 def _consume_request_iterator(request_iterator, state, call, request_serializer,
                               event_handler):
-    """Consume a request iterator supplied by the user."""
+    if cygrpc.is_fork_support_enabled():
+        condition_wait_timeout = 1.0
+    else:
+        condition_wait_timeout = None
 
     def consume_request_iterator():  # pylint: disable=too-many-branches
-        # Iterate over the request iterator until it is exhausted or an error
-        # condition is encountered.
         while True:
             return_from_user_request_generator_invoked = False
             try:
@@ -213,19 +224,14 @@
                             state.due.add(cygrpc.OperationType.send_message)
                         else:
                             return
-
-                        def _done():
-                            return (state.code is not None or
-                                    cygrpc.OperationType.send_message not in
-                                    state.due)
-
-                        _common.wait(
-                            state.condition.wait,
-                            _done,
-                            spin_cb=functools.partial(
-                                cygrpc.block_if_fork_in_progress, state))
-                        if state.code is not None:
-                            return
+                        while True:
+                            state.condition.wait(condition_wait_timeout)
+                            cygrpc.block_if_fork_in_progress(state)
+                            if state.code is None:
+                                if cygrpc.OperationType.send_message not in state.due:
+                                    break
+                            else:
+                                return
                 else:
                     return
         with state.condition:
@@ -275,21 +281,13 @@
         with self._state.condition:
             return self._state.code is not None
 
-    def _is_complete(self):
-        return self._state.code is not None
-
     def result(self, timeout=None):
-        """Returns the result of the computation or raises its exception.
-
-        See grpc.Future.result for the full API contract.
-        """
+        until = None if timeout is None else time.time() + timeout
         with self._state.condition:
-            timed_out = _common.wait(
-                self._state.condition.wait, self._is_complete, timeout=timeout)
-            if timed_out:
-                raise grpc.FutureTimeoutError()
-            else:
-                if self._state.code is grpc.StatusCode.OK:
+            while True:
+                if self._state.code is None:
+                    _wait_once_until(self._state.condition, until)
+                elif self._state.code is grpc.StatusCode.OK:
                     return self._state.response
                 elif self._state.cancelled:
                     raise grpc.FutureCancelledError()
@@ -297,17 +295,12 @@
                     raise self
 
     def exception(self, timeout=None):
-        """Return the exception raised by the computation.
-
-        See grpc.Future.exception for the full API contract.
-        """
+        until = None if timeout is None else time.time() + timeout
         with self._state.condition:
-            timed_out = _common.wait(
-                self._state.condition.wait, self._is_complete, timeout=timeout)
-            if timed_out:
-                raise grpc.FutureTimeoutError()
-            else:
-                if self._state.code is grpc.StatusCode.OK:
+            while True:
+                if self._state.code is None:
+                    _wait_once_until(self._state.condition, until)
+                elif self._state.code is grpc.StatusCode.OK:
                     return None
                 elif self._state.cancelled:
                     raise grpc.FutureCancelledError()
@@ -315,17 +308,12 @@
                     return self
 
     def traceback(self, timeout=None):
-        """Access the traceback of the exception raised by the computation.
-
-        See grpc.future.traceback for the full API contract.
-        """
+        until = None if timeout is None else time.time() + timeout
         with self._state.condition:
-            timed_out = _common.wait(
-                self._state.condition.wait, self._is_complete, timeout=timeout)
-            if timed_out:
-                raise grpc.FutureTimeoutError()
-            else:
-                if self._state.code is grpc.StatusCode.OK:
+            while True:
+                if self._state.code is None:
+                    _wait_once_until(self._state.condition, until)
+                elif self._state.code is grpc.StatusCode.OK:
                     return None
                 elif self._state.cancelled:
                     raise grpc.FutureCancelledError()
@@ -357,23 +345,17 @@
                 raise StopIteration()
             else:
                 raise self
-
-            def _response_ready():
-                return (
-                    self._state.response is not None or
-                    (cygrpc.OperationType.receive_message not in self._state.due
-                     and self._state.code is not None))
-
-            _common.wait(self._state.condition.wait, _response_ready)
-            if self._state.response is not None:
-                response = self._state.response
-                self._state.response = None
-                return response
-            elif cygrpc.OperationType.receive_message not in self._state.due:
-                if self._state.code is grpc.StatusCode.OK:
-                    raise StopIteration()
-                elif self._state.code is not None:
-                    raise self
+            while True:
+                self._state.condition.wait()
+                if self._state.response is not None:
+                    response = self._state.response
+                    self._state.response = None
+                    return response
+                elif cygrpc.OperationType.receive_message not in self._state.due:
+                    if self._state.code is grpc.StatusCode.OK:
+                        raise StopIteration()
+                    elif self._state.code is not None:
+                        raise self
 
     def __iter__(self):
         return self
@@ -404,47 +386,32 @@
 
     def initial_metadata(self):
         with self._state.condition:
-
-            def _done():
-                return self._state.initial_metadata is not None
-
-            _common.wait(self._state.condition.wait, _done)
+            while self._state.initial_metadata is None:
+                self._state.condition.wait()
             return self._state.initial_metadata
 
     def trailing_metadata(self):
         with self._state.condition:
-
-            def _done():
-                return self._state.trailing_metadata is not None
-
-            _common.wait(self._state.condition.wait, _done)
+            while self._state.trailing_metadata is None:
+                self._state.condition.wait()
             return self._state.trailing_metadata
 
     def code(self):
         with self._state.condition:
-
-            def _done():
-                return self._state.code is not None
-
-            _common.wait(self._state.condition.wait, _done)
+            while self._state.code is None:
+                self._state.condition.wait()
             return self._state.code
 
     def details(self):
         with self._state.condition:
-
-            def _done():
-                return self._state.details is not None
-
-            _common.wait(self._state.condition.wait, _done)
+            while self._state.details is None:
+                self._state.condition.wait()
             return _common.decode(self._state.details)
 
     def debug_error_string(self):
         with self._state.condition:
-
-            def _done():
-                return self._state.debug_error_string is not None
-
-            _common.wait(self._state.condition.wait, _done)
+            while self._state.debug_error_string is None:
+                self._state.condition.wait()
             return _common.decode(self._state.debug_error_string)
 
     def _repr(self):
diff --git a/src/python/grpcio/grpc/_common.py b/src/python/grpcio/grpc/_common.py
index b4b2473..f69127e 100644
--- a/src/python/grpcio/grpc/_common.py
+++ b/src/python/grpcio/grpc/_common.py
@@ -15,7 +15,6 @@
 
 import logging
 
-import time
 import six
 
 import grpc
@@ -61,8 +60,6 @@
         CYGRPC_STATUS_CODE_TO_STATUS_CODE)
 }
 
-MAXIMUM_WAIT_TIMEOUT = 0.1
-
 
 def encode(s):
     if isinstance(s, bytes):
@@ -99,50 +96,3 @@
 
 def fully_qualified_method(group, method):
     return '/{}/{}'.format(group, method)
-
-
-def _wait_once(wait_fn, timeout, spin_cb):
-    wait_fn(timeout=timeout)
-    if spin_cb is not None:
-        spin_cb()
-
-
-def wait(wait_fn, wait_complete_fn, timeout=None, spin_cb=None):
-    """Blocks waiting for an event without blocking the thread indefinitely.
-
-    See https://github.com/grpc/grpc/issues/19464 for full context. CPython's
-    `threading.Event.wait` and `threading.Condition.wait` methods, if invoked
-    without a timeout kwarg, may block the calling thread indefinitely. If the
-    call is made from the main thread, this means that signal handlers may not
-    run for an arbitrarily long period of time.
-
-    This wrapper calls the supplied wait function with an arbitrary short
-    timeout to ensure that no signal handler has to wait longer than
-    MAXIMUM_WAIT_TIMEOUT before executing.
-
-    Args:
-      wait_fn: A callable acceptable a single float-valued kwarg named
-        `timeout`. This function is expected to be one of `threading.Event.wait`
-        or `threading.Condition.wait`.
-      wait_complete_fn: A callable taking no arguments and returning a bool.
-        When this function returns true, it indicates that waiting should cease.
-      timeout: An optional float-valued number of seconds after which the wait
-        should cease.
-      spin_cb: An optional Callable taking no arguments and returning nothing.
-        This callback will be called on each iteration of the spin. This may be
-        used for, e.g. work related to forking.
-
-    Returns:
-      True if a timeout was supplied and it was reached. False otherwise.
-    """
-    if timeout is None:
-        while not wait_complete_fn():
-            _wait_once(wait_fn, MAXIMUM_WAIT_TIMEOUT, spin_cb)
-    else:
-        end = time.time() + timeout
-        while not wait_complete_fn():
-            remaining = min(end - time.time(), MAXIMUM_WAIT_TIMEOUT)
-            if remaining < 0:
-                return True
-            _wait_once(wait_fn, remaining, spin_cb)
-    return False
diff --git a/src/python/grpcio_tests/commands.py b/src/python/grpcio_tests/commands.py
index 166cea1..dc0795d 100644
--- a/src/python/grpcio_tests/commands.py
+++ b/src/python/grpcio_tests/commands.py
@@ -145,8 +145,6 @@
         'unit._exit_test.ExitTest.test_in_flight_partial_unary_stream_call',
         'unit._exit_test.ExitTest.test_in_flight_partial_stream_unary_call',
         'unit._exit_test.ExitTest.test_in_flight_partial_stream_stream_call',
-        # TODO(https://github.com/grpc/grpc/issues/18980): Reenable.
-        'unit._signal_handling_test.SignalHandlingTest',
         'unit._metadata_flags_test',
         'health_check._health_servicer_test.HealthServicerTest.test_cancelled_watch_removed_from_watch_list',
         # TODO(https://github.com/grpc/grpc/issues/17330) enable these three tests
diff --git a/src/python/grpcio_tests/tests/tests.json b/src/python/grpcio_tests/tests/tests.json
index 16ba484..cc08d56 100644
--- a/src/python/grpcio_tests/tests/tests.json
+++ b/src/python/grpcio_tests/tests/tests.json
@@ -67,7 +67,6 @@
   "unit._server_ssl_cert_config_test.ServerSSLCertReloadTestWithoutClientAuth",
   "unit._server_test.ServerTest",
   "unit._session_cache_test.SSLSessionCacheTest",
-  "unit._signal_handling_test.SignalHandlingTest",
   "unit._version_test.VersionTest",
   "unit.beta._beta_features_test.BetaFeaturesTest",
   "unit.beta._beta_features_test.ContextManagementAndLifecycleTest",
diff --git a/src/python/grpcio_tests/tests/unit/BUILD.bazel b/src/python/grpcio_tests/tests/unit/BUILD.bazel
index d21f5a5..a161794 100644
--- a/src/python/grpcio_tests/tests/unit/BUILD.bazel
+++ b/src/python/grpcio_tests/tests/unit/BUILD.bazel
@@ -16,7 +16,6 @@
     "_credentials_test.py",
     "_dns_resolver_test.py",
     "_empty_message_test.py",
-    "_error_message_encoding_test.py",
     "_exit_test.py",
     "_interceptor_test.py",
     "_invalid_metadata_test.py",
@@ -28,7 +27,6 @@
     # "_reconnect_test.py",
     "_resource_exhausted_test.py",
     "_rpc_test.py",
-    "_signal_handling_test.py",
     # TODO(ghostwriternr): To be added later.
     # "_server_ssl_cert_config_test.py",
     "_server_test.py",
@@ -42,11 +40,6 @@
 )
 
 py_library(
-    name = "_signal_client",
-    srcs = ["_signal_client.py"],
-)
-
-py_library(
     name = "resources",
     srcs = ["resources.py"],
     data=[
@@ -94,7 +87,6 @@
             ":_server_shutdown_scenarios",
             ":_from_grpc_import_star",
             ":_tcp_proxy",
-            ":_signal_client",
             "//src/python/grpcio_tests/tests/unit/framework/common",
             "//src/python/grpcio_tests/tests/testing",
             requirement('six'),
diff --git a/src/python/grpcio_tests/tests/unit/_signal_client.py b/src/python/grpcio_tests/tests/unit/_signal_client.py
deleted file mode 100644
index 65ddd6d..0000000
--- a/src/python/grpcio_tests/tests/unit/_signal_client.py
+++ /dev/null
@@ -1,84 +0,0 @@
-# Copyright 2019 the 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.
-"""Client for testing responsiveness to signals."""
-
-from __future__ import print_function
-
-import argparse
-import functools
-import logging
-import signal
-import sys
-
-import grpc
-
-SIGTERM_MESSAGE = "Handling sigterm!"
-
-UNARY_UNARY = "/test/Unary"
-UNARY_STREAM = "/test/ServerStreaming"
-
-_MESSAGE = b'\x00\x00\x00'
-
-_ASSERTION_MESSAGE = "Control flow should never reach here."
-
-# NOTE(gnossen): We use a global variable here so that the signal handler can be
-# installed before the RPC begins. If we do not do this, then we may receive the
-# SIGINT before the signal handler is installed. I'm not happy with per-process
-# global state, but the per-process global state that is signal handlers
-# somewhat forces my hand.
-per_process_rpc_future = None
-
-
-def handle_sigint(unused_signum, unused_frame):
-    print(SIGTERM_MESSAGE)
-    if per_process_rpc_future is not None:
-        per_process_rpc_future.cancel()
-    sys.stderr.flush()
-    sys.exit(0)
-
-
-def main_unary(server_target):
-    """Initiate a unary RPC to be interrupted by a SIGINT."""
-    global per_process_rpc_future  # pylint: disable=global-statement
-    with grpc.insecure_channel(server_target) as channel:
-        multicallable = channel.unary_unary(UNARY_UNARY)
-        signal.signal(signal.SIGINT, handle_sigint)
-        per_process_rpc_future = multicallable.future(
-            _MESSAGE, wait_for_ready=True)
-        result = per_process_rpc_future.result()
-        assert False, _ASSERTION_MESSAGE
-
-
-def main_streaming(server_target):
-    """Initiate a streaming RPC to be interrupted by a SIGINT."""
-    global per_process_rpc_future  # pylint: disable=global-statement
-    with grpc.insecure_channel(server_target) as channel:
-        signal.signal(signal.SIGINT, handle_sigint)
-        per_process_rpc_future = channel.unary_stream(UNARY_STREAM)(
-            _MESSAGE, wait_for_ready=True)
-        for result in per_process_rpc_future:
-            pass
-        assert False, _ASSERTION_MESSAGE
-
-
-if __name__ == '__main__':
-    parser = argparse.ArgumentParser(description='Signal test client.')
-    parser.add_argument('server', help='Server target')
-    parser.add_argument(
-        'arity', help='RPC arity', choices=('unary', 'streaming'))
-    args = parser.parse_args()
-    if args.arity == 'unary':
-        main_unary(args.server)
-    else:
-        main_streaming(args.server)
diff --git a/src/python/grpcio_tests/tests/unit/_signal_handling_test.py b/src/python/grpcio_tests/tests/unit/_signal_handling_test.py
deleted file mode 100644
index 9d0d647..0000000
--- a/src/python/grpcio_tests/tests/unit/_signal_handling_test.py
+++ /dev/null
@@ -1,158 +0,0 @@
-# Copyright 2019 the 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.
-"""Test of responsiveness to signals."""
-
-from __future__ import print_function
-
-import logging
-import os
-import signal
-import subprocess
-import tempfile
-import threading
-import unittest
-import sys
-
-import grpc
-
-from tests.unit import test_common
-from tests.unit import _signal_client
-
-_CLIENT_PATH = os.path.abspath(os.path.realpath(_signal_client.__file__))
-_HOST = 'localhost'
-
-
-class _GenericHandler(grpc.GenericRpcHandler):
-
-    def __init__(self):
-        self._connected_clients_lock = threading.RLock()
-        self._connected_clients_event = threading.Event()
-        self._connected_clients = 0
-
-        self._unary_unary_handler = grpc.unary_unary_rpc_method_handler(
-            self._handle_unary_unary)
-        self._unary_stream_handler = grpc.unary_stream_rpc_method_handler(
-            self._handle_unary_stream)
-
-    def _on_client_connect(self):
-        with self._connected_clients_lock:
-            self._connected_clients += 1
-            self._connected_clients_event.set()
-
-    def _on_client_disconnect(self):
-        with self._connected_clients_lock:
-            self._connected_clients -= 1
-            if self._connected_clients == 0:
-                self._connected_clients_event.clear()
-
-    def await_connected_client(self):
-        """Blocks until a client connects to the server."""
-        self._connected_clients_event.wait()
-
-    def _handle_unary_unary(self, request, servicer_context):
-        """Handles a unary RPC.
-
-        Blocks until the client disconnects and then echoes.
-        """
-        stop_event = threading.Event()
-
-        def on_rpc_end():
-            self._on_client_disconnect()
-            stop_event.set()
-
-        servicer_context.add_callback(on_rpc_end)
-        self._on_client_connect()
-        stop_event.wait()
-        return request
-
-    def _handle_unary_stream(self, request, servicer_context):
-        """Handles a server streaming RPC.
-
-        Blocks until the client disconnects and then echoes.
-        """
-        stop_event = threading.Event()
-
-        def on_rpc_end():
-            self._on_client_disconnect()
-            stop_event.set()
-
-        servicer_context.add_callback(on_rpc_end)
-        self._on_client_connect()
-        stop_event.wait()
-        yield request
-
-    def service(self, handler_call_details):
-        if handler_call_details.method == _signal_client.UNARY_UNARY:
-            return self._unary_unary_handler
-        elif handler_call_details.method == _signal_client.UNARY_STREAM:
-            return self._unary_stream_handler
-        else:
-            return None
-
-
-def _read_stream(stream):
-    stream.seek(0)
-    return stream.read()
-
-
-class SignalHandlingTest(unittest.TestCase):
-
-    def setUp(self):
-        self._server = test_common.test_server()
-        self._port = self._server.add_insecure_port('{}:0'.format(_HOST))
-        self._handler = _GenericHandler()
-        self._server.add_generic_rpc_handlers((self._handler,))
-        self._server.start()
-
-    def tearDown(self):
-        self._server.stop(None)
-
-    @unittest.skipIf(os.name == 'nt', 'SIGINT not supported on windows')
-    def testUnary(self):
-        """Tests that the server unary code path does not stall signal handlers."""
-        server_target = '{}:{}'.format(_HOST, self._port)
-        with tempfile.TemporaryFile(mode='r') as client_stdout:
-            with tempfile.TemporaryFile(mode='r') as client_stderr:
-                client = subprocess.Popen(
-                    (sys.executable, _CLIENT_PATH, server_target, 'unary'),
-                    stdout=client_stdout,
-                    stderr=client_stderr)
-                self._handler.await_connected_client()
-                client.send_signal(signal.SIGINT)
-                self.assertFalse(client.wait(), msg=_read_stream(client_stderr))
-                client_stdout.seek(0)
-                self.assertIn(_signal_client.SIGTERM_MESSAGE,
-                              client_stdout.read())
-
-    @unittest.skipIf(os.name == 'nt', 'SIGINT not supported on windows')
-    def testStreaming(self):
-        """Tests that the server streaming code path does not stall signal handlers."""
-        server_target = '{}:{}'.format(_HOST, self._port)
-        with tempfile.TemporaryFile(mode='r') as client_stdout:
-            with tempfile.TemporaryFile(mode='r') as client_stderr:
-                client = subprocess.Popen(
-                    (sys.executable, _CLIENT_PATH, server_target, 'streaming'),
-                    stdout=client_stdout,
-                    stderr=client_stderr)
-                self._handler.await_connected_client()
-                client.send_signal(signal.SIGINT)
-                self.assertFalse(client.wait(), msg=_read_stream(client_stderr))
-                client_stdout.seek(0)
-                self.assertIn(_signal_client.SIGTERM_MESSAGE,
-                              client_stdout.read())
-
-
-if __name__ == '__main__':
-    logging.basicConfig()
-    unittest.main(verbosity=2)
diff --git a/test/cpp/end2end/client_lb_end2end_test.cc b/test/cpp/end2end/client_lb_end2end_test.cc
index d499f13..8fc0eac 100644
--- a/test/cpp/end2end/client_lb_end2end_test.cc
+++ b/test/cpp/end2end/client_lb_end2end_test.cc
@@ -133,6 +133,59 @@
   std::set<grpc::string> clients_;
 };
 
+class FakeResolverResponseGeneratorWrapper {
+ public:
+  FakeResolverResponseGeneratorWrapper()
+      : response_generator_(grpc_core::MakeRefCounted<
+                            grpc_core::FakeResolverResponseGenerator>()) {}
+
+  FakeResolverResponseGeneratorWrapper(
+      FakeResolverResponseGeneratorWrapper&& other) {
+    response_generator_ = std::move(other.response_generator_);
+  }
+
+  void SetNextResolution(const std::vector<int>& ports) {
+    grpc_core::ExecCtx exec_ctx;
+    response_generator_->SetResponse(BuildFakeResults(ports));
+  }
+
+  void SetNextResolutionUponError(const std::vector<int>& ports) {
+    grpc_core::ExecCtx exec_ctx;
+    response_generator_->SetReresolutionResponse(BuildFakeResults(ports));
+  }
+
+  void SetFailureOnReresolution() {
+    grpc_core::ExecCtx exec_ctx;
+    response_generator_->SetFailureOnReresolution();
+  }
+
+  grpc_core::FakeResolverResponseGenerator* Get() const {
+    return response_generator_.get();
+  }
+
+ private:
+  static grpc_core::Resolver::Result BuildFakeResults(
+      const std::vector<int>& ports) {
+    grpc_core::Resolver::Result result;
+    for (const int& port : ports) {
+      char* lb_uri_str;
+      gpr_asprintf(&lb_uri_str, "ipv4:127.0.0.1:%d", port);
+      grpc_uri* lb_uri = grpc_uri_parse(lb_uri_str, true);
+      GPR_ASSERT(lb_uri != nullptr);
+      grpc_resolved_address address;
+      GPR_ASSERT(grpc_parse_uri(lb_uri, &address));
+      result.addresses.emplace_back(address.addr, address.len,
+                                    nullptr /* args */);
+      grpc_uri_destroy(lb_uri);
+      gpr_free(lb_uri_str);
+    }
+    return result;
+  }
+
+  grpc_core::RefCountedPtr<grpc_core::FakeResolverResponseGenerator>
+      response_generator_;
+};
+
 class ClientLbEnd2endTest : public ::testing::Test {
  protected:
   ClientLbEnd2endTest()
@@ -147,11 +200,7 @@
     GPR_GLOBAL_CONFIG_SET(grpc_client_channel_backup_poll_interval_ms, 1);
   }
 
-  void SetUp() override {
-    grpc_init();
-    response_generator_ =
-        grpc_core::MakeRefCounted<grpc_core::FakeResolverResponseGenerator>();
-  }
+  void SetUp() override { grpc_init(); }
 
   void TearDown() override {
     for (size_t i = 0; i < servers_.size(); ++i) {
@@ -186,38 +235,6 @@
     }
   }
 
-  grpc_core::Resolver::Result BuildFakeResults(const std::vector<int>& ports) {
-    grpc_core::Resolver::Result result;
-    for (const int& port : ports) {
-      char* lb_uri_str;
-      gpr_asprintf(&lb_uri_str, "ipv4:127.0.0.1:%d", port);
-      grpc_uri* lb_uri = grpc_uri_parse(lb_uri_str, true);
-      GPR_ASSERT(lb_uri != nullptr);
-      grpc_resolved_address address;
-      GPR_ASSERT(grpc_parse_uri(lb_uri, &address));
-      result.addresses.emplace_back(address.addr, address.len,
-                                    nullptr /* args */);
-      grpc_uri_destroy(lb_uri);
-      gpr_free(lb_uri_str);
-    }
-    return result;
-  }
-
-  void SetNextResolution(const std::vector<int>& ports) {
-    grpc_core::ExecCtx exec_ctx;
-    response_generator_->SetResponse(BuildFakeResults(ports));
-  }
-
-  void SetNextResolutionUponError(const std::vector<int>& ports) {
-    grpc_core::ExecCtx exec_ctx;
-    response_generator_->SetReresolutionResponse(BuildFakeResults(ports));
-  }
-
-  void SetFailureOnReresolution() {
-    grpc_core::ExecCtx exec_ctx;
-    response_generator_->SetFailureOnReresolution();
-  }
-
   std::vector<int> GetServersPorts(size_t start_index = 0) {
     std::vector<int> ports;
     for (size_t i = start_index; i < servers_.size(); ++i) {
@@ -226,6 +243,10 @@
     return ports;
   }
 
+  FakeResolverResponseGeneratorWrapper BuildResolverResponseGenerator() {
+    return FakeResolverResponseGeneratorWrapper();
+  }
+
   std::unique_ptr<grpc::testing::EchoTestService::Stub> BuildStub(
       const std::shared_ptr<Channel>& channel) {
     return grpc::testing::EchoTestService::NewStub(channel);
@@ -233,12 +254,13 @@
 
   std::shared_ptr<Channel> BuildChannel(
       const grpc::string& lb_policy_name,
+      const FakeResolverResponseGeneratorWrapper& response_generator,
       ChannelArguments args = ChannelArguments()) {
     if (lb_policy_name.size() > 0) {
       args.SetLoadBalancingPolicyName(lb_policy_name);
     }  // else, default to pick first
     args.SetPointer(GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR,
-                    response_generator_.get());
+                    response_generator.Get());
     return ::grpc::CreateCustomChannel("fake:///", creds_, args);
   }
 
@@ -401,8 +423,6 @@
   const grpc::string server_host_;
   std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_;
   std::vector<std::unique_ptr<ServerData>> servers_;
-  grpc_core::RefCountedPtr<grpc_core::FakeResolverResponseGenerator>
-      response_generator_;
   const grpc::string kRequestMessage_;
   std::shared_ptr<ChannelCredentials> creds_;
 };
@@ -410,7 +430,8 @@
 TEST_F(ClientLbEnd2endTest, ChannelStateConnectingWhenResolving) {
   const int kNumServers = 3;
   StartServers(kNumServers);
-  auto channel = BuildChannel("");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("", response_generator);
   auto stub = BuildStub(channel);
   // Initial state should be IDLE.
   EXPECT_EQ(channel->GetState(false /* try_to_connect */), GRPC_CHANNEL_IDLE);
@@ -423,7 +444,7 @@
   EXPECT_EQ(channel->GetState(false /* try_to_connect */),
             GRPC_CHANNEL_CONNECTING);
   // Return a resolver result, which allows the connection attempt to proceed.
-  SetNextResolution(GetServersPorts());
+  response_generator.SetNextResolution(GetServersPorts());
   // We should eventually transition into state READY.
   EXPECT_TRUE(WaitForChannelReady(channel.get()));
 }
@@ -432,9 +453,11 @@
   // Start servers and send one RPC per server.
   const int kNumServers = 3;
   StartServers(kNumServers);
-  auto channel = BuildChannel("");  // test that pick first is the default.
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel(
+      "", response_generator);  // test that pick first is the default.
   auto stub = BuildStub(channel);
-  SetNextResolution(GetServersPorts());
+  response_generator.SetNextResolution(GetServersPorts());
   for (size_t i = 0; i < servers_.size(); ++i) {
     CheckRpcSendOk(stub, DEBUG_LOCATION);
   }
@@ -454,19 +477,22 @@
 }
 
 TEST_F(ClientLbEnd2endTest, PickFirstProcessPending) {
-  StartServers(1);                  // Single server
-  auto channel = BuildChannel("");  // test that pick first is the default.
+  StartServers(1);  // Single server
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel(
+      "", response_generator);  // test that pick first is the default.
   auto stub = BuildStub(channel);
-  SetNextResolution({servers_[0]->port_});
+  response_generator.SetNextResolution({servers_[0]->port_});
   WaitForServer(stub, 0, DEBUG_LOCATION);
   // Create a new channel and its corresponding PF LB policy, which will pick
   // the subchannels in READY state from the previous RPC against the same
   // target (even if it happened over a different channel, because subchannels
   // are globally reused). Progress should happen without any transition from
   // this READY state.
-  auto second_channel = BuildChannel("");
+  auto second_response_generator = BuildResolverResponseGenerator();
+  auto second_channel = BuildChannel("", second_response_generator);
   auto second_stub = BuildStub(second_channel);
-  SetNextResolution({servers_[0]->port_});
+  second_response_generator.SetNextResolution({servers_[0]->port_});
   CheckRpcSendOk(second_stub, DEBUG_LOCATION);
 }
 
@@ -479,16 +505,18 @@
                             grpc_pick_unused_port_or_die()};
   CreateServers(2, ports);
   StartServer(1);
-  auto channel1 = BuildChannel("pick_first", args);
+  auto response_generator1 = BuildResolverResponseGenerator();
+  auto channel1 = BuildChannel("pick_first", response_generator1, args);
   auto stub1 = BuildStub(channel1);
-  SetNextResolution(ports);
+  response_generator1.SetNextResolution(ports);
   // Wait for second server to be ready.
   WaitForServer(stub1, 1, DEBUG_LOCATION);
   // Create a second channel with the same addresses.  Its PF instance
   // should immediately pick the second subchannel, since it's already
   // in READY state.
-  auto channel2 = BuildChannel("pick_first", args);
-  SetNextResolution(ports);
+  auto response_generator2 = BuildResolverResponseGenerator();
+  auto channel2 = BuildChannel("pick_first", response_generator2, args);
+  response_generator2.SetNextResolution(ports);
   // Check that the channel reports READY without waiting for the
   // initial backoff.
   EXPECT_TRUE(WaitForChannelReady(channel2.get(), 1 /* timeout_seconds */));
@@ -500,9 +528,10 @@
   args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs);
   const std::vector<int> ports = {grpc_pick_unused_port_or_die()};
   const gpr_timespec t0 = gpr_now(GPR_CLOCK_MONOTONIC);
-  auto channel = BuildChannel("pick_first", args);
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("pick_first", response_generator, args);
   auto stub = BuildStub(channel);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   // The channel won't become connected (there's no server).
   ASSERT_FALSE(channel->WaitForConnected(
       grpc_timeout_milliseconds_to_deadline(kInitialBackOffMs * 2)));
@@ -529,9 +558,10 @@
   constexpr int kMinReconnectBackOffMs = 1000;
   args.SetInt(GRPC_ARG_MIN_RECONNECT_BACKOFF_MS, kMinReconnectBackOffMs);
   const std::vector<int> ports = {grpc_pick_unused_port_or_die()};
-  auto channel = BuildChannel("pick_first", args);
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("pick_first", response_generator, args);
   auto stub = BuildStub(channel);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   // Make connection delay a 10% longer than it's willing to in order to make
   // sure we are hitting the codepath that waits for the min reconnect backoff.
   gpr_atm_rel_store(&g_connection_delay_ms, kMinReconnectBackOffMs * 1.10);
@@ -554,9 +584,10 @@
   constexpr int kInitialBackOffMs = 1000;
   args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs);
   const std::vector<int> ports = {grpc_pick_unused_port_or_die()};
-  auto channel = BuildChannel("pick_first", args);
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("pick_first", response_generator, args);
   auto stub = BuildStub(channel);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   // The channel won't become connected (there's no server).
   EXPECT_FALSE(
       channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(10)));
@@ -585,9 +616,10 @@
   constexpr int kInitialBackOffMs = 1000;
   args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs);
   const std::vector<int> ports = {grpc_pick_unused_port_or_die()};
-  auto channel = BuildChannel("pick_first", args);
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("pick_first", response_generator, args);
   auto stub = BuildStub(channel);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   // Wait for connect, which should fail ~immediately, because the server
   // is not up.
   gpr_log(GPR_INFO, "=== INITIAL CONNECTION ATTEMPT");
@@ -628,21 +660,22 @@
   // Start servers and send one RPC per server.
   const int kNumServers = 3;
   StartServers(kNumServers);
-  auto channel = BuildChannel("pick_first");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("pick_first", response_generator);
   auto stub = BuildStub(channel);
 
   std::vector<int> ports;
 
   // Perform one RPC against the first server.
   ports.emplace_back(servers_[0]->port_);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   gpr_log(GPR_INFO, "****** SET [0] *******");
   CheckRpcSendOk(stub, DEBUG_LOCATION);
   EXPECT_EQ(servers_[0]->service_.request_count(), 1);
 
   // An empty update will result in the channel going into TRANSIENT_FAILURE.
   ports.clear();
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   gpr_log(GPR_INFO, "****** SET none *******");
   grpc_connectivity_state channel_state;
   do {
@@ -654,7 +687,7 @@
   // Next update introduces servers_[1], making the channel recover.
   ports.clear();
   ports.emplace_back(servers_[1]->port_);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   gpr_log(GPR_INFO, "****** SET [1] *******");
   WaitForServer(stub, 1, DEBUG_LOCATION);
   EXPECT_EQ(servers_[0]->service_.request_count(), 0);
@@ -662,7 +695,7 @@
   // And again for servers_[2]
   ports.clear();
   ports.emplace_back(servers_[2]->port_);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   gpr_log(GPR_INFO, "****** SET [2] *******");
   WaitForServer(stub, 2, DEBUG_LOCATION);
   EXPECT_EQ(servers_[0]->service_.request_count(), 0);
@@ -676,14 +709,15 @@
   // Start servers and send one RPC per server.
   const int kNumServers = 3;
   StartServers(kNumServers);
-  auto channel = BuildChannel("pick_first");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("pick_first", response_generator);
   auto stub = BuildStub(channel);
 
   std::vector<int> ports;
 
   // Perform one RPC against the first server.
   ports.emplace_back(servers_[0]->port_);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   gpr_log(GPR_INFO, "****** SET [0] *******");
   CheckRpcSendOk(stub, DEBUG_LOCATION);
   EXPECT_EQ(servers_[0]->service_.request_count(), 1);
@@ -693,7 +727,7 @@
   ports.clear();
   ports.emplace_back(servers_[1]->port_);
   ports.emplace_back(servers_[0]->port_);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   gpr_log(GPR_INFO, "****** SET superset *******");
   CheckRpcSendOk(stub, DEBUG_LOCATION);
   // We stick to the previously connected server.
@@ -710,12 +744,14 @@
   StartServers(kNumServers);
   std::vector<int> ports = GetServersPorts();
   // Create two channels that (by default) use the global subchannel pool.
-  auto channel1 = BuildChannel("pick_first");
+  auto response_generator1 = BuildResolverResponseGenerator();
+  auto channel1 = BuildChannel("pick_first", response_generator1);
   auto stub1 = BuildStub(channel1);
-  SetNextResolution(ports);
-  auto channel2 = BuildChannel("pick_first");
+  response_generator1.SetNextResolution(ports);
+  auto response_generator2 = BuildResolverResponseGenerator();
+  auto channel2 = BuildChannel("pick_first", response_generator2);
   auto stub2 = BuildStub(channel2);
-  SetNextResolution(ports);
+  response_generator2.SetNextResolution(ports);
   WaitForServer(stub1, 0, DEBUG_LOCATION);
   // Send one RPC on each channel.
   CheckRpcSendOk(stub1, DEBUG_LOCATION);
@@ -735,12 +771,14 @@
   // Create two channels that use local subchannel pool.
   ChannelArguments args;
   args.SetInt(GRPC_ARG_USE_LOCAL_SUBCHANNEL_POOL, 1);
-  auto channel1 = BuildChannel("pick_first", args);
+  auto response_generator1 = BuildResolverResponseGenerator();
+  auto channel1 = BuildChannel("pick_first", response_generator1, args);
   auto stub1 = BuildStub(channel1);
-  SetNextResolution(ports);
-  auto channel2 = BuildChannel("pick_first", args);
+  response_generator1.SetNextResolution(ports);
+  auto response_generator2 = BuildResolverResponseGenerator();
+  auto channel2 = BuildChannel("pick_first", response_generator2, args);
   auto stub2 = BuildStub(channel2);
-  SetNextResolution(ports);
+  response_generator2.SetNextResolution(ports);
   WaitForServer(stub1, 0, DEBUG_LOCATION);
   // Send one RPC on each channel.
   CheckRpcSendOk(stub1, DEBUG_LOCATION);
@@ -756,13 +794,14 @@
   const int kNumUpdates = 1000;
   const int kNumServers = 3;
   StartServers(kNumServers);
-  auto channel = BuildChannel("pick_first");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("pick_first", response_generator);
   auto stub = BuildStub(channel);
   std::vector<int> ports = GetServersPorts();
   for (size_t i = 0; i < kNumUpdates; ++i) {
     std::shuffle(ports.begin(), ports.end(),
                  std::mt19937(std::random_device()()));
-    SetNextResolution(ports);
+    response_generator.SetNextResolution(ports);
     // We should re-enter core at the end of the loop to give the resolution
     // setting closure a chance to run.
     if ((i + 1) % 10 == 0) CheckRpcSendOk(stub, DEBUG_LOCATION);
@@ -784,16 +823,17 @@
       dead_ports.emplace_back(grpc_pick_unused_port_or_die());
     }
   }
-  auto channel = BuildChannel("pick_first");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("pick_first", response_generator);
   auto stub = BuildStub(channel);
   // The initial resolution only contains dead ports. There won't be any
   // selected subchannel. Re-resolution will return the same result.
-  SetNextResolution(dead_ports);
+  response_generator.SetNextResolution(dead_ports);
   gpr_log(GPR_INFO, "****** INITIAL RESOLUTION SET *******");
   for (size_t i = 0; i < 10; ++i) CheckRpcSendFailure(stub);
   // Set a re-resolution result that contains reachable ports, so that the
   // pick_first LB policy can recover soon.
-  SetNextResolutionUponError(alive_ports);
+  response_generator.SetNextResolutionUponError(alive_ports);
   gpr_log(GPR_INFO, "****** RE-RESOLUTION SET *******");
   WaitForServer(stub, 0, DEBUG_LOCATION, true /* ignore_failure */);
   CheckRpcSendOk(stub, DEBUG_LOCATION);
@@ -805,9 +845,10 @@
 TEST_F(ClientLbEnd2endTest, PickFirstReconnectWithoutNewResolverResult) {
   std::vector<int> ports = {grpc_pick_unused_port_or_die()};
   StartServers(1, ports);
-  auto channel = BuildChannel("pick_first");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("pick_first", response_generator);
   auto stub = BuildStub(channel);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   gpr_log(GPR_INFO, "****** INITIAL CONNECTION *******");
   WaitForServer(stub, 0, DEBUG_LOCATION);
   gpr_log(GPR_INFO, "****** STOPPING SERVER ******");
@@ -824,9 +865,10 @@
                             grpc_pick_unused_port_or_die()};
   CreateServers(2, ports);
   StartServer(1);
-  auto channel = BuildChannel("pick_first");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("pick_first", response_generator);
   auto stub = BuildStub(channel);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   gpr_log(GPR_INFO, "****** INITIAL CONNECTION *******");
   WaitForServer(stub, 1, DEBUG_LOCATION);
   gpr_log(GPR_INFO, "****** STOPPING SERVER ******");
@@ -840,9 +882,10 @@
 TEST_F(ClientLbEnd2endTest, PickFirstCheckStateBeforeStartWatch) {
   std::vector<int> ports = {grpc_pick_unused_port_or_die()};
   StartServers(1, ports);
-  auto channel_1 = BuildChannel("pick_first");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel_1 = BuildChannel("pick_first", response_generator);
   auto stub_1 = BuildStub(channel_1);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   gpr_log(GPR_INFO, "****** RESOLUTION SET FOR CHANNEL 1 *******");
   WaitForServer(stub_1, 0, DEBUG_LOCATION);
   gpr_log(GPR_INFO, "****** CHANNEL 1 CONNECTED *******");
@@ -851,13 +894,10 @@
   // create a new subchannel and hold a ref to it.
   StartServers(1, ports);
   gpr_log(GPR_INFO, "****** SERVER RESTARTED *******");
-  auto channel_2 = BuildChannel("pick_first");
+  auto response_generator_2 = BuildResolverResponseGenerator();
+  auto channel_2 = BuildChannel("pick_first", response_generator_2);
   auto stub_2 = BuildStub(channel_2);
-  // TODO(juanlishen): This resolution result will only be visible to channel 2
-  // since the response generator is only associated with channel 2 now. We
-  // should change the response generator to be able to deliver updates to
-  // multiple channels at once.
-  SetNextResolution(ports);
+  response_generator_2.SetNextResolution(ports);
   gpr_log(GPR_INFO, "****** RESOLUTION SET FOR CHANNEL 2 *******");
   WaitForServer(stub_2, 0, DEBUG_LOCATION, true);
   gpr_log(GPR_INFO, "****** CHANNEL 2 CONNECTED *******");
@@ -883,13 +923,15 @@
   // Start server, send RPC, and make sure channel is READY.
   const int kNumServers = 1;
   StartServers(kNumServers);
-  auto channel = BuildChannel("");  // pick_first is the default.
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel =
+      BuildChannel("", response_generator);  // pick_first is the default.
   auto stub = BuildStub(channel);
-  SetNextResolution(GetServersPorts());
+  response_generator.SetNextResolution(GetServersPorts());
   CheckRpcSendOk(stub, DEBUG_LOCATION);
   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
   // Stop server.  Channel should go into state IDLE.
-  SetFailureOnReresolution();
+  response_generator.SetFailureOnReresolution();
   servers_[0]->Shutdown();
   EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_IDLE);
@@ -897,14 +939,16 @@
 }
 
 TEST_F(ClientLbEnd2endTest, PickFirstPendingUpdateAndSelectedSubchannelFails) {
-  auto channel = BuildChannel("");  // pick_first is the default.
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel =
+      BuildChannel("", response_generator);  // pick_first is the default.
   auto stub = BuildStub(channel);
   // Create a number of servers, but only start 1 of them.
   CreateServers(10);
   StartServer(0);
   // Initially resolve to first server and make sure it connects.
   gpr_log(GPR_INFO, "Phase 1: Connect to first server.");
-  SetNextResolution({servers_[0]->port_});
+  response_generator.SetNextResolution({servers_[0]->port_});
   CheckRpcSendOk(stub, DEBUG_LOCATION, true /* wait_for_ready */);
   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
   // Send a resolution update with the remaining servers, none of which are
@@ -916,7 +960,7 @@
   gpr_log(GPR_INFO,
           "Phase 2: Resolver update pointing to remaining "
           "(not started) servers.");
-  SetNextResolution(GetServersPorts(1 /* start_index */));
+  response_generator.SetNextResolution(GetServersPorts(1 /* start_index */));
   // RPCs will continue to be sent to the first server.
   CheckRpcSendOk(stub, DEBUG_LOCATION);
   // Now stop the first server, so that the current subchannel list
@@ -947,9 +991,11 @@
   // Start server, send RPC, and make sure channel is READY.
   const int kNumServers = 1;
   StartServers(kNumServers);
-  auto channel = BuildChannel("");  // pick_first is the default.
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel =
+      BuildChannel("", response_generator);  // pick_first is the default.
   auto stub = BuildStub(channel);
-  SetNextResolution(GetServersPorts());
+  response_generator.SetNextResolution(GetServersPorts());
   CheckRpcSendOk(stub, DEBUG_LOCATION);
   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
   // Stop server.  Channel should go into state IDLE.
@@ -958,13 +1004,13 @@
   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_IDLE);
   // Now send resolver update that includes no addresses.  Channel
   // should stay in state IDLE.
-  SetNextResolution({});
+  response_generator.SetNextResolution({});
   EXPECT_FALSE(channel->WaitForStateChange(
       GRPC_CHANNEL_IDLE, grpc_timeout_seconds_to_deadline(3)));
   // Now bring the backend back up and send a non-empty resolver update,
   // and then try to send an RPC.  Channel should go back into state READY.
   StartServer(0);
-  SetNextResolution(GetServersPorts());
+  response_generator.SetNextResolution(GetServersPorts());
   CheckRpcSendOk(stub, DEBUG_LOCATION);
   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
 }
@@ -973,9 +1019,10 @@
   // Start servers and send one RPC per server.
   const int kNumServers = 3;
   StartServers(kNumServers);
-  auto channel = BuildChannel("round_robin");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("round_robin", response_generator);
   auto stub = BuildStub(channel);
-  SetNextResolution(GetServersPorts());
+  response_generator.SetNextResolution(GetServersPorts());
   // Wait until all backends are ready.
   do {
     CheckRpcSendOk(stub, DEBUG_LOCATION);
@@ -999,18 +1046,20 @@
 
 TEST_F(ClientLbEnd2endTest, RoundRobinProcessPending) {
   StartServers(1);  // Single server
-  auto channel = BuildChannel("round_robin");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("round_robin", response_generator);
   auto stub = BuildStub(channel);
-  SetNextResolution({servers_[0]->port_});
+  response_generator.SetNextResolution({servers_[0]->port_});
   WaitForServer(stub, 0, DEBUG_LOCATION);
   // Create a new channel and its corresponding RR LB policy, which will pick
   // the subchannels in READY state from the previous RPC against the same
   // target (even if it happened over a different channel, because subchannels
   // are globally reused). Progress should happen without any transition from
   // this READY state.
-  auto second_channel = BuildChannel("round_robin");
+  auto second_response_generator = BuildResolverResponseGenerator();
+  auto second_channel = BuildChannel("round_robin", second_response_generator);
   auto second_stub = BuildStub(second_channel);
-  SetNextResolution({servers_[0]->port_});
+  second_response_generator.SetNextResolution({servers_[0]->port_});
   CheckRpcSendOk(second_stub, DEBUG_LOCATION);
 }
 
@@ -1018,13 +1067,14 @@
   // Start servers and send one RPC per server.
   const int kNumServers = 3;
   StartServers(kNumServers);
-  auto channel = BuildChannel("round_robin");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("round_robin", response_generator);
   auto stub = BuildStub(channel);
   std::vector<int> ports;
   // Start with a single server.
   gpr_log(GPR_INFO, "*** FIRST BACKEND ***");
   ports.emplace_back(servers_[0]->port_);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   WaitForServer(stub, 0, DEBUG_LOCATION);
   // Send RPCs. They should all go servers_[0]
   for (size_t i = 0; i < 10; ++i) CheckRpcSendOk(stub, DEBUG_LOCATION);
@@ -1036,7 +1086,7 @@
   gpr_log(GPR_INFO, "*** SECOND BACKEND ***");
   ports.clear();
   ports.emplace_back(servers_[1]->port_);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   // Wait until update has been processed, as signaled by the second backend
   // receiving a request.
   EXPECT_EQ(0, servers_[1]->service_.request_count());
@@ -1050,7 +1100,7 @@
   gpr_log(GPR_INFO, "*** THIRD BACKEND ***");
   ports.clear();
   ports.emplace_back(servers_[2]->port_);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   WaitForServer(stub, 2, DEBUG_LOCATION);
   for (size_t i = 0; i < 10; ++i) CheckRpcSendOk(stub, DEBUG_LOCATION);
   EXPECT_EQ(0, servers_[0]->service_.request_count());
@@ -1063,7 +1113,7 @@
   ports.emplace_back(servers_[0]->port_);
   ports.emplace_back(servers_[1]->port_);
   ports.emplace_back(servers_[2]->port_);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   WaitForServer(stub, 0, DEBUG_LOCATION);
   WaitForServer(stub, 1, DEBUG_LOCATION);
   WaitForServer(stub, 2, DEBUG_LOCATION);
@@ -1075,7 +1125,7 @@
   // An empty update will result in the channel going into TRANSIENT_FAILURE.
   gpr_log(GPR_INFO, "*** NO BACKENDS ***");
   ports.clear();
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   grpc_connectivity_state channel_state;
   do {
     channel_state = channel->GetState(true /* try to connect */);
@@ -1086,7 +1136,7 @@
   gpr_log(GPR_INFO, "*** BACK TO SECOND BACKEND ***");
   ports.clear();
   ports.emplace_back(servers_[1]->port_);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   WaitForServer(stub, 1, DEBUG_LOCATION);
   channel_state = channel->GetState(false /* try to connect */);
   ASSERT_EQ(channel_state, GRPC_CHANNEL_READY);
@@ -1097,13 +1147,14 @@
 TEST_F(ClientLbEnd2endTest, RoundRobinUpdateInError) {
   const int kNumServers = 3;
   StartServers(kNumServers);
-  auto channel = BuildChannel("round_robin");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("round_robin", response_generator);
   auto stub = BuildStub(channel);
   std::vector<int> ports;
 
   // Start with a single server.
   ports.emplace_back(servers_[0]->port_);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   WaitForServer(stub, 0, DEBUG_LOCATION);
   // Send RPCs. They should all go to servers_[0]
   for (size_t i = 0; i < 10; ++i) SendRpc(stub);
@@ -1116,7 +1167,7 @@
   servers_[1]->Shutdown();
   ports.emplace_back(servers_[1]->port_);
   ports.emplace_back(servers_[2]->port_);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   WaitForServer(stub, 0, DEBUG_LOCATION);
   WaitForServer(stub, 2, DEBUG_LOCATION);
 
@@ -1130,13 +1181,14 @@
   // Start servers and send one RPC per server.
   const int kNumServers = 3;
   StartServers(kNumServers);
-  auto channel = BuildChannel("round_robin");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("round_robin", response_generator);
   auto stub = BuildStub(channel);
   std::vector<int> ports = GetServersPorts();
   for (size_t i = 0; i < 1000; ++i) {
     std::shuffle(ports.begin(), ports.end(),
                  std::mt19937(std::random_device()()));
-    SetNextResolution(ports);
+    response_generator.SetNextResolution(ports);
     if (i % 10 == 0) CheckRpcSendOk(stub, DEBUG_LOCATION);
   }
   // Check LB policy name for the channel.
@@ -1162,9 +1214,10 @@
     second_ports.push_back(grpc_pick_unused_port_or_die());
   }
   StartServers(kNumServers, first_ports);
-  auto channel = BuildChannel("round_robin");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("round_robin", response_generator);
   auto stub = BuildStub(channel);
-  SetNextResolution(first_ports);
+  response_generator.SetNextResolution(first_ports);
   // Send a number of RPCs, which succeed.
   for (size_t i = 0; i < 100; ++i) {
     CheckRpcSendOk(stub, DEBUG_LOCATION);
@@ -1188,7 +1241,7 @@
   StartServers(kNumServers, second_ports);
   // Don't notify of the update. Wait for the LB policy's re-resolution to
   // "pull" the new ports.
-  SetNextResolutionUponError(second_ports);
+  response_generator.SetNextResolutionUponError(second_ports);
   gpr_log(GPR_INFO, "****** SERVERS RESTARTED *******");
   gpr_log(GPR_INFO, "****** SENDING REQUEST TO SUCCEED *******");
   // Client request should eventually (but still fairly soon) succeed.
@@ -1205,9 +1258,10 @@
   const int kNumServers = 3;
   StartServers(kNumServers);
   const auto ports = GetServersPorts();
-  auto channel = BuildChannel("round_robin");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("round_robin", response_generator);
   auto stub = BuildStub(channel);
-  SetNextResolution(ports);
+  response_generator.SetNextResolution(ports);
   for (size_t i = 0; i < kNumServers; ++i) {
     WaitForServer(stub, i, DEBUG_LOCATION);
   }
@@ -1251,9 +1305,10 @@
   args.SetServiceConfigJSON(
       "{\"healthCheckConfig\": "
       "{\"serviceName\": \"health_check_service_name\"}}");
-  auto channel = BuildChannel("round_robin", args);
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("round_robin", response_generator, args);
   auto stub = BuildStub(channel);
-  SetNextResolution({servers_[0]->port_});
+  response_generator.SetNextResolution({servers_[0]->port_});
   EXPECT_TRUE(WaitForChannelReady(channel.get()));
   CheckRpcSendOk(stub, DEBUG_LOCATION);
 }
@@ -1267,9 +1322,10 @@
   args.SetServiceConfigJSON(
       "{\"healthCheckConfig\": "
       "{\"serviceName\": \"health_check_service_name\"}}");
-  auto channel = BuildChannel("round_robin", args);
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("round_robin", response_generator, args);
   auto stub = BuildStub(channel);
-  SetNextResolution(GetServersPorts());
+  response_generator.SetNextResolution(GetServersPorts());
   // Channel should not become READY, because health checks should be failing.
   gpr_log(GPR_INFO,
           "*** initial state: unknown health check service name for "
@@ -1341,15 +1397,17 @@
   args.SetServiceConfigJSON(
       "{\"healthCheckConfig\": "
       "{\"serviceName\": \"health_check_service_name\"}}");
-  auto channel1 = BuildChannel("round_robin", args);
+  auto response_generator1 = BuildResolverResponseGenerator();
+  auto channel1 = BuildChannel("round_robin", response_generator1, args);
   auto stub1 = BuildStub(channel1);
   std::vector<int> ports = GetServersPorts();
-  SetNextResolution(ports);
+  response_generator1.SetNextResolution(ports);
   // Create a channel with health checking enabled but inhibited.
   args.SetInt(GRPC_ARG_INHIBIT_HEALTH_CHECKING, 1);
-  auto channel2 = BuildChannel("round_robin", args);
+  auto response_generator2 = BuildResolverResponseGenerator();
+  auto channel2 = BuildChannel("round_robin", response_generator2, args);
   auto stub2 = BuildStub(channel2);
-  SetNextResolution(ports);
+  response_generator2.SetNextResolution(ports);
   // First channel should not become READY, because health checks should be
   // failing.
   EXPECT_FALSE(WaitForChannelReady(channel1.get(), 1));
@@ -1376,19 +1434,21 @@
   args.SetServiceConfigJSON(
       "{\"healthCheckConfig\": "
       "{\"serviceName\": \"health_check_service_name\"}}");
-  auto channel1 = BuildChannel("round_robin", args);
+  auto response_generator1 = BuildResolverResponseGenerator();
+  auto channel1 = BuildChannel("round_robin", response_generator1, args);
   auto stub1 = BuildStub(channel1);
   std::vector<int> ports = GetServersPorts();
-  SetNextResolution(ports);
+  response_generator1.SetNextResolution(ports);
   // Create a channel with health-checking enabled with a different
   // service name.
   ChannelArguments args2;
   args2.SetServiceConfigJSON(
       "{\"healthCheckConfig\": "
       "{\"serviceName\": \"health_check_service_name2\"}}");
-  auto channel2 = BuildChannel("round_robin", args2);
+  auto response_generator2 = BuildResolverResponseGenerator();
+  auto channel2 = BuildChannel("round_robin", response_generator2, args2);
   auto stub2 = BuildStub(channel2);
-  SetNextResolution(ports);
+  response_generator2.SetNextResolution(ports);
   // Allow health checks from channel 2 to succeed.
   servers_[0]->SetServingStatus("health_check_service_name2", true);
   // First channel should not become READY, because health checks should be
@@ -1438,9 +1498,11 @@
   const int kNumServers = 1;
   const int kNumRpcs = 10;
   StartServers(kNumServers);
-  auto channel = BuildChannel("intercept_trailing_metadata_lb");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel =
+      BuildChannel("intercept_trailing_metadata_lb", response_generator);
   auto stub = BuildStub(channel);
-  SetNextResolution(GetServersPorts());
+  response_generator.SetNextResolution(GetServersPorts());
   for (size_t i = 0; i < kNumRpcs; ++i) {
     CheckRpcSendOk(stub, DEBUG_LOCATION);
   }
@@ -1470,9 +1532,11 @@
       "    }\n"
       "  } ]\n"
       "}");
-  auto channel = BuildChannel("intercept_trailing_metadata_lb", args);
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel =
+      BuildChannel("intercept_trailing_metadata_lb", response_generator, args);
   auto stub = BuildStub(channel);
-  SetNextResolution(GetServersPorts());
+  response_generator.SetNextResolution(GetServersPorts());
   for (size_t i = 0; i < kNumRpcs; ++i) {
     CheckRpcSendOk(stub, DEBUG_LOCATION);
   }
diff --git a/tools/distrib/python/bazel_deps.sh b/tools/distrib/python/bazel_deps.sh
index 5f4ee1d..67896f1 100755
--- a/tools/distrib/python/bazel_deps.sh
+++ b/tools/distrib/python/bazel_deps.sh
@@ -26,6 +26,7 @@
   docker build -t bazel_local_img tools/dockerfile/test/sanity
   docker run -v "$(realpath .):/src/grpc/:ro" \
     -w /src/grpc/third_party/protobuf         \
+    --rm=true                                 \
     bazel_local_img                           \
     bazel query 'deps('$1')'
 fi