test: Fix flaky test TestServerStreaming_ClientCallSendMsgTwice in end2end_test.go. (grpc#9004)

Fixes: grpc#8993 

This PR will fix the flaky test
`TestServerStreaming_ClientCallSendMsgTwice` introduced in PR grpc#8385 .

The test expected a single call to `stream.SendMsg()` to return a
`codes.Canceled` error immediately after the stream's context was marked
as done.

Solution: Introduced a retry loop with a timeout for the
`stream.SendMsg()` call. Instead of testing a single invocation, the
server handler now continuously attempts to send a message until it
either receives an error or timeout expired.

Successfully run the test on forge for 1 million times without any
flake.

RELEASE NOTES: N/A

Author: Pranjali-2501

test/end2end_test.go

@@ -3960,7 +3960,18 @@ func (s) TestServerStreaming_ClientCallSendMsgTwice(t *testing.T) {
 			// Block until the stream’s context is done. Second call to client.SendMsg
 			// triggers a RST_STREAM which cancels the stream context on the server.
 			<-stream.Context().Done()
-			if err := stream.SendMsg(&testpb.StreamingOutputCallRequest{}); status.Code(err) != codes.Canceled {
+			var err error
+			waitCtx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+			defer cancel()
+			for ; waitCtx.Err() == nil; <-time.After(time.Millisecond) {
+				if err = stream.SendMsg(&testpb.StreamingOutputCallRequest{}); err != nil {
+					break
+				}
+			}
+			if waitCtx.Err() != nil {
+				t.Fatalf("Context timed out waiting for stream.SendMsg() to return an error")
+			}
+			if status.Code(err) != codes.Canceled {
 				t.Errorf("stream.SendMsg() = %v, want error %v", err, codes.Canceled)
 			}
 			close(handlerDone)