Repository: apache/flink
File: flink-runtime/src/main/java/org/apache/flink/streaming/api/transformations/StubTransformation.java (Java)

Code before:
        super(name, outputType, parallelism);
        this.upstreamFinder = upstreamFinder;
        this.inputAdjuster = inputAdjuster;
    }

    /**
     * Creates a new {@link StubTransformation} with the given type information and predicate to
     * find the upstream transformation.
     */
    public static <T> StubTransformation<T> create(
            TypeInformation<T> typeInformation, Predicate<Transformation<?>> upstreamFinder) {
        return create(typeInformation, upstreamFinder, Function.identity());
    }

    /**
     * Creates a new {@link StubTransformation} with the given type information and predicate to
     * find the upstream transformation and additionally allows the upstream transformation to be
     * adjusted before connecting it to the downstream transformations. A common use case for the
     * adjuster is to attach the downstream transformations to the side-output of the upstream
     * transformation.
     */
    public static <T> StubTransformation<T> create(
            TypeInformation<T> typeInformation,
            Predicate<Transformation<?>> upstreamFinder,
            Function<Transformation<?>, Transformation<?>> inputAdjuster) {
        checkNotNull(upstreamFinder, "connectionId must not be null");
        checkNotNull(inputAdjuster, "inputAdjuster must not be null");
        return new StubTransformation<>(
                "unconnected", typeInformation, 1, upstreamFinder, inputAdjuster);
    }

    public Function<Transformation<?>, Transformation<?>> getInputAdjuster() {
        return inputAdjuster;
    }

    public Predicate<Transformation<?>> getUpstreamFinder() {
        return upstreamFinder;
    }

    @Override
    protected List<Transformation<?>> getTransitivePredecessorsInternal() {
        // This transformation does not have any inputs.
        // It will be implicitly connected to an upstream transformation during translation.
        return List.of(this);
    }

    @Override
    public List<Transformation<?>> getInputs() {
        // This transformation does not have any inputs.
        // It will be implicitly connected to an upstream transformation during translation.
        return List.of();

Review comment (suggestion):
```suggestion
        checkNotNull(upstreamFinder, "upstreamFinder must not be null");
```

Code after:
        super(name, outputType, parallelism);
        this.upstreamFinder = upstreamFinder;
        this.inputAdjuster = inputAdjuster;
    }

    /**
     * Creates a new {@link StubTransformation} with the given type information and predicate to
     * find the upstream transformation.
     */
    public static <T> StubTransformation<T> create(
            TypeInformation<T> typeInformation, Predicate<Transformation<?>> upstreamFinder) {
        return create(typeInformation, upstreamFinder, Function.identity());
    }

    /**
     * Creates a new {@link StubTransformation} with the given type information and predicate to
     * find the upstream transformation and additionally allows the upstream transformation to be
     * adjusted before connecting it to the downstream transformations. A common use case for the
     * adjuster is to attach the downstream transformations to the side-output of the upstream
     * transformation.
     */
    public static <T> StubTransformation<T> create(
            TypeInformation<T> typeInformation,
            Predicate<Transformation<?>> upstreamFinder,
            Function<Transformation<?>, Transformation<?>> inputAdjuster) {
        checkNotNull(upstreamFinder, "upstreamFinder must not be null");
        checkNotNull(inputAdjuster, "inputAdjuster must not be null");
        return new StubTransformation<>(
                "unconnected", typeInformation, 1, upstreamFinder, inputAdjuster);
    }

    public Function<Transformation<?>, Transformation<?>> getInputAdjuster() {
        return inputAdjuster;
    }

    public Predicate<Transformation<?>> getUpstreamFinder() {
        return upstreamFinder;
    }

    @Override
    protected List<Transformation<?>> getTransitivePredecessorsInternal() {
        // This transformation does not have any inputs.
        // It will be implicitly connected to an upstream transformation during translation.
        return List.of(this);
    }

    @Override
    public List<Transformation<?>> getInputs() {
        // This transformation does not have any inputs.
        // It will be implicitly connected to an upstream transformation during translation.
        return List.of();