Repository: grpc/grpc-go

Issue:
stats/opentelemetry: Write e2e tests for tracing more than single message sent and received

In all of the [e2e tests](https://github.com/grpc/grpc-go/blob/732f3f32f5365f4f5b89a3f1be19534cd28db794/stats/opentelemetry/e2e_test.go), we only have single message being sent/received, which means that the logic on the sequence numbers incremented is untested. Write a separate test for streaming RPC where multiple messages are being sent and received and validate the traces accordingly

Hints from issue comments:
@vinothkumarr227: @purnesh42H Please assign the issue to me.

Solution patch:
diff --git a/credentials/alts/internal/conn/common.go b/credentials/alts/internal/conn/common.go
index 1795d0c9..46617132 100644
--- a/credentials/alts/internal/conn/common.go
+++ b/credentials/alts/internal/conn/common.go
@@ -54,11 +54,10 @@ func SliceForAppend(in []byte, n int) (head, tail []byte) {
 func ParseFramedMsg(b []byte, maxLen uint32) ([]byte, []byte, error) {
 	// If the size field is not complete, return the provided buffer as
 	// remaining buffer.
-	if len(b) < MsgLenFieldSize {
+	length, sufficientBytes := parseMessageLength(b)
+	if !sufficientBytes {
 		return nil, b, nil
 	}
-	msgLenField := b[:MsgLenFieldSize]
-	length := binary.LittleEndian.Uint32(msgLenField)
 	if length > maxLen {
 		return nil, nil, fmt.Errorf("received the frame length %d larger than the limit %d", length, maxLen)
 	}
@@ -68,3 +67,14 @@ func ParseFramedMsg(b []byte, maxLen uint32) ([]byte, []byte, error) {
 	}
 	return b[:MsgLenFieldSize+length], b[MsgLenFieldSize+length:], nil
 }
+
+// parseMessageLength returns the message length based on frame header. It also
+// returns a boolean indicating if the buffer contains sufficient bytes to parse
+// the length header. If there are insufficient bytes, (0, false) is returned.
+func parseMessageLength(b []byte) (uint32, bool) {
+	if len(b) < MsgLenFieldSize {
+		return 0, false
+	}
+	msgLenField := b[:MsgLenFieldSize]
+	return binary.LittleEndian.Uint32(msgLenField), true
+}
diff --git a/credentials/alts/internal/conn/record.go b/credentials/alts/internal/conn/record.go
index bd5249ed..f9d2646d 100644
--- a/credentials/alts/internal/conn/record.go
+++ b/credentials/alts/internal/conn/record.go
@@ -67,6 +67,8 @@ const (
 	// The maximum write buffer size. This *must* be multiple of
 	// altsRecordDefaultLength.
 	altsWriteBufferMaxSize = 512 * 1024 // 512KiB
+	// The initial buffer used to read from the network.
+	altsReadBufferInitialSize = 32 * 1024 // 32KiB
 )
 
 var (
@@ -87,7 +89,7 @@ type conn struct {
 	net.Conn
 	crypto ALTSRecordCrypto
 	// buf holds data that has been read from the connection and decrypted,
-	// but has not yet been returned by Read.
+	// but has not yet been returned by Read. It is a sub-slice of protected.
 	buf                []byte
 	payloadLengthLimit int
 	// protected holds data read from the network but have not yet been
@@ -115,21 +117,13 @@ func NewConn(c net.Conn, side core.Side, recordProtocol string, key []byte, prot
 	}
 	overhead := MsgLenFieldSize + msgTypeFieldSize + crypto.EncryptionOverhead()
 	payloadLengthLimit := altsRecordDefaultLength - overhead
-	var protectedBuf []byte
-	if protected == nil {
-		// We pre-allocate protected to be of size
-		// 2*altsRecordDefaultLength-1 during initialization. We only
-		// read from the network into protected when protected does not
-		// contain a complete frame, which is at most
-		// altsRecordDefaultLength-1 (bytes). And we read at most
-		// altsRecordDefaultLength (bytes) data into protected at one
-		// time. Therefore, 2*altsRecordDefaultLength-1 is large enough
-		// to buffer data read from the network.
-		protectedBuf = make([]byte, 0, 2*altsRecordDefaultLength-1)
-	} else {
-		protectedBuf = make([]byte, len(protected))
-		copy(protectedBuf, protected)
-	}
+	// We pre-allocate protected to be of size 32KB during initialization.
+	// We increase the size of the buffer by the required amount if it can't
+	// hold a complete encrypted record.
+	protectedBuf := make([]byte, max(altsReadBufferInitialSize, len(protected)))
+	// Copy additional data from hanshaker service.
+	copy(protectedBuf, protected)
+	protectedBuf = protectedBuf[:len(protected)]
 
 	altsConn := &conn{
 		Conn:               c,
@@ -166,11 +160,26 @@ func (p *conn) Read(b []byte) (n int, err error) {
 		// Check whether a complete frame has been received yet.
 		for len(framedMsg) == 0 {
 			if len(p.protected) == cap(p.protected) {
-				tmp := make([]byte, len(p.protected), cap(p.protected)+altsRecordDefaultLength)
-				copy(tmp, p.protected)
-				p.protected = tmp
+				// We can parse the length header to know exactly how large
+				// the buffer needs to be to hold the entire frame.
+				length, didParse := parseMessageLength(p.protected)
+				if !didParse {
+					// The protected buffer is initialized with a capacity of
+					// larger than 4B. It should always hold the message length
+					// header.
+					panic(fmt.Sprintf("protected buffer length shorter than expected: %d vs %d", len(p.protected), MsgLenFieldSize))
+				}
+				oldProtectedBuf := p.protected
+				// The new buffer must be able to hold the message length header
+				// and the entire message.
+				requiredCapacity := int(length) + MsgLenFieldSize
+				p.protected = make([]byte, requiredCapacity)
+				// Copy the contents of the old buffer and set the length of the
+				// new buffer to the number of bytes already read.
+				copy(p.protected, oldProtectedBuf)
+				p.protected = p.protected[:len(oldProtectedBuf)]
 			}
-			n, err = p.Conn.Read(p.protected[len(p.protected):min(cap(p.protected), len(p.protected)+altsRecordDefaultLength)])
+			n, err = p.Conn.Read(p.protected[len(p.protected):cap(p.protected)])
 			if err != nil {
 				return 0, err
 			}
@@ -189,6 +198,15 @@ func (p *conn) Read(b []byte) (n int, err error) {
 		}
 		ciphertext := msg[msgTypeFieldSize:]
 
+		// Decrypt directly into the buffer, avoiding a copy from p.buf if
+		// possible.
+		if len(b) >= len(ciphertext) {
+			dec, err := p.crypto.Decrypt(b[:0], ciphertext)
+			if err != nil {
+				return 0, err
+			}
+			return len(dec), nil
+		}
 		// Decrypt requires that if the dst and ciphertext alias, they
 		// must alias exactly. Code here used to use msg[:0], but msg
 		// starts MsgLenFieldSize+msgTypeFieldSize bytes earlier than
diff --git a/credentials/alts/internal/handshaker/handshaker.go b/credentials/alts/internal/handshaker/handshaker.go
index 50721f69..becd2f3b 100644
--- a/credentials/alts/internal/handshaker/handshaker.go
+++ b/credentials/alts/internal/handshaker/handshaker.go
@@ -308,6 +308,7 @@ func (h *altsHandshaker) accessHandshakerService(req *altspb.HandshakerReq) (*al
 // whatever received from the network and send it to the handshaker service.
 func (h *altsHandshaker) processUntilDone(resp *altspb.HandshakerResp, extra []byte) (*altspb.HandshakerResult, []byte, error) {
 	var lastWriteTime time.Time
+	buf := make([]byte, frameLimit)
 	for {
 		if len(resp.OutFrames) > 0 {
 			lastWriteTime = time.Now()
@@ -318,7 +319,6 @@ func (h *altsHandshaker) processUntilDone(resp *altspb.HandshakerResp, extra []b
 		if resp.Result != nil {
 			return resp.Result, extra, nil
 		}
-		buf := make([]byte, frameLimit)
 		n, err := h.conn.Read(buf)
 		if err != nil && err != io.EOF {
 			return nil, nil, err
diff --git a/credentials/alts/internal/proto/grpc_gcp/handshaker.pb.go b/credentials/alts/internal/proto/grpc_gcp/handshaker.pb.go
index b780024f..6370b2a6 100644
--- a/credentials/alts/internal/proto/grpc_gcp/handshaker.pb.go
+++ b/credentials/alts/internal/proto/grpc_gcp/handshaker.pb.go
@@ -331,9 +331,11 @@ type StartClientHandshakeReq struct {
 	// ALTS connections. The access token that should be used to authenticate to
 	// the peer. The access token MUST be strongly bound to the ALTS credentials
 	// used to establish the connection that the token is sent over.
-	AccessToken   string `protobuf:"bytes,11,opt,name=access_token,json=accessToken,proto3" json:"access_token,omitempty"`
-	unknownFields protoimpl.UnknownFields
-	sizeCache     protoimpl.SizeCache
+	AccessToken string `protobuf:"bytes,11,opt,name=access_token,json=accessToken,proto3" json:"access_token,omitempty"`
+	// (Optional) Ordered transport protocol preferences supported by the client.
+	TransportProtocolPreferences *TransportProtocolPreferences `protobuf:"bytes,12,opt,name=transport_protocol_preferences,json=transportProtocolPreferences,proto3" json:"transport_protocol_preferences,omitempty"`
+	unknownFields                protoimpl.UnknownFields
+	sizeCache                    protoimpl.SizeCache
 }
 
 func (x *StartClientHandshakeReq) Reset() {
@@ -443,6 +445,13 @@ func (x *StartClientHandshakeReq) GetAccessToken() string {
 	return ""
 }
 
+func (x *StartClientHandshakeReq) GetTransportProtocolPreferences() *TransportProtocolPreferences {
+	if x != nil {
+		return x.TransportProtocolPreferences
+	}
+	return nil
+}
+
 type ServerHandshakeParameters struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
 	// The record protocols supported by the server, e.g.,
@@ -534,9 +543,11 @@ type StartServerHandshakeReq struct {
 	// (Optional) RPC protocol versions supported by the server.
 	RpcVersions *RpcProtocolVersions `protobuf:"bytes,6,opt,name=rpc_versions,json=rpcVersions,proto3" json:"rpc_versions,omitempty"`
 	// (Optional) Maximum frame size supported by the server.
-	MaxFrameSize  uint32 `protobuf:"varint,7,opt,name=max_frame_size,json=maxFrameSize,proto3" json:"max_frame_size,omitempty"`
-	unknownFields protoimpl.UnknownFields
-	sizeCache     protoimpl.SizeCache
+	MaxFrameSize uint32 `protobuf:"varint,7,opt,name=max_frame_size,json=maxFrameSize,proto3" json:"max_frame_size,omitempty"`
+	// (Optional) Transport protocol preferences supported by the server.
+	TransportProtocolPreferences *TransportProtocolPreferences `protobuf:"bytes,8,opt,name=transport_protocol_preferences,json=transportProtocolPreferences,proto3" json:"transport_protocol_preferences,omitempty"`
+	unknownFields                protoimpl.UnknownFields
+	sizeCache                    protoimpl.SizeCache
 }
 
 func (x *StartServerHandshakeReq) Reset() {
@@ -618,6 +629,13 @@ func (x *StartServerHandshakeReq) GetMaxFrameSize() uint32 {
 	return 0
 }
 
+func (x *StartServerHandshakeReq) GetTransportProtocolPreferences() *TransportProtocolPreferences {
+	if x != nil {
+		return x.TransportProtocolPreferences
+	}
+	return nil
+}
+
 type NextHandshakeMessageReq struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
 	// Bytes in out_frames returned from the peer's HandshakerResp. It is possible
@@ -798,9 +816,11 @@ type HandshakerResult struct {
 	// The RPC protocol versions supported by the peer.
 	PeerRpcVersions *RpcProtocolVersions `protobuf:"bytes,7,opt,name=peer_rpc_versions,json=peerRpcVersions,proto3" json:"peer_rpc_versions,omitempty"`
 	// The maximum frame size of the peer.
-	MaxFrameSize  uint32 `protobuf:"varint,8,opt,name=max_frame_size,json=maxFrameSize,proto3" json:"max_frame_size,omitempty"`
-	unknownFields protoimpl.UnknownFields
-	sizeCache     protoimpl.SizeCache
+	MaxFrameSize uint32 `protobuf:"varint,8,opt,name=max_frame_size,json=maxFrameSize,proto3" json:"max_frame_size,omitempty"`
+	// (Optional) The transport protocol negotiated for this connection.
+	TransportProtocol *NegotiatedTransportProtocol `protobuf:"bytes,9,opt,name=transport_protocol,json=transportProtocol,proto3" json:"transport_protocol,omitempty"`
+	unknownFields     protoimpl.UnknownFields
+	sizeCache         protoimpl.SizeCache
 }
 
 func (x *HandshakerResult) Reset() {
@@ -889,6 +909,13 @@ func (x *HandshakerResult) GetMaxFrameSize() uint32 {
 	return 0
 }
 
+func (x *HandshakerResult) GetTransportProtocol() *NegotiatedTransportProtocol {
+	if x != nil {
+		return x.TransportProtocol
+	}
+	return nil
+}
+
 type HandshakerStatus struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
 	// The status code. This could be the gRPC status code.
@@ -1041,7 +1068,7 @@ const file_grpc_gcp_handshaker_proto_rawDesc = "" +
 	"\x0fAttributesEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
 	"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01B\x10\n" +
-	"\x0eidentity_oneof\"\xfb\x04\n" +
+	"\x0eidentity_oneof\"\xe9\x05\n" +
 	"\x17StartClientHandshakeReq\x12[\n" +
 	"\x1bhandshake_security_protocol\x18\x01 \x01(\x0e2\x1b.grpc.gcp.HandshakeProtocolR\x19handshakeSecurityProtocol\x123\n" +
 	"\x15application_protocols\x18\x02 \x03(\tR\x14applicationProtocols\x12)\n" +
@@ -1055,12 +1082,13 @@ const file_grpc_gcp_handshaker_proto_rawDesc = "" +
 	"\frpc_versions\x18\t \x01(\v2\x1d.grpc.gcp.RpcProtocolVersionsR\vrpcVersions\x12$\n" +
 	"\x0emax_frame_size\x18\n" +
 	" \x01(\rR\fmaxFrameSize\x12&\n" +
-	"\faccess_token\x18\v \x01(\tB\x03\x80\x01\x01R\vaccessToken\"\xaf\x01\n" +
+	"\faccess_token\x18\v \x01(\tB\x03\x80\x01\x01R\vaccessToken\x12l\n" +
+	"\x1etransport_protocol_preferences\x18\f \x01(\v2&.grpc.gcp.TransportProtocolPreferencesR\x1ctransportProtocolPreferences\"\xaf\x01\n" +
 	"\x19ServerHandshakeParameters\x12)\n" +
 	"\x10record_protocols\x18\x01 \x03(\tR\x0frecordProtocols\x12=\n" +
 	"\x10local_identities\x18\x02 \x03(\v2\x12.grpc.gcp.IdentityR\x0flocalIdentities\x12\x1e\n" +
 	"\x05token\x18\x03 \x01(\tB\x03\x80\x01\x01H\x00R\x05token\x88\x01\x01B\b\n" +
-	"\x06_token\"\xa5\x04\n" +
+	"\x06_token\"\x93\x05\n" +
 	"\x17StartServerHandshakeReq\x123\n" +
 	"\x15application_protocols\x18\x01 \x03(\tR\x14applicationProtocols\x12m\n" +
 	"\x14handshake_parameters\x18\x02 \x03(\v2:.grpc.gcp.StartServerHandshakeReq.HandshakeParametersEntryR\x13handshakeParameters\x12\x19\n" +
@@ -1068,7 +1096,8 @@ const file_grpc_gcp_handshaker_proto_rawDesc = "" +
 	"\x0elocal_endpoint\x18\x04 \x01(\v2\x12.grpc.gcp.EndpointR\rlocalEndpoint\x12;\n" +
 	"\x0fremote_endpoint\x18\x05 \x01(\v2\x12.grpc.gcp.EndpointR\x0eremoteEndpoint\x12@\n" +
 	"\frpc_versions\x18\x06 \x01(\v2\x1d.grpc.gcp.RpcProtocolVersionsR\vrpcVersions\x12$\n" +
-	"\x0emax_frame_size\x18\a \x01(\rR\fmaxFrameSize\x1ak\n" +
+	"\x0emax_frame_size\x18\a \x01(\rR\fmaxFrameSize\x12l\n" +
+	"\x1etransport_protocol_preferences\x18\b \x01(\v2&.grpc.gcp.TransportProtocolPreferencesR\x1ctransportProtocolPreferences\x1ak\n" +
 	"\x18HandshakeParametersEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\x05R\x03key\x129\n" +
 	"\x05value\x18\x02 \x01(\v2#.grpc.gcp.ServerHandshakeParametersR\x05value:\x028\x01\"b\n" +
@@ -1079,7 +1108,7 @@ const file_grpc_gcp_handshaker_proto_rawDesc = "" +
 	"\fclient_start\x18\x01 \x01(\v2!.grpc.gcp.StartClientHandshakeReqH\x00R\vclientStart\x12F\n" +
 	"\fserver_start\x18\x02 \x01(\v2!.grpc.gcp.StartServerHandshakeReqH\x00R\vserverStart\x127\n" +
 	"\x04next\x18\x03 \x01(\v2!.grpc.gcp.NextHandshakeMessageReqH\x00R\x04nextB\v\n" +
-	"\treq_oneof\"\x9a\x03\n" +
+	"\treq_oneof\"\xf0\x03\n" +
 	"\x10HandshakerResult\x121\n" +
 	"\x14application_protocol\x18\x01 \x01(\tR\x13applicationProtocol\x12'\n" +
 	"\x0frecord_protocol\x18\x02 \x01(\tR\x0erecordProtocol\x12\x19\n" +
@@ -1088,7 +1117,8 @@ const file_grpc_gcp_handshaker_proto_rawDesc = "" +
 	"\x0elocal_identity\x18\x05 \x01(\v2\x12.grpc.gcp.IdentityR\rlocalIdentity\x12*\n" +
 	"\x11keep_channel_open\x18\x06 \x01(\bR\x0fkeepChannelOpen\x12I\n" +
 	"\x11peer_rpc_versions\x18\a \x01(\v2\x1d.grpc.gcp.RpcProtocolVersionsR\x0fpeerRpcVersions\x12$\n" +
-	"\x0emax_frame_size\x18\b \x01(\rR\fmaxFrameSize\"@\n" +
+	"\x0emax_frame_size\x18\b \x01(\rR\fmaxFrameSize\x12T\n" +
+	"\x12transport_protocol\x18\t \x01(\v2%.grpc.gcp.NegotiatedTransportProtocolR\x11transportProtocol\"@\n" +
 	"\x10HandshakerStatus\x12\x12\n" +
 	"\x04code\x18\x01 \x01(\rR\x04code\x12\x18\n" +
 	"\adetails\x18\x02 \x01(\tR\adetails\"\xbe\x01\n" +
@@ -1125,21 +1155,23 @@ func file_grpc_gcp_handshaker_proto_rawDescGZIP() []byte {
 var file_grpc_gcp_handshaker_proto_enumTypes = make([]protoimpl.EnumInfo, 2)
 var file_grpc_gcp_handshaker_proto_msgTypes = make([]protoimpl.MessageInfo, 12)
 var file_grpc_gcp_handshaker_proto_goTypes = []any{
-	(HandshakeProtocol)(0),            // 0: grpc.gcp.HandshakeProtocol
-	(NetworkProtocol)(0),              // 1: grpc.gcp.NetworkProtocol
-	(*Endpoint)(nil),                  // 2: grpc.gcp.Endpoint
-	(*Identity)(nil),                  // 3: grpc.gcp.Identity
-	(*StartClientHandshakeReq)(nil),   // 4: grpc.gcp.StartClientHandshakeReq
-	(*ServerHandshakeParameters)(nil), // 5: grpc.gcp.ServerHandshakeParameters
-	(*StartServerHandshakeReq)(nil),   // 6: grpc.gcp.StartServerHandshakeReq
-	(*NextHandshakeMessageReq)(nil),   // 7: grpc.gcp.NextHandshakeMessageReq
-	(*HandshakerReq)(nil),             // 8: grpc.gcp.HandshakerReq
-	(*HandshakerResult)(nil),          // 9: grpc.gcp.HandshakerResult
-	(*HandshakerStatus)(nil),          // 10: grpc.gcp.HandshakerStatus
-	(*HandshakerResp)(nil),            // 11: grpc.gcp.HandshakerResp
-	nil,                               // 12: grpc.gcp.Identity.AttributesEntry
-	nil,                               // 13: grpc.gcp.StartServerHandshakeReq.HandshakeParametersEntry
-	(*RpcProtocolVersions)(nil),       // 14: grpc.gcp.RpcProtocolVersions
+	(HandshakeProtocol)(0),               // 0: grpc.gcp.HandshakeProtocol
+	(NetworkProtocol)(0),                 // 1: grpc.gcp.NetworkProtocol
+	(*Endpoint)(nil),                     // 2: grpc.gcp.Endpoint
+	(*Identity)(nil),                     // 3: grpc.gcp.Identity
+	(*StartClientHandshakeReq)(nil),      // 4: grpc.gcp.StartClientHandshakeReq
+	(*ServerHandshakeParameters)(nil),    // 5: grpc.gcp.ServerHandshakeParameters
+	(*StartServerHandshakeReq)(nil),      // 6: grpc.gcp.StartServerHandshakeReq
+	(*NextHandshakeMessageReq)(nil),      // 7: grpc.gcp.NextHandshakeMessageReq
+	(*HandshakerReq)(nil),                // 8: grpc.gcp.HandshakerReq
+	(*HandshakerResult)(nil),             // 9: grpc.gcp.HandshakerResult
+	(*HandshakerStatus)(nil),             // 10: grpc.gcp.HandshakerStatus
+	(*HandshakerResp)(nil),               // 11: grpc.gcp.HandshakerResp
+	nil,                                  // 12: grpc.gcp.Identity.AttributesEntry
+	nil,                                  // 13: grpc.gcp.StartServerHandshakeReq.HandshakeParametersEntry
+	(*RpcProtocolVersions)(nil),          // 14: grpc.gcp.RpcProtocolVersions
+	(*TransportProtocolPreferences)(nil), // 15: grpc.gcp.TransportProtocolPreferences
+	(*NegotiatedTransportProtocol)(nil),  // 16: grpc.gcp.NegotiatedTransportProtocol
 }
 var file_grpc_gcp_handshaker_proto_depIdxs = []int32{
 	1,  // 0: grpc.gcp.Endpoint.protocol:type_name -> grpc.gcp.NetworkProtocol
@@ -1150,27 +1182,30 @@ var file_grpc_gcp_handshaker_proto_depIdxs = []int32{
 	2,  // 5: grpc.gcp.StartClientHandshakeReq.local_endpoint:type_name -> grpc.gcp.Endpoint
 	2,  // 6: grpc.gcp.StartClientHandshakeReq.remote_endpoint:type_name -> grpc.gcp.Endpoint
 	14, // 7: grpc.gcp.StartClientHandshakeReq.rpc_versions:type_name -> grpc.gcp.RpcProtocolVersions
-	3,  // 8: grpc.gcp.ServerHandshakeParameters.local_identities:type_name -> grpc.gcp.Identity
-	13, // 9: grpc.gcp.StartServerHandshakeReq.handshake_parameters:type_name -> grpc.gcp.StartServerHandshakeReq.HandshakeParametersEntry
-	2,  // 10: grpc.gcp.StartServerHandshakeReq.local_endpoint:type_name -> grpc.gcp.Endpoint
-	2,  // 11: grpc.gcp.StartServerHandshakeReq.remote_endpoint:type_name -> grpc.gcp.Endpoint
-	14, // 12: grpc.gcp.StartServerHandshakeReq.rpc_versions:type_name -> grpc.gcp.RpcProtocolVersions
-	4,  // 13: grpc.gcp.HandshakerReq.client_start:type_name -> grpc.gcp.StartClientHandshakeReq
-	6,  // 14: grpc.gcp.HandshakerReq.server_start:type_name -> grpc.gcp.StartServerHandshakeReq
-	7,  // 15: grpc.gcp.HandshakerReq.next:type_name -> grpc.gcp.NextHandshakeMessageReq
-	3,  // 16: grpc.gcp.HandshakerResult.peer_identity:type_name -> grpc.gcp.Identity
-	3,  // 17: grpc.gcp.HandshakerResult.local_identity:type_name -> grpc.gcp.Identity
-	14, // 18: grpc.gcp.HandshakerResult.peer_rpc_versions:type_name -> grpc.gcp.RpcProtocolVersions
-	9,  // 19: grpc.gcp.HandshakerResp.result:type_name -> grpc.gcp.HandshakerResult
-	10, // 20: grpc.gcp.HandshakerResp.status:type_name -> grpc.gcp.HandshakerStatus
-	5,  // 21: grpc.gcp.StartServerHandshakeReq.HandshakeParametersEntry.value:type_name -> grpc.gcp.ServerHandshakeParameters
-	8,  // 22: grpc.gcp.HandshakerService.DoHandshake:input_type -> grpc.gcp.HandshakerReq
-	11, // 23: grpc.gcp.HandshakerService.DoHandshake:output_type -> grpc.gcp.HandshakerResp
-	23, // [23:24] is the sub-list for method output_type
-	22, // [22:23] is the sub-list for method input_type
-	22, // [22:22] is the sub-list for extension type_name
-	22, // [22:22] is the sub-list for extension extendee
-	0,  // [0:22] is the sub-list for field type_name
+	15, // 8: grpc.gcp.StartClientHandshakeReq.transport_protocol_preferences:type_name -> grpc.gcp.TransportProtocolPreferences
+	3,  // 9: grpc.gcp.ServerHandshakeParameters.local_identities:type_name -> grpc.gcp.Identity
+	13, // 10: grpc.gcp.StartServerHandshakeReq.handshake_parameters:type_name -> grpc.gcp.StartServerHandshakeReq.HandshakeParametersEntry
+	2,  // 11: grpc.gcp.StartServerHandshakeReq.local_endpoint:type_name -> grpc.gcp.Endpoint
+	2,  // 12: grpc.gcp.StartServerHandshakeReq.remote_endpoint:type_name -> grpc.gcp.Endpoint
+	14, // 13: grpc.gcp.StartServerHandshakeReq.rpc_versions:type_name -> grpc.gcp.RpcProtocolVersions
+	15, // 14: grpc.gcp.StartServerHandshakeReq.transport_protocol_preferences:type_name -> grpc.gcp.TransportProtocolPreferences
+	4,  // 15: grpc.gcp.HandshakerReq.client_start:type_name -> grpc.gcp.StartClientHandshakeReq
+	6,  // 16: grpc.gcp.HandshakerReq.server_start:type_name -> grpc.gcp.StartServerHandshakeReq
+	7,  // 17: grpc.gcp.HandshakerReq.next:type_name -> grpc.gcp.NextHandshakeMessageReq
+	3,  // 18: grpc.gcp.HandshakerResult.peer_identity:type_name -> grpc.gcp.Identity
+	3,  // 19: grpc.gcp.HandshakerResult.local_identity:type_name -> grpc.gcp.Identity
+	14, // 20: grpc.gcp.HandshakerResult.peer_rpc_versions:type_name -> grpc.gcp.RpcProtocolVersions
+	16, // 21: grpc.gcp.HandshakerResult.transport_protocol:type_name -> grpc.gcp.NegotiatedTransportProtocol
+	9,  // 22: grpc.gcp.HandshakerResp.result:type_name -> grpc.gcp.HandshakerResult
+	10, // 23: grpc.gcp.HandshakerResp.status:type_name -> grpc.gcp.HandshakerStatus
+	5,  // 24: grpc.gcp.StartServerHandshakeReq.HandshakeParametersEntry.value:type_name -> grpc.gcp.ServerHandshakeParameters
+	8,  // 25: grpc.gcp.HandshakerService.DoHandshake:input_type -> grpc.gcp.HandshakerReq
+	11, // 26: grpc.gcp.HandshakerService.DoHandshake:output_type -> grpc.gcp.HandshakerResp
+	26, // [26:27] is the sub-list for method output_type
+	25, // [25:26] is the sub-list for method input_type
+	25, // [25:25] is the sub-list for extension type_name
+	25, // [25:25] is the sub-list for extension extendee
+	0,  // [0:25] is the sub-list for field type_name
 }
 
 func init() { file_grpc_gcp_handshaker_proto_init() }
diff --git a/credentials/alts/internal/proto/grpc_gcp/transport_security_common.pb.go b/credentials/alts/internal/proto/grpc_gcp/transport_security_common.pb.go
index fcbbb8fa..cf48193c 100644
--- a/credentials/alts/internal/proto/grpc_gcp/transport_security_common.pb.go
+++ b/credentials/alts/internal/proto/grpc_gcp/transport_security_common.pb.go
@@ -144,6 +144,97 @@ func (x *RpcProtocolVersions) GetMinRpcVersion() *RpcProtocolVersions_Version {
 	return nil
 }
 
+// The ordered list of protocols that the client wishes to use, or the set
+// that the server supports.
+type TransportProtocolPreferences struct {
+	state             protoimpl.MessageState `protogen:"open.v1"`
+	TransportProtocol []string               `protobuf:"bytes,1,rep,name=transport_protocol,json=transportProtocol,proto3" json:"transport_protocol,omitempty"`
+	unknownFields     protoimpl.UnknownFields
+	sizeCache         protoimpl.SizeCache
+}
+
+func (x *TransportProtocolPreferences) Reset() {
+	*x = TransportProtocolPreferences{}
+	mi := &file_grpc_gcp_transport_security_common_proto_msgTypes[1]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *TransportProtocolPreferences) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*TransportProtocolPreferences) ProtoMessage() {}
+
+func (x *TransportProtocolPreferences) ProtoReflect() protoreflect.Message {
+	mi := &file_grpc_gcp_transport_security_common_proto_msgTypes[1]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use TransportProtocolPreferences.ProtoReflect.Descriptor instead.
+func (*TransportProtocolPreferences) Descriptor() ([]byte, []int) {
+	return file_grpc_gcp_transport_security_common_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *TransportProtocolPreferences) GetTransportProtocol() []string {
+	if x != nil {
+		return x.TransportProtocol
+	}
+	return nil
+}
+
+// The negotiated transport protocol.
+type NegotiatedTransportProtocol struct {
+	state             protoimpl.MessageState `protogen:"open.v1"`
+	TransportProtocol string                 `protobuf:"bytes,1,opt,name=transport_protocol,json=transportProtocol,proto3" json:"transport_protocol,omitempty"`
+	unknownFields     protoimpl.UnknownFields
+	sizeCache         protoimpl.SizeCache
+}
+
+func (x *NegotiatedTransportProtocol) Reset() {
+	*x = NegotiatedTransportProtocol{}
+	mi := &file_grpc_gcp_transport_security_common_proto_msgTypes[2]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *NegotiatedTransportProtocol) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*NegotiatedTransportProtocol) ProtoMessage() {}
+
+func (x *NegotiatedTransportProtocol) ProtoReflect() protoreflect.Message {
+	mi := &file_grpc_gcp_transport_security_common_proto_msgTypes[2]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use NegotiatedTransportProtocol.ProtoReflect.Descriptor instead.
+func (*NegotiatedTransportProtocol) Descriptor() ([]byte, []int) {
+	return file_grpc_gcp_transport_security_common_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *NegotiatedTransportProtocol) GetTransportProtocol() string {
+	if x != nil {
+		return x.TransportProtocol
+	}
+	return ""
+}
+
 // RPC version contains a major version and a minor version.
 type RpcProtocolVersions_Version struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
@@ -155,7 +246,7 @@ type RpcProtocolVersions_Version struct {
 
 func (x *RpcProtocolVersions_Version) Reset() {
 	*x = RpcProtocolVersions_Version{}
-	mi := &file_grpc_gcp_transport_security_common_proto_msgTypes[1]
+	mi := &file_grpc_gcp_transport_security_common_proto_msgTypes[3]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -167,7 +258,7 @@ func (x *RpcProtocolVersions_Version) String() string {
 func (*RpcProtocolVersions_Version) ProtoMessage() {}
 
 func (x *RpcProtocolVersions_Version) ProtoReflect() protoreflect.Message {
-	mi := &file_grpc_gcp_transport_security_common_proto_msgTypes[1]
+	mi := &file_grpc_gcp_transport_security_common_proto_msgTypes[3]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -207,7 +298,11 @@ const file_grpc_gcp_transport_security_common_proto_rawDesc = "" +
 	"\x0fmin_rpc_version\x18\x02 \x01(\v2%.grpc.gcp.RpcProtocolVersions.VersionR\rminRpcVersion\x1a5\n" +
 	"\aVersion\x12\x14\n" +
 	"\x05major\x18\x01 \x01(\rR\x05major\x12\x14\n" +
-	"\x05minor\x18\x02 \x01(\rR\x05minor*Q\n" +
+	"\x05minor\x18\x02 \x01(\rR\x05minor\"M\n" +
+	"\x1cTransportProtocolPreferences\x12-\n" +
+	"\x12transport_protocol\x18\x01 \x03(\tR\x11transportProtocol\"L\n" +
+	"\x1bNegotiatedTransportProtocol\x12-\n" +
+	"\x12transport_protocol\x18\x01 \x01(\tR\x11transportProtocol*Q\n" +
 	"\rSecurityLevel\x12\x11\n" +
 	"\rSECURITY_NONE\x10\x00\x12\x12\n" +
 	"\x0eINTEGRITY_ONLY\x10\x01\x12\x19\n" +
@@ -227,15 +322,17 @@ func file_grpc_gcp_transport_security_common_proto_rawDescGZIP() []byte {
 }
 
 var file_grpc_gcp_transport_security_common_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
-var file_grpc_gcp_transport_security_common_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
+var file_grpc_gcp_transport_security_common_proto_msgTypes = make([]protoimpl.MessageInfo, 4)
 var file_grpc_gcp_transport_security_common_proto_goTypes = []any{
-	(SecurityLevel)(0),                  // 0: grpc.gcp.SecurityLevel
-	(*RpcProtocolVersions)(nil),         // 1: grpc.gcp.RpcProtocolVersions
-	(*RpcProtocolVersions_Version)(nil), // 2: grpc.gcp.RpcProtocolVersions.Version
+	(SecurityLevel)(0),                   // 0: grpc.gcp.SecurityLevel
+	(*RpcProtocolVersions)(nil),          // 1: grpc.gcp.RpcProtocolVersions
+	(*TransportProtocolPreferences)(nil), // 2: grpc.gcp.TransportProtocolPreferences
+	(*NegotiatedTransportProtocol)(nil),  // 3: grpc.gcp.NegotiatedTransportProtocol
+	(*RpcProtocolVersions_Version)(nil),  // 4: grpc.gcp.RpcProtocolVersions.Version
 }
 var file_grpc_gcp_transport_security_common_proto_depIdxs = []int32{
-	2, // 0: grpc.gcp.RpcProtocolVersions.max_rpc_version:type_name -> grpc.gcp.RpcProtocolVersions.Version
-	2, // 1: grpc.gcp.RpcProtocolVersions.min_rpc_version:type_name -> grpc.gcp.RpcProtocolVersions.Version
+	4, // 0: grpc.gcp.RpcProtocolVersions.max_rpc_version:type_name -> grpc.gcp.RpcProtocolVersions.Version
+	4, // 1: grpc.gcp.RpcProtocolVersions.min_rpc_version:type_name -> grpc.gcp.RpcProtocolVersions.Version
 	2, // [2:2] is the sub-list for method output_type
 	2, // [2:2] is the sub-list for method input_type
 	2, // [2:2] is the sub-list for extension type_name
@@ -254,7 +351,7 @@ func file_grpc_gcp_transport_security_common_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_grpc_gcp_transport_security_common_proto_rawDesc), len(file_grpc_gcp_transport_security_common_proto_rawDesc)),
 			NumEnums:      1,
-			NumMessages:   2,
+			NumMessages:   4,
 			NumExtensions: 0,
 			NumServices:   0,
 		},
diff --git a/credentials/credentials.go b/credentials/credentials.go
index 6bba20f3..a63ab606 100644
--- a/credentials/credentials.go
+++ b/credentials/credentials.go
@@ -120,6 +120,20 @@ type AuthInfo interface {
 	AuthType() string
 }
 
+// AuthorityValidator validates the authority used to override the `:authority`
+// header. This is an optional interface that implementations of AuthInfo can
+// implement if they support per-RPC authority overrides. It is invoked when the
+// application attempts to override the HTTP/2 `:authority` header using the
+// CallAuthority call option.
+type AuthorityValidator interface {
+	// ValidateAuthority checks the authority value used to override the
+	// `:authority` header. The authority parameter is the override value
+	// provided by the application via the CallAuthority option. This value
+	// typically corresponds to the server hostname or endpoint the RPC is
+	// targeting. It returns non-nil error if the validation fails.
+	ValidateAuthority(authority string) error
+}
+
 // ErrConnDispatched indicates that rawConn has been dispatched out of gRPC
 // and the caller should not close rawConn.
 var ErrConnDispatched = errors.New("credentials: rawConn is dispatched out of gRPC")
diff --git a/credentials/insecure/insecure.go b/credentials/insecure/insecure.go
index 4c805c64..f45c04f7 100644
--- a/credentials/insecure/insecure.go
+++ b/credentials/insecure/insecure.go
@@ -71,6 +71,12 @@ func (info) AuthType() string {
 	return "insecure"
 }
 
+// ValidateAuthority allows any value to be overridden for the :authority
+// header.
+func (info) ValidateAuthority(string) error {
+	return nil
+}
+
 // insecureBundle implements an insecure bundle.
 // An insecure bundle provides a thin wrapper around insecureTC to support
 // the credentials.Bundle interface.
diff --git a/credentials/tls.go b/credentials/tls.go
index bd5fe22b..20f65f7b 100644
--- a/credentials/tls.go
+++ b/credentials/tls.go
@@ -22,6 +22,7 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"errors"
 	"fmt"
 	"net"
 	"net/url"
@@ -50,6 +51,21 @@ func (t TLSInfo) AuthType() string {
 	return "tls"
 }
 
+// ValidateAuthority validates the provided authority being used to override the
+// :authority header by verifying it against the peer certificates. It returns a
+// non-nil error if the validation fails.
+func (t TLSInfo) ValidateAuthority(authority string) error {
+	var errs []error
+	for _, cert := range t.State.PeerCertificates {
+		var err error
+		if err = cert.VerifyHostname(authority); err == nil {
+			return nil
+		}
+		errs = append(errs, err)
+	}
+	return fmt.Errorf("credentials: invalid authority %q: %v", authority, errors.Join(errs...))
+}
+
 // cipherSuiteLookup returns the string version of a TLS cipher suite ID.
 func cipherSuiteLookup(cipherSuiteID uint16) string {
 	for _, s := range tls.CipherSuites() {
diff --git a/internal/transport/http2_client.go b/internal/transport/http2_client.go
index 4c2d3057..32047128 100644
--- a/internal/transport/http2_client.go
+++ b/internal/transport/http2_client.go
@@ -749,6 +749,25 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*ClientS
 		callHdr = &newCallHdr
 	}
 
+	// The authority specified via the `CallAuthority` CallOption takes the
+	// highest precedence when determining the `:authority` header. It overrides
+	// any value present in the Host field of CallHdr. Before applying this
+	// override, the authority string is validated. If the credentials do not
+	// implement the AuthorityValidator interface, or if validation fails, the
+	// RPC is failed with a status code of `UNAVAILABLE`.
+	if callHdr.Authority != "" {
+		auth, ok := t.authInfo.(credentials.AuthorityValidator)
+		if !ok {
+			return nil, &NewStreamError{Err: status.Errorf(codes.Unavailable, "credentials type %q does not implement the AuthorityValidator interface, but authority override specified with CallAuthority call option", t.authInfo.AuthType())}
+		}
+		if err := auth.ValidateAuthority(callHdr.Authority); err != nil {
+			return nil, &NewStreamError{Err: status.Errorf(codes.Unavailable, "failed to validate authority %q : %v", callHdr.Authority, err)}
+		}
+		newCallHdr := *callHdr
+		newCallHdr.Host = callHdr.Authority
+		callHdr = &newCallHdr
+	}
+
 	headerFields, err := t.createHeaderFields(ctx, callHdr)
 	if err != nil {
 		return nil, &NewStreamError{Err: err, AllowTransparentRetry: false}
diff --git a/internal/transport/transport.go b/internal/transport/transport.go
index af4a4aea..1730a639 100644
--- a/internal/transport/transport.go
+++ b/internal/transport/transport.go
@@ -540,6 +540,11 @@ type CallHdr struct {
 	PreviousAttempts int // value of grpc-previous-rpc-attempts header to set
 
 	DoneFunc func() // called when the stream is finished
+
+	// Authority is used to explicitly override the `:authority` header. If set,
+	// this value takes precedence over the Host field and will be used as the
+	// value for the `:authority` header.
+	Authority string
 }
 
 // ClientTransport is the common interface for all gRPC client-side transport
diff --git a/rpc_util.go b/rpc_util.go
index ad20e9df..47ea09f5 100644
--- a/rpc_util.go
+++ b/rpc_util.go
@@ -160,6 +160,7 @@ type callInfo struct {
 	codec                 baseCodec
 	maxRetryRPCBufferSize int
 	onFinish              []func(err error)
+	authority             string
 }
 
 func defaultCallInfo() *callInfo {
@@ -365,6 +366,36 @@ func (o MaxRecvMsgSizeCallOption) before(c *callInfo) error {
 }
 func (o MaxRecvMsgSizeCallOption) after(*callInfo, *csAttempt) {}
 
+// CallAuthority returns a CallOption that sets the HTTP/2 :authority header of
+// an RPC to the specified value. When using CallAuthority, the credentials in
+// use must implement the AuthorityValidator interface.
+//
+// # Experimental
+//
+// Notice: This API is EXPERIMENTAL and may be changed or removed in a later
+// release.
+func CallAuthority(authority string) CallOption {
+	return AuthorityOverrideCallOption{Authority: authority}
+}
+
+// AuthorityOverrideCallOption is a CallOption that indicates the HTTP/2
+// :authority header value to use for the call.
+//
+// # Experimental
+//
+// Notice: This type is EXPERIMENTAL and may be changed or removed in a later
+// release.
+type AuthorityOverrideCallOption struct {
+	Authority string
+}
+
+func (o AuthorityOverrideCallOption) before(c *callInfo) error {
+	c.authority = o.Authority
+	return nil
+}
+
+func (o AuthorityOverrideCallOption) after(*callInfo, *csAttempt) {}
+
 // MaxCallSendMsgSize returns a CallOption which sets the maximum message size
 // in bytes the client can send. If this is not set, gRPC uses the default
 // `math.MaxInt32`.
diff --git a/stream.go b/stream.go
index 01e66c1e..bbbcbc72 100644
--- a/stream.go
+++ b/stream.go
@@ -297,6 +297,7 @@ func newClientStreamWithParams(ctx context.Context, desc *StreamDesc, cc *Client
 		Method:         method,
 		ContentSubtype: callInfo.contentSubtype,
 		DoneFunc:       doneFunc,
+		Authority:      callInfo.authority,
 	}
 
 	// Set our outgoing compression according to the UseCompressor CallOption, if
diff --git a/xds/internal/clients/lrsclient/load_store.go b/xds/internal/clients/lrsclient/load_store.go
index d52db0c7..fd363ad6 100644
--- a/xds/internal/clients/lrsclient/load_store.go
+++ b/xds/internal/clients/lrsclient/load_store.go
@@ -1,5 +1,3 @@
-//revive:disable:unused-parameter
-
 /*
  *
  * Copyright 2025 gRPC authors.
@@ -20,7 +18,12 @@
 
 package lrsclient
 
-import "context"
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"time"
+)
 
 // A LoadStore aggregates loads for multiple clusters and services that are
 // intended to be reported via LRS.
@@ -30,51 +33,394 @@ import "context"
 //
 // It is safe for concurrent use.
 type LoadStore struct {
+	stop func(ctx context.Context) // Function to call to Stop the LoadStore
+
+	// mu only protects the map (2 layers). The read/write to
+	// *PerClusterReporter doesn't need to hold the mu.
+	mu sync.Mutex
+	// clusters is a map with cluster name as the key. The second layer is a
+	// map with service name as the key. Each value (PerClusterReporter)
+	// contains data for a (cluster, service) pair.
+	//
+	// Note that new entries are added to this map, but never removed. This is
+	// potentially a memory leak. But the memory is allocated for each new
+	// (cluster,service) pair, and the memory allocated is just pointers and
+	// maps. So this shouldn't get too bad.
+	clusters map[string]map[string]*PerClusterReporter
 }
 
-// Stop stops the LRS stream associated with this LoadStore.
+// newLoadStore creates a LoadStore.
+func newLoadStore() *LoadStore {
+	return &LoadStore{
+		clusters: make(map[string]map[string]*PerClusterReporter),
+	}
+}
+
+// Stop signals the LoadStore to stop reporting.
 //
-// If this LoadStore is the only one using the underlying LRS stream, the
-// stream will be closed. If other LoadStores are also using the same stream,
-// the reference count to the stream is decremented, and the stream remains
-// open until all LoadStores have called Stop().
+// Before closing the underlying LRS stream, this method may block until a
+// final load report send attempt completes or the provided context `ctx` expires.
 //
-// If this is the last LoadStore for the stream, this method makes a last
-// attempt to flush any unreported load data to the LRS server. It will either
-// wait for this attempt to complete, or for the provided context to be done
-// before canceling the LRS stream.
-func (ls *LoadStore) Stop(ctx context.Context) error {
-	panic("unimplemented")
+// The provided context must have a deadline or timeout set to prevent Stop
+// from blocking indefinitely if the final send attempt fails to complete.
+func (ls *LoadStore) Stop(ctx context.Context) {
+	ls.stop(ctx)
 }
 
 // ReporterForCluster returns the PerClusterReporter for the given cluster and
 // service.
-func (ls *LoadStore) ReporterForCluster(clusterName, serviceName string) PerClusterReporter {
-	panic("unimplemented")
+func (ls *LoadStore) ReporterForCluster(clusterName, serviceName string) *PerClusterReporter {
+	ls.mu.Lock()
+	defer ls.mu.Unlock()
+	c, ok := ls.clusters[clusterName]
+	if !ok {
+		c = make(map[string]*PerClusterReporter)
+		ls.clusters[clusterName] = c
+	}
+
+	if p, ok := c[serviceName]; ok {
+		return p
+	}
+	p := &PerClusterReporter{
+		cluster: clusterName,
+		service: serviceName,
+	}
+	c[serviceName] = p
+	return p
+}
+
+// stats returns the load data for the given cluster names. Data is returned in
+// a slice with no specific order.
+//
+// If no clusterName is given (an empty slice), all data for all known clusters
+// is returned.
+//
+// If a cluster's loadData is empty (no load to report), it's not appended to
+// the returned slice.
+func (ls *LoadStore) stats(clusterNames []string) []*loadData {
+	ls.mu.Lock()
+	defer ls.mu.Unlock()
+
+	var ret []*loadData
+	if len(clusterNames) == 0 {
+		for _, c := range ls.clusters {
+			ret = appendClusterStats(ret, c)
+		}
+		return ret
+	}
+	for _, n := range clusterNames {
+		if c, ok := ls.clusters[n]; ok {
+			ret = appendClusterStats(ret, c)
+		}
+	}
+
+	return ret
 }
 
 // PerClusterReporter records load data pertaining to a single cluster. It
 // provides methods to record call starts, finishes, server-reported loads,
 // and dropped calls.
+//
+// It is safe for concurrent use.
+//
+// TODO(purnesh42h): Use regular maps with mutexes instead of sync.Map here.
+// The latter is optimized for two common use cases: (1) when the entry for a
+// given key is only ever written once but read many times, as in caches that
+// only grow, or (2) when multiple goroutines read, write, and overwrite
+// entries for disjoint sets of keys. In these two cases, use of a Map may
+// significantly reduce lock contention compared to a Go map paired with a
+// separate Mutex or RWMutex.
+// Neither of these conditions are met here, and we should transition to a
+// regular map with a mutex for better type safety.
 type PerClusterReporter struct {
+	cluster, service string
+	drops            sync.Map // map[string]*uint64
+	localityRPCCount sync.Map // map[string]*rpcCountData
+
+	mu               sync.Mutex
+	lastLoadReportAt time.Time
 }
 
 // CallStarted records a call started in the LoadStore.
 func (p *PerClusterReporter) CallStarted(locality string) {
-	panic("unimplemented")
+	s, ok := p.localityRPCCount.Load(locality)
+	if !ok {
+		tp := newRPCCountData()
+		s, _ = p.localityRPCCount.LoadOrStore(locality, tp)
+	}
+	s.(*rpcCountData).incrInProgress()
+	s.(*rpcCountData).incrIssued()
 }
 
 // CallFinished records a call finished in the LoadStore.
 func (p *PerClusterReporter) CallFinished(locality string, err error) {
-	panic("unimplemented")
+	f, ok := p.localityRPCCount.Load(locality)
+	if !ok {
+		// The map is never cleared, only values in the map are reset. So the
+		// case where entry for call-finish is not found should never happen.
+		return
+	}
+	f.(*rpcCountData).decrInProgress()
+	if err == nil {
+		f.(*rpcCountData).incrSucceeded()
+	} else {
+		f.(*rpcCountData).incrErrored()
+	}
 }
 
 // CallServerLoad records the server load in the LoadStore.
 func (p *PerClusterReporter) CallServerLoad(locality, name string, val float64) {
-	panic("unimplemented")
+	s, ok := p.localityRPCCount.Load(locality)
+	if !ok {
+		// The map is never cleared, only values in the map are reset. So the
+		// case where entry for callServerLoad is not found should never happen.
+		return
+	}
+	s.(*rpcCountData).addServerLoad(name, val)
 }
 
 // CallDropped records a call dropped in the LoadStore.
 func (p *PerClusterReporter) CallDropped(category string) {
-	panic("unimplemented")
+	d, ok := p.drops.Load(category)
+	if !ok {
+		tp := new(uint64)
+		d, _ = p.drops.LoadOrStore(category, tp)
+	}
+	atomic.AddUint64(d.(*uint64), 1)
+}
+
+// stats returns and resets all loads reported to the store, except inProgress
+// rpc counts.
+//
+// It returns nil if the store doesn't contain any (new) data.
+func (p *PerClusterReporter) stats() *loadData {
+	sd := newLoadData(p.cluster, p.service)
+	p.drops.Range(func(key, val any) bool {
+		d := atomic.SwapUint64(val.(*uint64), 0)
+		if d == 0 {
+			return true
+		}
+		sd.totalDrops += d
+		keyStr := key.(string)
+		if keyStr != "" {
+			// Skip drops without category. They are counted in total_drops, but
+			// not in per category. One example is drops by circuit breaking.
+			sd.drops[keyStr] = d
+		}
+		return true
+	})
+	p.localityRPCCount.Range(func(key, val any) bool {
+		countData := val.(*rpcCountData)
+		succeeded := countData.loadAndClearSucceeded()
+		inProgress := countData.loadInProgress()
+		errored := countData.loadAndClearErrored()
+		issued := countData.loadAndClearIssued()
+		if succeeded == 0 && inProgress == 0 && errored == 0 && issued == 0 {
+			return true
+		}
+
+		ld := localityData{
+			requestStats: requestData{
+				succeeded:  succeeded,
+				errored:    errored,
+				inProgress: inProgress,
+				issued:     issued,
+			},
+			loadStats: make(map[string]serverLoadData),
+		}
+		countData.serverLoads.Range(func(key, val any) bool {
+			sum, count := val.(*rpcLoadData).loadAndClear()
+			if count == 0 {
+				return true
+			}
+			ld.loadStats[key.(string)] = serverLoadData{
+				count: count,
+				sum:   sum,
+			}
+			return true
+		})
+		sd.localityStats[key.(string)] = ld
+		return true
+	})
+
+	p.mu.Lock()
+	sd.reportInterval = time.Since(p.lastLoadReportAt)
+	p.lastLoadReportAt = time.Now()
+	p.mu.Unlock()
+
+	if sd.totalDrops == 0 && len(sd.drops) == 0 && len(sd.localityStats) == 0 {
+		return nil
+	}
+	return sd
+}
+
+// loadData contains all load data reported to the LoadStore since the most recent
+// call to stats().
+type loadData struct {
+	// cluster is the name of the cluster this data is for.
+	cluster string
+	// service is the name of the EDS service this data is for.
+	service string
+	// totalDrops is the total number of dropped requests.
+	totalDrops uint64
+	// drops is the number of dropped requests per category.
+	drops map[string]uint64
+	// localityStats contains load reports per locality.
+	localityStats map[string]localityData
+	// reportInternal is the duration since last time load was reported (stats()
+	// was called).
+	reportInterval time.Duration
+}
+
+// localityData contains load data for a single locality.
+type localityData struct {
+	// requestStats contains counts of requests made to the locality.
+	requestStats requestData
+	// loadStats contains server load data for requests made to the locality,
+	// indexed by the load type.
+	loadStats map[string]serverLoadData
+}
+
+// requestData contains request counts.
+type requestData struct {
+	// succeeded is the number of succeeded requests.
+	succeeded uint64
+	// errored is the number of requests which ran into errors.
+	errored uint64
+	// inProgress is the number of requests in flight.
+	inProgress uint64
+	// issued is the total number requests that were sent.
+	issued uint64
+}
+
+// serverLoadData contains server load data.
+type serverLoadData struct {
+	// count is the number of load reports.
+	count uint64
+	// sum is the total value of all load reports.
+	sum float64
+}
+
+// appendClusterStats gets the Data for all the given clusters, append to ret,
+// and return the new slice.
+//
+// Data is only appended to ret if it's not empty.
+func appendClusterStats(ret []*loadData, clusters map[string]*PerClusterReporter) []*loadData {
+	for _, d := range clusters {
+		data := d.stats()
+		if data == nil {
+			// Skip this data if it doesn't contain any information.
+			continue
+		}
+		ret = append(ret, data)
+	}
+	return ret
+}
+
+func newLoadData(cluster, service string) *loadData {
+	return &loadData{
+		cluster:       cluster,
+		service:       service,
+		drops:         make(map[string]uint64),
+		localityStats: make(map[string]localityData),
+	}
+}
+
+type rpcCountData struct {
+	// Only atomic accesses are allowed for the fields.
+	succeeded  *uint64
+	errored    *uint64
+	inProgress *uint64
+	issued     *uint64
+
+	// Map from load desc to load data (sum+count). Loading data from map is
+	// atomic, but updating data takes a lock, which could cause contention when
+	// multiple RPCs try to report loads for the same desc.
+	//
+	// To fix the contention, shard this map.
+	serverLoads sync.Map // map[string]*rpcLoadData
+}
+
+func newRPCCountData() *rpcCountData {
+	return &rpcCountData{
+		succeeded:  new(uint64),
+		errored:    new(uint64),
+		inProgress: new(uint64),
+		issued:     new(uint64),
+	}
+}
+
+func (rcd *rpcCountData) incrSucceeded() {
+	atomic.AddUint64(rcd.succeeded, 1)
+}
+
+func (rcd *rpcCountData) loadAndClearSucceeded() uint64 {
+	return atomic.SwapUint64(rcd.succeeded, 0)
+}
+
+func (rcd *rpcCountData) incrErrored() {
+	atomic.AddUint64(rcd.errored, 1)
+}
+
+func (rcd *rpcCountData) loadAndClearErrored() uint64 {
+	return atomic.SwapUint64(rcd.errored, 0)
+}
+
+func (rcd *rpcCountData) incrInProgress() {
+	atomic.AddUint64(rcd.inProgress, 1)
+}
+
+func (rcd *rpcCountData) decrInProgress() {
+	atomic.AddUint64(rcd.inProgress, ^uint64(0)) // atomic.Add(x, -1)
+}
+
+func (rcd *rpcCountData) loadInProgress() uint64 {
+	return atomic.LoadUint64(rcd.inProgress) // InProgress count is not clear when reading.
+}
+
+func (rcd *rpcCountData) incrIssued() {
+	atomic.AddUint64(rcd.issued, 1)
+}
+
+func (rcd *rpcCountData) loadAndClearIssued() uint64 {
+	return atomic.SwapUint64(rcd.issued, 0)
+}
+
+func (rcd *rpcCountData) addServerLoad(name string, d float64) {
+	loads, ok := rcd.serverLoads.Load(name)
+	if !ok {
+		tl := newRPCLoadData()
+		loads, _ = rcd.serverLoads.LoadOrStore(name, tl)
+	}
+	loads.(*rpcLoadData).add(d)
+}
+
+// rpcLoadData is data for server loads (from trailers or oob). Fields in this
+// struct must be updated consistently.
+//
+// The current solution is to hold a lock, which could cause contention. To fix,
+// shard serverLoads map in rpcCountData.
+type rpcLoadData struct {
+	mu    sync.Mutex
+	sum   float64
+	count uint64
+}
+
+func newRPCLoadData() *rpcLoadData {
+	return &rpcLoadData{}
+}
+
+func (rld *rpcLoadData) add(v float64) {
+	rld.mu.Lock()
+	rld.sum += v
+	rld.count++
+	rld.mu.Unlock()
+}
+
+func (rld *rpcLoadData) loadAndClear() (s float64, c uint64) {
+	rld.mu.Lock()
+	s, rld.sum = rld.sum, 0
+	c, rld.count = rld.count, 0
+	rld.mu.Unlock()
+	return s, c
 }
diff --git a/xds/internal/clients/lrsclient/logging.go b/xds/internal/clients/lrsclient/logging.go
new file mode 100644
index 00000000..032b56bd
--- /dev/null
+++ b/xds/internal/clients/lrsclient/logging.go
@@ -0,0 +1,36 @@
+/*
+ *
+ * Copyright 2025 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.
+ *
+ */
+
+package lrsclient
+
+import (
+	"fmt"
+
+	"google.golang.org/grpc/grpclog"
+	internalgrpclog "google.golang.org/grpc/internal/grpclog"
+)
+
+var logger = grpclog.Component("xds")
+
+func prefixLogger(c *LRSClient) *internalgrpclog.PrefixLogger {
+	return internalgrpclog.NewPrefixLogger(logger, clientPrefix(c))
+}
+
+func clientPrefix(c *LRSClient) string {
+	return fmt.Sprintf("[lrs-client %p] ", c)
+}
diff --git a/xds/internal/clients/lrsclient/lrs_stream.go b/xds/internal/clients/lrsclient/lrs_stream.go
new file mode 100644
index 00000000..df761d40
--- /dev/null
+++ b/xds/internal/clients/lrsclient/lrs_stream.go
@@ -0,0 +1,318 @@
+/*
+ *
+ * Copyright 2025 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.
+ */
+
+package lrsclient
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"time"
+
+	"google.golang.org/grpc/grpclog"
+	"google.golang.org/grpc/internal/backoff"
+	igrpclog "google.golang.org/grpc/internal/grpclog"
+	"google.golang.org/grpc/internal/pretty"
+	"google.golang.org/grpc/xds/internal/clients"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/durationpb"
+
+	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
+	v3lrspb "github.com/envoyproxy/go-control-plane/envoy/service/load_stats/v3"
+)
+
+// Any per-RPC level logs which print complete request or response messages
+// should be gated at this verbosity level. Other per-RPC level logs which print
+// terse output should be at `INFO` and verbosity 2.
+const perRPCVerbosityLevel = 9
+
+// streamImpl provides all the functionality associated with an LRS (Load
+// Reporting Service) stream on the client-side. It manages the lifecycle of
+// the LRS stream, including starting, stopping, and retrying the stream. It
+// also provides a LoadStore that can be used to report load, with a Stop
+// function that should be called when the load reporting is no longer
+// needed.
+type streamImpl struct {
+	// The following fields are initialized when a stream instance is created
+	// and are read-only afterwards, and hence can be accessed without a mutex.
+	transport clients.Transport       // Transport to use for LRS stream.
+	backoff   func(int) time.Duration // Backoff for retries, after stream failures.
+	nodeProto *v3corepb.Node          // Identifies the gRPC application.
+	doneCh    chan struct{}           // To notify exit of LRS goroutine.
+	logger    *igrpclog.PrefixLogger
+
+	cancelStream context.CancelFunc // Cancel the stream. If nil, the stream is not active.
+	loadStore    *LoadStore         // LoadStore returned to user for pushing loads.
+
+	finalSendRequest chan struct{} // To request for the final attempt to send loads.
+	finalSendDone    chan error    // To signal completion of the final attempt of sending loads.
+}
+
+// streamOpts holds the options for creating an lrsStream.
+type streamOpts struct {
+	transport clients.Transport       // xDS transport to create the stream on.
+	backoff   func(int) time.Duration // Backoff for retries, after stream failures.
+	nodeProto *v3corepb.Node          // Node proto to identify the gRPC application.
+	logPrefix string                  // Prefix to be used for log messages.
+}
+
+// newStreamImpl creates a new StreamImpl with the provided options.
+//
+// The actual streaming RPC call is initiated when the first call to ReportLoad
+// is made, and is terminated when the last call to ReportLoad is canceled.
+func newStreamImpl(opts streamOpts) *streamImpl {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	lrs := &streamImpl{
+		transport:        opts.transport,
+		backoff:          opts.backoff,
+		nodeProto:        opts.nodeProto,
+		cancelStream:     cancel,
+		doneCh:           make(chan struct{}),
+		finalSendRequest: make(chan struct{}, 1),
+		finalSendDone:    make(chan error, 1),
+	}
+
+	l := grpclog.Component("xds")
+	lrs.logger = igrpclog.NewPrefixLogger(l, opts.logPrefix+fmt.Sprintf("[lrs-stream %p] ", lrs))
+	lrs.loadStore = newLoadStore()
+	go lrs.runner(ctx)
+	return lrs
+}
+
+// runner is responsible for managing the lifetime of an LRS streaming call. It
+// creates the stream, sends the initial LoadStatsRequest, receives the first
+// LoadStatsResponse, and then starts a goroutine to periodically send
+// LoadStatsRequests. The runner will restart the stream if it encounters any
+// errors.
+func (lrs *streamImpl) runner(ctx context.Context) {
+	defer close(lrs.doneCh)
+
+	// This feature indicates that the client supports the
+	// LoadStatsResponse.send_all_clusters field in the LRS response.
+	node := proto.Clone(lrs.nodeProto).(*v3corepb.Node)
+	node.ClientFeatures = append(node.ClientFeatures, "envoy.lrs.supports_send_all_clusters")
+
+	runLoadReportStream := func() error {
+		// streamCtx is created and canceled in case we terminate the stream
+		// early for any reason, to avoid gRPC-Go leaking the RPC's monitoring
+		// goroutine.
+		streamCtx, cancel := context.WithCancel(ctx)
+		defer cancel()
+
+		stream, err := lrs.transport.NewStream(streamCtx, "/envoy.service.load_stats.v3.LoadReportingService/StreamLoadStats")
+		if err != nil {
+			lrs.logger.Warningf("Failed to create new LRS streaming RPC: %v", err)
+			return nil
+		}
+		if lrs.logger.V(2) {
+			lrs.logger.Infof("LRS stream created")
+		}
+
+		if err := lrs.sendFirstLoadStatsRequest(stream, node); err != nil {
+			lrs.logger.Warningf("Sending first LRS request failed: %v", err)
+			return nil
+		}
+
+		clusters, interval, err := lrs.recvFirstLoadStatsResponse(stream)
+		if err != nil {
+			lrs.logger.Warningf("Reading from LRS streaming RPC failed: %v", err)
+			return nil
+		}
+
+		// We reset backoff state when we successfully receive at least one
+		// message from the server.
+		lrs.sendLoads(streamCtx, stream, clusters, interval)
+		return backoff.ErrResetBackoff
+	}
+	backoff.RunF(ctx, runLoadReportStream, lrs.backoff)
+}
+
+// sendLoads is responsible for periodically sending load reports to the LRS
+// server at the specified interval for the specified clusters, until the passed
+// in context is canceled.
+func (lrs *streamImpl) sendLoads(ctx context.Context, stream clients.Stream, clusterNames []string, interval time.Duration) {
+	tick := time.NewTicker(interval)
+	defer tick.Stop()
+	for {
+		select {
+		case <-tick.C:
+		case <-ctx.Done():
+			return
+		case <-lrs.finalSendRequest:
+			var finalSendErr error
+			if lrs.logger.V(2) {
+				lrs.logger.Infof("Final send request received. Attempting final LRS report.")
+			}
+			if err := lrs.sendLoadStatsRequest(stream, lrs.loadStore.stats(clusterNames)); err != nil {
+				lrs.logger.Warningf("Failed to send final load report. Writing to LRS stream failed: %v", err)
+				finalSendErr = err
+			}
+			if lrs.logger.V(2) {
+				lrs.logger.Infof("Successfully sent final load report.")
+			}
+			lrs.finalSendDone <- finalSendErr
+			return
+		}
+
+		if err := lrs.sendLoadStatsRequest(stream, lrs.loadStore.stats(clusterNames)); err != nil {
+			lrs.logger.Warningf("Failed to send periodic load report. Writing to LRS stream failed: %v", err)
+			return
+		}
+	}
+}
+
+func (lrs *streamImpl) sendFirstLoadStatsRequest(stream clients.Stream, node *v3corepb.Node) error {
+	req := &v3lrspb.LoadStatsRequest{Node: node}
+	if lrs.logger.V(perRPCVerbosityLevel) {
+		lrs.logger.Infof("Sending initial LoadStatsRequest: %s", pretty.ToJSON(req))
+	}
+	msg, err := proto.Marshal(req)
+	if err != nil {
+		lrs.logger.Warningf("Failed to marshal LoadStatsRequest: %v", err)
+		return err
+	}
+	err = stream.Send(msg)
+	if err == io.EOF {
+		return getStreamError(stream)
+	}
+	return err
+}
+
+// recvFirstLoadStatsResponse receives the first LoadStatsResponse from the LRS
+// server.  Returns the following:
+//   - a list of cluster names requested by the server or an empty slice if the
+//     server requested for load from all clusters
+//   - the load reporting interval, and
+//   - any error encountered
+func (lrs *streamImpl) recvFirstLoadStatsResponse(stream clients.Stream) ([]string, time.Duration, error) {
+	r, err := stream.Recv()
+	if err != nil {
+		return nil, 0, fmt.Errorf("lrs: failed to receive first LoadStatsResponse: %v", err)
+	}
+	var resp v3lrspb.LoadStatsResponse
+	if err := proto.Unmarshal(r, &resp); err != nil {
+		if lrs.logger.V(2) {
+			lrs.logger.Infof("Failed to unmarshal response to LoadStatsResponse: %v", err)
+		}
+		return nil, time.Duration(0), fmt.Errorf("lrs: unexpected message type %T", r)
+	}
+	if lrs.logger.V(perRPCVerbosityLevel) {
+		lrs.logger.Infof("Received first LoadStatsResponse: %s", pretty.ToJSON(&resp))
+	}
+
+	internal := resp.GetLoadReportingInterval()
+	if internal.CheckValid() != nil {
+		return nil, 0, fmt.Errorf("lrs: invalid load_reporting_interval: %v", err)
+	}
+	loadReportingInterval := internal.AsDuration()
+
+	clusters := resp.Clusters
+	if resp.SendAllClusters {
+		// Return an empty slice to send stats for all clusters.
+		clusters = []string{}
+	}
+
+	return clusters, loadReportingInterval, nil
+}
+
+func (lrs *streamImpl) sendLoadStatsRequest(stream clients.Stream, loads []*loadData) error {
+	clusterStats := make([]*v3endpointpb.ClusterStats, 0, len(loads))
+	for _, sd := range loads {
+		droppedReqs := make([]*v3endpointpb.ClusterStats_DroppedRequests, 0, len(sd.drops))
+		for category, count := range sd.drops {
+			droppedReqs = append(droppedReqs, &v3endpointpb.ClusterStats_DroppedRequests{
+				Category:     category,
+				DroppedCount: count,
+			})
+		}
+		localityStats := make([]*v3endpointpb.UpstreamLocalityStats, 0, len(sd.localityStats))
+		for l, localityData := range sd.localityStats {
+			lid, err := localityFromString(l)
+			if err != nil {
+				return err
+			}
+			loadMetricStats := make([]*v3endpointpb.EndpointLoadMetricStats, 0, len(localityData.loadStats))
+			for name, loadData := range localityData.loadStats {
+				loadMetricStats = append(loadMetricStats, &v3endpointpb.EndpointLoadMetricStats{
+					MetricName:                    name,
+					NumRequestsFinishedWithMetric: loadData.count,
+					TotalMetricValue:              loadData.sum,
+				})
+			}
+			localityStats = append(localityStats, &v3endpointpb.UpstreamLocalityStats{
+				Locality: &v3corepb.Locality{
+					Region:  lid.Region,
+					Zone:    lid.Zone,
+					SubZone: lid.SubZone,
+				},
+				TotalSuccessfulRequests: localityData.requestStats.succeeded,
+				TotalRequestsInProgress: localityData.requestStats.inProgress,
+				TotalErrorRequests:      localityData.requestStats.errored,
+				TotalIssuedRequests:     localityData.requestStats.issued,
+				LoadMetricStats:         loadMetricStats,
+				UpstreamEndpointStats:   nil, // TODO: populate for per endpoint loads.
+			})
+		}
+
+		clusterStats = append(clusterStats, &v3endpointpb.ClusterStats{
+			ClusterName:           sd.cluster,
+			ClusterServiceName:    sd.service,
+			UpstreamLocalityStats: localityStats,
+			TotalDroppedRequests:  sd.totalDrops,
+			DroppedRequests:       droppedReqs,
+			LoadReportInterval:    durationpb.New(sd.reportInterval),
+		})
+	}
+
+	req := &v3lrspb.LoadStatsRequest{ClusterStats: clusterStats}
+	if lrs.logger.V(perRPCVerbosityLevel) {
+		lrs.logger.Infof("Sending LRS loads: %s", pretty.ToJSON(req))
+	}
+	msg, err := proto.Marshal(req)
+	if err != nil {
+		if lrs.logger.V(2) {
+			lrs.logger.Infof("Failed to marshal LoadStatsRequest: %v", err)
+		}
+		return err
+	}
+	err = stream.Send(msg)
+	if err == io.EOF {
+		return getStreamError(stream)
+	}
+	return err
+}
+
+func getStreamError(stream clients.Stream) error {
+	for {
+		if _, err := stream.Recv(); err != nil {
+			return err
+		}
+	}
+}
+
+// localityFromString converts a json representation of locality, into a
+// clients.Locality struct.
+func localityFromString(s string) (ret clients.Locality, _ error) {
+	err := json.Unmarshal([]byte(s), &ret)
+	if err != nil {
+		return clients.Locality{}, fmt.Errorf("%s is not a well formatted locality, error: %v", s, err)
+	}
+	return ret, nil
+}
diff --git a/xds/internal/clients/lrsclient/lrsclient.go b/xds/internal/clients/lrsclient/lrsclient.go
index 5bd8aa60..af163e22 100644
--- a/xds/internal/clients/lrsclient/lrsclient.go
+++ b/xds/internal/clients/lrsclient/lrsclient.go
@@ -21,19 +21,163 @@
 // See: https://www.envoyproxy.io/docs/envoy/latest/api-v3/service/load_stats/v3/lrs.proto
 package lrsclient
 
-import "google.golang.org/grpc/xds/internal/clients"
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc/grpclog"
+	igrpclog "google.golang.org/grpc/internal/grpclog"
+	"google.golang.org/grpc/xds/internal/clients"
+	clientsinternal "google.golang.org/grpc/xds/internal/clients/internal"
+	"google.golang.org/grpc/xds/internal/clients/internal/backoff"
+)
+
+const (
+	clientFeatureNoOverprovisioning = "envoy.lb.does_not_support_overprovisioning"
+	clientFeatureResourceWrapper    = "xds.config.resource-in-sotw"
+)
+
+var (
+	defaultExponentialBackoff = backoff.DefaultExponential.Backoff
+)
 
 // LRSClient is an LRS (Load Reporting Service) client.
 type LRSClient struct {
+	transportBuilder clients.TransportBuilder
+	node             clients.Node
+	backoff          func(int) time.Duration // Backoff for LRS stream failures.
+	logger           *igrpclog.PrefixLogger
+
+	// The LRSClient owns a bunch of streams to individual LRS servers.
+	//
+	// Once all references to a stream are dropped, the stream is closed.
+	mu         sync.Mutex
+	lrsStreams map[clients.ServerIdentifier]*streamImpl // Map from server config to in-use streamImpls.
+	lrsRefs    map[clients.ServerIdentifier]int         // Map from server config to number of references.
 }
 
 // New returns a new LRS Client configured with the provided config.
-func New(_ Config) (*LRSClient, error) {
-	panic("unimplemented")
+func New(config Config) (*LRSClient, error) {
+	switch {
+	case config.Node.ID == "":
+		return nil, errors.New("lrsclient: node ID in node is empty")
+	case config.TransportBuilder == nil:
+		return nil, errors.New("lrsclient: transport builder is nil")
+	}
+
+	c := &LRSClient{
+		transportBuilder: config.TransportBuilder,
+		node:             config.Node,
+		backoff:          defaultExponentialBackoff,
+		lrsStreams:       make(map[clients.ServerIdentifier]*streamImpl),
+		lrsRefs:          make(map[clients.ServerIdentifier]int),
+	}
+	c.logger = prefixLogger(c)
+	return c, nil
+}
+
+// ReportLoad creates and returns a LoadStore for the caller to report loads
+// using a LoadReportingStream.
+//
+// Caller must call Stop on the returned LoadStore when they are done reporting
+// load to this server.
+func (c *LRSClient) ReportLoad(si clients.ServerIdentifier) (*LoadStore, error) {
+	lrs, err := c.getOrCreateLRSStream(si)
+	if err != nil {
+		return nil, err
+	}
+	return lrs.loadStore, nil
 }
 
-// ReportLoad creates a new load reporting stream for the provided server. It
-// creates and returns a LoadStore for the caller to report loads.
-func (*LRSClient) ReportLoad(_ clients.ServerIdentifier) *LoadStore {
-	panic("unimplemented")
+// getOrCreateLRSStream returns an lrs stream for the given server identifier.
+//
+// If an active lrs stream exists for the given server identifier, it is
+// returned. Otherwise, a new lrs stream is created and returned.
+func (c *LRSClient) getOrCreateLRSStream(serverIdentifier clients.ServerIdentifier) (*streamImpl, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.logger.V(2) {
+		c.logger.Infof("Received request for a reference to an lrs stream for server identifier %q", serverIdentifier)
+	}
+
+	// Use an existing stream, if one exists for this server identifier.
+	if s, ok := c.lrsStreams[serverIdentifier]; ok {
+		if c.logger.V(2) {
+			c.logger.Infof("Reusing an existing lrs stream for server identifier %q", serverIdentifier)
+		}
+		c.lrsRefs[serverIdentifier]++
+		return s, nil
+	}
+
+	if c.logger.V(2) {
+		c.logger.Infof("Creating a new lrs stream for server identifier %q", serverIdentifier)
+	}
+
+	l := grpclog.Component("xds")
+	logPrefix := clientPrefix(c)
+	c.logger = igrpclog.NewPrefixLogger(l, logPrefix)
+
+	// Create a new transport and create a new lrs stream, and add it to the
+	// map of lrs streams.
+	tr, err := c.transportBuilder.Build(serverIdentifier)
+	if err != nil {
+		return nil, fmt.Errorf("lrsclient: failed to create transport for server identifier %s: %v", serverIdentifier, err)
+	}
+
+	nodeProto := clientsinternal.NodeProto(c.node)
+	nodeProto.ClientFeatures = []string{clientFeatureNoOverprovisioning, clientFeatureResourceWrapper}
+	lrs := newStreamImpl(streamOpts{
+		transport: tr,
+		backoff:   c.backoff,
+		nodeProto: nodeProto,
+		logPrefix: logPrefix,
+	})
+
+	// Register a stop function that decrements the reference count, stops
+	// the LRS stream when the last reference is removed and closes the
+	// transport and removes the lrs stream and its references from the
+	// respective maps. Before closing the stream, it waits for the provided
+	// context to be done (timeout or cancellation).
+	stop := func(ctx context.Context) {
+		c.mu.Lock()
+		defer c.mu.Unlock()
+
+		if r, ok := c.lrsRefs[serverIdentifier]; !ok || r == 0 {
+			c.logger.Errorf("Attempting to stop already stopped StreamImpl")
+			return
+		}
+		c.lrsRefs[serverIdentifier]--
+		if c.lrsRefs[serverIdentifier] != 0 {
+			return
+		}
+
+		lrs.finalSendRequest <- struct{}{}
+
+		select {
+		case err := <-lrs.finalSendDone:
+			if err != nil {
+				c.logger.Warningf("Final send attempt failed: %v", err)
+			}
+		case <-ctx.Done():
+			c.logger.Warningf("Context canceled before finishing the final send attempt: %v", err)
+		}
+
+		lrs.cancelStream()
+		lrs.cancelStream = nil
+		lrs.logger.Infof("Stopping LRS stream")
+		<-lrs.doneCh
+
+		delete(c.lrsStreams, serverIdentifier)
+		tr.Close()
+	}
+	lrs.loadStore.stop = stop
+
+	c.lrsStreams[serverIdentifier] = lrs
+	c.lrsRefs[serverIdentifier] = 1
+
+	return lrs, nil
 }
diff --git a/xds/internal/xdsclient/xdsresource/unmarshal_lds.go b/xds/internal/xdsclient/xdsresource/unmarshal_lds.go
index 1b0d4599..475300ce 100644
--- a/xds/internal/xdsclient/xdsresource/unmarshal_lds.go
+++ b/xds/internal/xdsclient/xdsresource/unmarshal_lds.go
@@ -250,7 +250,7 @@ func processServerSideListener(lis *v3listenerpb.Listener) (*ListenerUpdate, err
 	if n := len(lis.ListenerFilters); n != 0 {
 		return nil, fmt.Errorf("unsupported field 'listener_filters' contains %d entries", n)
 	}
-	if useOrigDst := lis.GetUseOriginalDst(); useOrigDst != nil && useOrigDst.GetValue() {
+	if lis.GetUseOriginalDst().GetValue() {
 		return nil, errors.New("unsupported field 'use_original_dst' is present and set to true")
 	}
 	addr := lis.GetAddress()



Test patch:
diff --git a/credentials/alts/internal/conn/record_test.go b/credentials/alts/internal/conn/record_test.go
index c50fb4c8..e4992489 100644
--- a/credentials/alts/internal/conn/record_test.go
+++ b/credentials/alts/internal/conn/record_test.go
@@ -26,6 +26,7 @@ import (
 	"math"
 	"net"
 	"reflect"
+	"strings"
 	"testing"
 
 	core "google.golang.org/grpc/credentials/alts/internal"
@@ -188,6 +189,48 @@ func (s) TestLargeMsg(t *testing.T) {
 	}
 }
 
+// TestLargeRecord writes a very large ALTS record and verifies that the server
+// receives it correctly. The large ALTS record should cause the reader to
+// expand it's read buffer to hold the entire record and store the decrypted
+// message until the receiver reads all of the bytes.
+func (s) TestLargeRecord(t *testing.T) {
+	clientConn, serverConn := newConnPair(rekeyRecordProtocol, nil, nil)
+	msg := []byte(strings.Repeat("a", 2*altsReadBufferInitialSize))
+	// Increase the size of ALTS records written by the client.
+	clientConn.payloadLengthLimit = math.MaxInt32
+	if n, err := clientConn.Write(msg); n != len(msg) || err != nil {
+		t.Fatalf("Write() = %v, %v; want %v, <nil>", n, err, len(msg))
+	}
+	rcvMsg := make([]byte, len(msg))
+	if n, err := io.ReadFull(serverConn, rcvMsg); n != len(rcvMsg) || err != nil {
+		t.Fatalf("Read() = %v, %v; want %v, <nil>", n, err, len(rcvMsg))
+	}
+	if !reflect.DeepEqual(msg, rcvMsg) {
+		t.Fatalf("Write()/Server Read() = %v, want %v", rcvMsg, msg)
+	}
+}
+
+// BenchmarkLargeMessage measures the performance of ALTS conns for sending and
+// receiving a large message.
+func BenchmarkLargeMessage(b *testing.B) {
+	msgLen := 20 * 1024 * 1024 // 20 MiB
+	msg := make([]byte, msgLen)
+	rcvMsg := make([]byte, len(msg))
+	b.ResetTimer()
+	clientConn, serverConn := newConnPair(rekeyRecordProtocol, nil, nil)
+	for range b.N {
+		// Write 20 MiB 5 times to transfer a total of 100 MiB.
+		for range 5 {
+			if n, err := clientConn.Write(msg); n != len(msg) || err != nil {
+				b.Fatalf("Write() = %v, %v; want %v, <nil>", n, err, len(msg))
+			}
+			if n, err := io.ReadFull(serverConn, rcvMsg); n != len(rcvMsg) || err != nil {
+				b.Fatalf("Read() = %v, %v; want %v, <nil>", n, err, len(rcvMsg))
+			}
+		}
+	}
+}
+
 func testIncorrectMsgType(t *testing.T, rp string) {
 	// framedMsg is an empty ciphertext with correct framing but wrong
 	// message type.
diff --git a/credentials/credentials_ext_test.go b/credentials/credentials_ext_test.go
new file mode 100644
index 00000000..27da3a26
--- /dev/null
+++ b/credentials/credentials_ext_test.go
@@ -0,0 +1,332 @@
+/*
+ *
+ * Copyright 2025 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.
+ *
+ */
+
+package credentials_test
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"net"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/internal/stubserver"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+	"google.golang.org/grpc/testdata"
+
+	testgrpc "google.golang.org/grpc/interop/grpc_testing"
+	testpb "google.golang.org/grpc/interop/grpc_testing"
+)
+
+func authorityChecker(ctx context.Context, wantAuthority string) error {
+	md, ok := metadata.FromIncomingContext(ctx)
+	if !ok {
+		return status.Error(codes.InvalidArgument, "failed to parse metadata")
+	}
+	auths, ok := md[":authority"]
+	if !ok {
+		return status.Error(codes.InvalidArgument, "no authority header")
+	}
+	if len(auths) != 1 {
+		return status.Errorf(codes.InvalidArgument, "expected exactly one authority header, got %v", auths)
+	}
+	if auths[0] != wantAuthority {
+		return status.Errorf(codes.InvalidArgument, "invalid authority header %q, want %q", auths[0], wantAuthority)
+	}
+	return nil
+}
+
+// Tests the `grpc.CallAuthority` option with TLS credentials. This test verifies
+// that the provided authority is correctly propagated to the server when a
+// correct authority is used.
+func (s) TestCorrectAuthorityWithTLSCreds(t *testing.T) {
+	cert, err := tls.LoadX509KeyPair(testdata.Path("x509/server1_cert.pem"), testdata.Path("x509/server1_key.pem"))
+	if err != nil {
+		t.Fatalf("Failed to load key pair: %s", err)
+	}
+	creds, err := credentials.NewClientTLSFromFile(testdata.Path("x509/server_ca_cert.pem"), "x.test.example.com")
+	if err != nil {
+		t.Fatalf("Failed to create credentials %v", err)
+	}
+	const authority = "auth.test.example.com"
+	ss := &stubserver.StubServer{
+		EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
+			if err := authorityChecker(ctx, authority); err != nil {
+				return nil, err
+			}
+			return &testpb.Empty{}, nil
+		},
+	}
+	if err := ss.StartServer(grpc.Creds(credentials.NewServerTLSFromCert(&cert))); err != nil {
+		t.Fatalf("Error starting endpoint server: %v", err)
+	}
+	defer ss.Stop()
+
+	cc, err := grpc.NewClient(ss.Address, grpc.WithTransportCredentials(creds))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(%q) = %v", ss.Address, err)
+	}
+	defer cc.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	if _, err = testgrpc.NewTestServiceClient(cc).EmptyCall(ctx, &testpb.Empty{}, grpc.CallAuthority(authority)); status.Code(err) != codes.OK {
+		t.Fatalf("EmptyCall() returned status %v, want %v", status.Code(err), codes.OK)
+	}
+
+}
+
+// Tests the `grpc.CallAuthority` option with TLS credentials. This test verifies
+// that the RPC fails with `UNAVAILABLE` status code and doesn't reach the server
+// when an incorrect authority is used.
+func (s) TestIncorrectAuthorityWithTLS(t *testing.T) {
+	cert, err := tls.LoadX509KeyPair(testdata.Path("x509/server1_cert.pem"), testdata.Path("x509/server1_key.pem"))
+	if err != nil {
+		t.Fatalf("Failed to load key pair: %s", err)
+	}
+	creds, err := credentials.NewClientTLSFromFile(testdata.Path("x509/server_ca_cert.pem"), "x.test.example.com")
+	if err != nil {
+		t.Fatalf("Failed to create credentials %v", err)
+	}
+
+	serverCalled := make(chan struct{})
+	ss := &stubserver.StubServer{
+		EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
+			close(serverCalled)
+			return nil, nil
+		},
+	}
+	if err := ss.StartServer(grpc.Creds(credentials.NewServerTLSFromCert(&cert))); err != nil {
+		t.Fatalf("Error starting endpoint server: %v", err)
+	}
+	defer ss.Stop()
+
+	cc, err := grpc.NewClient(ss.Address, grpc.WithTransportCredentials(creds))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(%q) = %v", ss.Address, err)
+	}
+	defer cc.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	const authority = "auth.example.com"
+	if _, err = testgrpc.NewTestServiceClient(cc).EmptyCall(ctx, &testpb.Empty{}, grpc.CallAuthority(authority)); status.Code(err) != codes.Unavailable {
+		t.Fatalf("EmptyCall() returned status %v, want %v", status.Code(err), codes.Unavailable)
+	}
+	select {
+	case <-serverCalled:
+		t.Fatalf("Server handler should not have been called")
+	case <-time.After(defaultTestShortTimeout):
+	}
+}
+
+// Tests the scenario where the `grpc.CallAuthority` call option is used with
+// insecure transport credentials. The test verifies that the specified
+// authority is correctly propagated to the server.
+func (s) TestAuthorityCallOptionWithInsecureCreds(t *testing.T) {
+	const authority = "test.server.name"
+
+	ss := &stubserver.StubServer{
+		EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
+			if err := authorityChecker(ctx, authority); err != nil {
+				return nil, err
+			}
+			return &testpb.Empty{}, nil
+		},
+	}
+	if err := ss.Start(nil); err != nil {
+		t.Fatalf("Error starting endpoint server: %v", err)
+	}
+	defer ss.Stop()
+
+	cc, err := grpc.NewClient(ss.Address, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(%q) = %v", ss.Address, err)
+	}
+	defer cc.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	if _, err = testgrpc.NewTestServiceClient(cc).EmptyCall(ctx, &testpb.Empty{}, grpc.CallAuthority(authority)); err != nil {
+		t.Fatalf("EmptyCall() rpc failed: %v", err)
+	}
+}
+
+// testAuthInfoNoValidator implements only credentials.AuthInfo and not
+// credentials.AuthorityValidator.
+type testAuthInfoNoValidator struct{}
+
+// AuthType returns the authentication type.
+func (testAuthInfoNoValidator) AuthType() string {
+	return "test"
+}
+
+// testAuthInfoWithValidator implements both credentials.AuthInfo and
+// credentials.AuthorityValidator.
+type testAuthInfoWithValidator struct {
+	validAuthority string
+}
+
+// AuthType returns the authentication type.
+func (testAuthInfoWithValidator) AuthType() string {
+	return "test"
+}
+
+// ValidateAuthority implements credentials.AuthorityValidator.
+func (v testAuthInfoWithValidator) ValidateAuthority(authority string) error {
+	if authority == v.validAuthority {
+		return nil
+	}
+	return fmt.Errorf("invalid authority %q, want %q", authority, v.validAuthority)
+}
+
+// testCreds is a test TransportCredentials that can optionally support
+// authority validation.
+type testCreds struct {
+	authority string
+}
+
+// ClientHandshake performs the client-side handshake.
+func (c *testCreds) ClientHandshake(ctx context.Context, authority string, rawConn net.Conn) (net.Conn, credentials.AuthInfo, error) {
+	if c.authority != "" {
+		return rawConn, testAuthInfoWithValidator{validAuthority: c.authority}, nil
+	}
+	return rawConn, testAuthInfoNoValidator{}, nil
+}
+
+// ServerHandshake performs the server-side handshake.
+func (c *testCreds) ServerHandshake(rawConn net.Conn) (net.Conn, credentials.AuthInfo, error) {
+	if c.authority != "" {
+		return rawConn, testAuthInfoWithValidator{validAuthority: c.authority}, nil
+	}
+	return rawConn, testAuthInfoNoValidator{}, nil
+}
+
+// Clone creates a copy of testCreds.
+func (c *testCreds) Clone() credentials.TransportCredentials {
+	return &testCreds{authority: c.authority}
+}
+
+// Info provides protocol information.
+func (c *testCreds) Info() credentials.ProtocolInfo {
+	return credentials.ProtocolInfo{}
+}
+
+// OverrideServerName overrides the server name used for verification.
+func (c *testCreds) OverrideServerName(serverName string) error {
+	return nil
+}
+
+// TestAuthorityValidationFailureWithCustomCreds tests the `grpc.CallAuthority`
+// call option using custom credentials. It covers two failure scenarios:
+// - The credentials implement AuthorityValidator but authority used to override
+// is not valid.
+// - The credentials do not implement AuthorityValidator, but an authority
+// override is specified.
+// In both cases, the RPC is expected to fail with an `UNAVAILABLE` status code.
+func (s) TestAuthorityValidationFailureWithCustomCreds(t *testing.T) {
+	tests := []struct {
+		name      string
+		creds     credentials.TransportCredentials
+		authority string
+	}{
+		{
+			name:      "IncorrectAuthorityWithFakeCreds",
+			authority: "auth.example.com",
+			creds:     &testCreds{authority: "auth.test.example.com"},
+		},
+		{
+			name:      "FakeCredsWithNoAuthValidator",
+			creds:     &testCreds{},
+			authority: "auth.test.example.com",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			serverCalled := make(chan struct{})
+			ss := stubserver.StubServer{
+				EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
+					close(serverCalled)
+					return nil, nil
+				},
+			}
+			if err := ss.StartServer(); err != nil {
+				t.Fatalf("Failed to start stub server: %v", err)
+			}
+			defer ss.Stop()
+
+			cc, err := grpc.NewClient(ss.Address, grpc.WithTransportCredentials(tt.creds))
+			if err != nil {
+				t.Fatalf("grpc.NewClient(%q) = %v", ss.Address, err)
+			}
+			defer cc.Close()
+
+			ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+			defer cancel()
+			if _, err = testgrpc.NewTestServiceClient(cc).EmptyCall(ctx, &testpb.Empty{}, grpc.CallAuthority(tt.authority)); status.Code(err) != codes.Unavailable {
+				t.Fatalf("EmptyCall() returned status %v, want %v", status.Code(err), codes.Unavailable)
+			}
+			select {
+			case <-serverCalled:
+				t.Fatalf("Server should not have been called")
+			case <-time.After(defaultTestShortTimeout):
+			}
+		})
+	}
+
+}
+
+// TestCorrectAuthorityWithCustomCreds tests the `grpc.CallAuthority` call
+// option using custom credentials. It verifies that the provided authority is
+// correctly propagated to the server when a correct authority is used.
+func (s) TestCorrectAuthorityWithCustomCreds(t *testing.T) {
+	const authority = "auth.test.example.com"
+	creds := &testCreds{authority: "auth.test.example.com"}
+	ss := stubserver.StubServer{
+		EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
+			if err := authorityChecker(ctx, authority); err != nil {
+				return nil, err
+			}
+			return &testpb.Empty{}, nil
+		},
+	}
+	if err := ss.StartServer(); err != nil {
+		t.Fatalf("Failed to start stub server: %v", err)
+	}
+	defer ss.Stop()
+
+	cc, err := grpc.NewClient(ss.Address, grpc.WithTransportCredentials(creds))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(%q) = %v", ss.Address, err)
+	}
+	defer cc.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	if _, err = testgrpc.NewTestServiceClient(cc).EmptyCall(ctx, &testpb.Empty{}, grpc.CallAuthority(authority)); status.Code(err) != codes.OK {
+		t.Fatalf("EmptyCall() returned status %v, want %v", status.Code(err), codes.OK)
+	}
+}
diff --git a/credentials/tls_ext_test.go b/credentials/tls_ext_test.go
index 22881a6f..ceb810a4 100644
--- a/credentials/tls_ext_test.go
+++ b/credentials/tls_ext_test.go
@@ -43,6 +43,7 @@ import (
 )
 
 const defaultTestTimeout = 10 * time.Second
+const defaultTestShortTimeout = 10 * time.Millisecond
 
 type s struct {
 	grpctest.Tester
diff --git a/stats/opentelemetry/e2e_test.go b/stats/opentelemetry/e2e_test.go
index 01fe5a9e..4dbaadb2 100644
--- a/stats/opentelemetry/e2e_test.go
+++ b/stats/opentelemetry/e2e_test.go
@@ -1707,3 +1707,85 @@ func verifyTrace(t *testing.T, spans tracetest.SpanStubs, want traceSpanInfo) {
 		t.Errorf("Expected span not found: %q (kind: %s)", want.name, want.spanKind)
 	}
 }
+
+// TestStreamingRPC_TraceSequenceNumbers verifies that sequence numbers
+// are incremented correctly for multiple messages sent and received
+// during a streaming RPC.
+func (s) TestStreamingRPC_TraceSequenceNumbers(t *testing.T) {
+	mo, _ := defaultMetricsOptions(t, nil)
+	to, exporter := defaultTraceOptions(t)
+	ss := setupStubServer(t, mo, to)
+	defer ss.Stop()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	stream, err := ss.Client.FullDuplexCall(ctx)
+	if err != nil {
+		t.Fatalf("ss.Client.FullDuplexCall failed: %f", err)
+	}
+
+	const numMessages = 3
+	var wantOutboundEvents, wantInboundEvents []trace.Event
+	for i := range numMessages {
+		if err := stream.Send(&testpb.StreamingOutputCallRequest{}); err != nil {
+			t.Fatalf("stream.Send() failed at message %d: %v", i, err)
+		}
+		wantOutboundEvents = append(wantOutboundEvents, trace.Event{
+			Name: "Outbound message",
+			Attributes: []attribute.KeyValue{
+				attribute.Int("sequence-number", i),
+				attribute.Int("message-size", 0),
+			},
+		})
+		wantInboundEvents = append(wantInboundEvents, trace.Event{
+			Name: "Inbound message",
+			Attributes: []attribute.KeyValue{
+				attribute.Int("sequence-number", i),
+				attribute.Int("message-size", 0),
+			},
+		})
+	}
+	stream.CloseSend()
+	_, err = stream.Recv()
+	if err != io.EOF {
+		t.Fatalf("stream.Recv() got unexpected err=%v; want io.EOF", err)
+	}
+
+	wantSpanInfos := []traceSpanInfo{
+		{
+			name:       "Sent.grpc.testing.TestService.FullDuplexCall",
+			spanKind:   oteltrace.SpanKindClient.String(),
+			events:     nil,
+			attributes: nil,
+		},
+		{
+			name:     "Recv.grpc.testing.TestService.FullDuplexCall",
+			spanKind: oteltrace.SpanKindServer.String(),
+			events:   wantInboundEvents,
+			attributes: []attribute.KeyValue{
+				attribute.Bool("Client", false),
+				attribute.Bool("FailFast", false),
+				attribute.Int("previous-rpc-attempts", 0),
+				attribute.Bool("transparent-retry", false),
+			},
+		},
+		{
+			name:     "Attempt.grpc.testing.TestService.FullDuplexCall",
+			spanKind: oteltrace.SpanKindInternal.String(),
+			events:   wantOutboundEvents,
+			attributes: []attribute.KeyValue{
+				attribute.Bool("Client", true),
+				attribute.Bool("FailFast", true),
+				attribute.Int("previous-rpc-attempts", 0),
+				attribute.Bool("transparent-retry", false),
+			},
+		},
+	}
+
+	spans, err := waitForTraceSpans(ctx, exporter, wantSpanInfos)
+	if err != nil {
+		t.Fatal(err)
+	}
+	validateTraces(t, spans, wantSpanInfos)
+}
diff --git a/test/end2end_test.go b/test/end2end_test.go
index 906546ef..a4258771 100644
--- a/test/end2end_test.go
+++ b/test/end2end_test.go
@@ -136,6 +136,7 @@ var (
 
 var raceMode bool // set by race.go in race mode
 
+// Note : Do not use this for further tests.
 type testServer struct {
 	testgrpc.UnimplementedTestServiceServer
 
@@ -3584,6 +3585,160 @@ func testClientStreamingError(t *testing.T, e env) {
 	}
 }
 
+// Tests that a client receives a cardinality violation error for client-streaming
+// RPCs if the server doesn't send a message before returning status OK.
+func (s) TestClientStreamingCardinalityViolation_ServerHandlerMissingSendAndClose(t *testing.T) {
+	// TODO : https://github.com/grpc/grpc-go/issues/8119 - remove `t.Skip()`
+	// after this is fixed.
+	t.Skip()
+	ss := &stubserver.StubServer{
+		StreamingInputCallF: func(_ testgrpc.TestService_StreamingInputCallServer) error {
+			// Returning status OK without sending a response message.This is a
+			// cardinality violation.
+			return nil
+		},
+	}
+	if err := ss.Start(nil); err != nil {
+		t.Fatalf("Error starting endpoint server: %v", err)
+	}
+	defer ss.Stop()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	stream, err := ss.Client.StreamingInputCall(ctx)
+	if err != nil {
+		t.Fatalf(".StreamingInputCall(_) = _, %v, want <nil>", err)
+	}
+
+	_, err = stream.CloseAndRecv()
+	if err == nil {
+		t.Fatalf("stream.CloseAndRecv() = %v, want an error", err)
+	}
+	if status.Code(err) != codes.Internal {
+		t.Fatalf("stream.CloseAndRecv() = %v, want error %s", err, codes.Internal)
+	}
+}
+
+// Tests that the server can continue to receive messages after calling SendAndClose. Although
+// this is unexpected, we retain it for backward compatibility.
+func (s) TestClientStreaming_ServerHandlerRecvAfterSendAndClose(t *testing.T) {
+	ss := stubserver.StubServer{
+		StreamingInputCallF: func(stream testgrpc.TestService_StreamingInputCallServer) error {
+			if err := stream.SendAndClose(&testpb.StreamingInputCallResponse{}); err != nil {
+				t.Errorf("stream.SendAndClose(_) = %v, want <nil>", err)
+			}
+			if resp, err := stream.Recv(); err != nil || !proto.Equal(resp, &testpb.StreamingInputCallRequest{}) {
+				t.Errorf("stream.Recv() = %s, %v, want non-nil empty response, <nil>", resp, err)
+			}
+			return nil
+		},
+	}
+	if err := ss.Start(nil); err != nil {
+		t.Fatal("Error starting server:", err)
+	}
+	defer ss.Stop()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	stream, err := ss.Client.StreamingInputCall(ctx)
+	if err != nil {
+		t.Fatalf(".StreamingInputCall(_) = _, %v, want <nil>", err)
+	}
+	if err := stream.Send(&testpb.StreamingInputCallRequest{}); err != nil {
+		t.Fatalf("stream.Send(_) = %v, want <nil>", err)
+	}
+	if resp, err := stream.CloseAndRecv(); err != nil || !proto.Equal(resp, &testpb.StreamingInputCallResponse{}) {
+		t.Fatalf("stream.CloseSend() = %v , %v, want non-nil empty response, <nil>", resp, err)
+	}
+}
+
+// Tests that Recv() on client streaming client blocks till the server handler
+// returns even after calling SendAndClose from the server handler.
+func (s) TestClientStreaming_RecvWaitsForServerHandlerRetrun(t *testing.T) {
+	waitForReturn := make(chan struct{})
+	ss := stubserver.StubServer{
+		StreamingInputCallF: func(stream testgrpc.TestService_StreamingInputCallServer) error {
+			if err := stream.SendAndClose(&testpb.StreamingInputCallResponse{}); err != nil {
+				t.Errorf("stream.SendAndClose(_) = %v, want <nil>", err)
+			}
+			<-waitForReturn
+			return nil
+		},
+	}
+	if err := ss.Start(nil); err != nil {
+		t.Fatal("Error starting server:", err)
+	}
+	defer ss.Stop()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	stream, err := ss.Client.StreamingInputCall(ctx)
+	if err != nil {
+		t.Fatalf(".StreamingInputCall(_) = _, %v, want <nil>", err)
+	}
+	// Start Recv in a goroutine to test if it blocks until the server handler returns.
+	errCh := make(chan error, 1)
+	go func() {
+		resp := new(testpb.StreamingInputCallResponse)
+		err := stream.RecvMsg(resp)
+		errCh <- err
+	}()
+
+	// Check that Recv() is blocked.
+	select {
+	case err := <-errCh:
+		t.Fatalf("stream.RecvMsg(_) = %v returned unexpectedly", err)
+	case <-time.After(defaultTestShortTimeout):
+	}
+
+	close(waitForReturn)
+
+	// Recv() should return after the server handler returns.
+	select {
+	case err := <-errCh:
+		if err != nil {
+			t.Fatalf("stream.RecvMsg(_) = %v, want <nil>", err)
+		}
+	case <-time.After(defaultTestTimeout):
+		t.Fatal("Timed out waiting for stream.RecvMsg(_) to return")
+	}
+}
+
+// Tests the behavior where server handler returns an error after calling
+// SendAndClose. It verifies the that client receives nil message and
+// non-nil error.
+func (s) TestClientStreaming_ReturnErrorAfterSendAndClose(t *testing.T) {
+	wantError := status.Error(codes.Internal, "error for testing")
+	ss := stubserver.StubServer{
+		StreamingInputCallF: func(stream testgrpc.TestService_StreamingInputCallServer) error {
+			if err := stream.SendAndClose(&testpb.StreamingInputCallResponse{}); err != nil {
+				t.Errorf("stream.SendAndClose(_) = %v, want <nil>", err)
+			}
+			return wantError
+		},
+	}
+	if err := ss.Start(nil); err != nil {
+		t.Fatal("Error starting server:", err)
+	}
+	defer ss.Stop()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	stream, err := ss.Client.StreamingInputCall(ctx)
+	if err != nil {
+		t.Fatalf(".StreamingInputCall(_) = _, %v, want <nil>", err)
+	}
+	resp, err := stream.CloseAndRecv()
+
+	wantStatus, _ := status.FromError(wantError)
+	gotStatus, _ := status.FromError(err)
+
+	if gotStatus.Code() != wantStatus.Code() || gotStatus.Message() != wantStatus.Message() || resp != nil {
+		t.Fatalf("stream.CloseSend() = %v , %v, want <nil>, %s", resp, err, wantError)
+	}
+}
+
 func (s) TestExceedMaxStreamsLimit(t *testing.T) {
 	for _, e := range listTestEnv() {
 		testExceedMaxStreamsLimit(t, e)
diff --git a/xds/internal/clients/lrsclient/load_store_test.go b/xds/internal/clients/lrsclient/load_store_test.go
new file mode 100644
index 00000000..a21ac71d
--- /dev/null
+++ b/xds/internal/clients/lrsclient/load_store_test.go
@@ -0,0 +1,473 @@
+/*
+ *
+ * Copyright 2020 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.
+ */
+
+package lrsclient
+
+import (
+	"fmt"
+	"sort"
+	"sync"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+)
+
+var (
+	dropCategories = []string{"drop_for_real", "drop_for_fun"}
+	localities     = []string{"locality-A", "locality-B"}
+	errTest        = fmt.Errorf("test error")
+)
+
+// rpcData wraps the rpc counts and load data to be pushed to the store.
+type rpcData struct {
+	start, success, failure int
+	serverData              map[string]float64 // Will be reported with successful RPCs.
+}
+
+func verifyLoadStoreData(wantStoreData, gotStoreData []*loadData) error {
+	if diff := cmp.Diff(wantStoreData, gotStoreData, cmpopts.EquateEmpty(), cmp.AllowUnexported(loadData{}, localityData{}, requestData{}, serverLoadData{}), cmpopts.IgnoreFields(loadData{}, "reportInterval"), sortDataSlice); diff != "" {
+		return fmt.Errorf("store.stats() returned unexpected diff (-want +got):\n%s", diff)
+	}
+	return nil
+}
+
+// TestDrops spawns a bunch of goroutines which report drop data. After the
+// goroutines have exited, the test dumps the stats from the Store and makes
+// sure they are as expected.
+func TestDrops(t *testing.T) {
+	var (
+		drops = map[string]int{
+			dropCategories[0]: 30,
+			dropCategories[1]: 40,
+			"":                10,
+		}
+		wantStoreData = &loadData{
+			totalDrops: 80,
+			drops: map[string]uint64{
+				dropCategories[0]: 30,
+				dropCategories[1]: 40,
+			},
+		}
+	)
+
+	ls := PerClusterReporter{}
+	var wg sync.WaitGroup
+	for category, count := range drops {
+		for i := 0; i < count; i++ {
+			wg.Add(1)
+			go func(c string) {
+				ls.CallDropped(c)
+				wg.Done()
+			}(category)
+		}
+	}
+	wg.Wait()
+
+	gotStoreData := ls.stats()
+	if err := verifyLoadStoreData([]*loadData{wantStoreData}, []*loadData{gotStoreData}); err != nil {
+		t.Error(err)
+	}
+}
+
+// TestLocalityStats spawns a bunch of goroutines which report rpc and load
+// data. After the goroutines have exited, the test dumps the stats from the
+// Store and makes sure they are as expected.
+func TestLocalityStats(t *testing.T) {
+	var (
+		ld = map[string]rpcData{
+			localities[0]: {
+				start:      40,
+				success:    20,
+				failure:    10,
+				serverData: map[string]float64{"net": 1, "disk": 2, "cpu": 3, "mem": 4},
+			},
+			localities[1]: {
+				start:      80,
+				success:    40,
+				failure:    20,
+				serverData: map[string]float64{"net": 1, "disk": 2, "cpu": 3, "mem": 4},
+			},
+		}
+		wantStoreData = &loadData{
+			localityStats: map[string]localityData{
+				localities[0]: {
+					requestStats: requestData{
+						succeeded:  20,
+						errored:    10,
+						inProgress: 10,
+						issued:     40,
+					},
+					loadStats: map[string]serverLoadData{
+						"net":  {count: 20, sum: 20},
+						"disk": {count: 20, sum: 40},
+						"cpu":  {count: 20, sum: 60},
+						"mem":  {count: 20, sum: 80},
+					},
+				},
+				localities[1]: {
+					requestStats: requestData{
+						succeeded:  40,
+						errored:    20,
+						inProgress: 20,
+						issued:     80,
+					},
+					loadStats: map[string]serverLoadData{
+						"net":  {count: 40, sum: 40},
+						"disk": {count: 40, sum: 80},
+						"cpu":  {count: 40, sum: 120},
+						"mem":  {count: 40, sum: 160},
+					},
+				},
+			},
+		}
+	)
+
+	ls := PerClusterReporter{}
+	var wg sync.WaitGroup
+	for locality, data := range ld {
+		wg.Add(data.start)
+		for i := 0; i < data.start; i++ {
+			go func(l string) {
+				ls.CallStarted(l)
+				wg.Done()
+			}(locality)
+		}
+		// The calls to callStarted() need to happen before the other calls are
+		// made. Hence the wait here.
+		wg.Wait()
+
+		wg.Add(data.success)
+		for i := 0; i < data.success; i++ {
+			go func(l string, serverData map[string]float64) {
+				ls.CallFinished(l, nil)
+				for n, d := range serverData {
+					ls.CallServerLoad(l, n, d)
+				}
+				wg.Done()
+			}(locality, data.serverData)
+		}
+		wg.Add(data.failure)
+		for i := 0; i < data.failure; i++ {
+			go func(l string) {
+				ls.CallFinished(l, errTest)
+				wg.Done()
+			}(locality)
+		}
+		wg.Wait()
+	}
+
+	gotStoreData := ls.stats()
+	if err := verifyLoadStoreData([]*loadData{wantStoreData}, []*loadData{gotStoreData}); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestResetAfterStats(t *testing.T) {
+	// Push a bunch of drops, call stats and load stats, and leave inProgress to be non-zero.
+	// Dump the stats. Verify expected
+	// Push the same set of loads as before
+	// Now dump and verify the newly expected ones.
+	var (
+		drops = map[string]int{
+			dropCategories[0]: 30,
+			dropCategories[1]: 40,
+		}
+		ld = map[string]rpcData{
+			localities[0]: {
+				start:      40,
+				success:    20,
+				failure:    10,
+				serverData: map[string]float64{"net": 1, "disk": 2, "cpu": 3, "mem": 4},
+			},
+			localities[1]: {
+				start:      80,
+				success:    40,
+				failure:    20,
+				serverData: map[string]float64{"net": 1, "disk": 2, "cpu": 3, "mem": 4},
+			},
+		}
+		wantStoreData = &loadData{
+			totalDrops: 70,
+			drops: map[string]uint64{
+				dropCategories[0]: 30,
+				dropCategories[1]: 40,
+			},
+			localityStats: map[string]localityData{
+				localities[0]: {
+					requestStats: requestData{
+						succeeded:  20,
+						errored:    10,
+						inProgress: 10,
+						issued:     40,
+					},
+
+					loadStats: map[string]serverLoadData{
+						"net":  {count: 20, sum: 20},
+						"disk": {count: 20, sum: 40},
+						"cpu":  {count: 20, sum: 60},
+						"mem":  {count: 20, sum: 80},
+					},
+				},
+				localities[1]: {
+					requestStats: requestData{
+						succeeded:  40,
+						errored:    20,
+						inProgress: 20,
+						issued:     80,
+					},
+
+					loadStats: map[string]serverLoadData{
+						"net":  {count: 40, sum: 40},
+						"disk": {count: 40, sum: 80},
+						"cpu":  {count: 40, sum: 120},
+						"mem":  {count: 40, sum: 160},
+					},
+				},
+			},
+		}
+	)
+
+	reportLoad := func(ls *PerClusterReporter) {
+		for category, count := range drops {
+			for i := 0; i < count; i++ {
+				ls.CallDropped(category)
+			}
+		}
+		for locality, data := range ld {
+			for i := 0; i < data.start; i++ {
+				ls.CallStarted(locality)
+			}
+			for i := 0; i < data.success; i++ {
+				ls.CallFinished(locality, nil)
+				for n, d := range data.serverData {
+					ls.CallServerLoad(locality, n, d)
+				}
+			}
+			for i := 0; i < data.failure; i++ {
+				ls.CallFinished(locality, errTest)
+			}
+		}
+	}
+
+	ls := PerClusterReporter{}
+	reportLoad(&ls)
+	gotStoreData := ls.stats()
+	if err := verifyLoadStoreData([]*loadData{wantStoreData}, []*loadData{gotStoreData}); err != nil {
+		t.Error(err)
+	}
+
+	// The above call to stats() should have reset all load reports except the
+	// inProgress rpc count. We are now going to push the same load data into
+	// the store. So, we should expect to see twice the count for inProgress.
+	for _, l := range localities {
+		ls := wantStoreData.localityStats[l]
+		ls.requestStats.inProgress *= 2
+		wantStoreData.localityStats[l] = ls
+	}
+	reportLoad(&ls)
+	gotStoreData = ls.stats()
+	if err := verifyLoadStoreData([]*loadData{wantStoreData}, []*loadData{gotStoreData}); err != nil {
+		t.Error(err)
+	}
+}
+
+var sortDataSlice = cmp.Transformer("SortDataSlice", func(in []*loadData) []*loadData {
+	out := append([]*loadData(nil), in...) // Copy input to avoid mutating it
+	sort.Slice(out,
+		func(i, j int) bool {
+			if out[i].cluster < out[j].cluster {
+				return true
+			}
+			if out[i].cluster == out[j].cluster {
+				return out[i].service < out[j].service
+			}
+			return false
+		},
+	)
+	return out
+})
+
+// Test all load are returned for the given clusters, and all clusters are
+// reported if no cluster is specified.
+func TestStoreStats(t *testing.T) {
+	var (
+		testClusters = []string{"c0", "c1", "c2"}
+		testServices = []string{"s0", "s1"}
+		testLocality = "test-locality"
+	)
+
+	store := newLoadStore()
+	for _, c := range testClusters {
+		for _, s := range testServices {
+			store.ReporterForCluster(c, s).CallStarted(testLocality)
+			store.ReporterForCluster(c, s).CallServerLoad(testLocality, "abc", 123)
+			store.ReporterForCluster(c, s).CallDropped("dropped")
+			store.ReporterForCluster(c, s).CallFinished(testLocality, nil)
+		}
+	}
+
+	wantC0 := []*loadData{
+		{
+			cluster: "c0", service: "s0",
+			totalDrops: 1, drops: map[string]uint64{"dropped": 1},
+			localityStats: map[string]localityData{
+				"test-locality": {
+					requestStats: requestData{succeeded: 1, issued: 1},
+					loadStats:    map[string]serverLoadData{"abc": {count: 1, sum: 123}},
+				},
+			},
+		},
+		{
+			cluster: "c0", service: "s1",
+			totalDrops: 1, drops: map[string]uint64{"dropped": 1},
+			localityStats: map[string]localityData{
+				"test-locality": {
+					requestStats: requestData{succeeded: 1, issued: 1},
+					loadStats:    map[string]serverLoadData{"abc": {count: 1, sum: 123}},
+				},
+			},
+		},
+	}
+	// Call Stats with just "c0", this should return data for "c0", and not
+	// touch data for other clusters.
+	gotC0 := store.stats([]string{"c0"})
+	verifyLoadStoreData(wantC0, gotC0)
+
+	wantOther := []*loadData{
+		{
+			cluster: "c1", service: "s0",
+			totalDrops: 1, drops: map[string]uint64{"dropped": 1},
+			localityStats: map[string]localityData{
+				"test-locality": {
+					requestStats: requestData{succeeded: 1, issued: 1},
+					loadStats:    map[string]serverLoadData{"abc": {count: 1, sum: 123}},
+				},
+			},
+		},
+		{
+			cluster: "c1", service: "s1",
+			totalDrops: 1, drops: map[string]uint64{"dropped": 1},
+			localityStats: map[string]localityData{
+				"test-locality": {
+					requestStats: requestData{succeeded: 1, issued: 1},
+					loadStats:    map[string]serverLoadData{"abc": {count: 1, sum: 123}},
+				},
+			},
+		},
+		{
+			cluster: "c2", service: "s0",
+			totalDrops: 1, drops: map[string]uint64{"dropped": 1},
+			localityStats: map[string]localityData{
+				"test-locality": {
+					requestStats: requestData{succeeded: 1, issued: 1},
+					loadStats:    map[string]serverLoadData{"abc": {count: 1, sum: 123}},
+				},
+			},
+		},
+		{
+			cluster: "c2", service: "s1",
+			totalDrops: 1, drops: map[string]uint64{"dropped": 1},
+			localityStats: map[string]localityData{
+				"test-locality": {
+					requestStats: requestData{succeeded: 1, issued: 1},
+					loadStats:    map[string]serverLoadData{"abc": {count: 1, sum: 123}},
+				},
+			},
+		},
+	}
+	// Call Stats with empty slice, this should return data for all the
+	// remaining clusters, and not include c0 (because c0 data was cleared).
+	gotOther := store.stats(nil)
+	if err := verifyLoadStoreData(wantOther, gotOther); err != nil {
+		t.Error(err)
+	}
+}
+
+// Test the cases that if a cluster doesn't have load to report, its data is not
+// appended to the slice returned by Stats().
+func TestStoreStatsEmptyDataNotReported(t *testing.T) {
+	var (
+		testServices = []string{"s0", "s1"}
+		testLocality = "test-locality"
+	)
+
+	store := newLoadStore()
+	// "c0"'s RPCs all finish with success.
+	for _, s := range testServices {
+		store.ReporterForCluster("c0", s).CallStarted(testLocality)
+		store.ReporterForCluster("c0", s).CallFinished(testLocality, nil)
+	}
+	// "c1"'s RPCs never finish (always inprocess).
+	for _, s := range testServices {
+		store.ReporterForCluster("c1", s).CallStarted(testLocality)
+	}
+
+	want0 := []*loadData{
+		{
+			cluster: "c0", service: "s0",
+			localityStats: map[string]localityData{
+				"test-locality": {requestStats: requestData{succeeded: 1, issued: 1}},
+			},
+		},
+		{
+			cluster: "c0", service: "s1",
+			localityStats: map[string]localityData{
+				"test-locality": {requestStats: requestData{succeeded: 1, issued: 1}},
+			},
+		},
+		{
+			cluster: "c1", service: "s0",
+			localityStats: map[string]localityData{
+				"test-locality": {requestStats: requestData{inProgress: 1, issued: 1}},
+			},
+		},
+		{
+			cluster: "c1", service: "s1",
+			localityStats: map[string]localityData{
+				"test-locality": {requestStats: requestData{inProgress: 1, issued: 1}},
+			},
+		},
+	}
+	// Call Stats with empty slice, this should return data for all the
+	// clusters.
+	got0 := store.stats(nil)
+	if err := verifyLoadStoreData(want0, got0); err != nil {
+		t.Error(err)
+	}
+
+	want1 := []*loadData{
+		{
+			cluster: "c1", service: "s0",
+			localityStats: map[string]localityData{
+				"test-locality": {requestStats: requestData{inProgress: 1}},
+			},
+		},
+		{
+			cluster: "c1", service: "s1",
+			localityStats: map[string]localityData{
+				"test-locality": {requestStats: requestData{inProgress: 1}},
+			},
+		},
+	}
+	// Call Stats with empty slice again, this should return data only for "c1",
+	// because "c0" data was cleared, but "c1" has in-progress RPCs.
+	got1 := store.stats(nil)
+	if err := verifyLoadStoreData(want1, got1); err != nil {
+		t.Error(err)
+	}
+}
diff --git a/xds/internal/clients/lrsclient/loadreport_test.go b/xds/internal/clients/lrsclient/loadreport_test.go
new file mode 100644
index 00000000..1c5072ac
--- /dev/null
+++ b/xds/internal/clients/lrsclient/loadreport_test.go
@@ -0,0 +1,613 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+package lrsclient_test
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/google/uuid"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/internal/grpctest"
+	"google.golang.org/grpc/status"
+	"google.golang.org/grpc/xds/internal/clients"
+	"google.golang.org/grpc/xds/internal/clients/grpctransport"
+	"google.golang.org/grpc/xds/internal/clients/internal/testutils"
+	"google.golang.org/grpc/xds/internal/clients/internal/testutils/e2e"
+	"google.golang.org/grpc/xds/internal/clients/internal/testutils/fakeserver"
+	"google.golang.org/grpc/xds/internal/clients/lrsclient"
+	"google.golang.org/protobuf/testing/protocmp"
+	"google.golang.org/protobuf/types/known/durationpb"
+
+	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
+	v3lrspb "github.com/envoyproxy/go-control-plane/envoy/service/load_stats/v3"
+)
+
+type s struct {
+	grpctest.Tester
+}
+
+func Test(t *testing.T) {
+	grpctest.RunSubTests(t, s{})
+}
+
+const (
+	testLocality1                 = `{"region":"test-region1"}`
+	testLocality2                 = `{"region":"test-region2"}`
+	testKey1                      = "test-key1"
+	testKey2                      = "test-key2"
+	defaultTestWatchExpiryTimeout = 100 * time.Millisecond
+	defaultTestTimeout            = 5 * time.Second
+	defaultTestShortTimeout       = 10 * time.Millisecond // For events expected to *not* happen.
+)
+
+var (
+	toleranceCmpOpt   = cmpopts.EquateApprox(0, 1e-5)
+	ignoreOrderCmpOpt = protocmp.FilterField(&v3endpointpb.ClusterStats{}, "upstream_locality_stats",
+		cmpopts.SortSlices(func(a, b protocmp.Message) bool {
+			return a.String() < b.String()
+		}),
+	)
+)
+
+type wrappedListener struct {
+	net.Listener
+	newConnChan *testutils.Channel // Connection attempts are pushed here.
+}
+
+func (wl *wrappedListener) Accept() (net.Conn, error) {
+	c, err := wl.Listener.Accept()
+	if err != nil {
+		return nil, err
+	}
+	wl.newConnChan.Send(struct{}{})
+	return c, err
+}
+
+// Tests a load reporting scenario where the LRS client is reporting loads to
+// multiple servers. Verifies the following:
+//   - calling the load reporting API with different server configuration
+//     results in connections being created to those corresponding servers
+//   - the same load.Store is not returned when the load reporting API called
+//     with different server configurations
+//   - canceling the load reporting from the client results in the LRS stream
+//     being canceled on the server
+func (s) TestReportLoad_ConnectionCreation(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Create two management servers that also serve LRS.
+	l, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("net.Listen() failed: %v", err)
+	}
+	newConnChan1 := testutils.NewChannelWithSize(1)
+	lis1 := &wrappedListener{
+		Listener:    l,
+		newConnChan: newConnChan1,
+	}
+	mgmtServer1 := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
+		Listener:                    lis1,
+		SupportLoadReportingService: true,
+	})
+	l, err = net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("net.Listen() failed: %v", err)
+	}
+	newConnChan2 := testutils.NewChannelWithSize(1)
+	lis2 := &wrappedListener{
+		Listener:    l,
+		newConnChan: newConnChan2,
+	}
+	mgmtServer2 := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
+		Listener:                    lis2,
+		SupportLoadReportingService: true,
+	})
+
+	// Create an LRS client with a configuration that contains both of
+	// the above two servers. The authority name is immaterial here since load
+	// reporting is per-server and not per-authority.
+	nodeID := uuid.New().String()
+
+	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	config := lrsclient.Config{
+		Node:             clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
+		TransportBuilder: grpctransport.NewBuilder(credentials),
+	}
+	client, err := lrsclient.New(config)
+	if err != nil {
+		t.Fatalf("lrsclient.New() failed: %v", err)
+	}
+
+	serverIdentifier1 := clients.ServerIdentifier{ServerURI: mgmtServer1.Address, Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"}}
+	loadStore1, err := client.ReportLoad(serverIdentifier1)
+	if err != nil {
+		t.Fatalf("client.ReportLoad() failed: %v", err)
+	}
+	ssCtx, ssCancel := context.WithTimeout(context.Background(), time.Millisecond)
+	defer ssCancel()
+	defer loadStore1.Stop(ssCtx)
+
+	// Call the load reporting API to report load to the first management
+	// server, and ensure that a connection to the server is created.
+	if _, err := newConnChan1.Receive(ctx); err != nil {
+		t.Fatal("Timeout when waiting for a connection to the first management server, after starting load reporting")
+	}
+	if _, err := mgmtServer1.LRSServer.LRSStreamOpenChan.Receive(ctx); err != nil {
+		t.Fatal("Timeout when waiting for LRS stream to be created")
+	}
+
+	// Call the load reporting API to report load to the first management
+	// server, and ensure that a connection to the server is created.
+	serverIdentifier2 := clients.ServerIdentifier{ServerURI: mgmtServer2.Address, Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"}}
+	loadStore2, err := client.ReportLoad(serverIdentifier2)
+	if err != nil {
+		t.Fatalf("client.ReportLoad() failed: %v", err)
+	}
+	if _, err := newConnChan2.Receive(ctx); err != nil {
+		t.Fatal("Timeout when waiting for a connection to the second management server, after starting load reporting")
+	}
+	if _, err := mgmtServer2.LRSServer.LRSStreamOpenChan.Receive(ctx); err != nil {
+		t.Fatal("Timeout when waiting for LRS stream to be created")
+	}
+
+	if loadStore1 == loadStore2 {
+		t.Fatalf("Got same store for different servers, want different")
+	}
+
+	// Push some loads on the received store.
+	loadStore2.ReporterForCluster("cluster", "eds").CallDropped("test")
+
+	// Ensure the initial load reporting request is received at the server.
+	lrsServer := mgmtServer2.LRSServer
+	req, err := lrsServer.LRSRequestChan.Receive(ctx)
+	if err != nil {
+		t.Fatalf("Timeout when waiting for initial LRS request: %v", err)
+	}
+	gotInitialReq := req.(*fakeserver.Request).Req.(*v3lrspb.LoadStatsRequest)
+	nodeProto := &v3corepb.Node{
+		Id:                   nodeID,
+		UserAgentName:        "user-agent",
+		UserAgentVersionType: &v3corepb.Node_UserAgentVersion{UserAgentVersion: "0.0.0.0"},
+		ClientFeatures:       []string{"envoy.lb.does_not_support_overprovisioning", "xds.config.resource-in-sotw", "envoy.lrs.supports_send_all_clusters"},
+	}
+	wantInitialReq := &v3lrspb.LoadStatsRequest{Node: nodeProto}
+	if diff := cmp.Diff(gotInitialReq, wantInitialReq, protocmp.Transform()); diff != "" {
+		t.Fatalf("Unexpected diff in initial LRS request (-got, +want):\n%s", diff)
+	}
+
+	// Send a response from the server with a small deadline.
+	lrsServer.LRSResponseChan <- &fakeserver.Response{
+		Resp: &v3lrspb.LoadStatsResponse{
+			SendAllClusters:       true,
+			LoadReportingInterval: &durationpb.Duration{Nanos: 50000000}, // 50ms
+		},
+	}
+
+	// Ensure that loads are seen on the server.
+	req, err = lrsServer.LRSRequestChan.Receive(ctx)
+	if err != nil {
+		t.Fatalf("Timeout when waiting for LRS request with loads: %v", err)
+	}
+	gotLoad := req.(*fakeserver.Request).Req.(*v3lrspb.LoadStatsRequest).ClusterStats
+	if l := len(gotLoad); l != 1 {
+		t.Fatalf("Received load for %d clusters, want 1", l)
+	}
+
+	// This field is set by the client to indicate the actual time elapsed since
+	// the last report was sent. We cannot deterministically compare this, and
+	// we cannot use the cmpopts.IgnoreFields() option on proto structs, since
+	// we already use the protocmp.Transform() which marshals the struct into
+	// another message. Hence setting this field to nil is the best option here.
+	gotLoad[0].LoadReportInterval = nil
+	wantLoad := &v3endpointpb.ClusterStats{
+		ClusterName:          "cluster",
+		ClusterServiceName:   "eds",
+		TotalDroppedRequests: 1,
+		DroppedRequests:      []*v3endpointpb.ClusterStats_DroppedRequests{{Category: "test", DroppedCount: 1}},
+	}
+	if diff := cmp.Diff(wantLoad, gotLoad[0], protocmp.Transform(), toleranceCmpOpt, ignoreOrderCmpOpt); diff != "" {
+		t.Fatalf("Unexpected diff in LRS request (-got, +want):\n%s", diff)
+	}
+
+	// Stop this load reporting stream, server should see error canceled.
+	ssCtx, ssCancel = context.WithTimeout(context.Background(), time.Millisecond)
+	defer ssCancel()
+	loadStore2.Stop(ssCtx)
+
+	// Server should receive a stream canceled error. There may be additional
+	// load reports from the client in the channel.
+	for {
+		if ctx.Err() != nil {
+			t.Fatal("Timeout when waiting for the LRS stream to be canceled on the server")
+		}
+		u, err := lrsServer.LRSRequestChan.Receive(ctx)
+		if err != nil {
+			continue
+		}
+		// Ignore load reports sent before the stream was cancelled.
+		if u.(*fakeserver.Request).Err == nil {
+			continue
+		}
+		if status.Code(u.(*fakeserver.Request).Err) != codes.Canceled {
+			t.Fatalf("Unexpected LRS request: %v, want error canceled", u)
+		}
+		break
+	}
+}
+
+// Tests a load reporting scenario where the load reporting API is called
+// multiple times for the same server. The test verifies the following:
+//   - calling the load reporting API the second time for the same server
+//     configuration does not create a new LRS stream
+//   - the LRS stream is closed *only* after all the API calls invoke their
+//     cancel functions
+//   - creating new streams after the previous one was closed works
+func (s) TestReportLoad_StreamCreation(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Create a management server that serves LRS.
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{SupportLoadReportingService: true})
+
+	// Create an LRS client with configuration pointing to the above server.
+	nodeID := uuid.New().String()
+
+	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	config := lrsclient.Config{
+		Node:             clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
+		TransportBuilder: grpctransport.NewBuilder(credentials),
+	}
+	client, err := lrsclient.New(config)
+	if err != nil {
+		t.Fatalf("lrsclient.New() failed: %v", err)
+	}
+
+	// Call the load reporting API, and ensure that an LRS stream is created.
+	serverIdentifier := clients.ServerIdentifier{ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"}}
+	loadStore1, err := client.ReportLoad(serverIdentifier)
+	if err != nil {
+		t.Fatalf("client.ReportLoad() failed: %v", err)
+	}
+	lrsServer := mgmtServer.LRSServer
+	if _, err := lrsServer.LRSStreamOpenChan.Receive(ctx); err != nil {
+		t.Fatalf("Timeout when waiting for LRS stream to be created: %v", err)
+	}
+
+	// Push some loads on the received store.
+	loadStore1.ReporterForCluster("cluster1", "eds1").CallDropped("test")
+	loadStore1.ReporterForCluster("cluster1", "eds1").CallStarted(testLocality1)
+	loadStore1.ReporterForCluster("cluster1", "eds1").CallServerLoad(testLocality1, testKey1, 3.14)
+	loadStore1.ReporterForCluster("cluster1", "eds1").CallServerLoad(testLocality1, testKey1, 2.718)
+	loadStore1.ReporterForCluster("cluster1", "eds1").CallFinished(testLocality1, nil)
+	loadStore1.ReporterForCluster("cluster1", "eds1").CallStarted(testLocality2)
+	loadStore1.ReporterForCluster("cluster1", "eds1").CallServerLoad(testLocality2, testKey2, 1.618)
+	loadStore1.ReporterForCluster("cluster1", "eds1").CallFinished(testLocality2, nil)
+
+	// Ensure the initial load reporting request is received at the server.
+	req, err := lrsServer.LRSRequestChan.Receive(ctx)
+	if err != nil {
+		t.Fatalf("Timeout when waiting for initial LRS request: %v", err)
+	}
+	gotInitialReq := req.(*fakeserver.Request).Req.(*v3lrspb.LoadStatsRequest)
+	nodeProto := &v3corepb.Node{
+		Id:                   nodeID,
+		UserAgentName:        "user-agent",
+		UserAgentVersionType: &v3corepb.Node_UserAgentVersion{UserAgentVersion: "0.0.0.0"},
+		ClientFeatures:       []string{"envoy.lb.does_not_support_overprovisioning", "xds.config.resource-in-sotw", "envoy.lrs.supports_send_all_clusters"},
+	}
+	wantInitialReq := &v3lrspb.LoadStatsRequest{Node: nodeProto}
+	if diff := cmp.Diff(gotInitialReq, wantInitialReq, protocmp.Transform()); diff != "" {
+		t.Fatalf("Unexpected diff in initial LRS request (-got, +want):\n%s", diff)
+	}
+
+	// Send a response from the server with a small deadline.
+	lrsServer.LRSResponseChan <- &fakeserver.Response{
+		Resp: &v3lrspb.LoadStatsResponse{
+			SendAllClusters:       true,
+			LoadReportingInterval: &durationpb.Duration{Nanos: 50000000}, // 50ms
+		},
+	}
+
+	// Ensure that loads are seen on the server.
+	req, err = lrsServer.LRSRequestChan.Receive(ctx)
+	if err != nil {
+		t.Fatal("Timeout when waiting for LRS request with loads")
+	}
+	gotLoad := req.(*fakeserver.Request).Req.(*v3lrspb.LoadStatsRequest).ClusterStats
+	if l := len(gotLoad); l != 1 {
+		t.Fatalf("Received load for %d clusters, want 1", l)
+	}
+
+	// This field is set by the client to indicate the actual time elapsed since
+	// the last report was sent. We cannot deterministically compare this, and
+	// we cannot use the cmpopts.IgnoreFields() option on proto structs, since
+	// we already use the protocmp.Transform() which marshals the struct into
+	// another message. Hence setting this field to nil is the best option here.
+	gotLoad[0].LoadReportInterval = nil
+	wantLoad := &v3endpointpb.ClusterStats{
+		ClusterName:          "cluster1",
+		ClusterServiceName:   "eds1",
+		TotalDroppedRequests: 1,
+		DroppedRequests:      []*v3endpointpb.ClusterStats_DroppedRequests{{Category: "test", DroppedCount: 1}},
+		UpstreamLocalityStats: []*v3endpointpb.UpstreamLocalityStats{
+			{
+				Locality: &v3corepb.Locality{Region: "test-region1"},
+				LoadMetricStats: []*v3endpointpb.EndpointLoadMetricStats{
+					// TotalMetricValue is the aggregation of 3.14 + 2.718 = 5.858
+					{MetricName: testKey1, NumRequestsFinishedWithMetric: 2, TotalMetricValue: 5.858}},
+				TotalSuccessfulRequests: 1,
+				TotalIssuedRequests:     1,
+			},
+			{
+				Locality: &v3corepb.Locality{Region: "test-region2"},
+				LoadMetricStats: []*v3endpointpb.EndpointLoadMetricStats{
+					{MetricName: testKey2, NumRequestsFinishedWithMetric: 1, TotalMetricValue: 1.618}},
+				TotalSuccessfulRequests: 1,
+				TotalIssuedRequests:     1,
+			},
+		},
+	}
+	if diff := cmp.Diff(wantLoad, gotLoad[0], protocmp.Transform(), toleranceCmpOpt, ignoreOrderCmpOpt); diff != "" {
+		t.Fatalf("Unexpected diff in LRS request (-got, +want):\n%s", diff)
+	}
+
+	// Make another call to the load reporting API, and ensure that a new LRS
+	// stream is not created.
+	loadStore2, err := client.ReportLoad(serverIdentifier)
+	if err != nil {
+		t.Fatalf("client.ReportLoad() failed: %v", err)
+	}
+	sCtx, sCancel := context.WithTimeout(context.Background(), defaultTestShortTimeout)
+	defer sCancel()
+	if _, err := lrsServer.LRSStreamOpenChan.Receive(sCtx); err != context.DeadlineExceeded {
+		t.Fatal("New LRS stream created when expected to use an existing one")
+	}
+
+	// Push more loads.
+	loadStore2.ReporterForCluster("cluster2", "eds2").CallDropped("test")
+
+	// Ensure that loads are seen on the server. We need a loop here because
+	// there could have been some requests from the client in the time between
+	// us reading the first request and now. Those would have been queued in the
+	// request channel that we read out of.
+	for {
+		if ctx.Err() != nil {
+			t.Fatalf("Timeout when waiting for new loads to be seen on the server")
+		}
+
+		req, err = lrsServer.LRSRequestChan.Receive(ctx)
+		if err != nil {
+			continue
+		}
+		gotLoad = req.(*fakeserver.Request).Req.(*v3lrspb.LoadStatsRequest).ClusterStats
+		if l := len(gotLoad); l != 1 {
+			continue
+		}
+		gotLoad[0].LoadReportInterval = nil
+		wantLoad := &v3endpointpb.ClusterStats{
+			ClusterName:          "cluster2",
+			ClusterServiceName:   "eds2",
+			TotalDroppedRequests: 1,
+			DroppedRequests:      []*v3endpointpb.ClusterStats_DroppedRequests{{Category: "test", DroppedCount: 1}},
+		}
+		if diff := cmp.Diff(wantLoad, gotLoad[0], protocmp.Transform()); diff != "" {
+			t.Logf("Unexpected diff in LRS request (-got, +want):\n%s", diff)
+			continue
+		}
+		break
+	}
+
+	// Cancel the first load reporting call, and ensure that the stream does not
+	// close (because we have another call open).
+	ssCtx, ssCancel := context.WithTimeout(context.Background(), time.Millisecond)
+	defer ssCancel()
+	loadStore1.Stop(ssCtx)
+	sCtx, sCancel = context.WithTimeout(context.Background(), defaultTestShortTimeout)
+	defer sCancel()
+	if _, err := lrsServer.LRSStreamCloseChan.Receive(sCtx); err != context.DeadlineExceeded {
+		t.Fatal("LRS stream closed when expected to stay open")
+	}
+
+	// Stop the second load reporting call, and ensure the stream is closed.
+	ssCtx, ssCancel = context.WithTimeout(context.Background(), time.Millisecond)
+	defer ssCancel()
+	loadStore2.Stop(ssCtx)
+	if _, err := lrsServer.LRSStreamCloseChan.Receive(ctx); err != nil {
+		t.Fatal("Timeout waiting for LRS stream to close")
+	}
+
+	// Calling the load reporting API again should result in the creation of a
+	// new LRS stream. This ensures that creating and closing multiple streams
+	// works smoothly.
+	loadStore3, err := client.ReportLoad(serverIdentifier)
+	if err != nil {
+		t.Fatalf("client.ReportLoad() failed: %v", err)
+	}
+	if _, err := lrsServer.LRSStreamOpenChan.Receive(ctx); err != nil {
+		t.Fatalf("Timeout when waiting for LRS stream to be created: %v", err)
+	}
+	ssCtx, ssCancel = context.WithTimeout(context.Background(), time.Millisecond)
+	defer ssCancel()
+	loadStore3.Stop(ssCtx)
+}
+
+// TestReportLoad_StopWithContext tests the behavior of LoadStore.Stop() when
+// called with a context. It verifies that:
+//   - Stop() blocks until the context expires or final load send attempt is
+//     made.
+//   - Final load report is seen on the server after stop is called.
+//   - The stream is closed after Stop() returns.
+func (s) TestReportLoad_StopWithContext(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Create a management server that serves LRS.
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{SupportLoadReportingService: true})
+
+	// Create an LRS client with configuration pointing to the above server.
+	nodeID := uuid.New().String()
+
+	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	config := lrsclient.Config{
+		Node:             clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
+		TransportBuilder: grpctransport.NewBuilder(credentials),
+	}
+	client, err := lrsclient.New(config)
+	if err != nil {
+		t.Fatalf("lrsclient.New() failed: %v", err)
+	}
+
+	// Call the load reporting API, and ensure that an LRS stream is created.
+	serverIdentifier := clients.ServerIdentifier{ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"}}
+	loadStore, err := client.ReportLoad(serverIdentifier)
+	if err != nil {
+		t.Fatalf("client.ReportLoad() failed: %v", err)
+	}
+	lrsServer := mgmtServer.LRSServer
+	if _, err := lrsServer.LRSStreamOpenChan.Receive(ctx); err != nil {
+		t.Fatalf("Timeout when waiting for LRS stream to be created: %v", err)
+	}
+
+	// Push some loads on the received store.
+	loadStore.ReporterForCluster("cluster1", "eds1").CallDropped("test")
+
+	// Ensure the initial load reporting request is received at the server.
+	req, err := lrsServer.LRSRequestChan.Receive(ctx)
+	if err != nil {
+		t.Fatalf("Timeout when waiting for initial LRS request: %v", err)
+	}
+	gotInitialReq := req.(*fakeserver.Request).Req.(*v3lrspb.LoadStatsRequest)
+	nodeProto := &v3corepb.Node{
+		Id:                   nodeID,
+		UserAgentName:        "user-agent",
+		UserAgentVersionType: &v3corepb.Node_UserAgentVersion{UserAgentVersion: "0.0.0.0"},
+		ClientFeatures:       []string{"envoy.lb.does_not_support_overprovisioning", "xds.config.resource-in-sotw", "envoy.lrs.supports_send_all_clusters"},
+	}
+	wantInitialReq := &v3lrspb.LoadStatsRequest{Node: nodeProto}
+	if diff := cmp.Diff(gotInitialReq, wantInitialReq, protocmp.Transform()); diff != "" {
+		t.Fatalf("Unexpected diff in initial LRS request (-got, +want):\n%s", diff)
+	}
+
+	// Send a response from the server with a small deadline.
+	lrsServer.LRSResponseChan <- &fakeserver.Response{
+		Resp: &v3lrspb.LoadStatsResponse{
+			SendAllClusters:       true,
+			LoadReportingInterval: &durationpb.Duration{Nanos: 50000000}, // 50ms
+		},
+	}
+
+	// Ensure that loads are seen on the server.
+	req, err = lrsServer.LRSRequestChan.Receive(ctx)
+	if err != nil {
+		t.Fatal("Timeout when waiting for LRS request with loads")
+	}
+	gotLoad := req.(*fakeserver.Request).Req.(*v3lrspb.LoadStatsRequest).ClusterStats
+	if l := len(gotLoad); l != 1 {
+		t.Fatalf("Received load for %d clusters, want 1", l)
+	}
+
+	// This field is set by the client to indicate the actual time elapsed since
+	// the last report was sent. We cannot deterministically compare this, and
+	// we cannot use the cmpopts.IgnoreFields() option on proto structs, since
+	// we already use the protocmp.Transform() which marshals the struct into
+	// another message. Hence setting this field to nil is the best option here.
+	gotLoad[0].LoadReportInterval = nil
+	wantLoad := &v3endpointpb.ClusterStats{
+		ClusterName:          "cluster1",
+		ClusterServiceName:   "eds1",
+		TotalDroppedRequests: 1,
+		DroppedRequests:      []*v3endpointpb.ClusterStats_DroppedRequests{{Category: "test", DroppedCount: 1}},
+	}
+	if diff := cmp.Diff(wantLoad, gotLoad[0], protocmp.Transform(), toleranceCmpOpt, ignoreOrderCmpOpt); diff != "" {
+		t.Fatalf("Unexpected diff in LRS request (-got, +want):\n%s", diff)
+	}
+
+	// Create a context for Stop() that remains until the end of test to ensure
+	// that only possibility of Stop()s to finish is if final load send attempt
+	// is made. If final load attempt is not made, test will timeout.
+	stopCtx, stopCancel := context.WithCancel(ctx)
+	defer stopCancel()
+
+	// Push more loads.
+	loadStore.ReporterForCluster("cluster2", "eds2").CallDropped("test")
+
+	stopCalled := make(chan struct{})
+	// Call Stop in a separate goroutine. It will block until
+	// final load send attempt is made.
+	go func() {
+		loadStore.Stop(stopCtx)
+		close(stopCalled)
+	}()
+	<-stopCalled
+
+	// Ensure that loads are seen on the server. We need a loop here because
+	// there could have been some requests from the client in the time between
+	// us reading the first request and now. Those would have been queued in the
+	// request channel that we read out of.
+	for {
+		if ctx.Err() != nil {
+			t.Fatalf("Timeout when waiting for new loads to be seen on the server")
+		}
+
+		req, err = lrsServer.LRSRequestChan.Receive(ctx)
+		if err != nil {
+			continue
+		}
+		if req.(*fakeserver.Request).Req.(*v3lrspb.LoadStatsRequest) == nil {
+			// This can happen due to a race:
+			// 1. Load for "cluster2" is reported just before Stop().
+			// 2. The periodic ticker might send this load before Stop()'s
+			//    final send mechanism processes it, clearing the data.
+			// 3. Stop()'s final send might then send an empty report.
+			//    This is acceptable for this test because we only need to verify
+			//    if the final load report send attempt was made.
+			t.Logf("Empty final load report sent on server")
+			break
+		}
+		gotLoad = req.(*fakeserver.Request).Req.(*v3lrspb.LoadStatsRequest).ClusterStats
+		if l := len(gotLoad); l != 1 {
+			continue
+		}
+		gotLoad[0].LoadReportInterval = nil
+		wantLoad := &v3endpointpb.ClusterStats{
+			ClusterName:          "cluster2",
+			ClusterServiceName:   "eds2",
+			TotalDroppedRequests: 1,
+			DroppedRequests:      []*v3endpointpb.ClusterStats_DroppedRequests{{Category: "test", DroppedCount: 1}},
+		}
+		if diff := cmp.Diff(wantLoad, gotLoad[0], protocmp.Transform()); diff != "" {
+			t.Logf("Unexpected diff in LRS request (-got, +want):\n%s", diff)
+			continue
+		}
+		break
+	}
+
+	// Verify the stream is eventually closed on the server side.
+	if _, err := lrsServer.LRSStreamCloseChan.Receive(ctx); err != nil {
+		t.Fatal("Timeout waiting for LRS stream to close")
+	}
+}
diff --git a/xds/internal/clients/xdsclient/helpers_test.go b/xds/internal/clients/xdsclient/helpers_test.go
index 03b40fd4..d84e0a4b 100644
--- a/xds/internal/clients/xdsclient/helpers_test.go
+++ b/xds/internal/clients/xdsclient/helpers_test.go
@@ -129,7 +129,7 @@ func processServerSideListener(lis *v3listenerpb.Listener) (*listenerUpdate, err
 	if n := len(lis.ListenerFilters); n != 0 {
 		return nil, fmt.Errorf("unsupported field 'listener_filters' contains %d entries", n)
 	}
-	if useOrigDst := lis.GetUseOriginalDst(); useOrigDst != nil && useOrigDst.GetValue() {
+	if lis.GetUseOriginalDst().GetValue() {
 		return nil, errors.New("unsupported field 'use_original_dst' is present and set to true")
 	}
 	addr := lis.GetAddress()
diff --git a/xds/internal/clients/xdsclient/test/helpers_test.go b/xds/internal/clients/xdsclient/test/helpers_test.go
index e911ec59..d296fee6 100644
--- a/xds/internal/clients/xdsclient/test/helpers_test.go
+++ b/xds/internal/clients/xdsclient/test/helpers_test.go
@@ -137,7 +137,7 @@ func processServerSideListener(lis *v3listenerpb.Listener) (*listenerUpdate, err
 	if n := len(lis.ListenerFilters); n != 0 {
 		return nil, fmt.Errorf("unsupported field 'listener_filters' contains %d entries", n)
 	}
-	if useOrigDst := lis.GetUseOriginalDst(); useOrigDst != nil && useOrigDst.GetValue() {
+	if lis.GetUseOriginalDst().GetValue() {
 		return nil, errors.New("unsupported field 'use_original_dst' is present and set to true")
 	}
 	addr := lis.GetAddress()