@VisibleForTesting
    static IndexRange computeConsumedSubpartitionRange(
            int consumerSubtaskIndex,
            int numConsumers,
            Supplier<Integer> numOfSubpartitionsSupplier,
            boolean isDynamicGraph,
            boolean isBroadcast) {
        int consumerIndex = consumerSubtaskIndex % numConsumers;
        if (!isDynamicGraph) {
            return new IndexRange(consumerIndex, consumerIndex);
        } else {
            int numSubpartitions = numOfSubpartitionsSupplier.get();
            if (isBroadcast) {
                // broadcast results have only one subpartition, and be consumed multiple times.
                checkArgument(numSubpartitions == 1);
                return new IndexRange(0, 0);
            } else {
                checkArgument(consumerIndex < numConsumers);
                checkArgument(numConsumers <= numSubpartitions);

                int start = consumerIndex * numSubpartitions / numConsumers;
                int nextStart = (consumerIndex + 1) * numSubpartitions / numConsumers;

                return new IndexRange(start, nextStart - 1);
            }
        }
    }
@Test
    void testComputeBroadcastConsumedSubpartitionRange() {
        final IndexRange range1 = computeConsumedSubpartitionRange(0, 3, 1, true, true);
        assertThat(range1).isEqualTo(new IndexRange(0, 0));

        final IndexRange range2 = computeConsumedSubpartitionRange(1, 3, 1, true, true);
        assertThat(range2).isEqualTo(new IndexRange(0, 0));

        final IndexRange range3 = computeConsumedSubpartitionRange(2, 3, 1, true, true);
        assertThat(range3).isEqualTo(new IndexRange(0, 0));
    }