Small fixes

Author: blootsvoets

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

@@ -70,6 +70,14 @@ public class FitbitRestSourceConnectorConfig extends RestSourceConnectorConfig {
   private static final boolean FITBIT_API_INTRADAY_ACCESS_DEFAULT = false;
   private static final String FITBIT_API_INTRADAY_ACCESS_DISPLAY = "Is Fitbit Intraday API available?";
 
+  public static final String FITBIT_USER_POLL_INTERVAL = "fitbit.user.poll.interval";
+  private static final String FITBIT_USER_POLL_INTERVAL_DOC = "Polling interval per Fitbit user per request route in seconds.";
+  // 150 requests per hour -> 2.5 per minute. There are currently 5 paths, that limits us to 1
+  // call per route per 2 minutes.
+  private static final int FITBIT_USER_POLL_INTERVAL_DEFAULT = 150;
+  private static final String FITBIT_USER_POLL_INTERVAL_DISPLAY = "Per-user per-route polling interval.";
+
+
   public static final String FITBIT_USER_CREDENTIALS_DIR_CONFIG = "fitbit.user.dir";
   private static final String FITBIT_USER_CREDENTIALS_DIR_DOC = "Directory containing Fitbit user information and credentials. Only used if a file-based user repository is configured.";
   private static final String FITBIT_USER_CREDENTIALS_DIR_DISPLAY = "User directory";
@@ -178,9 +186,19 @@ public String toString() {
             Width.SHORT,
             FITBIT_API_SECRET_DISPLAY)
 
+        .define(FITBIT_USER_POLL_INTERVAL,
+            Type.INT,
+            FITBIT_USER_POLL_INTERVAL_DEFAULT,
+            Importance.MEDIUM,
+            FITBIT_USER_POLL_INTERVAL_DOC,
+            group,
+            ++orderInGroup,
+            Width.SHORT,
+            FITBIT_USER_POLL_INTERVAL_DISPLAY)
+
         .define(FITBIT_API_INTRADAY_ACCESS_CONFIG,
             Type.BOOLEAN,
-            true,
+            FITBIT_API_INTRADAY_ACCESS_DEFAULT,
             Importance.MEDIUM,
             FITBIT_API_INTRADAY_ACCESS_DOC,
             group,
@@ -352,4 +370,12 @@ public String getActivityLogTopic() {
   public boolean hasIntradayAccess() {
     return getBoolean(FITBIT_API_INTRADAY_ACCESS_CONFIG);
   }
+
+  public Duration getPollIntervalPerUser() {
+    return Duration.ofSeconds(getInt(FITBIT_USER_POLL_INTERVAL));
+  }
+
+  public Duration getTooManyRequestsCooldownInterval() {
+    return Duration.ofHours(1);
+  }
 }

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

@@ -140,13 +140,15 @@ public void initialize(RestSourceConnectorConfig config) {
 
   @Override
   public void requestSucceeded(RestRequest request, SourceRecord record) {
+    lastPollPerUser.put(((FitbitRestRequest) request).getUser().getId(), lastPoll);
     String userKey = ((FitbitRestRequest) request).getUser().getId();
     Instant offset = Instant.ofEpochMilli((Long) record.sourceOffset().get(TIMESTAMP_OFFSET_KEY));
     offsets.put(userKey, offset);
   }
 
   @Override
   public void requestEmpty(RestRequest request) {
+    lastPollPerUser.put(((FitbitRestRequest) request).getUser().getId(), lastPoll);
     FitbitRestRequest fitbitRequest = (FitbitRestRequest) request;
     Instant endOffset = fitbitRequest.getDateRange().end().toInstant();
     if (DAYS.between(endOffset, lastPoll) >= HISTORICAL_TIME_DAYS) {
@@ -160,7 +162,16 @@ public void requestFailed(RestRequest request, Response response) {
     if (response != null && response.code() == 429) {
       User user = ((FitbitRestRequest)request).getUser();
       tooManyRequestsForUser.add(user);
-      Instant backOff = lastPoll.plus(getTooManyRequestsCooldown());
+      String cooldownString = response.header("Retry-After");
+      Duration cooldown = getTooManyRequestsCooldown();
+      if (cooldownString != null) {
+        try {
+          cooldown = Duration.ofSeconds(Long.parseLong(cooldownString));
+        } catch (NumberFormatException ex) {
+          cooldown = getTooManyRequestsCooldown();
+        }
+      }
+      Instant backOff = lastPoll.plus(cooldown);
       lastPollPerUser.put(user.getId(), backOff);
       logger.info("Too many requests for user {}. Backing off until {}",
           user, backOff.plus(getPollIntervalPerUser()));
@@ -185,10 +196,7 @@ public Stream<FitbitRestRequest> requests() {
           .map(u -> new AbstractMap.SimpleImmutableEntry<>(u, nextPoll(u)))
           .filter(u -> lastPoll.isAfter(u.getValue()))
           .sorted(Comparator.comparing(Map.Entry::getValue))
-          .flatMap(u -> {
-            lastPollPerUser.put(u.getKey().getId(), lastPoll);
-            return this.createRequests(u.getKey());
-          })
+          .flatMap(u -> this.createRequests(u.getKey()))
           .filter(Objects::nonNull);
     } catch (IOException e) {
       logger.warn("Cannot read users");

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

@@ -214,12 +214,4 @@ public RequestGenerator getRequestGenerator() {
     requestGenerator.initialize(this);
     return requestGenerator;
   }
-
-  public Duration getPollIntervalPerUser() {
-    return Duration.ofMinutes(30);
-  }
-
-  public Duration getTooManyRequestsCooldownInterval() {
-    return Duration.ofHours(1);
-  }
 }

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

@@ -76,13 +76,16 @@ public List<SourceRecord> poll() throws InterruptedException {
       }
 
       Iterator<? extends RestRequest> requestIterator = requestGenerator.requests()
-          .filter(RestRequest::isStillValid)
           .iterator();
 
 
       while (requests.isEmpty() && requestIterator.hasNext()) {
         RestRequest request = requestIterator.next();
 
+        if (!request.isStillValid()) {
+          continue;
+        }
+
         logger.info("Requesting {}", request.getRequest().url());
         requestsGenerated++;
 

kafka-connect-rest-source/src/main/java/org/radarbase/connect/rest/request/RestRequest.java

@@ -72,7 +72,7 @@ public boolean isStillValid() {
 
   /**
    * Handle the request using the internal client, using the request route converter.
-   * @return stream of resulting source records, or {@code null} if the response was not successful.
+   * @return stream of resulting source records.
    * @throws IOException if making or parsing the request failed.
    */
   public Stream<SourceRecord> handleRequest() throws IOException {
@@ -83,7 +83,7 @@ public Stream<SourceRecord> handleRequest() throws IOException {
     try (Response response = client.newCall(request).execute()) {
       if (!response.isSuccessful()) {
         route.requestFailed(this, response);
-        return null;
+        return Stream.empty();
       }
 
       Collection<SourceRecord> records = route.converter().convert(this, response);