Filter zero length slices passed to grpc passthru endpoint (#28205)

* updating passthru endpoint to filter 0 length slices passed to endpoint write

* adding testcase which caused the failure

* fix minor bug in passthru endpoint to handle rare case where endpoint write is called with no data to be written

test/core/end2end/fuzzers/api_fuzzer_corpus/testcase-4563554393260032

@@ -0,0 +1,46 @@
+actions {
+  create_server {
+  }
+}
+actions {
+  create_channel {
+    target: "unix::::::::::::::::::::::::::;:::::;:::::::::::::>:::::::::::::::::::::::9:::::\026I:::::::c::,:::\332\261::::::::::::::::"
+    channel_actions {
+      add_n_bytes_writable: 18446744073709551607
+      add_n_bytes_readable: 18446744073709551607
+    }
+  }
+}
+actions {
+  create_call {
+    propagation_mask: 6881280
+    method {
+      value: "/foo"
+    }
+    timeout: 1953002608
+  }
+}
+actions {
+  queue_batch {
+    operations {
+      send_initial_metadata {
+      }
+    }
+    operations {
+      send_message {
+        message {
+          intern: true
+        }
+        message {
+          value: "Bh\007-600"
+          intern: true
+        }
+      }
+    }
+  }
+}
+actions {
+  advance_time: 6881280
+}
+actions {
+}

test/core/util/passthru_endpoint.cc

@@ -235,6 +235,7 @@ static void do_pending_write_op_locked(half* m, grpc_error_handle error) {
     GPR_ASSERT(max_writable >= static_cast<uint64_t>(split_length));
     max_writable -= split_length;
   }
+
   if (immediate_bytes_read > 0) {
     GPR_ASSERT(!other->pending_read_op.is_armed);
     if (m->parent->simulate_channel_actions) {
@@ -264,17 +265,24 @@ static void me_write(grpc_endpoint* ep, grpc_slice_buffer* slices,
         GRPC_ERROR_CREATE_FROM_STATIC_STRING("Endpoint already shutdown"));
   } else {
     GPR_ASSERT(!m->pending_write_op.is_armed);
-    m->pending_write_op.is_armed = true;
-    m->pending_write_op.cb = cb;
-    m->pending_write_op.ep = ep;
     // Copy slices into m->pending_write_op.slices
     m->pending_write_op.slices = &m->write_buffer;
     GPR_ASSERT(m->pending_write_op.slices->count == 0);
     for (int i = 0; i < static_cast<int>(slices->count); i++) {
-      grpc_slice_buffer_add_indexed(m->pending_write_op.slices,
-                                    grpc_slice_copy(slices->slices[i]));
+      if (GPR_SLICE_LENGTH(slices->slices[i]) > 0) {
+        grpc_slice_buffer_add_indexed(m->pending_write_op.slices,
+                                      grpc_slice_copy(slices->slices[i]));
+      }
+    }
+    if (m->pending_write_op.slices->count > 0) {
+      m->pending_write_op.is_armed = true;
+      m->pending_write_op.cb = cb;
+      m->pending_write_op.ep = ep;
+      do_pending_write_op_locked(m, GRPC_ERROR_NONE);
+    } else {
+      // There is nothing to write. Schedule callback to be run right away.
+      grpc_core::ExecCtx::Run(DEBUG_LOCATION, cb, GRPC_ERROR_NONE);
     }
-    do_pending_write_op_locked(m, GRPC_ERROR_NONE);
   }
   gpr_mu_unlock(&m->parent->mu);
 }