Fix incorrect usage of grpc API

Signed-off-by: joshlee <joshlee@anyscale.com>

Author: Sparks0219

src/ray/ray_syncer/ray_syncer_bidi_reactor_base.h

@@ -175,10 +175,12 @@ class RaySyncerBidiReactorBase : public RaySyncerBidiReactor, public T {
          ok,
          disconnected = IsDisconnected(),
          msg = std::move(receiving_message_)]() mutable {
-          if (*disconnected) {
-            return;
-          }
-
+          // NOTE: According to the grpc callback streaming api best practices 3.)
+          // https://grpc.io/docs/languages/cpp/best_practices/#callback-streaming-api
+          // The client must read all incoming data i.e. until OnReadDone(ok = false)
+          // happens for OnDone to be called. Hence even if disconnected_ is true, we
+          // still need to allow OnReadDone to repeatedly execute until StartReadData has
+          // consumed all the data for OnDone to be called.
           if (!ok) {
             RAY_LOG_EVERY_MS(INFO, 1000) << "Failed to read a message from node: "
                                          << NodeID::FromBinary(GetRemoteNodeID());