[FLINK-16913] Migrate StateBackends to use ReadableConfig instead of …

…Configuration

StateBackendFactories do not need a full read and write access to the
Configuration object. It's sufficient to have read only access. Moreover
the ReadableConfig is a lightweight interface that can be implemented in
other ways, not just through the Configuration. Lastly we exposed this
lightweight interface as a configuration entry point for
ExecutionEnvironments. This change will make it possible to pass the
ReadableConfig directly to the StateBackendFactories without fragile
adapters.

docs/_includes/generated/rocks_db_configuration.html

@@ -58,8 +58,8 @@
  </tr>
  <tr>
  <td><h5>state.backend.rocksdb.timer-service.factory</h5></td>
- <td style="word-wrap: break-word;">"ROCKSDB"</td>
- <td>String</td>
+ <td style="word-wrap: break-word;">ROCKSDB</td>
+ <td><p>Enum</p>Possible values: [HEAP, ROCKSDB]</td>
  <td>This determines the factory for timer service state implementation. Options are either HEAP (heap-based, default) or ROCKSDB for an implementation based on RocksDB .</td>
  </tr>
  </tbody>

docs/_includes/generated/state_backend_rocksdb_section.html

@@ -34,8 +34,8 @@
  </tr>
  <tr>
  <td><h5>state.backend.rocksdb.timer-service.factory</h5></td>
- <td style="word-wrap: break-word;">"ROCKSDB"</td>
- <td>String</td>
+ <td style="word-wrap: break-word;">ROCKSDB</td>
+ <td><p>Enum</p>Possible values: [HEAP, ROCKSDB]</td>
  <td>This determines the factory for timer service state implementation. Options are either HEAP (heap-based, default) or ROCKSDB for an implementation based on RocksDB .</td>
  </tr>
  </tbody>

flink-runtime/src/main/java/org/apache/flink/runtime/state/ConfigurableStateBackend.java

@@ -19,8 +19,8 @@
 package org.apache.flink.runtime.state;
 
 import org.apache.flink.annotation.Internal;
-import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.IllegalConfigurationException;
+import org.apache.flink.configuration.ReadableConfig;
 
 /**
  * An interface for state backends that pick up additional parameters from a configuration.
@@ -44,5 +44,5 @@ public interface ConfigurableStateBackend {
  *
  * @throws IllegalConfigurationException Thrown if the configuration contained invalid entries.
  */
- StateBackend configure(Configuration config, ClassLoader classLoader) throws IllegalConfigurationException;
+ StateBackend configure(ReadableConfig config, ClassLoader classLoader) throws IllegalConfigurationException;
 }

flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendFactory.java

@@ -19,8 +19,8 @@
 package org.apache.flink.runtime.state;
 
 import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.IllegalConfigurationException;
+import org.apache.flink.configuration.ReadableConfig;
 
 import java.io.IOException;
 
@@ -48,5 +48,5 @@ public interface StateBackendFactory<T extends StateBackend> {
  * @throws IOException
  * If the state backend initialization failed due to an I/O exception
  */
- T createFromConfig(Configuration config, ClassLoader classLoader) throws IllegalConfigurationException, IOException;
+ T createFromConfig(ReadableConfig config, ClassLoader classLoader) throws IllegalConfigurationException, IOException;
 }