Apply backpressure for ext proc calls for response body messages too by overriding ClientCall.requestMessages(int) in ExtProcClientCall. Also introduce null check for extProcClientCallRequestObserver in isReady since it may be called on the call even before start is called that initializes it.

Author: kannanjgithub

xds/src/main/java/io/grpc/xds/ExternalProcessorFilter.java

@@ -497,11 +497,20 @@ public boolean isReady() {
           return false;
         }
         if (config.getObservabilityMode()) {
-          return super.isReady() && extProcClientCallRequestObserver.isReady();
+          return super.isReady() && extProcClientCallRequestObserver != null
+              && extProcClientCallRequestObserver.isReady();
         }
         return super.isReady();
       }
 
+      @Override
+      public void request(int numMessages) {
+        if (config.getObservabilityMode() && !isReady()) {
+          return;
+        }
+        super.request(numMessages);
+      }
+
       @Override
       public void sendMessage(InputStream message) {
         if (extProcStreamCompleted.get()) {