// given that memory allocation may trigger GC and block the thread.
	// Also note that emptyFunc is a placeholder, until we will be able
	// to compute watcher.forget function (which has to happen under lock).
	watcher := newCacheWatcher(
		chanSize,
		filterWithAttrsFunction(key, pred),
		emptyFunc,
		c.versioner,
		deadline,
		pred.AllowWatchBookmarks,
		c.groupResource,
		identifier,
	)

	}

	deadline := time.Now().Add(time.Hour)
	// After that, verifies the cacheWatcher.process goroutine works correctly.
	for i := 0; i < maxRetriesToProduceTheRaceCondition; i++ {
		w = newCacheWatcher(2, filter, emptyFunc, storage.APIObjectVersioner{}, deadline, false, schema.GroupResource{Resource: "pods"}, "")
		w.input <- &watchCacheEvent{Object: &v1.Pod{}, ResourceVersion: uint64(i + 1)}

  OwningOpRef<mlir::ModuleOp> mlir_module_;
};

TEST_F(FunctionClusterTensorflowDialectTest, ClustersTfReplicatedBridge) {
  CellReader<int64_t> compilation_status(kCompilationStreamz);

  TF_ASSERT_OK(CreateMlirModule("empty_func.mlir"));

  TF_EXPECT_OK(RunFunctionTf2xlaClusteringBridge(
      *mlir_module_, /*is_supported_by_replicated_brige*/ true,
      /*is_in_fallback_enabled_mode=*/false));

    "tf_device.cluster_func"() {_xla_compile_device_type = "TPU", _replication_info = "cluster0", func = @empty_func, num_cores_per_replica = 1, step_marker_location = "", topology = "", device_assignment = [], input_sharding_configuration = [], output_sharding_configuration = [], use_spmd_for_xla_partitioning = false} : () -> ()
    func.return
  }
  func.func @empty_func() {
    func.return
  }
}

// -----

        "@local_tsl//tsl/platform:errors",
    ],
)

tf_cc_test(
    name = "cluster_tf_test",
    srcs = ["cluster_tf_test.cc"],
    data = [
        "testdata/empty_func.mlir",
        "testdata/invalid_executor.mlir",
        "testdata/outside_compilation.mlir",
    ],
    deps = [
        ":cluster_tf",
        "//tensorflow/compiler/mlir:register_common_dialects",

        "@local_tsl//tsl/platform:errors",
    ],
)

tf_cc_test(
    name = "cluster_tf_test",
    srcs = ["cluster_tf_test.cc"],
    data = [
        "testdata/empty_func.mlir",
        "testdata/invalid_executor.mlir",
        "testdata/multiple_submodules.mlir",
    ],
    deps = [
        ":cluster_tf",
        "//tensorflow/compiler/mlir:register_common_dialects",

  OwningOpRef<mlir::ModuleOp> mlir_module_;
};

TEST_F(SessionClusterTensorflowDialectTest, ClustersTf) {
  CellReader<int64_t> compilation_status(kCompilationStreamz);

  TF_ASSERT_OK(CreateMlirModule("empty_func.mlir"));

  TF_EXPECT_OK(
      RunSessionTf2xlaClusteringBridge(*mlir_module_,
                                       /*is_in_fallback_enabled_mode=*/false));

  OwningOpRef<mlir::ModuleOp> mlir_module_;
};

TEST_F(TensorflowDialectToExecutorTest, ConvertsToExecutor) {
  CellReader<int64_t> compilation_status(kCompilationStreamz);

  TF_ASSERT_OK(CreateMlirModule("empty_func.mlir"));

  TF_EXPECT_OK(ExportFromTensorflowDialectToExecutor(*mlir_module_));

  EXPECT_EQ(compilation_status.Delta("success"), 1);
}

TEST_F(TensorflowDialectToExecutorTest, ErrorsWhenCannotConvert) {

				if fname != p.funcs[u.fnfid].fname {
					// New function; emit entry for previous one.
					if err := emitFunc(u); err != nil {
						return err
					}
				}
			}
			tstmts += uint64(u.NxStmts)
			count := p.unitTable[u]
			if count != 0 {
				cstmts += uint64(u.NxStmts)
			}
		}
		if err := emitFunc(extcu{}); err != nil {
			return err
		}
	}
	if _, err := fmt.Fprintf(tabber, "%s\t%s\t%.1f%%\n",

  OwningOpRef<mlir::ModuleOp> mlir_module_;
};

TEST_F(TensorflowDialectToExecutorTest, ConvertsToExecutor) {
  CellReader<int64_t> compilation_status(kExportStreamzName);

  TF_ASSERT_OK(CreateMlirModule("empty_func.mlir"));

  TF_EXPECT_OK(ExportFromTensorflowDialectToExecutor(*mlir_module_));

  EXPECT_EQ(compilation_status.Delta(kExportSuccess), 1);
  EXPECT_EQ(compilation_status.Delta(kExportFailed), 0);
}