}
        try {
            return createNow(mavenSettings, pomFile);
        } catch (Exception e) {
            throw new MavenConversionException(String.format("Unable to create Maven project model using POM %s.", pomFile), e);
        }
    }

		})
		ads.ExpectEmptyResponse()

		// Create pod, due to wildcard subscribe we should receive it
		createPod(s, "pod", "sa", "127.0.0.1", "not-node")
		expect(ads.ExpectResponse(), "Kubernetes//Pod/default/pod")

		// A new pod should push only that one
		createPod(s, "pod2", "sa", "127.0.0.2", "node")
		expect(ads.ExpectResponse(), "Kubernetes//Pod/default/pod2")

}

func TestHasInitialEventsEndBookmarkAnnotation(t *testing.T) {
	createPod := func(name string) *example.Pod {
		return &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: name}}
	}
	createAnnotatedPod := func(name, value string) *example.Pod {
		p := createPod(name)
		p.Annotations = map[string]string{}
		p.Annotations[metav1.InitialEventsAnnotationKey] = value
		return p
	}

  %ch0 = tfrt.new.chain

  %ch1 = tfrt_fallback_async.createop(%ch0) key(0) device("/CPU:0") "tf.Const"()
         { dtype = i32, value = dense<[2]> : tensor<1xi32> } num_args(0)
  %ch2 = tfrt_fallback_async.createop(%ch1) key(1) device("/CPU:0")
         "tf.TestAsyncTfrtAsyncThread"() {T = i32} num_args(1)
  %ch3 = tfrt_fallback_async.createop(%ch2) key(2) device("/CPU:0")
         "tf.TestPrintThreadName"() num_args(0)

#     # copybara:uncomment driver = "//tensorflow/compiler/mlir:run_lit.sh",
#     exclude = [
#         "compile.benchmark.large.mlir",
#         "batch_function_fallback.mlir",
#         "create_op.mlir",
#         "custom_thread_pool.mlir",
#     ],
#     # copybara:uncomment flaky = ["compile.error.mlir"],
#     size_override = {
#         "compile.benchmark.small.mlir": "medium",

		client := &mockRecyclerClient{
			events: test.eventSequence,
			pod:    test.existingPod,
		}
		err := internalRecycleVolumeByWatchingPodUntilCompletion(test.createPod.Name, test.createPod, client)
		receivedError := ""
		if err != nil {
			receivedError = err.Error()
		}
		if receivedError != test.expectedError {
			t.Errorf("Test %q failed, expected error %q, got %q", test.name, test.expectedError, receivedError)

	})
	initTestEnv(t, c.client.Kube(), fx)
	createPod := func(ip, name string) {
		addPods(t, c, fx, generatePod(ip, name, "ns", "1", "", map[string]string{}, map[string]string{}))
	}

	createPod("128.0.0.1", "pod1")
	assert.Equal(t, c.pods.getPodKeys("128.0.0.1"), []types.NamespacedName{{Name: "pod1", Namespace: "ns"}})
	events.WaitOrdered("pod1/add", "pod1/update")
	createPod("128.0.0.1", "pod2")

// created as an interface to allow testing.
type jobControlInterface interface {
	// GetJob retrieves a Job.
	GetJob(namespace, name string) (*batchv1.Job, error)
	// CreateJob creates new Jobs according to the spec.
	CreateJob(namespace string, job *batchv1.Job) (*batchv1.Job, error)
	// DeleteJob deletes the Job identified by name.
	// TODO: delete by UID?
	DeleteJob(namespace string, name string) error
}

  );

  let results = (outs
    Variadic<TensorType>:$results
  );

  let assemblyFormat = "attr-dict `:` type($results)";
}

def CreateOp: FallbackSync_Op<"createop", []> {
  let summary = "The Fallback CreateOp";

  let description = [{
    The CreateOp creates the tensorflow::OpKernel in the fallback context.

    op_key: The unique key to identify this op.
    num_args: The number of inputs.

  tfrt.return %1 : i1
}

// CHECK-LABEL: func @createop
func.func @createop(%in_ch: !tfrt.chain) -> !tfrt.chain {

  // CHECK: [[ch:%.*]] = tfrt_fallback_async.createop(%{{.*}}) key(100) device("cpu") "tf.AddV2"() {T = i32} num_args(2)
  %out_ch = tfrt_fallback_async.createop(%in_ch) key(100) device("cpu") "tf.AddV2"() {T = i32} num_args(2)

  // CHECK: tfrt.return [[ch]]