azagrebin commented on a change in pull request #7549: [FLINK-11403][network] Remove ResultPartitionConsumableNotifier from ResultPartition URL: https://github.com/apache/flink/pull/7549#discussion_r266905690
########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/NetworkResultPartition.java ########## @@ -0,0 +1,414 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.io.network.partition; + +import org.apache.flink.runtime.executiongraph.IntermediateResultPartition; +import org.apache.flink.runtime.io.disk.iomanager.IOManager; +import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter; +import org.apache.flink.runtime.io.network.buffer.Buffer; +import org.apache.flink.runtime.io.network.buffer.BufferConsumer; +import org.apache.flink.runtime.io.network.buffer.BufferPool; +import org.apache.flink.runtime.io.network.buffer.BufferPoolOwner; +import org.apache.flink.runtime.io.network.buffer.BufferProvider; +import org.apache.flink.runtime.io.network.partition.consumer.LocalInputChannel; +import org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel; +import org.apache.flink.runtime.jobgraph.DistributionPattern; +import org.apache.flink.runtime.taskmanager.TaskManager; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.apache.flink.util.Preconditions.checkArgument; +import static org.apache.flink.util.Preconditions.checkElementIndex; +import static org.apache.flink.util.Preconditions.checkNotNull; +import static org.apache.flink.util.Preconditions.checkState; + +/** + * A result partition for data produced by a single task. + * + * <p>This class is the runtime part of a logical {@link IntermediateResultPartition}. Essentially, + * a result partition is a collection of {@link Buffer} instances. The buffers are organized in one + * or more {@link ResultSubpartition} instances, which further partition the data depending on the + * number of consuming tasks and the data {@link DistributionPattern}. + * + * <p>Tasks, which consume a result partition have to request one of its subpartitions. The request + * happens either remotely (see {@link RemoteInputChannel}) or locally (see {@link LocalInputChannel}) + * + * <h2>Life-cycle</h2> + * + * <p>The life-cycle of each result partition has three (possibly overlapping) phases: + * <ol> + * <li><strong>Produce</strong>: </li> + * <li><strong>Consume</strong>: </li> + * <li><strong>Release</strong>: </li> + * </ol> + * + * <h2>Lazy deployment and updates of consuming tasks</h2> + * + * <p>Before a consuming task can request the result, it has to be deployed. The time of deployment + * depends on the PIPELINED vs. BLOCKING characteristic of the result partition. With pipelined + * results, receivers are deployed as soon as the first buffer is added to the result partition. + * With blocking results on the other hand, receivers are deployed after the partition is finished. + * + * <h2>Buffer management</h2> + * + * <h2>State management</h2> + */ +public class NetworkResultPartition implements ResultPartitionWriter, BufferPoolOwner { + + private static final Logger LOG = LoggerFactory.getLogger(NetworkResultPartition.class); + + private final String owningTaskName; + + private final ResultPartitionID partitionId; + + /** Type of this partition. Defines the concrete subpartition implementation to use. */ + private final ResultPartitionType partitionType; + + /** The subpartitions of this partition. At least one. */ + private final ResultSubpartition[] subpartitions; + + private final ResultPartitionManager partitionManager; + + public final int numTargetKeyGroups; Review comment: can be private ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
