flink-ml-iteration/flink-ml-iteration-1.15/src/main/java/org/apache/flink/iteration/operator/coordinator/HeadOperatorCoordinator.java [42:61]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
public class HeadOperatorCoordinator implements OperatorCoordinator, SharedProgressAlignerListener {

    private final Context context;

    private final SubtaskGateway[] subtaskGateways;

    private final SharedProgressAligner sharedProgressAligner;

    public HeadOperatorCoordinator(Context context, SharedProgressAligner sharedProgressAligner) {
        this.context = context;
        this.sharedProgressAligner = Objects.requireNonNull(sharedProgressAligner);
        this.subtaskGateways = new SubtaskGateway[context.currentParallelism()];

        sharedProgressAligner.registerAlignedListener(context.getOperatorId(), this);
    }

    @Override
    public void start() {}

    @Override
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/coordinator/HeadOperatorCoordinator.java [42:61]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
public class HeadOperatorCoordinator implements OperatorCoordinator, SharedProgressAlignerListener {

    private final Context context;

    private final SubtaskGateway[] subtaskGateways;

    private final SharedProgressAligner sharedProgressAligner;

    public HeadOperatorCoordinator(Context context, SharedProgressAligner sharedProgressAligner) {
        this.context = context;
        this.sharedProgressAligner = Objects.requireNonNull(sharedProgressAligner);
        this.subtaskGateways = new SubtaskGateway[context.currentParallelism()];

        sharedProgressAligner.registerAlignedListener(context.getOperatorId(), this);
    }

    @Override
    public void start() {}

    @Override
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -