public void consume(Inbox inbox) {
        ensureNotDone();
        if (limit <= 0) {
            done.compareAndSet(null, new ResultLimitReachedException());
            ensureNotDone();
        }

        while (offset > 0 && inbox.poll() != null) {
            offset--;
        }

        for (JetSqlRow row; (row = (JetSqlRow) inbox.peek()) != null && rows.offer(row); ) {
            inbox.remove();
            if (limit != Long.MAX_VALUE) {
                limit -= 1;
                if (limit < 1) {
                    done.compareAndSet(null, new ResultLimitReachedException());
                    ensureNotDone();
                }
            }
        }
    }
@Test
    public void when_queueCapacityExceeded_then_inboxNotConsumed() {
        initProducer(false);
        int numExcessItems = 2;
        for (int i = 0; i < QueryResultProducerImpl.QUEUE_CAPACITY + numExcessItems; i++) {
            inbox.queue().add(jetRow());
        }
        producer.consume(inbox);
        assertThat(inbox).hasSize(2);
    }