public final class AdaptiveLoadBasedRecordWriter<T extends IOReadableWritable>
        extends RecordWriter<T> {

    // Core attributions
    // Max traverse channel size when selecting an idler channel.
    private final int maxTraverseSize;
    // To restore the current channel index.
    private int currentChannel;
    // The available sub-partitions to send records.
    private final int numberOfSubpartitions;

    // Place holders of other auxiliary methods or attributions.

    void traverseToTheIdlestChannel() {
        // main logic overview
        // 1. compute the size of bytes per subpartition result in the next #maxTraverseSize channels
        //     It can be considered that all partitions form an abstract probation queue, 
        //     where each calculation covers a range of #maxTraverseSize items traversed backward.
        // 2. Get the channel whose bytes in queue is the smallest in 
        //    the #maxTraverseSize channels as the target channel.
        // 3. Assign the currentChannel with the target channel(in step2)
    }

    @Override
    public void emit(T record) throws IOException {
        // Placeholders to check or prepare.
        // ... 

        // Main calling logic
        traverseToTheIdlestChannel();
        
        // Placeholders to send record
        // ... 
    }
}

private static <OUT> RecordWriter<SerializationDelegate<StreamRecord<OUT>>> createRecordWriter(
            NonChainedOutput streamOutput,
            int outputIndex,
            Environment environment,
            String taskNameWithSubtask,
            long bufferTimeout) {

        // The old lines placeholders.
		// ...

        // The new adjusted lines.
        Configuration conf = environment.getJobConfiguration();
        final boolean adaptable = conf.get(NettyShuffleEnvironmentOptions.ADAPTIVE_PARTITIONER_ENABLE) && bufferWriter.getNumberOfSubpartitions() > 1;
        final int maxTraverseSize = conf.get(NettyShuffleEnvironmentOptions.ADAPTIVE_PARTITIONER_MAX_TRAVERSE_SIZE);

        RecordWriter<SerializationDelegate<StreamRecord<OUT>>> output =
                new RecordWriterBuilder<SerializationDelegate<StreamRecord<OUT>>>()
                        .setChannelSelector(outputPartitioner)
                        .setTimeout(bufferTimeout)
                        .setTaskName(taskNameWithSubtask)
                        .setAdaptable(adaptable)
                        .setMaxTraverseSize(maxTraverseSize)
                        .build(bufferWriter);
        output.setMetricGroup(environment.getMetricGroup().getIOMetricGroup());
        return output;
}

    // old placeholders lines
    // ...

    // The new introduced methods.


	// Keep the original partitionCustom 
    /**
     * Partitions a DataStream on the key returned by the selector, using a custom partitioner. This
     * method takes the key selector to get the key to partition on, and a partitioner that accepts
     * the key type.
     *
     * <p>Note: This method works only on single field keys, i.e. the selector cannot return tuples
     * of fields.
     *
     * @param partitioner The partitioner to assign partitions to keys.
     * @param keySelector The KeySelector with which the DataStream is partitioned.
     * @return The partitioned DataStream.
     * @see KeySelector
     */
    public <K> DataStream<T> partitionCustom(
            Partitioner<K> partitioner, KeySelector<T, K> keySelector) {
        return setConnectionType(
                new CustomPartitionerWrapper<>(clean(partitioner), clean(keySelector)));
    }

    // The new introduced for partitionCustom.
    /**
     * Partitions a DataStream on the key returned by the selector, using a custom partitioner. This
     * method takes the key selector to get the key to partition on, and a partitioner that accepts
     * the key type.
     *
     * <p>Note: This method works only on single field keys, i.e. the selector cannot return tuples
     * of fields.
     *
     * @param partitioner The partitioner to assign partitions to keys.
     * @param keySelector The KeySelector with which the DataStream is partitioned.
     * @param adaptable If enabled the load based adaptive partition. Note, This parameter can only
     *     be used when the partition logic is not bound to downstream subtasks; otherwise, it would
     *     cause semantic errors in data partitioning.
     * @return The partitioned DataStream.
     * @see KeySelector
     */
    public <K> DataStream<T> partitionCustom(
            Partitioner<K> partitioner, KeySelector<T, K> keySelector, boolean adaptable) {
        return setConnectionType(
                new CustomPartitionerWrapper<>(clean(partitioner), clean(keySelector), adaptable));
    }

    /**
     * Partitions a DataStream on the key returned by the selector, using a custom partitioner. This
     * method takes the key selector to get the key to partition on, and a partitioner that accepts
     * the key type.
     *
     * <p>Note: This method works only on single field keys, i.e. the selector cannot return tuples
     * of fields.
     *
     * @param partitioner The partitioner to assign partitions to keys.
     * @param keySelector The KeySelector with which the DataStream is partitioned.
     * @param enabledAdaptableAndMaxTraverseSize Enabled the load based adaptive partition and
     *     specify how many channels to traverse at most when looking for the idlest channel. Note,
     *     this parameter can only be used when the partition logic is not bound to downstream
     *     subtasks, otherwise, it would cause semantic errors in data partitioning.
     * @return The partitioned DataStream.
     * @see KeySelector
     */
    public <K> DataStream<T> partitionCustom(
            Partitioner<K> partitioner,
            KeySelector<T, K> keySelector,
            int enabledAdaptableAndMaxTraverseSize) {
        return setConnectionType(
                new CustomPartitionerWrapper<>(
                        clean(partitioner),
                        clean(keySelector),
                        true,
                        enabledAdaptableAndMaxTraverseSize));
    }

	// Keep the original shuffle
    /**
     * Sets the partitioning of the {@link DataStream} so that the output elements are shuffled
     * uniformly randomly to the next operation.
     *
     * @return The DataStream with shuffle partitioning set.
     */
    @PublicEvolving
    public DataStream<T> shuffle() {
        return setConnectionType(new ShufflePartitioner<T>());
    }

    // The new introduced for shuffle.
    /**
     * Sets the partitioning of the {@link DataStream} so that the output elements are shuffled
     * uniformly randomly to the next operation.
     *
     * @param adaptable If enabled the load based adaptive partition.
     * @return The DataStream with shuffle partitioning set.
     */
    public DataStream<T> shuffle(boolean adaptable) {
        return setConnectionType(new ShufflePartitioner<T>(adaptable));
    }

    /**
     * Sets the partitioning of the {@link DataStream} so that the output elements are shuffled
     * uniformly randomly to the next operation.
     *
     * @param enabledAdaptableAndMaxTraverseSize Enabled the load based adaptive partition and
     *     specify how many channels to traverse at most when looking for the idlest channel.
     * @return The DataStream with shuffle partitioning set.
     */
    public DataStream<T> shuffle(int enabledAdaptableAndMaxTraverseSize) {
        return setConnectionType(
                new ShufflePartitioner<T>(true, enabledAdaptableAndMaxTraverseSize));
    }

 	// Keep the original rebalance
    /**
     * Sets the partitioning of the {@link DataStream} so that the output elements are distributed
     * evenly to instances of the next operation in a round-robin fashion.
     *
     * @return The DataStream with rebalance partitioning set.
     */
    public DataStream<T> rebalance() {
        return setConnectionType(new RebalancePartitioner<T>());
    } 

    // The new introduced for rebalance.
    /**
     * Sets the partitioning of the {@link DataStream} so that the output elements are distributed
     * evenly to instances of the next operation in a round-robin fashion.
     *
     * @param adaptable If enabled the load based adaptive partition.
     * @return The DataStream with rebalance partitioning set.
     */
    public DataStream<T> rebalance(boolean adaptable) {
        return setConnectionType(new RebalancePartitioner<T>(adaptable));
    }

    /**
     * Sets the partitioning of the {@link DataStream} so that the output elements are distributed
     * evenly to instances of the next operation in a round-robin fashion.
     *
     * @param enabledAdaptableAndMaxTraverseSize Enabled the load based adaptive partition and
     *     specify how many channels to traverse at most when looking for the idlest channel.
     * @return The DataStream with rebalance partitioning set.
     */
    public DataStream<T> rebalance(int enabledAdaptableAndMaxTraverseSize) {
        return setConnectionType(
                new RebalancePartitioner<T>(true, enabledAdaptableAndMaxTraverseSize));
    }

 	// Keep the original rescale 
    /**
     * Sets the partitioning of the {@link DataStream} so that the output elements are distributed
     * evenly to a subset of instances of the next operation in a round-robin fashion.
     *
     * <p>The subset of downstream operations to which the upstream operation sends elements depends
     * on the degree of parallelism of both the upstream and downstream operation. For example, if
     * the upstream operation has parallelism 2 and the downstream operation has parallelism 4, then
     * one upstream operation would distribute elements to two downstream operations while the other
     * upstream operation would distribute to the other two downstream operations. If, on the other
     * hand, the downstream operation has parallelism 2 while the upstream operation has parallelism
     * 4 then two upstream operations will distribute to one downstream operation while the other
     * two upstream operations will distribute to the other downstream operations.
     *
     * <p>In cases where the different parallelisms are not multiples of each other one or several
     * downstream operations will have a differing number of inputs from upstream operations.
     *
     * @return The DataStream with rescale partitioning set.
     */
    @PublicEvolving
    public DataStream<T> rescale() {
        return setConnectionType(new RescalePartitioner<T>());
    }

    // The new introduced for rescale. 
    /**
     * Sets the partitioning of the {@link DataStream} so that the output elements are distributed
     * evenly to a subset of instances of the next operation in a round-robin fashion.
     *
     * <p>The subset of downstream operations to which the upstream operation sends elements depends
     * on the degree of parallelism of both the upstream and downstream operation. For example, if
     * the upstream operation has parallelism 2 and the downstream operation has parallelism 4, then
     * one upstream operation would distribute elements to two downstream operations while the other
     * upstream operation would distribute to the other two downstream operations. If, on the other
     * hand, the downstream operation has parallelism 2 while the upstream operation has parallelism
     * 4 then two upstream operations will distribute to one downstream operation while the other
     * two upstream operations will distribute to the other downstream operations.
     *
     * <p>In cases where the different parallelisms are not multiples of each other one or several
     * downstream operations will have a differing number of inputs from upstream operations.
     *
     * @param adaptable If enabled the load based adaptive partition.
     * @return The DataStream with rescale partitioning set.
     */
    public DataStream<T> rescale(boolean adaptable) {
        return setConnectionType(new RescalePartitioner<T>(adaptable));
    }

    /**
     * Sets the partitioning of the {@link DataStream} so that the output elements are distributed
     * evenly to a subset of instances of the next operation in a round-robin fashion.
     *
     * <p>The subset of downstream operations to which the upstream operation sends elements depends
     * on the degree of parallelism of both the upstream and downstream operation. For example, if
     * the upstream operation has parallelism 2 and the downstream operation has parallelism 4, then
     * one upstream operation would distribute elements to two downstream operations while the other
     * upstream operation would distribute to the other two downstream operations. If, on the other
     * hand, the downstream operation has parallelism 2 while the upstream operation has parallelism
     * 4 then two upstream operations will distribute to one downstream operation while the other
     * two upstream operations will distribute to the other downstream operations.
     *
     * <p>In cases where the different parallelisms are not multiples of each other one or several
     * downstream operations will have a differing number of inputs from upstream operations.
     *
     * @param enabledAdaptableAndMaxTraverseSize Enabled the load based adaptive partition and
     *     specify how many channels to traverse at most when looking for the idlest channel.
     * @return The DataStream with rescale partitioning set.
     */
    public DataStream<T> rescale(int enabledAdaptableAndMaxTraverseSize) {
        return setConnectionType(
                new RescalePartitioner<T>(true, enabledAdaptableAndMaxTraverseSize));
    }