fixes needed to enable retries for all tests (#26839)

src/core/ext/filters/client_channel/retry_filter.cc

@@ -1366,6 +1366,9 @@ void RetryFilter::CallData::CallAttempt::BatchData::
   grpc_metadata_batch_move(
       &call_attempt_->recv_initial_metadata_,
       pending->batch->payload->recv_initial_metadata.recv_initial_metadata);
+  // Propagate trailing_metadata_available.
+  *pending->batch->payload->recv_initial_metadata.trailing_metadata_available =
+      call_attempt_->trailing_metadata_available_;
   // Update bookkeeping.
   // Note: Need to do this before invoking the callback, since invoking
   // the callback will result in yielding the call combiner.
@@ -1593,6 +1596,10 @@ void RetryFilter::CallData::CallAttempt::BatchData::
     call_attempt_->recv_trailing_metadata_error_ = error;
     return;
   }
+  // Copy transport stats to be delivered up to the surface.
+  grpc_transport_move_stats(
+      &call_attempt_->collect_stats_,
+      pending->batch->payload->recv_trailing_metadata.collect_stats);
   // Return metadata.
   grpc_metadata_batch_move(
       &call_attempt_->recv_trailing_metadata_,
@@ -1762,6 +1769,11 @@ void RetryFilter::CallData::CallAttempt::BatchData::
     GRPC_ERROR_UNREF(error);
     return;
   }
+  // Propagate payload.
+  if (batch_.send_message) {
+    pending->batch->payload->send_message.stream_write_closed =
+        batch_.payload->send_message.stream_write_closed;
+  }
   // Add closure.
   closures->Add(pending->batch->on_complete, error,
                 "on_complete for pending batch");

src/core/ext/filters/http/client/http_client_filter.cc

@@ -232,6 +232,9 @@ static void send_message_on_complete(void* arg, grpc_error_handle error) {
   grpc_call_element* elem = static_cast<grpc_call_element*>(arg);
   call_data* calld = static_cast<call_data*>(elem->call_data);
   calld->send_message_cache.Destroy();
+  // Set the batch's send_message bit back to true, so the retry code
+  // above knows what was in this batch.
+  calld->send_message_batch->send_message = true;
   grpc_core::Closure::Run(DEBUG_LOCATION,
                           calld->original_send_message_on_complete,
                           GRPC_ERROR_REF(error));
@@ -474,8 +477,8 @@ static void http_client_start_transport_stream_op_batch(
 
 done:
   if (error != GRPC_ERROR_NONE) {
-    grpc_transport_stream_op_batch_finish_with_failure(
-        calld->send_message_batch, error, calld->call_combiner);
+    grpc_transport_stream_op_batch_finish_with_failure(batch, error,
+                                                       calld->call_combiner);
   } else if (!batch_will_be_handled_asynchronously) {
     grpc_call_next_op(elem, batch);
   }