}

  // Populate return_nodes
  DCHECK(tf_results->return_nodes.empty());
  tf_results->return_nodes.resize(results.return_nodes.size());
  for (int i = 0; i < results.return_nodes.size(); ++i) {
    tf_results->return_nodes[i] = ToOperation(results.return_nodes[i]);
  }

  // Populate missing unused map keys

    }
  }

  // Compute body nodes.
  std::vector<const Node*> body_nodes;
  status->status = tensorflow::ComputeBodyNodes(
      fn_body, fn_name, num_opers, opers, input_nodes, &body_nodes);
  if (TF_GetCode(status) != TF_OK) return nullptr;

  // Compute body nodes.
  std::vector<const Node*> control_output_nodes;
  control_output_nodes.reserve(ncontrol_outputs);

                            nodes_preserved.size(), storage.get(), storage_size,
                            status);
  EXPECT_EQ(TF_OK, TF_GetCode(status)) << TF_Message(status);

  for (size_t i = 0; i < nodes_preserved.size(); ++i) {
    EXPECT_EQ(nodes_preserved.find(string(static_cast<const char*>(values[i]),
                                          lens[i])) != nodes_preserved.end(),
              true);
  }

        // parallel device.
        device_(device),
        executor_(
            TFE_NewExecutor(is_async, /*enable_streaming_enqueue=*/true,
                            /*in_flight_nodes_limit=*/in_flight_nodes_limit)),
        op_(nullptr),
        thread_(tensorflow::Env::Default()->StartThread(
            tensorflow::ThreadOptions(), "parallel_device_execute",
            std::bind(&DeviceThread::Run, this))) {}

#include "tensorflow/core/tfrt/common/pjrt_util.h"
#include "tsl/lib/core/status_test_util.h"
#include "tsl/platform/casts.h"
#include "tsl/platform/status_matchers.h"
#include "tsl/protobuf/error_codes.pb.h"

namespace tensorflow {
namespace {

using ::testing::HasSubstr;
using ::testing::NotNull;
using ::tsl::testing::StatusIs;

PJRT_Buffer* CreateCBuffer() {
  auto status = pjrt::PjrtApi(DEVICE_CPU);

    // Inject failure to function instantiation if finding a node that contains
    // the given node name (error_node_) and requested device (error_device_).
    for (const auto node : graph->get()->nodes()) {
      if (node->name().find(error_node_) != string::npos &&
          node->requested_device() == error_device_) {
        return tensorflow::errors::Internal("Injected graph pass error.");
      }
    }

      // Name
      "AssertAndCollective",
      // Args
      {"x: float", "condition: bool"},
      // Return values
      {"y: float"},
      // Attr def
      {},
      // Nodes
      {
          {{"assert"},
           "Assert",
           {"condition", "x"},
           {{"T", std::vector<DataType>{DT_FLOAT}}}},
          {{"y"},
           "CollectiveReduce",
           {"x"},

  ASSERT_EQ(1, TF_OperationNumInputs(neg));
  TF_Output neg_input = TF_OperationInput({neg, 0});
  EXPECT_EQ(scalar, neg_input.oper);
  EXPECT_EQ(0, neg_input.index);

  // Test that we can't see control edges involving the source and sink nodes.
  TF_Operation* control_ops[100];
  EXPECT_EQ(0, TF_OperationNumControlInputs(scalar));
  EXPECT_EQ(0, TF_OperationGetControlInputs(scalar, control_ops, 100));

    TFE_DeleteContextOptions(opts);
    TFE_Executor* executor = TFE_NewExecutor(
        /*is_async=*/false, /*enable_streaming_enqueue=*/true,
        /*in_flight_nodes_limit=*/0);
    TFE_ContextSetExecutorForThread(ctx, executor);

    TFE_DeleteContext(ctx);
    TFE_DeleteExecutor(executor);
  }

  {
    TFE_ContextOptions* opts = TFE_NewContextOptions();

    tensorflow::ConfigProto config;
    // Do not skip grappler optimization even for small graphs.
    config.mutable_graph_options()
        ->mutable_rewrite_options()
        ->set_min_graph_nodes(-1);
    string serialized_config;
    ASSERT_TRUE(config.SerializeToString(&serialized_config));
    TFE_OpSetAttrString(
        op, "config_proto",
        reinterpret_cast<const void*>(serialized_config.c_str()),