[FLINK-23600] Rework remote module component parsing and binding

This closes #247.

Author: tzulitai

statefun-e2e-tests/statefun-exactly-once-remote-e2e/src/test/resources/remote-module/module.yaml

@@ -13,42 +13,29 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-version: "3.0"
-
-module:
-  meta:
-    type: remote
-  spec:
-    endpoints:
-      - endpoint:
-          meta:
-            kind: http
-          spec:
-            functions: org.apache.flink.statefun.e2e.remote/*
-            urlPathTemplate: http://remote-function:8000/service
-            maxNumBatchRequests: 10000
-    ingresses:
-      - ingress:
-          meta:
-            type: io.statefun.kafka/ingress
-            id: org.apache.flink.statefun.e2e.remote/invoke
-          spec:
-            address: kafka-broker:9092
-            consumerGroupId: remote-module-e2e
-            startupPosition:
-              type: earliest
-            topics:
-              - topic: invoke
-                valueType: statefun.e2e/org.apache.flink.statefun.e2e.remote.Invoke
-                targets:
-                  - org.apache.flink.statefun.e2e.remote/counter
-    egresses:
-      - egress:
-          meta:
-            type: io.statefun.kafka/egress
-            id: org.apache.flink.statefun.e2e.remote/invoke-results
-          spec:
-            address: kafka-broker:9092
-            deliverySemantic:
-              type: exactly-once
-              transactionTimeout: 15min
+kind: io.statefun.endpoints.v2/http
+spec:
+  functions: org.apache.flink.statefun.e2e.remote/*
+  urlPathTemplate: http://remote-function:8000/service
+  maxNumBatchRequests: 10000
+---
+kind: io.statefun.kafka.v1/ingress
+spec:
+  id: org.apache.flink.statefun.e2e.remote/invoke
+  address: kafka-broker:9092
+  consumerGroupId: remote-module-e2e
+  startupPosition:
+    type: earliest
+  topics:
+    - topic: invoke
+      valueType: statefun.e2e/org.apache.flink.statefun.e2e.remote.Invoke
+      targets:
+        - org.apache.flink.statefun.e2e.remote/counter
+---
+kind: io.statefun.kafka.v1/egress
+spec:
+  id: org.apache.flink.statefun.e2e.remote/invoke-results
+  address: kafka-broker:9092
+  deliverySemantic:
+    type: exactly-once
+    transactionTimeout: 15min

statefun-e2e-tests/statefun-smoke-e2e-java/src/test/resources/remote-module/module.yaml

@@ -13,17 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-version: "3.0"
-
-module:
-  meta:
-    type: remote
-  spec:
-    endpoints:
-      - endpoint:
-          meta:
-            kind: http
-          spec:
-            functions: statefun.smoke.e2e/*
-            urlPathTemplate: http://remote-function-host:8000
-            maxNumBatchRequests: 10000
+kind: io.statefun.endpoints.v2/http
+spec:
+  functions: statefun.smoke.e2e/command-interpreter-fn
+  urlPathTemplate: http://remote-function-host:8000
+  maxNumBatchRequests: 10000

statefun-e2e-tests/statefun-smoke-e2e-multilang-harness/src/test/resources/module.yaml

@@ -13,17 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-version: "3.0"
-
-module:
-  meta:
-    type: remote
-  spec:
-    endpoints:
-      - endpoint:
-          meta:
-            kind: http
-          spec:
-            functions: statefun.smoke.e2e/command-interpreter-fn
-            urlPathTemplate: http://localhost:8000
-            maxNumBatchRequests: 10000
+kind: io.statefun.endpoints.v2/http
+spec:
+  functions: statefun.smoke.e2e/command-interpreter-fn
+  urlPathTemplate: http://localhost:8000
+  maxNumBatchRequests: 10000

statefun-flink/pom.xml

@@ -32,6 +32,7 @@ under the License.
 
     <modules>
         <module>statefun-flink-common</module>
+        <module>statefun-flink-extensions</module>
         <module>statefun-flink-launcher</module>
         <module>statefun-flink-io</module>
         <module>statefun-flink-io-bundle</module>

statefun-flink/statefun-flink-common/pom.xml

@@ -34,6 +34,12 @@ under the License.
     </properties>
 
     <dependencies>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>statefun-sdk-embedded</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
         <!-- flink runtime -->
         <dependency>
             <groupId>org.apache.flink</groupId>

statefun-flink/statefun-flink-common/src/main/java/org/apache/flink/statefun/flink/common/json/StateFunObjectMapper.java

@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.statefun.flink.common.json;
+
+import java.io.IOException;
+import java.time.Duration;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationContext;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationFeature;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonDeserializer;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.module.SimpleModule;
+import org.apache.flink.statefun.sdk.TypeName;
+import org.apache.flink.util.TimeUtils;
+
+public final class StateFunObjectMapper {
+
+  public static ObjectMapper create() {
+    final ObjectMapper mapper =
+        new ObjectMapper().configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+
+    final SimpleModule module = new SimpleModule("statefun");
+    module.addDeserializer(Duration.class, new DurationJsonDeserializer());
+    module.addDeserializer(TypeName.class, new TypeNameJsonDeserializer());
+
+    mapper.registerModule(module);
+    return mapper;
+  }
+
+  private static final class DurationJsonDeserializer extends JsonDeserializer<Duration> {
+    @Override
+    public Duration deserialize(
+        JsonParser jsonParser, DeserializationContext deserializationContext) throws IOException {
+      return TimeUtils.parseDuration(jsonParser.getText());
+    }
+  }
+
+  private static final class TypeNameJsonDeserializer extends JsonDeserializer<TypeName> {
+    @Override
+    public TypeName deserialize(
+        JsonParser jsonParser, DeserializationContext deserializationContext) throws IOException {
+      return TypeName.parseFrom(jsonParser.getText());
+    }
+  }
+}