Nullable KTable FK extractor (#17)

* allow for nullable values from a KTable FK extractor

* bump versions, enable Gradle build&config caches

build.gradle.kts

@@ -27,6 +27,6 @@ dependencies {
 }
 
 tasks.wrapper {
-  gradleVersion = "7.4"
+  gradleVersion = "7.4.2"
   distributionType = Wrapper.DistributionType.ALL
 }

buildSrc/build.gradle.kts

@@ -2,12 +2,12 @@ import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
 
 plugins {
   `kotlin-dsl`
-  kotlin("jvm") version "1.6.10"
+  kotlin("jvm") version "1.6.21"
 }
 
 dependencies {
 
-  val kotlinVersion = "1.6.10"
+  val kotlinVersion = libs.versions.kotlin.get()
   implementation(enforcedPlatform("org.jetbrains.kotlin:kotlin-bom:$kotlinVersion"))
   implementation("org.jetbrains.kotlin:kotlin-serialization")
   implementation("org.jetbrains.kotlin:kotlin-reflect")
@@ -16,18 +16,20 @@ dependencies {
   val kotlinXSerializationVersion = "1.3.2"
   implementation(enforcedPlatform("org.jetbrains.kotlinx:kotlinx-serialization-bom:$kotlinXSerializationVersion"))
 
-  val gitVersioningPluginVersion = "5.1.3"
+  val gitVersioningPluginVersion = "5.2.0"
   implementation("me.qoomon:gradle-git-versioning-plugin:$gitVersioningPluginVersion")
 }
 
-val projectJvmTarget = "11"
+val gradleJvmTarget = "1.8"
+val gradleJvmVersion = "8"
+val gradleKotlinTarget = "1.6"
 
 tasks.withType<KotlinCompile>().configureEach {
 
   kotlinOptions {
-    jvmTarget = projectJvmTarget
-    apiVersion = "1.6"
-    languageVersion = "1.6"
+    jvmTarget = gradleJvmTarget
+    apiVersion = gradleKotlinTarget
+    languageVersion = gradleKotlinTarget
   }
 
   kotlinOptions.freeCompilerArgs += listOf(
@@ -41,10 +43,10 @@ tasks.withType<KotlinCompile>().configureEach {
 
 kotlin {
   jvmToolchain {
-    (this as JavaToolchainSpec).languageVersion.set(JavaLanguageVersion.of(projectJvmTarget))
+    (this as JavaToolchainSpec).languageVersion.set(JavaLanguageVersion.of(gradleJvmVersion))
   }
 
   kotlinDslPluginOptions {
-    jvmTarget.set(projectJvmTarget)
+    jvmTarget.set(gradleJvmTarget)
   }
 }

buildSrc/settings.gradle.kts

@@ -1 +1,10 @@
 apply(from = "./repositories.settings.gradle.kts")
+
+dependencyResolutionManagement {
+  @Suppress("UnstableApiUsage")
+  versionCatalogs {
+    create("libs") {
+      from(files("../gradle/libs.versions.toml"))
+    }
+  }
+}

buildSrc/src/main/kotlin/kotka/convention/kotlin-jvm.gradle.kts

@@ -18,31 +18,33 @@ dependencies {
     because("Only needed to run tests in a version of IntelliJ IDEA that bundles older versions")
   }
 
-  val kotestVersion = "5.0.3"
+  val kotestVersion = "5.2.3"
   testImplementation(platform("io.kotest:kotest-bom:$kotestVersion"))
   testImplementation("io.kotest:kotest-runner-junit5")
   testImplementation("io.kotest:kotest-assertions-core")
   testImplementation("io.kotest:kotest-property")
   testImplementation("io.kotest:kotest-assertions-json")
 
-  testImplementation("io.mockk:mockk:1.12.1")
+  testImplementation("io.mockk:mockk:1.12.3")
 
 }
 
-val projectJvmTarget = "11"
+val projectJvmTarget = "1.8"
+val projectJvmVersion = "8"
+val projectKotlinTarget = "1.6"
 
 kotlin {
   jvmToolchain {
-    (this as JavaToolchainSpec).languageVersion.set(JavaLanguageVersion.of(projectJvmTarget))
+    (this as JavaToolchainSpec).languageVersion.set(JavaLanguageVersion.of(projectJvmVersion))
   }
 }
 
 tasks.withType<KotlinCompile>().configureEach {
 
   kotlinOptions {
     jvmTarget = projectJvmTarget
-    apiVersion = "1.6"
-    languageVersion = "1.6"
+    apiVersion = projectKotlinTarget
+    languageVersion = projectKotlinTarget
   }
 
   kotlinOptions.freeCompilerArgs += listOf(

buildSrc/src/main/kotlin/kotka/convention/subproject.gradle.kts

@@ -4,5 +4,7 @@ plugins {
   base
 }
 
-group = rootProject.group
-version = rootProject.version
+if (project != rootProject) {
+  project.group = rootProject.group
+  project.version = rootProject.version
+}

gradle.properties

@@ -0,0 +1,5 @@
+org.gradle.caching=true
+org.gradle.unsafe.configuration-cache=true
+org.gradle.unsafe.configuration-cache-problems=warn
+
+org.gradle.kotlin.dsl.precompiled.accessors.strict=true

gradle/libs.versions.toml

@@ -2,7 +2,7 @@
 
 [versions]
 
-kotlin = "1.6.10"
+kotlin = "1.6.21"
 jvm    = "11"
 
 kafka  = "3.1.0"

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.4-all.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.4.2-all.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

gradlew

@@ -1,7 +1,7 @@
 #!/bin/sh
 
 #
-# Copyright © 2015-2021 the original authors.
+# Copyright © 2015-2021 the original authors.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -32,10 +32,10 @@
 #       Busybox and similar reduced shells will NOT work, because this script
 #       requires all of these POSIX shell features:
 #         * functions;
-#         * expansions «$var», «${var}», «${var:-default}», «${var+SET}»,
-#           «${var#prefix}», «${var%suffix}», and «$( cmd )»;
-#         * compound commands having a testable exit status, especially «case»;
-#         * various built-in commands including «command», «set», and «ulimit».
+#         * expansions «$var», «${var}», «${var:-default}», «${var+SET}»,
+#           «${var#prefix}», «${var%suffix}», and «$( cmd )»;
+#         * compound commands having a testable exit status, especially «case»;
+#         * various built-in commands including «command», «set», and «ulimit».
 #
 #   Important for patching:
 #

jitpack.yml

@@ -2,4 +2,4 @@
 
 # https://jitpack.io/docs/BUILDING/#java-version
 jdk:
-  - openjdk11
+  - openjdk8

modules/kotka-streams-extensions/src/main/kotlin/dev/adamko/kotka/extensions/tables/KTable.kt

@@ -55,13 +55,21 @@ fun <K, V, otherV, outV> KTable<K, V>.join(
   }
 }
 
+/**
+ * A function that extracts the key ([otherK]) from this table's value ([V]).
+ * If the result is null, the update is ignored as invalid.
+ * @see [KTable.join]
+ */
+fun interface ForeignKeyExtractor<V, otherK> : (V) -> otherK?
+
 
 /** @See [KTable.join] */
+
 fun <K, V, otherK, otherV, outV> KTable<K, V>.join(
   other: KTable<otherK, otherV>,
   name: String? = null,
   materialized: Materialized<K, outV, KeyValueStore<Bytes, ByteArray>>? = null,
-  foreignKeyExtractor: (V) -> otherK,
+  foreignKeyExtractor: ForeignKeyExtractor<V, otherK>,
   joiner: ValueJoiner<V, otherV, outV>,
 ): KTable<K, outV> =
   join(
@@ -78,7 +86,7 @@ fun <K, V, otherK, otherV, outV> KTable<K, V>.join(
   other: KTable<otherK, otherV>,
   tableJoined: TableJoined<K, otherK>? = null,
   materialized: Materialized<K, outV, KeyValueStore<Bytes, ByteArray>>? = null,
-  foreignKeyExtractor: (V) -> otherK,
+  foreignKeyExtractor: ForeignKeyExtractor<V, otherK>,
   joiner: ValueJoiner<V, otherV, outV>,
 ): KTable<K, outV> {
   return when {

modules/kotka-streams-kotlinx-serialization/build.gradle.kts

@@ -13,5 +13,4 @@ dependencies {
 
   api(platform(libs.kotlinx.serialization.bom))
   api(libs.kotlinx.serialization.core)
-
 }

settings.gradle.kts

@@ -7,7 +7,6 @@ include(
   ":modules:kotka-streams-kotlinx-serialization",
 )
 
-enableFeaturePreview("VERSION_CATALOGS")
 enableFeaturePreview("TYPESAFE_PROJECT_ACCESSORS")
 
 apply(from = "./buildSrc/repositories.settings.gradle.kts")