@Override
    public ResultSubpartitionView createSubpartitionView(
            ResultSubpartitionIndexSet indexSet, BufferAvailabilityListener availabilityListener)
            throws IOException {
        if (indexSet.size() == 1) {
            return createSubpartitionView(
                    indexSet.values().iterator().next(), availabilityListener);
        } else {
            UnionResultSubpartitionView unionView =
                    new UnionResultSubpartitionView(availabilityListener, indexSet.size());
            try {
                for (int i : indexSet.values()) {
                    ResultSubpartitionView view = createSubpartitionView(i, unionView);
                    unionView.notifyViewCreated(i, view);
                }
                return unionView;
            } catch (Exception e) {
                unionView.releaseAllResources();
                throw e;
            }
        }
    }
@Test
    void testReleaseAllResourcesAtFailure() {
        final int maxNumSubpartitions = 4;
        final ResultSubpartitionIndexSet indexSet =
                new ResultSubpartitionIndexSet(0, maxNumSubpartitions);
        final BufferAvailabilityListener availabilityListener = (ResultSubpartitionView view) -> {};

        for (int numSubpartitions = 1; numSubpartitions < maxNumSubpartitions; numSubpartitions++) {
            List<ResultSubpartitionView> views = new ArrayList<>();
            for (int i = 0; i < numSubpartitions; i++) {
                views.add(new NoOpResultSubpartitionViewWithReleaseListener());
            }

            ResultPartition partition =
                    TestingResultPartition.newBuilder()
                            .setCreateSubpartitionViewFunction(
                                    (index, listener) -> views.get(index))
                            .build();

            assertThatThrownBy(
                            () -> partition.createSubpartitionView(indexSet, availabilityListener))
                    .isInstanceOf(IndexOutOfBoundsException.class);

            assertThat(views).allMatch(ResultSubpartitionView::isReleased);
        }
    }