public static CheckpointedInputGate[] createCheckpointedMultipleInputGate(
            MailboxExecutor mailboxExecutor,
            List<IndexedInputGate>[] inputGates,
            TaskIOMetricGroup taskIOMetricGroup,
            CheckpointBarrierHandler barrierHandler,
            StreamConfig config) {

        registerCheckpointMetrics(taskIOMetricGroup, barrierHandler);

        InputGate[] unionedInputGates =
                Arrays.stream(inputGates)
                        .map(InputGateUtil::createInputGate)
                        .toArray(InputGate[]::new);

        return Arrays.stream(unionedInputGates)
                .map(
                        unionedInputGate ->
                                new CheckpointedInputGate(
                                        unionedInputGate,
                                        barrierHandler,
                                        mailboxExecutor,
                                        config.isGraphContainingLoops()
                                                ? UpstreamRecoveryTracker.NO_OP
                                                : UpstreamRecoveryTracker.forInputGate(
                                                        unionedInputGate)))
                .toArray(CheckpointedInputGate[]::new);
    }
@Test
    void testCreateCheckpointedMultipleInputGate() throws Exception {
        try (CloseableRegistry registry = new CloseableRegistry()) {
            MockEnvironment environment = new MockEnvironmentBuilder().build();
            MockStreamTask streamTask = new MockStreamTaskBuilder(environment).build();
            StreamConfig streamConfig = new StreamConfig(environment.getJobConfiguration());
            streamConfig.setCheckpointMode(CheckpointingMode.EXACTLY_ONCE);
            streamConfig.setUnalignedCheckpointsEnabled(true);

            // First input gate has index larger than the second
            List<IndexedInputGate>[] inputGates =
                    new List[] {
                        Collections.singletonList(getGate(1, 4)),
                        Collections.singletonList(getGate(0, 2)),
                    };

            CheckpointBarrierHandler barrierHandler =
                    InputProcessorUtil.createCheckpointBarrierHandler(
                            streamTask,
                            streamConfig,
                            new TestSubtaskCheckpointCoordinator(new MockChannelStateWriter()),
                            streamTask.getName(),
                            inputGates,
                            Collections.emptyList(),
                            new SyncMailboxExecutor(),
                            new TestProcessingTimeService());

            CheckpointedInputGate[] checkpointedMultipleInputGate =
                    InputProcessorUtil.createCheckpointedMultipleInputGate(
                            new SyncMailboxExecutor(),
                            inputGates,
                            environment.getMetricGroup().getIOMetricGroup(),
                            barrierHandler,
                            streamConfig);

            for (CheckpointedInputGate checkpointedInputGate : checkpointedMultipleInputGate) {
                registry.registerCloseable(checkpointedInputGate);
            }

            List<IndexedInputGate> allInputGates =
                    Arrays.stream(inputGates)
                            .flatMap(gates -> gates.stream())
                            .collect(Collectors.toList());
            for (IndexedInputGate inputGate : allInputGates) {
                for (int channelId = 0;
                        channelId < inputGate.getNumberOfInputChannels();
                        channelId++) {
                    barrierHandler.processBarrier(
                            new CheckpointBarrier(
                                    1,
                                    42,
                                    CheckpointOptions.unaligned(
                                            CheckpointType.CHECKPOINT,
                                            CheckpointStorageLocationReference.getDefault())),
                            new InputChannelInfo(inputGate.getGateIndex(), channelId),
                            false);
                }
            }
            assertThat(barrierHandler.getAllBarriersReceivedFuture(1)).isDone();
        }
    }