@Override
    public List<InputChannelInfo> getUnfinishedChannels() {
        List<InputChannelInfo> unfinishedChannels =
                new ArrayList<>(
                        numberOfInputChannels - channelsWithEndOfPartitionEvents.cardinality());
        synchronized (inputChannelsWithData) {
            for (int i = channelsWithEndOfPartitionEvents.nextClearBit(0);
                    i < numberOfInputChannels;
                    i = channelsWithEndOfPartitionEvents.nextClearBit(i + 1)) {
                unfinishedChannels.add(getChannel(i).getChannelInfo());
            }
        }

        return unfinishedChannels;
    }
@Test
    void testGetUnfinishedChannels() throws IOException, InterruptedException {
        SingleInputGate inputGate =
                new SingleInputGateBuilder()
                        .setSingleInputGateIndex(1)
                        .setNumberOfChannels(3)
                        .build();
        final TestInputChannel[] inputChannels =
                new TestInputChannel[] {
                    new TestInputChannel(inputGate, 0),
                    new TestInputChannel(inputGate, 1),
                    new TestInputChannel(inputGate, 2)
                };
        inputGate.setInputChannels(inputChannels);

        assertThat(inputGate.getUnfinishedChannels())
                .isEqualTo(
                        Arrays.asList(
                                inputChannels[0].getChannelInfo(),
                                inputChannels[1].getChannelInfo(),
                                inputChannels[2].getChannelInfo()));

        inputChannels[1].readEndOfPartitionEvent();
        inputGate.notifyChannelNonEmpty(inputChannels[1]);
        inputGate.getNext();
        assertThat(inputGate.getUnfinishedChannels())
                .isEqualTo(
                        Arrays.asList(
                                inputChannels[0].getChannelInfo(),
                                inputChannels[2].getChannelInfo()));

        inputChannels[0].readEndOfPartitionEvent();
        inputGate.notifyChannelNonEmpty(inputChannels[0]);
        inputGate.getNext();
        assertThat(inputGate.getUnfinishedChannels())
                .isEqualTo(Collections.singletonList(inputChannels[2].getChannelInfo()));

        inputChannels[2].readEndOfPartitionEvent();
        inputGate.notifyChannelNonEmpty(inputChannels[2]);
        inputGate.getNext();
        assertThat(inputGate.getUnfinishedChannels()).isEqualTo(Collections.emptyList());
    }