std::unique_ptr<AsyncValueAllocator> pjrt_allocator_ TF_GUARDED_BY(mu_);

  // Stream associated with this device. Operations enqueued on this
  // stream are executed on the device. Operations include data
  // copying back and forth between CPU and the device, and
  // computations enqueued by XLA.
  std::shared_ptr<se::Stream> stream_ TF_GUARDED_BY(mu_);
  // If false, only stream_ is valid and all computation and transfers use

// RUN: tf-tfrt-opt -split-input-file -tfrt-reconfig-batch-op="tfrt-min-num-batch-threads=2 tfrt-min-max-enqueued-batches=3" %s | FileCheck %s --dump-input=always

// -----

// The num_batch_threads is lowered bound to 2 from the original attribute of 1

// CHECK-LABEL: func private @batched_function
func.func private @batched_function(%arg0: tensor<1x3xf32>) -> tensor<1x3xf32> {
  %2 = "tf.Identity"(%arg0) : (tensor<1x3xf32>) -> tensor<1x3xf32>

		t.Fatalf("Failed with err: %v", err)
	}
	if len(gots) != len(wants) {
		t.Fatalf("Expected %d objects but got %d", len(wants), len(gots))
	}

	// Close expiry state's channel to inspect object versions enqueued for expiration
	close(workers[0])
	wg.Wait()
	for _, obj := range expired {
		switch obj.ObjectV.VersionID {
		case uuids[2].String(), uuids[3].String(), uuids[4].String():
		default:

						{
							PluginName:     "fooPlugin1",
							QueueingHintFn: queueHintReturnQueue,
						},
					},
				},
			},
		},
		{
			name:                         "Pod, registered in inFlightPods, is enqueued back to activeQ",
			isSchedulingQueueHintEnabled: true,
			initialPods:                  []*v1.Pod{pod, pod2},
			actions: []action{
				// This won't be added to inFlightEvents because no inFlightPods at this point.

    }
    for (existingCall in readyAsyncCalls) {
      if (existingCall.host == host) return existingCall
    }
    return null
  }

  /**
   * Cancel all calls currently enqueued or executing. Includes calls executed both
   * [synchronously][Call.execute] and [asynchronously][Call.enqueue].
   */
  fun cancelAll() {
    this.withLock {
      for (call in readyAsyncCalls) {

      "FINE: Q10000 scheduled after 100 µs: task",
      "FINE: Q10000 starting              : task",
      "FINE: Q10000 finished run in   0 µs: task",
    )
  }

  /** Inspect how many runnables have been enqueued. If none then we're truly sequential. */
  @Test fun singleQueueIsSerial() {
    redQueue.execute("task one", 100.µs) {
      log += "one:run@${taskFaker.nanoTime} parallel=${taskFaker.isParallel}"
    }

	set := newStatefulSet(3)
	pod := newStatefulSetPod(set, 0)
	ssc.addPod(logger, pod)
	ssc.queue.ShutDown()
	key, _ := ssc.queue.Get()
	if key != "" {
		t.Errorf("StatefulSet enqueued key for Pod with no Set %s", key)
	}
}

func TestStatefulSetControllerUpdatePod(t *testing.T) {
	logger, ctx := ktesting.NewTestContext(t)
	ssc, _, om, _ := newFakeStatefulSetController(ctx)

			manager.enqueueDaemonSet = func(ds *apps.DaemonSet) {
				if ds.Name == "ds" {
					enqueued = true
				}
			}

			enqueued = false
			manager.updateNode(logger, c.oldNode, c.newNode)
			if enqueued != c.shouldEnqueue {
				t.Errorf("Test case: '%s', expected: %t, got: %t", c.test, c.shouldEnqueue, enqueued)
			}
		}
	}
}

	start := uintptr(unsafe.Pointer(&pp.wbBuf.buf[0]))
	n := (pp.wbBuf.next - start) / unsafe.Sizeof(pp.wbBuf.buf[0])
	ptrs := pp.wbBuf.buf[:n]

	// Poison the buffer to make extra sure nothing is enqueued
	// while we're processing the buffer.
	pp.wbBuf.next = 0

	if useCheckmark {
		// Slow path for checkmark mode.
		for _, ptr := range ptrs {
			shade(ptr)
		}
		pp.wbBuf.reset()
		return

    }

    /**
     * Returns true if the request contains conditions that save the server from sending a response
     * that the client has locally. When a request is enqueued with its own conditions, the built-in
     * response cache won't be used.
     */
    private fun hasConditions(request: Request): Boolean =