flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/allround/AllRoundOperatorWrapper.java [70:101]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        } else {
            throw new UnsupportedOperationException(
                    "Unsupported operator class for all-round wrapper: " + operatorClass);
        }
    }

    @Override
    public <KEY> KeySelector<IterationRecord<T>, KEY> wrapKeySelector(
            KeySelector<T, KEY> keySelector) {
        return new ProxyKeySelector<>(keySelector);
    }

    @Override
    public StreamPartitioner<IterationRecord<T>> wrapStreamPartitioner(
            StreamPartitioner<T> streamPartitioner) {
        // Do not wrap the BroadcastPartitioner since it executes differently.
        if (streamPartitioner instanceof BroadcastPartitioner) {
            return new BroadcastPartitioner<>();
        }

        return new ProxyStreamPartitioner<>(streamPartitioner);
    }

    @Override
    public OutputTag<IterationRecord<T>> wrapOutputTag(OutputTag<T> outputTag) {
        return new OutputTag<>(
                outputTag.getId(), new IterationRecordTypeInfo<>(outputTag.getTypeInfo()));
    }

    @Override
    public TypeInformation<IterationRecord<T>> getWrappedTypeInfo(TypeInformation<T> typeInfo) {
        return new IterationRecordTypeInfo<>(typeInfo);
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/perround/PerRoundOperatorWrapper.java [70:100]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        } else {
            throw new UnsupportedOperationException(
                    "Unsupported operator class for all-round wrapper: " + operatorClass);
        }
    }

    @Override
    public <KEY> KeySelector<IterationRecord<T>, KEY> wrapKeySelector(
            KeySelector<T, KEY> keySelector) {
        return new ProxyKeySelector<>(keySelector);
    }

    @Override
    public StreamPartitioner<IterationRecord<T>> wrapStreamPartitioner(
            StreamPartitioner<T> streamPartitioner) {
        if (streamPartitioner instanceof BroadcastPartitioner) {
            return new BroadcastPartitioner<>();
        }

        return new ProxyStreamPartitioner<>(streamPartitioner);
    }

    @Override
    public OutputTag<IterationRecord<T>> wrapOutputTag(OutputTag<T> outputTag) {
        return new OutputTag<>(
                outputTag.getId(), new IterationRecordTypeInfo<>(outputTag.getTypeInfo()));
    }

    @Override
    public TypeInformation<IterationRecord<T>> getWrappedTypeInfo(TypeInformation<T> typeInfo) {
        return new IterationRecordTypeInfo<>(typeInfo);
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -