%instance_key = "tf.Const"() { value = dense<3> : tensor<i32> } : () -> tensor<i32>
  // CHECK: %[[REDUCE:.*]] = "mhlo.all_reduce"
  // CHECK-SAME{LITERAL}: replica_groups = dense<[[0, 1]]> : tensor<1x2xi64>
  // CHECK: mhlo.add
  // CHECK: mhlo.return
  // CHECK: return %[[REDUCE]]

  Pattern<
    (TF_CollectiveReduceV2Op:$src_op
       $input, $group_size, $group_key, $instance_key, $ordering_token,
       $merge_op, $final_op, $communication_hint, $timeout_seconds,
       $is_stateless, $max_subdivs_per_device
    ),
    [(TF_CollectiveReduceV2Op
       $input, $group_size, $group_key, $instance_key, (EmptyList),
       $merge_op, $final_op, $communication_hint, $timeout_seconds,

  %group_key: tensor<i32>,
  %group_size: tensor<i32>,
  %instance_key: tensor<i32>) {
  tf_executor.graph {
    // expected-remark@above {{ID: 5}}
    %island = tf_executor.island {
        // expected-remark@above {{ID: 3}}
        // expected-remark@above {{Successors: {4}}}

  TFE_OpSetAttrType(op.get(), "T", TFE_TensorHandleDataType(input));
  TFE_OpSetAttrInt(op.get(), "group_size", group_size);
  TFE_OpSetAttrInt(op.get(), "group_key", 0);
  TFE_OpSetAttrInt(op.get(), "instance_key", 0);
  const std::string merge_op("Add");
  TFE_OpSetAttrString(op.get(), "merge_op", merge_op.c_str(),
                      merge_op.length());
  const std::string final_op("Id");

          {{"y"},
           "CollectiveReduce",
           {"x"},
           {{"T", DT_FLOAT},
            {"group_size", static_cast<int>(devices.size())},
            {"group_key", 0},
            {"instance_key", 0},
            {"merge_op", "Add"},
            {"final_op", "Id"},
            {"subdiv_offsets", std::vector<int>()}},
           /*dep=*/{"assert"}},
      });

  TF_DeleteStatus(status);

  TFE_OpSetAttrType(op, "T", TFE_TensorHandleDataType(in));
  TFE_OpSetAttrInt(op, "group_size", group_size);
  TFE_OpSetAttrInt(op, "group_key", 123);
  TFE_OpSetAttrInt(op, "instance_key", 456);
  TFE_OpSetAttrString(op, "merge_op", "Add", 3);
  TFE_OpSetAttrString(op, "final_op", "Id", 2);
  std::vector<int64_t> subdiv_offsets;
  TFE_OpSetAttrIntList(op, "subdiv_offsets", subdiv_offsets.data(),

  %group_key: tensor<i32>,
  %instance_key: tensor<i32>,
  %token: tensor<!tf_type.resource<tensor<f32>>>
) -> tensor<3xi64> {
  %0 = "tf_device.cluster"() ({
    // arg4 shall have been removed.
    // CHECK: tf.CollectiveReduceV2
    // CHECK-NOT: [[TOKEN]]
    // CHECK-SAME: merge_op
    %0 = "tf.CollectiveReduceV2"(%input, %group_size, %group_key, %instance_key, %token) {

	"istio.io/istio/pkg/util/sets"
)

type hostPort struct {
	host instancesKey
	port int
}

// stores all the service instances from SE, WLE and pods
type serviceInstancesStore struct {
	ip2instance map[string][]*model.ServiceInstance
	// service instances by hostname -> config
	instances map[instancesKey]map[configKeyWithParent][]*model.ServiceInstance
	// instances only for serviceentry

// For `CollectiveReduceV2Op` we have 3 cases:
// 1) `is_stateless` is true turns off automatic ordering and we purely rely on
//    instance_key to distinguish collective groups. In this case, ordering
//    tokens are irrelevant. Each collective group should have a unique
//    instance_key at runtime.
// 2) If at least one ordering token is present, then we purely rely on ordering

collective ops. In this case, keys that are unique at runtime
(e.g. `instance_key`) should be used to distinguish collective groups.
  }];

  let arguments = (ins
    TF_FpOrI32OrI64Tensor:$input,
    TF_Int32Tensor:$group_size,
    TF_Int32Tensor:$group_key,
    TF_Int32Tensor:$instance_key,
    Variadic<TF_ResourceTensor>:$ordering_token,