public void removeExpiration(
            long now, long timeout, Collection<PartitionRequestListener> timeoutListeners) {
        Iterator<Map.Entry<InputChannelID, PartitionRequestListener>> iterator =
                listeners.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<InputChannelID, PartitionRequestListener> entry = iterator.next();
            PartitionRequestListener partitionRequestListener = entry.getValue();
            if ((now - partitionRequestListener.getCreateTimestamp()) > timeout) {
                timeoutListeners.add(partitionRequestListener);
                iterator.remove();
            }
        }
    }
@Test
    void testRemoveExpiration() {
        PartitionRequestListenerManager partitionRequestListenerManager =
                new PartitionRequestListenerManager();
        assertThat(partitionRequestListenerManager.isEmpty()).isTrue();

        List<PartitionRequestListener> listenerList = new ArrayList<>();
        List<PartitionRequestListener> expireListenerList = new ArrayList<>();
        NettyPartitionRequestListener listener1 =
                new NettyPartitionRequestListener(
                        TestingResultPartitionProvider.newBuilder().build(),
                        TestingSubpartitionCreatedViewReader.newBuilder()
                                .setReceiverId(new InputChannelID())
                                .build(),
                        new ResultSubpartitionIndexSet(0),
                        new ResultPartitionID(),
                        0L);
        partitionRequestListenerManager.registerListener(listener1);
        expireListenerList.add(listener1);

        NettyPartitionRequestListener listener2 =
                new NettyPartitionRequestListener(
                        TestingResultPartitionProvider.newBuilder().build(),
                        TestingSubpartitionCreatedViewReader.newBuilder()
                                .setReceiverId(new InputChannelID())
                                .build(),
                        new ResultSubpartitionIndexSet(1),
                        new ResultPartitionID(),
                        0L);
        partitionRequestListenerManager.registerListener(listener2);
        expireListenerList.add(listener2);

        long currentTimestamp = System.currentTimeMillis();
        NettyPartitionRequestListener listener3 =
                new NettyPartitionRequestListener(
                        TestingResultPartitionProvider.newBuilder().build(),
                        TestingSubpartitionCreatedViewReader.newBuilder()
                                .setReceiverId(new InputChannelID())
                                .build(),
                        new ResultSubpartitionIndexSet(2),
                        new ResultPartitionID(),
                        currentTimestamp);
        partitionRequestListenerManager.registerListener(listener3);
        listenerList.add(listener3);

        List<PartitionRequestListener> removeExpireListenerList = new ArrayList<>();
        partitionRequestListenerManager.removeExpiration(
                currentTimestamp, 1L, removeExpireListenerList);

        assertThat(partitionRequestListenerManager.getPartitionRequestListeners())
                .hasSize(listenerList.size());
        assertThat(listenerList)
                .containsAll(partitionRequestListenerManager.getPartitionRequestListeners());

        assertThat(removeExpireListenerList).hasSize(expireListenerList.size());
        assertThat(expireListenerList).containsAll(removeExpireListenerList);
    }