ReferenceQueue<Object> queue,
      PhantomReference<Object> frqReference) {
    this.queue = queue;

    this.finalizableReferenceClassReference = new WeakReference<>(finalizableReferenceClass);

    // Keep track of the FRQ that started us so we know when to stop.
    this.frqReference = frqReference;
  }

  /** Loops continuously, pulling references off the queue and cleaning them up. */

	}
}

func SetDefaults_QueuingConfiguration(cfg *v1beta3.QueuingConfiguration) {
	if cfg.HandSize == 0 {
		cfg.HandSize = PriorityLevelConfigurationDefaultHandSize
	}
	if cfg.Queues == 0 {
		cfg.Queues = PriorityLevelConfigurationDefaultQueues
	}
	if cfg.QueueLengthLimit == 0 {
		cfg.QueueLengthLimit = PriorityLevelConfigurationDefaultQueueLengthLimit
	}

type traceBufQueue struct {
	head, tail *traceBuf
}

// push queues buf into queue of buffers.
func (q *traceBufQueue) push(buf *traceBuf) {
	buf.link = nil
	if q.head == nil {
		q.head = buf
	} else {
		q.tail.link = buf
	}
	q.tail = buf
}

// pop dequeues from the queue of buffers.
func (q *traceBufQueue) pop() *traceBuf {
	buf := q.head
	if buf == nil {

	if end < start || start < 0 || end > len(b.old) {
		panic("invalid edit position")
	}
	b.q = append(b.q, edit{start, end, new})
}

// Bytes returns a new byte slice containing the original data
// with the queued edits applied.
func (b *Buffer) Bytes() []byte {
	// Sort edits by starting position and then by ending position.
	// Breaking ties by ending position allows insertions at point x

	key, quit := jm.queue.Get()
	if quit {
		return false
	}
	defer jm.queue.Done(key)

	requeueAfter, err := jm.sync(ctx, key)
	switch {
	case err != nil:
		utilruntime.HandleError(fmt.Errorf("error syncing CronJobController %v, requeuing: %w", key, err))
		jm.queue.AddRateLimited(key)
	case requeueAfter != nil:
		jm.queue.Forget(key)
		jm.queue.AddAfter(key, *requeueAfter)
	}

| `minio_replication_average_queued_bytes`          | `gauge` | Average number of bytes queued for replication since server start           | `server` |
| `minio_replication_average_queued_count`          | `gauge` | Average number of objects queued for replication since server start         | `server` |

        where:
        isolationMode << ISOLATION_MODES
    }

    def "errors in work submitted to other queues cause a task failure when waiting for work in #isolationMode"() {
        buildFile << """
            task runWork(type: WorkItemTask) {
                doLast {

            T operation = firstOperation;
            while (operation != null) {
                if (queueState == QueueState.Cancelled) {
                    // If an operation was pulled from the queue, but the queue was cancelled before this operation could start
                    // (for instance, because this worker was waiting on a worker lease) discard it without running.
                    return ++operationCount;

    def listener = Mock(OutputEventListener)
    def clock = new MockClock()
    def executor = new MockExecutor()

    @Subject renderer = new ThrottlingOutputEventListener(listener, 100, executor, clock)

    def "queues events until update event received"() {
        def event1 = event('message')
        def event2 = event('message')

        when:
        renderer.onOutput(event1)
        renderer.onOutput(event2)

        then:

}

func (t *trackingWorkqueue[T]) queue(item T) {
	if _, queued := t.pendingMap[item]; queued {
		// fmt.Printf("already queued: %#v\n", item)
		return
	}
	t.pendingMap[item] = struct{}{}
	t.pendingList = append(t.pendingList, item)
}
func (t *trackingWorkqueue[T]) dequeue(item T) {
	if _, queued := t.pendingMap[item]; !queued {
		// fmt.Printf("not queued: %#v\n", item)
		return
	}