Merge branch 'dev' into fitbit-hrv

Author: blootsvoets

Dockerfile

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM --platform=$BUILDPLATFORM gradle:8.1-jdk11 as builder
+FROM --platform=$BUILDPLATFORM gradle:8.4-jdk11 as builder
 
 RUN mkdir /code
 WORKDIR /code
@@ -32,7 +32,7 @@ COPY ./kafka-connect-fitbit-source/src/ /code/kafka-connect-fitbit-source/src
 
 RUN gradle jar
 
-FROM confluentinc/cp-kafka-connect-base:7.4.0
+FROM confluentinc/cp-kafka-connect-base:7.5.0
 
 MAINTAINER Joris Borgdorff <[email protected]>
 

build.gradle.kts

@@ -10,6 +10,7 @@ description = "Kafka connector for REST API sources"
 
 radarRootProject {
     projectVersion.set(Versions.project)
+    gradleVersion.set(Versions.wrapper)
 }
 
 subprojects {

buildSrc/build.gradle.kts

@@ -1,5 +1,5 @@
 plugins {
-    kotlin("jvm") version "1.8.10"
+    kotlin("jvm") version "1.9.10"
 }
 
 repositories {

buildSrc/src/main/kotlin/Versions.kt

@@ -1,8 +1,10 @@
+@Suppress("ConstPropertyName", "MemberVisibilityCanBePrivate")
 object Versions {
     const val project = "0.4.2-SNAPSHOT"
 
     const val java = 11
     const val kotlin = "1.9.10"
+    const val wrapper = "8.4"
 
     const val radarCommons = "1.1.1"
     const val confluent = "7.5.0"
@@ -16,10 +18,9 @@ object Versions {
 
     const val okhttp = "4.11.0"
 
-    const val managementPortal = "2.0.0"
-
     const val firebaseAdmin = "9.1.0"
     const val radarSchemas = "0.8.6-SNAPSHOT"
+    const val ktor = "2.3.5"
 
     const val junit = "5.9.3"
     const val wiremock = "2.27.2"

gradle/wrapper/gradle-wrapper.properties

@@ -1,6 +1,6 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.3-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.4-bin.zip
 networkTimeout=10000
 validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME

gradlew

@@ -145,15 +145,15 @@ if ! "$cygwin" && ! "$darwin" && ! "$nonstop" ; then
     case $MAX_FD in #(
       max*)
         # In POSIX sh, ulimit -H is undefined. That's why the result is checked to see if it worked.
-        # shellcheck disable=SC3045
+        # shellcheck disable=SC2039,SC3045
         MAX_FD=$( ulimit -H -n ) ||
             warn "Could not query maximum file descriptor limit"
     esac
     case $MAX_FD in  #(
       '' | soft) :;; #(
       *)
         # In POSIX sh, ulimit -n is undefined. That's why the result is checked to see if it worked.
-        # shellcheck disable=SC3045
+        # shellcheck disable=SC2039,SC3045
         ulimit -n "$MAX_FD" ||
             warn "Could not set maximum file descriptor limit to $MAX_FD"
     esac
@@ -202,11 +202,11 @@ fi
 # Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
 DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
 
-# Collect all arguments for the java command;
-#   * $DEFAULT_JVM_OPTS, $JAVA_OPTS, and $GRADLE_OPTS can contain fragments of
-#     shell script including quotes and variable substitutions, so put them in
-#     double quotes to make sure that they get re-expanded; and
-#   * put everything else in single quotes, so that it's not re-expanded.
+# Collect all arguments for the java command:
+#   * DEFAULT_JVM_OPTS, JAVA_OPTS, JAVA_OPTS, and optsEnvironmentVar are not allowed to contain shell fragments,
+#     and any embedded shellness will be escaped.
+#   * For example: A user cannot expect ${Hostname} to be expanded, as it is an environment variable and will be
+#     treated as '${Hostname}' itself on the command line.
 
 set -- \
         "-Dorg.gradle.appname=$APP_BASE_NAME" \

kafka-connect-fitbit-source/build.gradle.kts

@@ -4,13 +4,18 @@ dependencies {
     api(project(":kafka-connect-rest-source"))
     api("io.confluent:kafka-connect-avro-converter:${Versions.confluent}")
     api("org.radarbase:radar-schemas-commons:${Versions.radarSchemas}")
-    implementation("org.radarbase:oauth-client-util:${Versions.managementPortal}")
+    implementation("org.radarbase:radar-commons-kotlin:${Versions.radarCommons}")
 
     implementation(platform("com.fasterxml.jackson:jackson-bom:${Versions.jackson}"))
     implementation("com.fasterxml.jackson.dataformat:jackson-dataformat-yaml")
     implementation("com.fasterxml.jackson.datatype:jackson-datatype-jsr310")
     implementation("com.google.firebase:firebase-admin:${Versions.firebaseAdmin}")
 
+    implementation("io.ktor:ktor-client-auth:${Versions.ktor}")
+    implementation("io.ktor:ktor-client-content-negotiation:${Versions.ktor}")
+    implementation("io.ktor:ktor-serialization-jackson:${Versions.ktor}")
+    implementation("io.ktor:ktor-client-cio-jvm:${Versions.ktor}")
+
     // Included in connector runtime
     compileOnly("org.apache.kafka:connect-api:${Versions.kafka}")
     compileOnly(platform("com.fasterxml.jackson:jackson-bom:${Versions.jackson}"))

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/FitbitRestSourceConnectorConfig.java

@@ -17,11 +17,10 @@
 
 package org.radarbase.connect.rest.fitbit;
 
+import static io.ktor.http.URLUtilsKt.URLBuilder;
 import static org.apache.kafka.common.config.ConfigDef.NO_DEFAULT_VALUE;
 
 import java.lang.reflect.InvocationTargetException;
-import java.net.MalformedURLException;
-import java.net.URL;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -31,8 +30,9 @@
 import java.util.List;
 import java.util.Map;
 
+import io.ktor.http.URLParserException;
+import io.ktor.http.Url;
 import okhttp3.Headers;
-import okhttp3.HttpUrl;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigDef.Importance;
 import org.apache.kafka.common.config.ConfigDef.NonEmptyString;
@@ -479,18 +479,18 @@ public Path getFitbitUserCredentialsPath() {
     return Paths.get(getString(FITBIT_USER_CREDENTIALS_DIR_CONFIG));
   }
 
-  public HttpUrl getFitbitUserRepositoryUrl() {
+  public Url getFitbitUserRepositoryUrl() {
     String urlString = getString(FITBIT_USER_REPOSITORY_URL_CONFIG).trim();
     if (urlString.charAt(urlString.length() - 1) != '/') {
       urlString += '/';
     }
-    HttpUrl url = HttpUrl.parse(urlString);
-    if (url == null) {
+    try {
+      return URLBuilder(urlString).build();
+    } catch (URLParserException ex) {
       throw new ConfigException(FITBIT_USER_REPOSITORY_URL_CONFIG,
           getString(FITBIT_USER_REPOSITORY_URL_CONFIG),
-          "User repository URL " + urlString + " cannot be parsed as URL.");
+          "User repository URL " + urlString + " cannot be parsed as URL: " + ex);
     }
-    return url;
   }
 
   public Headers getClientCredentials() {
@@ -533,15 +533,17 @@ public String getFitbitUserRepositoryClientSecret() {
     return getPassword(FITBIT_USER_REPOSITORY_CLIENT_SECRET_CONFIG).value();
   }
 
-  public URL getFitbitUserRepositoryTokenUrl() {
+  public Url getFitbitUserRepositoryTokenUrl() {
     String value = getString(FITBIT_USER_REPOSITORY_TOKEN_URL_CONFIG);
     if (value == null || value.isEmpty()) {
       return null;
     } else {
       try {
-        return new URL(getString(FITBIT_USER_REPOSITORY_TOKEN_URL_CONFIG));
-      } catch (MalformedURLException e) {
-        throw new ConfigException("Fitbit user repository token URL is invalid.");
+        return URLBuilder(value).build();
+      } catch (URLParserException ex) {
+        throw new ConfigException(FITBIT_USER_REPOSITORY_URL_CONFIG,
+                getString(FITBIT_USER_REPOSITORY_URL_CONFIG),
+                "Fitbit user repository token URL " + value + " cannot be parsed as URL: " + ex);
       }
     }
   }