| 555 | } |
| 556 | |
| 557 | Status KrpcDataStreamSender::Channel::DoTransmitDataRpc() { |
| 558 | DCHECK(rpc_in_flight_batch_ != nullptr); |
| 559 | DCHECK(rpc_in_flight_batch_->IsInitialized()); |
| 560 | |
| 561 | // Initialize some constant fields in the request protobuf. |
| 562 | TransmitDataRequestPB req; |
| 563 | *req.mutable_dest_fragment_instance_id() = fragment_instance_id_; |
| 564 | req.set_sender_id(parent_->sender_id_); |
| 565 | req.set_dest_node_id(dest_node_id_); |
| 566 | |
| 567 | // Set the RowBatchHeader in the request. |
| 568 | req.set_allocated_row_batch_header( |
| 569 | const_cast<RowBatchHeaderPB*>(rpc_in_flight_batch_->header())); |
| 570 | |
| 571 | rpc_controller_.Reset(); |
| 572 | int sidecar_idx; |
| 573 | // Add 'tuple_offsets_' as sidecar. |
| 574 | KUDU_RETURN_IF_ERROR(rpc_controller_.AddOutboundSidecar(RpcSidecar::FromSlice( |
| 575 | rpc_in_flight_batch_->TupleOffsetsAsSlice()), &sidecar_idx), |
| 576 | "Unable to add tuple offsets to sidecar"); |
| 577 | req.set_tuple_offsets_sidecar_idx(sidecar_idx); |
| 578 | |
| 579 | // Add 'tuple_data_' as sidecar. |
| 580 | rpc_start_time_ns_ = MonotonicNanos(); |
| 581 | KUDU_RETURN_IF_ERROR(rpc_controller_.AddOutboundSidecar( |
| 582 | RpcSidecar::FromSlice(rpc_in_flight_batch_->TupleDataAsSlice()), &sidecar_idx), |
| 583 | "Unable to add tuple data to sidecar"); |
| 584 | req.set_tuple_data_sidecar_idx(sidecar_idx); |
| 585 | |
| 586 | resp_.Clear(); |
| 587 | proxy_->TransmitDataAsync(req, &resp_, &rpc_controller_, |
| 588 | boost::bind(&KrpcDataStreamSender::Channel::TransmitDataCompleteCb, this)); |
| 589 | // 'req' took ownership of 'header'. Need to release its ownership or 'header' will be |
| 590 | // deleted by destructor. |
| 591 | req.release_row_batch_header(); |
| 592 | return Status::OK(); |
| 593 | } |
| 594 | |
| 595 | Status KrpcDataStreamSender::Channel::TransmitData( |
| 596 | unique_ptr<OutboundRowBatch>* outbound_batch, bool swap_batch) { |
nothing calls this directly
no test coverage detected