Modified Scala Getting Started example to work both locally and on MSF (aws-samples#62)

Scala getting started working locally and on MSF (downgraded to Scala 2.12)

Author: nicusX

scala/GettingStarted/README.md

@@ -3,7 +3,7 @@
 Skeleton project for a basic Flink Java application to run on Amazon Managed Service for Apache Flink.
 
 * Flink version: 1.20
-* Scala version: 3.3.0
+* Scala version: 2.12.20
 * Flink API: DataStream API
 * Language: Java (11)
 
@@ -12,11 +12,13 @@ The project can run both on Amazon Managed Service for Apache Flink, and locally
 The application shows how to get runtime configuration, and sets up a Kinesis Data Stream source and a sink.
 
 ### Pre-requisites
+
 You need to have `sbt` tool installed on you machine to build a Scala project. Use [steps from official guide](https://www.scala-sbt.org/download.html) to do that.
 
 ### Build
+
 - Run `sbt assembly` to build an uber jar
-- Use `target/scala-3.3.0/ScalaGettingStarted-flink_1.20.jar` in your MSF application
+- Use `target/scala-2.12/ScalaGettingStarted-flink_1.20.jar` in your MSF application
 
 ### Runtime configuration
 
@@ -25,12 +27,19 @@ Apache Flink.
 
 The following runtime properties are required:
 
-| Group ID       | Key           | Description              |
-|----------------|---------------|--------------------------|
-| `InputStream0` | `stream.name` | Name of the input stream |
-| `InputStream0` | `aws.region`  | Region of the input stream, e.g. `us-east-1` |
+| Group ID        | Key           | Description                                   |
+|-----------------|---------------|-----------------------------------------------|
+| `InputStream0`  | `stream.name` | Name of the input stream                      |
+| `InputStream0`  | `aws.region`  | Region of the input stream, e.g. `us-east-1`  |
+| `OutputStream0` | `stream.name` | Name of the output stream                     |
+| `OutputStream0` | `aws.region`  | Region of the output stream, e.g. `us-east-1` |
+
+
+### Running in IntelliJ
+
+You can run this example directly in IntelliJ, without any local Flink cluster or local Flink installation.
 
-Runtime Properties are expected in the Group IDs `InputStream0` and `OutputStream0`.
+See [Running examples locally](../../java/running-examples-locally.md) for details.
 
 ### Generating data
 

scala/GettingStarted/build.sbt

@@ -4,21 +4,28 @@ lazy val root = (project in file(".")).
   settings(
     name := "getting-started-scala",
     version := "1.0",
-    scalaVersion := "3.3.0",
-    mainClass := Some("com.amazonaws.services.kinesisanalytics.main"),
+    scalaVersion := "2.12.20",
+    mainClass := Some("com.amazonaws.services.msf.BasicStreamingJob"),
     javacOptions ++= Seq("-source", "11", "-target", "11")
   )
 
 val jarName = "ScalaGettingStarted-flink_1.20.jar"
 val flinkVersion = "1.20.0"
 val msfRuntimeVersion = "1.2.0"
 val connectorVersion = "4.3.0-1.19"
+val log4jVersion = "2.17.2"
 
 libraryDependencies ++= Seq(
   "com.amazonaws" % "aws-kinesisanalytics-runtime" % msfRuntimeVersion,
   "org.apache.flink" % "flink-connector-kinesis" % connectorVersion,
   "org.apache.flink" % "flink-streaming-java" % flinkVersion % "provided",
+  "org.apache.flink" %% "flink-streaming-scala" % flinkVersion % "provided",
   "org.apache.flink" % "flink-connector-base" % flinkVersion % "provided",
+  "org.apache.flink" % "flink-clients" % flinkVersion % "provided",
+  // Logging
+  "org.apache.logging.log4j" % "log4j-slf4j-impl" % log4jVersion % "provided",
+  "org.apache.logging.log4j" % "log4j-api" % log4jVersion % "provided",
+  "org.apache.logging.log4j" % "log4j-core" % log4jVersion % "provided",
 )
 
 artifactName := { (_: ScalaVersion, _: ModuleID, _: Artifact) => jarName }

scala/GettingStarted/src/main/resources/flink-application-properties-dev.json

@@ -0,0 +1,16 @@
+[
+  {
+    "PropertyGroupId": "InputStream0",
+    "PropertyMap": {
+      "aws.region": "us-east-1",
+      "stream.name": "ExampleInputStream"
+    }
+  },
+  {
+    "PropertyGroupId": "OutputStream0",
+    "PropertyMap": {
+      "aws.region": "us-east-1",
+      "stream.name": "ExampleOutputStream"
+    }
+  }
+]

scala/GettingStarted/src/main/resources/log4j2.properties

@@ -0,0 +1,7 @@
+rootLogger.level = INFO
+rootLogger.appenderRef.console.ref = ConsoleAppender
+
+appender.console.name = ConsoleAppender
+appender.console.type = CONSOLE
+appender.console.layout.type = PatternLayout
+appender.console.layout.pattern = %d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n

scala/GettingStarted/src/main/scala/com/amazonaws/services/kinesisanalytics/BasicStreamingJob.scala

@@ -0,0 +1,57 @@
+package com.amazonaws.services.msf
+
+import com.amazonaws.services.kinesisanalytics.runtime.KinesisAnalyticsRuntime
+import org.apache.flink.api.common.serialization.SimpleStringSchema
+import org.apache.flink.connector.kinesis.sink.KinesisStreamsSink
+import org.apache.flink.streaming.api.environment.{LocalStreamEnvironment, StreamExecutionEnvironment}
+import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer
+
+import java.util
+import java.util.Properties
+
+
+object BasicStreamingJob {
+
+  private val streamNameKey = "stream.name"
+  private val defaultInputStreamName = "ExampleInputStream"
+  private val defaultOutputStreamName = "ExampleOutputStream"
+  private val localApplicationPropertiesResource = "/flink-application-properties-dev.json"
+
+  private def getApplicationProperties(env: StreamExecutionEnvironment): util.Map[String, Properties] = {
+    env match {
+      case localEnv: LocalStreamEnvironment =>
+        KinesisAnalyticsRuntime.getApplicationProperties(getClass.getResource(localApplicationPropertiesResource).getPath)
+      case _ =>
+        KinesisAnalyticsRuntime.getApplicationProperties
+    }
+  }
+
+  private def createSource(env: StreamExecutionEnvironment): FlinkKinesisConsumer[String] = {
+    val applicationProperties = getApplicationProperties(env)
+    val inputProperties = applicationProperties.get("InputStream0")
+
+    new FlinkKinesisConsumer[String](inputProperties.getProperty(streamNameKey, defaultInputStreamName),
+      new SimpleStringSchema, inputProperties)
+  }
+
+  private def createSink(env: StreamExecutionEnvironment): KinesisStreamsSink[String] = {
+    val applicationProperties = getApplicationProperties(env)
+    val outputProperties = applicationProperties.get("OutputStream0")
+
+    KinesisStreamsSink.builder[String]
+      .setKinesisClientProperties(outputProperties)
+      .setSerializationSchema(new SimpleStringSchema)
+      .setStreamName(outputProperties.getProperty(streamNameKey, defaultOutputStreamName))
+      .setPartitionKeyGenerator((element: String) => String.valueOf(element.hashCode))
+      .build
+  }
+
+  def main(args: Array[String]): Unit = {
+    val environment = StreamExecutionEnvironment.getExecutionEnvironment
+    
+    environment
+      .addSource(createSource(environment))
+      .sinkTo(createSink(environment))
+    environment.execute("Flink Streaming Scala Example")
+  }
+}