@Override
    public AvailabilityWithBacklog getAvailabilityAndBacklog(boolean isCreditAvailable) {
        synchronized (lock) {
            try {
                cacheBuffer();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            if (cachedBuffers.isEmpty()) {
                return new AvailabilityWithBacklog(false, 0);
            }

            return new AvailabilityWithBacklog(
                    isCreditAvailable || cachedBuffers.peek().f0.buffer().getDataType().isEvent(),
                    (int)
                            cachedBuffers.stream()
                                    .filter(x -> x.f0.buffer().getDataType().isBuffer())
                                    .count());
        }
    }
@Test
    void testGetAvailabilityAndBacklog() throws IOException {
        view0.notifyDataAvailable();
        view1.notifyDataAvailable();

        ResultSubpartitionView.AvailabilityWithBacklog availabilityAndBacklog1 =
                view.getAvailabilityAndBacklog(false);
        assertThat(availabilityAndBacklog1.getBacklog()).isPositive();
        assertThat(availabilityAndBacklog1.isAvailable()).isFalse();
        ResultSubpartitionView.AvailabilityWithBacklog availabilityAndBacklog2 =
                view.getAvailabilityAndBacklog(true);
        assertThat(availabilityAndBacklog2.getBacklog()).isPositive();
        assertThat(availabilityAndBacklog2.isAvailable()).isTrue();

        for (int i = 1; i < buffers0.size() + buffers1.size(); i++) {
            view.getNextBuffer();
        }

        ResultSubpartitionView.AvailabilityWithBacklog availabilityAndBacklog3 =
                view.getAvailabilityAndBacklog(false);
        assertThat(availabilityAndBacklog3.getBacklog()).isZero();
        assertThat(availabilityAndBacklog3.isAvailable()).isTrue();
        ResultSubpartitionView.AvailabilityWithBacklog availabilityAndBacklog4 =
                view.getAvailabilityAndBacklog(true);
        assertThat(availabilityAndBacklog4.getBacklog()).isZero();
        assertThat(availabilityAndBacklog4.isAvailable()).isTrue();
    }