TFE_ExecutorWaitForAllPendingNodes(executor, status);
  ASSERT_EQ(TF_OK, TF_GetCode(status)) << TF_Message(status);

  // TODO(b/136478427): Figure out how to correctly shut the server down.
  worker_server.release();

  // Update the server def with a new set of names (worker instead of
  // localhost).
  tensorflow::ServerDef updated_server_def = GetServerDef("worker", 2);
  serialized = updated_server_def.SerializeAsString();

  forward_op_->attrs.BuildNodeDef();
  std::unique_ptr<GradientFunction> gradient_fn;
  TF_RETURN_IF_ERROR(registry.Lookup(*forward_op_, &gradient_fn));
  tape->RecordOperation(forward_op_->inputs, retvals, gradient_fn.release(),
                        op_->Name());
  return absl::OkStatus();
}
}  // namespace internal

}  // namespace gradients

  return TensorHandlePtr(TFE_NewCustomDeviceTensorHandle(
      context, parallel_device_name.c_str(), t_released->dtype(), t_released,
      handle_methods, status));
}

// For TFE_CustomDevice::copy_tensor_to_device in the parallel device
// registration.
//
// Since this function is used to satisfy the TFE_CustomDevice C API,

  TFE_DeleteContext(ctx);

  TF_DeleteStatus(status);

  // TODO(b/136478427): Figure out how to correctly shut the server down.
  worker_server1.release();
  worker_server2.release();
}

TEST(CAPI, TestLocalFunctionWithPackedInput) {
  TestFunctionWithPackedInput(/*remote=*/false);
}

TEST(CAPI, TestRemoteFunctionWithPackedInput) {

                                    TF_Status* status) {
  status->status = value.status();
  if (!value.ok()) {
    return nullptr;
  }
  // Caller is responsible to call `TF_DeleteBuffer` to release the buffer.
  TF_Buffer* result = TF_NewBuffer();
  const std::string& value_str = *value;
  void* data = malloc(value_str.length());
  value_str.copy(static_cast<char*>(data), value_str.length(), 0);

};

// GraphOperation wraps and populates a TF_OperationDescription.
class GraphOperation : public TracingOperation {
 public:
  explicit GraphOperation(TF_Graph* g) : TracingOperation(kGraph), g_(g) {}
  void Release() override { delete this; }
  Status Reset(const char* op, const char* raw_device_name) override {
    if (op_) {
      return errors::FailedPrecondition("Reset called on already built op.");
    }

  status->status = new_op->Reset(op_or_function_name, nullptr);
  if (!status->status.ok()) {
    new_op->Release();
    new_op = nullptr;
  }
  return tensorflow::wrap(new_op);
}

void TFE_DeleteOp(TFE_Op* op) {
  if (op == nullptr) {
    return;
  }

  tensorflow::unwrap(op)->Release();
}

const char* TFE_OpGetName(const TFE_Op* op, TF_Status* status) {

    AbstractTensorHandle* conj_output;
    std::string name = "Conj_Exp_Grad";
    TF_RETURN_IF_ERROR(SafeConj(ctx, exp_.get(), &conj_output, name.c_str()));
    AbstractTensorHandlePtr conj_output_releaser(conj_output);

    name = "Mul_Exp_Grad";
    TF_RETURN_IF_ERROR(
        Mul(ctx, conj_output, grad_outputs[0], &grad_inputs[0], name.c_str()));
    return absl::OkStatus();
  }
  ~ExpGradientFunction() override {}

                                 status);
  EXPECT_EQ(TF_GetCode(status), TF_OK) << TF_Message(status);

  TFE_DeleteContextOptions(opts);
  TFE_DeleteContext(ctx);
  TF_DeleteStatus(status);

  w0.release();
  w1.release();
}

TEST(CAPI, ShareVariableAcrossContextsAfterUpdateContextWorksWithTimeout) {
  tensorflow::ServerDef server_def_0 = GetServerDef(3);

    switch (block->state) {
      case FetchState::ERROR:
        // TF_FALLTHROUGH_INTENDED
      case FetchState::CREATED:
        block->state = FetchState::FETCHING;
        block->mu.Unlock();  // Release the lock while making the API call.
        block->data.clear();
        block->data.resize(block_size_, 0);
        int64_t bytes_transferred;
        bytes_transferred = block_fetcher_(key.first, key.second, block_size_,