RocMarshal commented on code in PR #28500:
URL: https://github.com/apache/flink/pull/28500#discussion_r3736736228
##########
flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/StreamTableEnvironment.java:
##########
@@ -88,7 +90,13 @@ public interface StreamTableEnvironment extends
TableEnvironment {
* TableEnvironment}.
*/
static StreamTableEnvironment create(StreamExecutionEnvironment
executionEnvironment) {
- return create(executionEnvironment,
EnvironmentSettings.newInstance().build());
+ final RuntimeExecutionMode runtimeMode =
+
executionEnvironment.getConfiguration().get(ExecutionOptions.RUNTIME_MODE)
+ == RuntimeExecutionMode.BATCH
+ ? RuntimeExecutionMode.BATCH
+ : RuntimeExecutionMode.STREAMING;
+ return create(
+ executionEnvironment,
EnvironmentSettings.newInstance().inMode(runtimeMode).build());
Review Comment:
```suggestion
executionEnvironment,
EnvironmentSettings.newInstance().inRuntimeExecutionMode(runtimeMode).build());
```
##########
flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/bridge/scala/StreamTableEnvironment.scala:
##########
@@ -833,7 +835,13 @@ object StreamTableEnvironment {
* The Scala [[StreamExecutionEnvironment]] of the [[TableEnvironment]].
*/
def create(executionEnvironment: StreamExecutionEnvironment):
StreamTableEnvironment = {
- create(executionEnvironment, EnvironmentSettings.newInstance().build)
+ val runtimeMode =
+ if
(executionEnvironment.getConfiguration.get(ExecutionOptions.RUNTIME_MODE) ==
+ RuntimeExecutionMode.BATCH) RuntimeExecutionMode.BATCH
+ else RuntimeExecutionMode.STREAMING
+ create(
+ executionEnvironment,
+ EnvironmentSettings.newInstance().inMode(runtimeMode).build)
Review Comment:
```suggestion
val runtimeMode =
if (
executionEnvironment.getConfiguration.get(ExecutionOptions.RUNTIME_MODE) ==
RuntimeExecutionMode.BATCH
) RuntimeExecutionMode.BATCH
else RuntimeExecutionMode.STREAMING
create(executionEnvironment,
EnvironmentSettings.newInstance().inRuntimeExecutionMode(runtimeMode).build)
```
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java:
##########
@@ -187,6 +188,12 @@ public Builder inStreamingMode() {
return this;
}
+ /** Sets the {@link RuntimeExecutionMode} that the components should
work in. */
+ public Builder inMode(RuntimeExecutionMode mode) {
+ configuration.set(RUNTIME_MODE, mode);
+ return this;
+ }
Review Comment:
```suggestion
/** Sets the {@link RuntimeExecutionMode} that the components should
work in. */
public Builder inRuntimeExecutionMode(RuntimeExecutionMode mode) {
Preconditions.checkArgument(
mode == STREAMING || mode == BATCH,
"Unsupported value '%s' for '%s'. Only an explicit %s or
%s mode is supported in Table API.",
mode,
RUNTIME_MODE.key(),
STREAMING,
BATCH);
configuration.set(RUNTIME_MODE, mode);
return this;
}
```
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
