Migrate Flink benchmark implementation

This merge request will fix #20 (closed). It is based on the existing implementation in the apache-flink branch but has to tackle the following requirements:

• Migrate the individual benchmarks with their new enumeration
  • Migrate UC1
  • Migrate UC2 (f.k.a. UC3)
  • Migrate UC3 (f.k.a. UC4)
  • Migrate UC4 (f.k.a. UC2)
    • Migrate to latest architecture including dedicated feedback loop
      • Moved to #173
• Migrate to Titan Control Center Commons 0.1.0
• Fix code quality issues
• Support Docker builds (already available in apache-flink)
• Build and push Docker images in CI pipeline
• Update Docker Compose files in docker-test
• Enhance Gradle configuration sharing (see #107 (closed), might be tackled in a follow up MR)
  • Moved to !95 (merged)
• Clean up Gradle build files (see #59, might be tackled in a follow up MR)
  • Moved to #59
• Rename Gradle subprojects. Envisaged naming is already available in settings.gradle files as comments)
  • Moved to #174 (closed)
• Provide corresponding Kubernetes benchmark definitions (might be tackled in a follow up ticket)
  • Moved to #176 (closed)

theodolite-benchmarks/flink-commons/src/main/java/theodolite/commons/flink/StateBackends.java

@@ -17,8 +17,9 @@ public final class StateBackends {
   public static final String STATE_BACKEND_TYPE_MEMORY = "memory";
   public static final String STATE_BACKEND_TYPE_FILESYSTEM = "filesystem";
   public static final String STATE_BACKEND_TYPE_ROCKSDB = "rocksdb";
-  public static final String STATE_BACKEND_TYPE_DEFAULT = STATE_BACKEND_TYPE_ROCKSDB;
-  public static final String DEFAULT_STATE_BACKEND_PATH = "/opt/flink/statebackend";
+  // public static final String STATE_BACKEND_TYPE_DEFAULT = STATE_BACKEND_TYPE_ROCKSDB;
+  public static final String STATE_BACKEND_TYPE_DEFAULT = STATE_BACKEND_TYPE_MEMORY;
+  public static final String DEFAULT_STATE_BACKEND_PATH = "file:///opt/flink/statebackend";
 
   private static final Logger LOGGER = LoggerFactory.getLogger(StateBackends.class);
 
@@ -35,7 +36,9 @@ public final class StateBackends {
    * {@code StateBackendFactory#STATE_BACKEND_TYPE_ROCKSDB} is the default.
    */
   public static StateBackend fromConfiguration(final Configuration configuration) {
-    switch (configuration.getString(ConfigurationKeys.FLINK_STATE_BACKEND)) {
+    final String stateBackendType =
+        configuration.getString(ConfigurationKeys.FLINK_STATE_BACKEND, STATE_BACKEND_TYPE_DEFAULT);
+    switch (stateBackendType) {
       case STATE_BACKEND_TYPE_MEMORY:
         final int memoryStateBackendSize = configuration.getInt(
             ConfigurationKeys.FLINK_STATE_BACKEND_MEMORY_SIZE,
@@ -47,7 +50,6 @@ public final class StateBackends {
             DEFAULT_STATE_BACKEND_PATH);
         return new FsStateBackend(stateBackendPath);
       case STATE_BACKEND_TYPE_ROCKSDB:
-      default:
         final String stateBackendPath2 = configuration.getString(
             ConfigurationKeys.FLINK_STATE_BACKEND_PATH,
             DEFAULT_STATE_BACKEND_PATH);
@@ -57,6 +59,9 @@ public final class StateBackends {
           LOGGER.error("Cannot create RocksDB state backend.", e);
           throw new IllegalStateException(e);
         }
+      default:
+        throw new IllegalArgumentException(
+            "Unsupported state backend '" + stateBackendType + "' configured.");
     }
   }