Bumped Spark to 3.2.0.

build.sbt

@@ -48,6 +48,7 @@ lazy val flinkSbtPlugin =
       buildInfoKeys := Seq[BuildInfoKey](version),
       addSbtPlugin("se.marcuslonnberg" % "sbt-docker" % "1.8.2"),
       addSbtPlugin("com.typesafe.sbt" % "sbt-native-packager" % "1.3.25"),
+      addSbtPlugin("com.lightbend.cloudflow" % "sbt-cloudflow" % Dependencies.Versions.cloudflowVersion),
       scriptedLaunchOpts := {
         scriptedLaunchOpts.value ++
         Seq("-Xmx1024M", "-Dplugin.version=" + version.value)
@@ -108,6 +109,7 @@ lazy val sparkSbtPlugin =
       buildInfoKeys := Seq[BuildInfoKey](version),
       addSbtPlugin("se.marcuslonnberg" % "sbt-docker" % "1.8.2"),
       addSbtPlugin("com.typesafe.sbt" % "sbt-native-packager" % "1.3.25"),
+      addSbtPlugin("com.lightbend.cloudflow" % "sbt-cloudflow" % Dependencies.Versions.cloudflowVersion),
       scriptedLaunchOpts := {
         scriptedLaunchOpts.value ++
         Seq("-Xmx1024M", "-Dplugin.version=" + version.value)

cloudflow-it/Makefile

@@ -28,14 +28,16 @@ prepare-swiss-knife:
 .PHONY: prepare-clis
 prepare-clis:
 	@echo '****** Prepare the runtimes Clis'
-	wget https://downloads.apache.org/spark/spark-3.1.2/spark-3.1.2-bin-hadoop2.7.tgz
-	tar -xf spark-3.1.2-bin-hadoop2.7.tgz
-	mv spark-3.1.2-bin-hadoop2.7 spark
-	rm spark-3.1.2-bin-hadoop2.7.tgz
-	wget https://archive.apache.org/dist/flink/flink-1.13.5/flink-1.13.5-bin-scala_2.12.tgz
-	tar -xf flink-1.13.5-bin-scala_2.12.tgz
-	mv flink-1.13.5 flink
-	rm flink-1.13.5-bin-scala_2.12.tgz
+	rm -rf spark
+	wget https://downloads.apache.org/spark/spark-3.2.0/spark-3.2.0-bin-hadoop2.7.tgz
+	tar -xf spark-3.2.0-bin-hadoop2.7.tgz
+	mv spark-3.2.0-bin-hadoop2.7 spark
+	rm spark-3.2.0-bin-hadoop2.7.tgz
+	rm -rf flink
+	wget https://archive.apache.org/dist/flink/flink-1.14.2/flink-1.14.2-bin-scala_2.12.tgz
+	tar -xf flink-1.14.2-bin-scala_2.12.tgz
+	mv flink-1.14.2 flink
+	rm flink-1.14.2-bin-scala_2.12.tgz
 
 .PHONY: prepare-cluster
 prepare-cluster:

cloudflow-it/swiss-knife/build.sbt

@@ -43,28 +43,24 @@ lazy val akka = (project in file("./akka"))
   .dependsOn(datamodel)
 
 lazy val spark = (project in file("./spark"))
-  .enablePlugins(CloudflowSparkPlugin, CloudflowNativeSparkPlugin)
+  .enablePlugins(CloudflowNativeSparkPlugin)
   .settings(commonSettings)  
   .settings(
     name := "swiss-knife-spark",
-    baseDockerInstructions := sparkNativeCloudflowDockerInstructions.value,
-    libraryDependencies ~= fixSparkNativeCloudflowDeps
   )
   .dependsOn(datamodel)
 
 lazy val flink = (project in file("./flink"))
-  .enablePlugins(CloudflowFlinkPlugin, CloudflowNativeFlinkPlugin)
+  .enablePlugins(CloudflowNativeFlinkPlugin)
   .settings(commonSettings)
   .settings(
     name := "swiss-knife-flink",
-    baseDockerInstructions := flinkNativeCloudflowDockerInstructions.value,
-    libraryDependencies ~= fixFlinkNativeCloudflowDeps
   )
   .dependsOn(datamodel)
 
 lazy val commonSettings = Seq(
   headerLicense := Some(HeaderLicense.ALv2("(C) 2016-2020", "Lightbend Inc. <https://www.lightbend.com>")),
-  scalaVersion := "2.12.11",
+  scalaVersion := "2.12.15",
   scalacOptions ++= Seq(
     "-encoding", "UTF-8",
     "-target:jvm-1.8",

cloudflow-it/swiss-knife/project/cloudflow-plugins.sbt

@@ -6,6 +6,6 @@ val currentVersion = {
   )(identity)
 }
 
-addSbtPlugin("com.lightbend.cloudflow" % "sbt-cloudflow" % "2.1.0")
+addSbtPlugin("com.lightbend.cloudflow" % "sbt-cloudflow" % "2.3.0-RC30")
 addSbtPlugin("com.lightbend.cloudflow" % "contrib-sbt-flink" % currentVersion)
 addSbtPlugin("com.lightbend.cloudflow" % "contrib-sbt-spark" % currentVersion)

cloudflow-sbt-flink/src/main/scala/cloudflow/contrib/CloudflowNativeFlinkPlugin.scala

@@ -23,9 +23,13 @@ import sbtdocker.DockerKeys._
 import sbtdocker.Instructions
 import sbtdocker.staging.CopyFile
 import com.typesafe.sbt.packager.Keys._
+import cloudflow.sbt.CloudflowBasePlugin
+import cloudflow.sbt.CloudflowKeys._
+import cloudflow.sbt.CloudflowBasePlugin._
 
 object CloudflowNativeFlinkPlugin extends AutoPlugin {
   val FlinkHome = "/opt/flink"
+  val FlinkVersion = "1.14.2"
   val FlinkUsrLib = s"$FlinkHome/usrlib"
 
   val AppJarsDir: String = "app-jars"
@@ -35,93 +39,125 @@ object CloudflowNativeFlinkPlugin extends AutoPlugin {
 
   val contribVersion = buildinfo.BuildInfo.version
 
-  object autoImport {
-    val flinkNativeCloudflowDockerInstructions =
-      taskKey[Seq[sbtdocker.Instruction]]("Docker instructions to build the Cloudflow Flink Native Streamlet")
-
-    val flinkNativeCloudflowDeps =
-      Seq(
-        "com.lightbend.cloudflow" %% "contrib-flink" % contribVersion,
-        "com.lightbend.cloudflow" %% "contrib-flink-testkit" % contribVersion % "test")
-
-    def fixFlinkNativeCloudflowDeps(ld: Seq[ModuleID]): Seq[ModuleID] = {
-      ld.filter { dep =>
-        !(dep.organization == "com.lightbend.cloudflow" && dep.name.startsWith("cloudflow-flink"))
-      } ++ flinkNativeCloudflowDeps
-    }
+  override def requires = CloudflowBasePlugin
 
+  object autoImport {
+    val stageAppJars = taskKey[Unit]("Stages the jars for the application.")
   }
 
   import autoImport._
 
   override def trigger = noTrigger
 
-  override lazy val projectSettings = Seq(flinkNativeCloudflowDockerInstructions := {
-    val appDir: File = stage.value
-    val appJarsDir: File = new File(appDir, AppJarsDir)
-    val depJarsDir: File = new File(appDir, DepJarsDir)
-
-    val flinkEntrypoint = (ThisProject / target).value / "cloudflow-flink" / "docker-entrypoint.sh"
-    IO.write(flinkEntrypoint, flinkEntrypointContent)
-
-    val scalaVersion = (ThisProject / scalaBinaryVersion).value
-    val flinkVersion = "1.13.5"
-    val flinkPackageVersion = "1.13.5"
-
-    val flinkTgz = s"flink-${flinkVersion}-bin-scala_${scalaVersion}.tgz"
-
-    val flinkTgzUrl = s"https://archive.apache.org/dist/flink/flink-${flinkPackageVersion}/$flinkTgz"
-
+  override lazy val projectSettings =
     Seq(
-      Instructions.Env("FLINK_VERSION", flinkVersion),
-      Instructions.Env("SCALA_VERSION", scalaVersion),
-      Instructions.Env("FLINK_HOME", FlinkHome),
-      Instructions.Env("PATH", "$FLINK_HOME/bin:$PATH"),
-      Instructions.Env("FLINK_ENV_JAVA_OPTS", "-Dlogback.configurationFile=/opt/logging/logback.xml"),
-      Instructions.User("root"),
-      Instructions.Copy(CopyFile(flinkEntrypoint), "/docker-entrypoint.sh"),
-      Instructions.Run.shell(
+      libraryDependencies ++=
+        Seq(
+          "com.lightbend.cloudflow" %% "contrib-flink" % contribVersion,
+          "com.lightbend.cloudflow" %% "contrib-flink-testkit" % contribVersion % "test"),
+      stageAppJars := Def.taskDyn {
+          Def.task {
+            val stagingDir = stage.value
+            val projectJars = (Runtime / internalDependencyAsJars).value.map(_.data)
+            val depJars = (Runtime / externalDependencyClasspath).value.map(_.data)
+
+            val appJarDir = new File(stagingDir, AppJarsDir)
+            val depJarDir = new File(stagingDir, DepJarsDir)
+
+            IO.delete(appJarDir)
+            IO.delete(depJarDir)
+
+            projectJars.foreach { jar =>
+              IO.copyFile(jar, new File(appJarDir, jar.getName))
+            }
+            depJars.foreach { jar =>
+              IO.copyFile(jar, new File(depJarDir, jar.getName))
+            }
+          }
+        }.value,
+      baseDockerInstructions := {
+        val appDir: File = stage.value
+        val appJarsDir: File = new File(appDir, AppJarsDir)
+        val depJarsDir: File = new File(appDir, DepJarsDir)
+
+        val flinkEntrypoint = (ThisProject / target).value / "cloudflow-flink" / "docker-entrypoint.sh"
+        IO.write(flinkEntrypoint, flinkEntrypointContent)
+
+        val scalaVersion = (ThisProject / scalaBinaryVersion).value
+        val flinkVersion = FlinkVersion
+        val flinkPackageVersion = FlinkVersion
+
+        val flinkTgz = s"flink-${flinkVersion}-bin-scala_${scalaVersion}.tgz"
+
+        val flinkTgzUrl = s"https://archive.apache.org/dist/flink/flink-${flinkPackageVersion}/$flinkTgz"
+
         Seq(
-          Seq("apk", "add", "curl", "wget", "bash", "snappy-dev", "gettext-dev"),
-          Seq("wget", flinkTgzUrl),
-          Seq("tar", "-xvzf", flinkTgz),
-          Seq("mv", s"flink-${flinkVersion}", FlinkHome),
-          Seq("rm", flinkTgz),
-          Seq("addgroup", "-S", "-g", "9999", "flink"),
-          Seq("adduser", "-S", "-h", FlinkHome, "-u", "9999", "flink", "flink"),
-          Seq("addgroup", "-S", "-g", "185", "cloudflow"),
-          Seq("adduser", "-u", "185", "-S", "-h", "/home/cloudflow", "-s", "/sbin/nologin", "cloudflow", "root"),
-          Seq("adduser", "cloudflow", "cloudflow"),
-          Seq("rm", "-rf", "/var/lib/apt/lists/*"),
-          Seq("chown", "-R", "flink:flink", "/var"),
-          Seq("chown", "-R", "flink:root", "/usr/local"),
-          Seq("chmod", "775", "/usr/local"),
-          Seq("mkdir", FlinkUsrLib),
-          Seq(
-            "mv",
-            s"${FlinkHome}/opt/flink-queryable-state-runtime_${scalaVersion}-${flinkVersion}.jar",
-            s"${FlinkHome}/lib"),
-          Seq("mkdir", "-p", "/prometheus"),
-          Seq(
-            "curl",
-            "https://repo1.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/0.11.0/jmx_prometheus_javaagent-0.11.0.jar",
-            "-o",
-            "/prometheus/jmx_prometheus_javaagent.jar"),
-          Seq("chmod", "-R", "777", FlinkHome),
-          Seq("chmod", "a+x", "/docker-entrypoint.sh")).reduce(_ ++ Seq("&&") ++ _)),
-      Instructions.EntryPoint.exec(Seq("bash", "/docker-entrypoint.sh")),
-      Instructions.User(UserInImage),
-      Instructions.WorkDir(FlinkHome),
-      Instructions
-        .Copy(sources = Seq(CopyFile(depJarsDir)), destination = FlinkUsrLib, chown = Some(userAsOwner(UserInImage))),
-      Instructions
-        .Copy(sources = Seq(CopyFile(appJarsDir)), destination = FlinkUsrLib, chown = Some(userAsOwner(UserInImage))),
-      Instructions.Run(
-        s"cp ${FlinkUsrLib}/cloudflow-runner_${(ThisProject / scalaBinaryVersion).value}*.jar  $FlinkUsrLib/cloudflow-runner.jar"),
-      Instructions.Run(s"rm ${FlinkUsrLib}/cloudflow-runner_${(ThisProject / scalaBinaryVersion).value}*.jar"),
-      Instructions.Cmd("help"),
-      Instructions.Expose(Seq(6123, 8081)))
-  })
+          Instructions.Env("FLINK_VERSION", flinkVersion),
+          Instructions.Env("SCALA_VERSION", scalaVersion),
+          Instructions.Env("FLINK_HOME", FlinkHome),
+          Instructions.Env("PATH", "$FLINK_HOME/bin:$PATH"),
+          Instructions.Env("FLINK_ENV_JAVA_OPTS", "-Dlogback.configurationFile=/opt/logging/logback.xml"),
+          Instructions.User("root"),
+          Instructions.Copy(CopyFile(flinkEntrypoint), "/docker-entrypoint.sh"),
+          Instructions.Run.shell(
+            Seq(
+              Seq("apk", "add", "curl", "wget", "bash", "snappy-dev", "gettext-dev"),
+              Seq("wget", flinkTgzUrl),
+              Seq("tar", "-xvzf", flinkTgz),
+              Seq("mv", s"flink-${flinkVersion}", FlinkHome),
+              Seq("rm", flinkTgz),
+              Seq("addgroup", "-S", "-g", "9999", "flink"),
+              Seq("adduser", "-S", "-h", FlinkHome, "-u", "9999", "flink", "flink"),
+              Seq("addgroup", "-S", "-g", "185", "cloudflow"),
+              Seq("adduser", "-u", "185", "-S", "-h", "/home/cloudflow", "-s", "/sbin/nologin", "cloudflow", "root"),
+              Seq("adduser", "cloudflow", "cloudflow"),
+              Seq("rm", "-rf", "/var/lib/apt/lists/*"),
+              Seq("chown", "-R", "flink:flink", "/var"),
+              Seq("chown", "-R", "flink:root", "/usr/local"),
+              Seq("chmod", "775", "/usr/local"),
+              Seq("mkdir", FlinkUsrLib),
+              Seq("mv", s"${FlinkHome}/opt/flink-queryable-state-runtime-${flinkVersion}.jar", s"${FlinkHome}/lib"),
+              Seq("mkdir", "-p", "/prometheus"),
+              Seq(
+                "curl",
+                "https://repo1.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/0.11.0/jmx_prometheus_javaagent-0.11.0.jar",
+                "-o",
+                "/prometheus/jmx_prometheus_javaagent.jar"),
+              Seq("chmod", "-R", "777", FlinkHome),
+              Seq("chmod", "a+x", "/docker-entrypoint.sh")).reduce(_ ++ Seq("&&") ++ _)),
+          Instructions.EntryPoint.exec(Seq("bash", "/docker-entrypoint.sh")),
+          Instructions.User(UserInImage),
+          Instructions.WorkDir(FlinkHome),
+          Instructions
+            .Copy(
+              sources = Seq(CopyFile(depJarsDir)),
+              destination = FlinkUsrLib,
+              chown = Some(userAsOwner(UserInImage))),
+          Instructions
+            .Copy(
+              sources = Seq(CopyFile(appJarsDir)),
+              destination = FlinkUsrLib,
+              chown = Some(userAsOwner(UserInImage))),
+          Instructions.Run(
+            s"cp ${FlinkUsrLib}/cloudflow-runner_${(ThisProject / scalaBinaryVersion).value}*.jar  $FlinkUsrLib/cloudflow-runner.jar"),
+          Instructions.Run(s"rm ${FlinkUsrLib}/cloudflow-runner_${(ThisProject / scalaBinaryVersion).value}*.jar"),
+          Instructions.Cmd("help"),
+          Instructions.Expose(Seq(6123, 8081)))
+      },
+      dockerfile in docker := {
+        val log = streams.value.log
+
+        IO.delete(((ThisProject / target).value / "docker"))
+
+        // this triggers side-effects, e.g. files being created in the staging area
+        stageAppJars.value
+
+        new Dockerfile {
+          from(cloudflowDockerBaseImage.value)
+          addInstructions((ThisProject / baseDockerInstructions).value)
+          addInstructions((ThisProject / extraDockerInstructions).value)
+        }
+      })
 
   private lazy val flinkEntrypointContent =
     scala.io.Source

cloudflow-sbt-flink/src/sbt-test/sbt-cloudflow/simple-flink/build.sbt

@@ -1,12 +1,10 @@
 lazy val helloWorld =  (project in file("."))
-    .enablePlugins(CloudflowApplicationPlugin, CloudflowFlinkPlugin, CloudflowNativeFlinkPlugin)
+    .enablePlugins(CloudflowApplicationPlugin, CloudflowNativeFlinkPlugin)
     .settings(
-      scalaVersion := "2.12.11",
+      scalaVersion := "2.12.15",
       name := "hello-world",
       version := "0.0.1",
       resolvers += "Flink 13.0".at("https://repository.apache.org/content/repositories/orgapacheflink-1420/"),
-      baseDockerInstructions := flinkNativeCloudflowDockerInstructions.value,
-      libraryDependencies ~= fixFlinkNativeCloudflowDeps
     )
 
 val checkCRFile = taskKey[Unit]("Testing the CR file")