Merge pull request #147 from RADAR-base/feat/offset-times

Fix Oura offset times

Author: mpgxvii

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

@@ -63,6 +63,7 @@ public class OuraSourceTask extends SourceTask {
   private AvroData avroData = new AvroData(20);
   private KafkaOffsetManager offsetManager;
   String TIMESTAMP_OFFSET_KEY = "timestamp";
+  long TIMEOUT = 60000L;
 
   public void initialize(OuraRestSourceConnectorConfig config, OffsetStorageReader offsetStorageReader) {
     OuraRestSourceConnectorConfig ouraConfig = (OuraRestSourceConnectorConfig) config;
@@ -144,6 +145,8 @@ public List<SourceRecord> poll() throws InterruptedException {
     List<SourceRecord> sourceRecords = Collections.emptyList();
 
     do {
+      Thread.sleep(TIMEOUT);
+
       Map<String, String> configs = context.configs();
       Iterator<? extends RestRequest> requestIterator = this.requests()
           .iterator();

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

@@ -91,8 +91,7 @@ constructor(
             }
         val endDate = user.endDate?.coerceAtMost(Instant.now()) ?: Instant.now()
         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)
+            logger.info("Interval between dates is too short. Not requesting..")
             return emptySequence()
         }
         val endTime = (startOffset + defaultQueryRange).coerceAtMost(endDate)
@@ -130,19 +129,12 @@ constructor(
             ouraOffsetManager.updateOffsets(
                 request.route,
                 request.user,
-                Instant.ofEpochSecond(offset).plus(Duration.ofMillis(500)),
+                Instant.ofEpochSecond(offset).plus(ONE_DAY),
             )
-            val currentNextRequestTime = userNextRequest[request.user.versionedId]
             val nextRequestTime = Instant.now().plus(SUCCESS_BACK_OFF_TIME)
-            userNextRequest[request.user.versionedId] =
-                currentNextRequestTime?.let {
-                    if (currentNextRequestTime > nextRequestTime) {
-                        currentNextRequestTime
-                    } else {
-                        nextRequestTime
-                    }
-                }
-                    ?: nextRequestTime
+            userNextRequest[request.user.versionedId] = userNextRequest[request.user.versionedId]?.let {
+                if (it > nextRequestTime) it else nextRequestTime
+            } ?: nextRequestTime
         } else {
             if (request.startDate.plus(TIME_AFTER_REQUEST).isBefore(Instant.now())) {
                 logger.info("No records found, updating offsets to end date..")
@@ -151,7 +143,9 @@ constructor(
                     request.user,
                     request.endDate,
                 )
-                userNextRequest[request.user.versionedId] = Instant.now().plus(ONE_DAY)
+                userNextRequest[request.user.versionedId] = Instant.now().plus(BACK_OFF_TIME)
+            } else {
+                userNextRequest[request.user.versionedId] = Instant.now().plus(BACK_OFF_TIME)
             }
         }
         return records