mxm commented on a change in pull request #13116:
URL: https://github.com/apache/beam/pull/13116#discussion_r511985337
##########
File path:
runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java
##########
@@ -259,28 +278,44 @@ static StreamExecutionEnvironment
createStreamExecutionEnvironment(
options.setShutdownSourcesAfterIdleMs(0L);
}
}
+ }
- applyLatencyTrackingInterval(flinkStreamEnv.getConfig(), options);
-
- if (options.getAutoWatermarkInterval() != null) {
-
flinkStreamEnv.getConfig().setAutoWatermarkInterval(options.getAutoWatermarkInterval());
- }
-
- // State backend
- if (options.getStateBackendFactory() != null) {
+ private static void configureStateBackend(
+ FlinkPipelineOptions options, StreamExecutionEnvironment env) {
+ if (options.getStateBackend() != null) {
+ final String storagePath = options.getStateBackendStoragePath();
+ Preconditions.checkArgument(
+ storagePath != null,
+ "State backend was set to '%s' but no storage path was provided.",
+ options.getStateBackend());
+
+ final StateBackend stateBackend;
+ if (options.getStateBackend().equalsIgnoreCase("rocksdb")) {
Review comment:
There is no factory method available for state backends, they all have
different constructors. We can't use the class name.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
