[flink-statefun] 06/11: [FLINK-16124] [kinesis] Implement runtime GenericKinesisSinkProvider

[tz...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git

commit d0c74a754280aa31949b5aea96e50ac645aa9c52
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Fri Mar 20 13:23:50 2020 +0800

    [FLINK-16124] [kinesis] Implement runtime GenericKinesisSinkProvider
---
 .../flink/io/kinesis/KinesisSinkProvider.java      |  2 +-
 .../polyglot/GenericKinesisEgressSerializer.java   | 62 +++++++++++++++
 .../polyglot/GenericKinesisSinkProvider.java       | 92 ++++++++++++++++++++++
 .../polyglot/KinesisEgressSpecJsonParser.java      | 43 ++++++++++
 .../io/kinesis/GenericKinesisSinkProviderTest.java | 50 ++++++++++++
 .../src/test/resources/generic-kinesis-egress.yaml | 32 ++++++++
 .../flink/io/kinesis/PolyglotKinesisIOTypes.java   |  4 +
 7 files changed, 284 insertions(+), 1 deletion(-)

diff --git a/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kinesis/KinesisSinkProvider.java b/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kinesis/KinesisSinkProvider.java
index c1156b6..c969a0d 100644
--- a/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kinesis/KinesisSinkProvider.java
+++ b/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kinesis/KinesisSinkProvider.java
@@ -26,7 +26,7 @@ import org.apache.flink.statefun.sdk.kinesis.egress.KinesisEgressSpec;
 import org.apache.flink.streaming.api.functions.sink.SinkFunction;
 import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisProducer;
 
-final class KinesisSinkProvider implements SinkProvider {
+public final class KinesisSinkProvider implements SinkProvider {
 
   @Override
   public <T> SinkFunction<T> forSpec(EgressSpec<T> spec) {
diff --git a/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kinesis/polyglot/GenericKinesisEgressSerializer.java b/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kinesis/polyglot/GenericKinesisEgressSerializer.java
new file mode 100644
index 0000000..db3db58
--- /dev/null
+++ b/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kinesis/polyglot/GenericKinesisEgressSerializer.java
@@ -0,0 +1,62 @@
+/*
+ * 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.io.kinesis.polyglot;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.InvalidProtocolBufferException;
+import org.apache.flink.statefun.flink.io.generated.KinesisEgressRecord;
+import org.apache.flink.statefun.sdk.kinesis.egress.EgressRecord;
+import org.apache.flink.statefun.sdk.kinesis.egress.KinesisEgressSerializer;
+
+public final class GenericKinesisEgressSerializer implements KinesisEgressSerializer<Any> {
+
+  private static final long serialVersionUID = 1L;
+
+  @Override
+  public EgressRecord serialize(Any value) {
+    final KinesisEgressRecord kinesisEgressRecord = asKinesisEgressRecord(value);
+
+    final EgressRecord.Builder builder =
+        EgressRecord.newBuilder()
+            .withData(kinesisEgressRecord.getValueBytes().toByteArray())
+            .withStream(kinesisEgressRecord.getStream())
+            .withPartitionKey(kinesisEgressRecord.getPartitionKey());
+
+    final String explicitHashKey = kinesisEgressRecord.getExplicitHashKey();
+    if (explicitHashKey != null && !explicitHashKey.isEmpty()) {
+      builder.withExplicitHashKey(explicitHashKey);
+    }
+
+    return builder.build();
+  }
+
+  private static KinesisEgressRecord asKinesisEgressRecord(Any message) {
+    if (!message.is(KinesisEgressRecord.class)) {
+      throw new IllegalStateException(
+          "The generic Kinesis egress expects only messages of type "
+              + KinesisEgressRecord.class.getName());
+    }
+    try {
+      return message.unpack(KinesisEgressRecord.class);
+    } catch (InvalidProtocolBufferException e) {
+      throw new RuntimeException(
+          "Unable to unpack message as a " + KinesisEgressRecord.class.getName(), e);
+    }
+  }
+}
diff --git a/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kinesis/polyglot/GenericKinesisSinkProvider.java b/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kinesis/polyglot/GenericKinesisSinkProvider.java
new file mode 100644
index 0000000..ad8fc1f
--- /dev/null
+++ b/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kinesis/polyglot/GenericKinesisSinkProvider.java
@@ -0,0 +1,92 @@
+/*
+ * 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.io.kinesis.polyglot;
+
+import static org.apache.flink.statefun.flink.io.kinesis.polyglot.AwsAuthSpecJsonParser.optionalAwsCredentials;
+import static org.apache.flink.statefun.flink.io.kinesis.polyglot.AwsAuthSpecJsonParser.optionalAwsRegion;
+import static org.apache.flink.statefun.flink.io.kinesis.polyglot.KinesisEgressSpecJsonParser.clientConfigProperties;
+import static org.apache.flink.statefun.flink.io.kinesis.polyglot.KinesisEgressSpecJsonParser.optionalMaxOutstandingRecords;
+
+import com.google.protobuf.Any;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.flink.statefun.flink.io.kinesis.KinesisSinkProvider;
+import org.apache.flink.statefun.flink.io.spi.JsonEgressSpec;
+import org.apache.flink.statefun.flink.io.spi.SinkProvider;
+import org.apache.flink.statefun.sdk.io.EgressIdentifier;
+import org.apache.flink.statefun.sdk.io.EgressSpec;
+import org.apache.flink.statefun.sdk.kinesis.egress.KinesisEgressBuilder;
+import org.apache.flink.statefun.sdk.kinesis.egress.KinesisEgressSpec;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+
+public final class GenericKinesisSinkProvider implements SinkProvider {
+
+  private final KinesisSinkProvider delegateProvider = new KinesisSinkProvider();
+
+  @Override
+  public <T> SinkFunction<T> forSpec(EgressSpec<T> spec) {
+    final KinesisEgressSpec<T> kinesisEgressSpec = asKinesisEgressSpec(spec);
+    return delegateProvider.forSpec(kinesisEgressSpec);
+  }
+
+  private static <T> KinesisEgressSpec<T> asKinesisEgressSpec(EgressSpec<T> spec) {
+    if (!(spec instanceof JsonEgressSpec)) {
+      throw new IllegalArgumentException("Wrong type " + spec.type());
+    }
+    JsonEgressSpec<T> casted = (JsonEgressSpec<T>) spec;
+
+    EgressIdentifier<T> id = casted.id();
+    validateConsumedType(id);
+
+    JsonNode specJson = casted.specJson();
+
+    KinesisEgressBuilder<T> kinesisEgressBuilder = KinesisEgressBuilder.forIdentifier(id);
+
+    optionalAwsRegion(specJson).ifPresent(kinesisEgressBuilder::withAwsRegion);
+    optionalAwsCredentials(specJson).ifPresent(kinesisEgressBuilder::withAwsCredentials);
+    optionalMaxOutstandingRecords(specJson)
+        .ifPresent(kinesisEgressBuilder::withMaxOutstandingRecords);
+    clientConfigProperties(specJson)
+        .entrySet()
+        .forEach(
+            entry ->
+                kinesisEgressBuilder.withClientConfigurationProperty(
+                    entry.getKey(), entry.getValue()));
+
+    kinesisEgressBuilder.withSerializer(serializerClass());
+
+    return kinesisEgressBuilder.build();
+  }
+
+  private static void validateConsumedType(EgressIdentifier<?> id) {
+    Class<?> consumedType = id.consumedType();
+    if (Any.class != consumedType) {
+      throw new IllegalArgumentException(
+          "Generic Kinesis egress is only able to consume messages types of "
+              + Any.class.getName()
+              + " but "
+              + consumedType.getName()
+              + " is provided.");
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static <T> Class<T> serializerClass() {
+    // this cast is safe, because we've already validated that the consumed type is Any.
+    return (Class<T>) GenericKinesisEgressSerializer.class;
+  }
+}
diff --git a/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kinesis/polyglot/KinesisEgressSpecJsonParser.java b/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kinesis/polyglot/KinesisEgressSpecJsonParser.java
new file mode 100644
index 0000000..594623c
--- /dev/null
+++ b/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kinesis/polyglot/KinesisEgressSpecJsonParser.java
@@ -0,0 +1,43 @@
+/*
+ * 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.io.kinesis.polyglot;
+
+import java.util.Map;
+import java.util.OptionalInt;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonPointer;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.flink.statefun.flink.common.json.Selectors;
+
+final class KinesisEgressSpecJsonParser {
+
+  private KinesisEgressSpecJsonParser() {}
+
+  private static final JsonPointer MAX_OUTSTANDING_RECORDS_POINTER =
+      JsonPointer.compile("/maxOutstandingRecords");
+  private static final JsonPointer CLIENT_CONFIG_PROPS_POINTER =
+      JsonPointer.compile("/clientConfigProperties");
+
+  static OptionalInt optionalMaxOutstandingRecords(JsonNode ingressSpecNode) {
+    return Selectors.optionalIntegerAt(ingressSpecNode, MAX_OUTSTANDING_RECORDS_POINTER);
+  }
+
+  static Map<String, String> clientConfigProperties(JsonNode ingressSpecNode) {
+    return Selectors.propertiesAt(ingressSpecNode, CLIENT_CONFIG_PROPS_POINTER);
+  }
+}
diff --git a/statefun-flink/statefun-flink-io-bundle/src/test/java/org/apache/flink/statefun/flink/io/kinesis/GenericKinesisSinkProviderTest.java b/statefun-flink/statefun-flink-io-bundle/src/test/java/org/apache/flink/statefun/flink/io/kinesis/GenericKinesisSinkProviderTest.java
new file mode 100644
index 0000000..2a6b19b
--- /dev/null
+++ b/statefun-flink/statefun-flink-io-bundle/src/test/java/org/apache/flink/statefun/flink/io/kinesis/GenericKinesisSinkProviderTest.java
@@ -0,0 +1,50 @@
+/*
+ * 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.io.kinesis;
+
+import static org.apache.flink.statefun.flink.io.testutils.YamlUtils.loadAsJsonFromClassResource;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import com.google.protobuf.Any;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.flink.statefun.flink.io.kinesis.polyglot.GenericKinesisSinkProvider;
+import org.apache.flink.statefun.flink.io.spi.JsonEgressSpec;
+import org.apache.flink.statefun.sdk.io.EgressIdentifier;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisProducer;
+import org.junit.Test;
+
+public class GenericKinesisSinkProviderTest {
+
+  @Test
+  public void exampleUsage() {
+    JsonNode egressDefinition =
+        loadAsJsonFromClassResource(getClass().getClassLoader(), "generic-kinesis-egress.yaml");
+    JsonEgressSpec<?> spec =
+        new JsonEgressSpec<>(
+            PolyglotKinesisIOTypes.GENERIC_KINESIS_EGRESS_TYPE,
+            new EgressIdentifier<>("foo", "bar", Any.class),
+            egressDefinition);
+
+    GenericKinesisSinkProvider provider = new GenericKinesisSinkProvider();
+    SinkFunction<?> source = provider.forSpec(spec);
+
+    assertThat(source, instanceOf(FlinkKinesisProducer.class));
+  }
+}
diff --git a/statefun-flink/statefun-flink-io-bundle/src/test/resources/generic-kinesis-egress.yaml b/statefun-flink/statefun-flink-io-bundle/src/test/resources/generic-kinesis-egress.yaml
new file mode 100644
index 0000000..902822d
--- /dev/null
+++ b/statefun-flink/statefun-flink-io-bundle/src/test/resources/generic-kinesis-egress.yaml
@@ -0,0 +1,32 @@
+# 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.
+
+egress:
+  meta:
+    type: statefun.kinesis.io/generic-egress
+    id: com.mycomp.foo/bar
+  spec:
+    awsRegion:
+      type: custom-endpoint
+      endpoint: https://localhost:4567
+      id: us-west-1
+    awsCredentials:
+      type: profile
+      profileName: john-doe
+      profilePath: /path/to/profile/config
+    maxOutstandingRecords: 9999
+    clientConfigProperties:
+      - ThreadingModel: POOLED
+      - ThreadPoolSize: 10
diff --git a/statefun-flink/statefun-flink-io/src/main/java/org/apache/flink/statefun/flink/io/kinesis/PolyglotKinesisIOTypes.java b/statefun-flink/statefun-flink-io/src/main/java/org/apache/flink/statefun/flink/io/kinesis/PolyglotKinesisIOTypes.java
index c6b384e..252e59d 100644
--- a/statefun-flink/statefun-flink-io/src/main/java/org/apache/flink/statefun/flink/io/kinesis/PolyglotKinesisIOTypes.java
+++ b/statefun-flink/statefun-flink-io/src/main/java/org/apache/flink/statefun/flink/io/kinesis/PolyglotKinesisIOTypes.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.statefun.flink.io.kinesis;
 
+import org.apache.flink.statefun.sdk.EgressType;
 import org.apache.flink.statefun.sdk.IngressType;
 
 public final class PolyglotKinesisIOTypes {
@@ -26,4 +27,7 @@ public final class PolyglotKinesisIOTypes {
 
   public static final IngressType ROUTABLE_PROTOBUF_KINESIS_INGRESS_TYPE =
       new IngressType("statefun.kinesis.io", "routable-protobuf-ingress");
+
+  public static final EgressType GENERIC_KINESIS_EGRESS_TYPE =
+      new EgressType("statefun.kinesis.io", "generic-egress");
 }