mxm commented on a change in pull request #13116:
URL: https://github.com/apache/beam/pull/13116#discussion_r519346341
##########
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")) {
+ try {
+ stateBackend = new RocksDBStateBackend(storagePath);
+ } catch (IOException e) {
+ throw new RuntimeException("Could not create RocksDB state
backend.", e);
+ }
+ } else if (options.getStateBackend().equalsIgnoreCase("filesystem")) {
+ stateBackend = new FsStateBackend(storagePath);
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unknown state backend '%s'. Use 'rocksdb' or 'filesystem' or
configure via Flink config file.",
+ options.getStateBackend()));
+ }
+ env.setStateBackend(stateBackend);
+ } else if (options.getStateBackendFactory() != null) {
+ // Legacy way of setting the state backend
final StateBackend stateBackend =
InstanceBuilder.ofType(FlinkStateBackendFactory.class)
.fromClass(options.getStateBackendFactory())
.build()
.createStateBackend(options);
- flinkStreamEnv.setStateBackend(stateBackend);
+ env.setStateBackend(stateBackend);
Review comment:
Actually we can't because the if blocks are non-exhaustive. If nothing
is configured, we don't want to call `setStateBackend´.
----------------------------------------------------------------
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]
