[FLINK-20897][streaming-java] Make boundedness mutable in LegacySourc…

…eTransformation

InputFormat, SourceFunction, and DataStream API are exposed via table source runtime
providers that might not set the boundedness correctly. However, the table planner has
the correct information and needs a way to enrich these transformations. The new source
stack will set this property correctly.

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/LegacySourceTransformation.java

@@ -45,7 +45,7 @@ public class LegacySourceTransformation<T> extends PhysicalTransformation<T>
 
  private final StreamOperatorFactory<T> operatorFactory;
 
- private final Boundedness boundedness;
+ private Boundedness boundedness;
 
  /**
  * Creates a new {@code LegacySourceTransformation} from the given operator.
@@ -77,6 +77,11 @@ public LegacySourceTransformation(
  this.boundedness = checkNotNull(boundedness);
  }
 
+ /** Mutable for legacy sources in the Table API. */
+ public void setBoundedness(Boundedness boundedness) {
+ this.boundedness = boundedness;
+ }
+
  @Override
  public Boundedness getBoundedness() {
  return boundedness;