Repository: apache/flink
File: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/InflightDataRescalingDescriptor.java (Java)

Code before:
     *   Upstream task has a corresponding {@link #oldSubtaskIndexes} where it sends data over
     *   virtual channel while specifying the subtask index in the VirtualChannelSelector. This
     *   subtask then demultiplexes over channel indexes.
     * </ul>
     *
     * <p>For outputs, it's vice-versa. The information must be kept in sync but they are used in
     * opposite ways for multiplexing/demultiplexing.
     *
     * <p>Note that in the common rescaling case both information is set and need to be
     * simultaneously used. If the input subtask subsumes the state of 3 old subtasks and a channel
     * corresponds to 2 old channels, then there are 6 virtual channels to be demultiplexed.
     */
    public static class InflightDataGateOrPartitionRescalingDescriptor implements Serializable {

        public static final InflightDataGateOrPartitionRescalingDescriptor NO_STATE =
                new InflightDataGateOrPartitionRescalingDescriptor(
                        new int[0],
                        RescaleMappings.SYMMETRIC_IDENTITY,
                        Collections.emptySet(),
                        MappingType.IDENTITY) {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public int[] getOldSubtaskInstances() {
                        return new int[0];
                    }

                    @Override
                    public RescaleMappings getRescaleMappings() {
                        return RescaleMappings.SYMMETRIC_IDENTITY;
                    }
                };

        private static final long serialVersionUID = 1L;

        /** Set when several operator instances are merged into one. */
        private final int[] oldSubtaskIndexes;

        /**
         * Set when channels are merged because the connected operator has been rescaled for each
         * gate/partition.
         */
        private final RescaleMappings rescaledChannelsMappings;

        /** All channels where upstream duplicates data (only valid for downstream mappings). */
        private final Set<Integer> ambiguousSubtaskIndexes;

        private final MappingType mappingType;

        /** Type of mapping which should be used for this in-flight data. */

Review comment (suggestion):
Maybe we can eliminate some GC:

```
private static final int[] EMPTY_INT_ARRAY = new int[0];

@Override
public int[] getOldSubtaskInstances() {
    return EMPTY_INT_ARRAY;
}
```

Code after:
     * <ul>
     *   {@link #rescaledChannelsMappings} is set when there is a downscale of the upstream task.
     *   Upstream task has a corresponding {@link #oldSubtaskIndexes} where it sends data over
     *   virtual channel while specifying the subtask index in the VirtualChannelSelector. This
     *   subtask then demultiplexes over channel indexes.
     * </ul>
     *
     * <p>For outputs, it's vice-versa. The information must be kept in sync but they are used in
     * opposite ways for multiplexing/demultiplexing.
     *
     * <p>Note that in the common rescaling case both information is set and need to be
     * simultaneously used. If the input subtask subsumes the state of 3 old subtasks and a channel
     * corresponds to 2 old channels, then there are 6 virtual channels to be demultiplexed.
     */
    public static class InflightDataGateOrPartitionRescalingDescriptor implements Serializable {

        public static final InflightDataGateOrPartitionRescalingDescriptor NO_STATE =
                new InflightDataGateOrPartitionRescalingDescriptor(
                        EMPTY_INT_ARRAY,
                        RescaleMappings.SYMMETRIC_IDENTITY,
                        Collections.emptySet(),
                        MappingType.IDENTITY) {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public int[] getOldSubtaskInstances() {
                        return EMPTY_INT_ARRAY;
                    }

                    @Override
                    public RescaleMappings getRescaleMappings() {
                        return RescaleMappings.SYMMETRIC_IDENTITY;
                    }
                };

        private static final long serialVersionUID = 1L;

        /** Set when several operator instances are merged into one. */
        private final int[] oldSubtaskIndexes;

        /**
         * Set when channels are merged because the connected operator has been rescaled for each
         * gate/partition.
         */
        private final RescaleMappings rescaledChannelsMappings;

        /** All channels where upstream duplicates data (only valid for downstream mappings). */
        private final Set<Integer> ambiguousSubtaskIndexes;

        private final MappingType mappingType;