Release 0.5.3

.github/workflows/snyk.yaml

@@ -0,0 +1,34 @@
+name: Snyk test
+
+on:
+  - pull_request
+
+jobs:
+  security:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v3
+      - uses: snyk/actions/setup@master
+        with:
+          snyk-version: v1.1032.0
+
+      - uses: actions/setup-java@v3
+        with:
+          distribution: temurin
+          java-version: 17
+
+      - name: Setup Gradle
+        uses: gradle/gradle-build-action@v2
+
+      - name: Run Snyk to check for vulnerabilities
+        env:
+          SNYK_TOKEN: ${{ secrets.SNYK_TOKEN }}
+        run: >
+          snyk test
+          --all-projects
+          --configuration-matching="^runtimeClasspath$"
+          --fail-on=upgradable
+          --org=radar-base
+          --policy-path=.snyk
+          --severity-threshold=high

buildSrc/src/main/kotlin/Versions.kt

@@ -1,9 +1,9 @@
 @Suppress("ConstPropertyName", "MemberVisibilityCanBePrivate")
 object Versions {
-    const val project = "0.5.2"
+    const val project = "0.5.3"
 
     const val java = 17
-    const val kotlin = "1.9.10"
+    const val kotlin = "1.9.22"
     const val wrapper = "8.4"
 
     const val radarCommons = "1.1.2"
@@ -14,20 +14,18 @@ object Versions {
     const val managementPortal = "2.0.0"
 
     // From image
-    const val jackson = "2.16.1"
+    const val jackson = "2.17.0"
 
-    const val log4j2 = "2.20.0"
-    const val slf4j = "2.0.9"
+    const val log4j2 = "2.23.1"
+    const val slf4j = "2.0.13"
 
     const val okhttp = "4.12.0"
 
-    const val firebaseAdmin = "9.1.0"
+    const val firebaseAdmin = "9.2.0"
     const val radarSchemas = "0.8.7-hotfix"
-    const val ktor = "2.3.5"
+    const val ktor = "2.3.10"
 
-    const val junit = "5.9.3"
-    const val wiremock = "2.27.2"
-    const val mockito = "5.3.1"
-
-    const val kotlinVersion = "1.9.10"
+    const val junit = "5.10.2"
+    const val wiremock = "3.0.1"
+    const val mockito = "5.11.0"
 }

kafka-connect-fitbit-source/Dockerfile

@@ -34,6 +34,13 @@ RUN gradle jar
 
 FROM confluentinc/cp-kafka-connect-base:7.5.0
 
+USER root
+
+RUN yum remove -y zulu11-ca-jdk-headless && yum remove -y zulu11-ca-jre-headless
+RUN yum install -y zulu17-ca-jdk-headless && yum install -y zulu17-ca-jre-headless
+
+USER appuser
+
 MAINTAINER Joris Borgdorff <joris@thehyve.nl>
 
 LABEL description="Kafka REST API Source connector"

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/user/ServiceUserRepository.kt

@@ -240,7 +240,10 @@ class ServiceUserRepository : UserRepository {
     ): T = withContext(Dispatchers.IO) {
         val response = client.request(builder)
         val contentLength = response.contentLength()
-        val hasBody = contentLength != null && contentLength > 0
+        // if Transfer-Encoding: chunked, then the request has data but contentLength will be null.
+        val transferEncoding = response.headers["Transfer-Encoding"]
+        val hasBody = (contentLength != null && contentLength > 0) ||
+            (transferEncoding != null && transferEncoding.contains("chunked"))
         if (response.status == HttpStatusCode.NotFound) {
             throw NoSuchElementException("URL " + response.request.url + " does not exist")
         } else if (!response.status.isSuccess() || !hasBody) {

kafka-connect-oura-source/Dockerfile

@@ -34,6 +34,13 @@ RUN gradle jar
 
 FROM confluentinc/cp-kafka-connect-base:7.5.0
 
+USER root
+
+RUN yum remove -y zulu11-ca-jdk-headless && yum remove -y zulu11-ca-jre-headless
+RUN yum install -y zulu17-ca-jdk-headless && yum install -y zulu17-ca-jre-headless
+
+USER appuser
+
 MAINTAINER Pauline Conde <pauline.conde@kcl.ac.uk>
 
 LABEL description="Kafka Oura REST API Source connector"

kafka-connect-oura-source/src/main/java/org/radarbase/connect/rest/oura/offset/KafkaOffsetManager.java

@@ -36,7 +36,7 @@ public void initialize(List<Map<String, Object>> partitions) {
       .filter(e -> e.getValue() != null && e.getValue().containsKey(TIMESTAMP_OFFSET_KEY))
       .collect(Collectors.toMap(
           e -> (String) e.getKey().get("user") + "-" + e.getKey().get("route"),
-          e -> Instant.ofEpochMilli(((Number) e.getValue().get(TIMESTAMP_OFFSET_KEY)).longValue())));
+          e -> Instant.ofEpochSecond(((Number) e.getValue().get(TIMESTAMP_OFFSET_KEY)).longValue())));
     } else {
       logger.warn("Offset storage reader is null, will resume from an empty state.");
     }

kafka-connect-rest-source/src/test/java/org/radarbase/connect/rest/RestTaskTest.java

@@ -17,27 +17,11 @@
 
 package org.radarbase.connect.rest;
 
-import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
-import static com.github.tomakehurst.wiremock.client.WireMock.equalTo;
-import static com.github.tomakehurst.wiremock.client.WireMock.matching;
-import static com.github.tomakehurst.wiremock.client.WireMock.post;
-import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor;
-import static com.github.tomakehurst.wiremock.client.WireMock.resetAllRequests;
-import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
-import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
-import static com.github.tomakehurst.wiremock.client.WireMock.urlMatching;
-import static com.github.tomakehurst.wiremock.client.WireMock.verify;
-import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
 import com.github.tomakehurst.wiremock.WireMockServer;
 import com.github.tomakehurst.wiremock.client.VerificationException;
 import com.github.tomakehurst.wiremock.client.WireMock;
 import com.github.tomakehurst.wiremock.verification.LoggedRequest;
 import com.github.tomakehurst.wiremock.verification.NearMiss;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.apache.kafka.connect.source.SourceTaskContext;
 import org.apache.kafka.connect.storage.OffsetStorageReader;
@@ -56,6 +40,23 @@
 import org.radarbase.connect.rest.single.SingleRestSourceConnector;
 import org.radarbase.connect.rest.single.SingleRestSourceConnectorConfig;
 
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.equalTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.matching;
+import static com.github.tomakehurst.wiremock.client.WireMock.post;
+import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.resetAllRequests;
+import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlMatching;
+import static com.github.tomakehurst.wiremock.client.WireMock.verify;
+import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
 @ExtendWith(WireMockRule.class)
 public class RestTaskTest {
 
@@ -167,7 +168,7 @@ public void afterEach(ExtensionContext context) {
       }
     }
 
-    private void checkForUnmatchedRequests() {
+    public void checkForUnmatchedRequests() {
       List<LoggedRequest> unmatchedRequests = findAllUnmatchedRequests();
       if (!unmatchedRequests.isEmpty()) {
         List<NearMiss> nearMisses = findNearMissesForAllUnmatchedRequests();

oura-library/src/main/kotlin/org/radarbase/oura/request/OuraRequestGenerator.kt

@@ -90,7 +90,7 @@ constructor(
                 offsetTime.coerceAtLeast(startDate)
             }
         val endDate = if (user.endDate >= Instant.now()) Instant.now() else user.endDate
-        if (Duration.between(startOffset, endDate).toDays() <= ONE_DAY) {
+        if (Duration.between(startOffset, endDate) <= ONE_DAY) {
             logger.info("Interval between dates is too short. Backing off..")
             userNextRequest[user.versionedId] = Instant.now().plus(USER_BACK_OFF_TIME)
             return emptySequence()
@@ -145,11 +145,13 @@ constructor(
                     ?: nextRequestTime
         } else {
             if (request.startDate.plus(TIME_AFTER_REQUEST).isBefore(Instant.now())) {
+                logger.info("No records found, updating offsets to end date..")
                 ouraOffsetManager.updateOffsets(
                     request.route,
                     request.user,
                     request.endDate,
                 )
+                userNextRequest[request.user.versionedId] = Instant.now().plus(BACK_OFF_TIME)
             }
         }
         return records
@@ -240,10 +242,10 @@ constructor(
     companion object {
         private val logger = LoggerFactory.getLogger(OuraRequestGenerator::class.java)
         private val BACK_OFF_TIME = Duration.ofMinutes(10L)
-        private val ONE_DAY = 1L
+        private val ONE_DAY = Duration.ofDays(1L)
         private val TIME_AFTER_REQUEST = Duration.ofDays(30)
-        private val USER_BACK_OFF_TIME = Duration.ofMinutes(2L)
-        private val SUCCESS_BACK_OFF_TIME = Duration.ofSeconds(3L)
+        private val USER_BACK_OFF_TIME = Duration.ofHours(12L)
+        private val SUCCESS_BACK_OFF_TIME = Duration.ofMinutes(1L)
         private val USER_MAX_REQUESTS = 20
         val JSON_FACTORY = JsonFactory()
         val JSON_READER = ObjectMapper(JSON_FACTORY).registerModule(JavaTimeModule()).reader()