Merge pull request #42 from RADAR-base/alternative_poll

Alternative poll algorithm

Author: blootsvoets

README.md

@@ -61,8 +61,6 @@ your Fitbit App client ID and client secret. The following tables shows the poss
 <tr>
 <td>fitbit.user.repository.url</td></td><td>URL for webservice containing user credentials. Only used if a webservice-based user repository is configured.</td></td><td>string</td></td><td>""</td></td><td></td></td><td>low</td></td></tr>
 <tr>
-<td>fitbit.max.users.per.poll</td></td><td>Maximum number of users to query in a single poll operation. Decrease this if memory constrains are pressing.</td></td><td>int</td></td><td>100</td></td><td>[1,...]</td></td><td>low</td></td></tr>
-<tr>
 <td>fitbit.intraday.steps.topic</td></td><td>Topic for Fitbit intraday steps</td></td><td>string</td></td><td>connect_fitbit_intraday_steps</td></td><td>non-empty string without control characters</td></td><td>low</td></td></tr>
 <tr>
 <td>fitbit.intraday.heart.rate.topic</td></td><td>Topic for Fitbit intraday heart_rate</td></td><td>string</td></td><td>connect_fitbit_intraday_heart_rate</td></td><td>non-empty string without control characters</td></td><td>low</td></td></tr>

docker/source-fitbit.properties.template

@@ -6,6 +6,5 @@ rest.source.poll.interval.ms=5000
 rest.source.request.generator.class=org.radarbase.connect.rest.fitbit.request.FitbitRequestGenerator
 fitbit.api.client=?
 fitbit.api.secret=?
-fitbit.max.users.per.poll=10
 fitbit.user.repository.class=org.radarbase.connect.rest.fitbit.user.ServiceUserRepository
-fitbit.user.repository.url=http://radar-device-auth-backend:8080/
+fitbit.user.repository.url=http://localhost:8080/

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

@@ -105,11 +105,6 @@ public class FitbitRestSourceConnectorConfig extends RestSourceConnectorConfig {
   private static final String FITBIT_TIME_ZONE_TOPIC_DEFAULT = "connect_fitbit_time_zone";
   private static final String FITBIT_TIME_ZONE_TOPIC_DISPLAY = "Time zone topic";
 
-  private static final String FITBIT_MAX_USERS_PER_POLL_CONFIG = "fitbit.max.users.per.poll";
-  private static final String FITBIT_MAX_USERS_PER_POLL_DOC = "Maximum number of users to query in a single poll operation. Decrease this if memory constrains are pressing.";
-  private static final int FITBIT_MAX_USERS_PER_POLL_DEFAULT = 100;
-  private static final String FITBIT_MAX_USERS_PER_POLL_DISPLAY = "Maximum users per poll";
-
   private static final String FITBIT_ACTIVITY_LOG_TOPIC_CONFIG = "fitbit.activity.log.topic";
   private static final String FITBIT_ACTIVITY_LOG_TOPIC_DOC = "Topic for Fitbit activity log.";
   private static final String FITBIT_ACTIVITY_LOG_TOPIC_DEFAULT = "connect_fitbit_activity_log";
@@ -152,7 +147,6 @@ public String toString() {
     };
 
     return RestSourceConnectorConfig.conf()
-
         .define(FITBIT_USERS_CONFIG,
             Type.LIST,
             Collections.emptyList(),
@@ -225,17 +219,6 @@ public String toString() {
             Width.SHORT,
             FITBIT_USER_REPOSITORY_URL_DISPLAY)
 
-        .define(FITBIT_MAX_USERS_PER_POLL_CONFIG,
-            Type.INT,
-            FITBIT_MAX_USERS_PER_POLL_DEFAULT,
-            Range.atLeast(1),
-            Importance.LOW,
-            FITBIT_MAX_USERS_PER_POLL_DOC,
-            group,
-            ++orderInGroup,
-            Width.SHORT,
-            FITBIT_MAX_USERS_PER_POLL_DISPLAY)
-
         .define(FITBIT_INTRADAY_STEPS_TOPIC_CONFIG,
             Type.STRING,
             FITBIT_INTRADAY_STEPS_TOPIC_DEFAULT,
@@ -362,10 +345,6 @@ public Headers getClientCredentials() {
     return clientCredentials;
   }
 
-  public long getMaxUsersPerPoll() {
-    return getInt(FITBIT_MAX_USERS_PER_POLL_CONFIG);
-  }
-
   public String getActivityLogTopic() {
     return getString(FITBIT_ACTIVITY_LOG_TOPIC_CONFIG);
   }

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

@@ -109,7 +109,6 @@ public abstract class FitbitPollingRoute implements PollingRequestRoute {
   private Duration pollInterval;
   private Instant lastPoll;
   private String baseUrl;
-  private long maxUsersPerPoll;
   private Duration pollIntervalPerUser;
   private final Set<User> tooManyRequestsForUser;
   private Duration tooManyRequestsCooldown;
@@ -133,7 +132,6 @@ public void initialize(RestSourceConnectorConfig config) {
     FitbitRestSourceConnectorConfig fitbitConfig = (FitbitRestSourceConnectorConfig) config;
     this.pollInterval = fitbitConfig.getPollInterval();
     this.baseUrl = fitbitConfig.getUrl();
-    this.maxUsersPerPoll = fitbitConfig.getMaxUsersPerPoll();
     this.pollIntervalPerUser = fitbitConfig.getPollIntervalPerUser();
     this.tooManyRequestsCooldown = fitbitConfig.getTooManyRequestsCooldownInterval()
         .minus(getPollIntervalPerUser());
@@ -187,7 +185,6 @@ public Stream<FitbitRestRequest> requests() {
           .map(u -> new AbstractMap.SimpleImmutableEntry<>(u, nextPoll(u)))
           .filter(u -> lastPoll.isAfter(u.getValue()))
           .sorted(Comparator.comparing(Map.Entry::getValue))
-          .limit(maxUsersPerPoll)
           .flatMap(u -> {
             lastPollPerUser.put(u.getKey().getId(), lastPoll);
             return this.createRequests(u.getKey());
@@ -199,6 +196,15 @@ public Stream<FitbitRestRequest> requests() {
     }
   }
 
+
+  /** Get the time that this route should be polled again. */
+  @Override
+  public Instant getTimeOfNextRequest() {
+    return nextPolls()
+            .min(Comparator.naturalOrder())
+            .orElse(nearFuture());
+  }
+
   private Map<String, Object> getPartition(User user) {
     return partitions.computeIfAbsent(user.getId(),
         k -> generator.getPartition(routeName, user));

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

@@ -22,10 +22,13 @@
 
 import java.lang.reflect.InvocationTargetException;
 import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.atomic.LongAdder;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.source.SourceRecord;
@@ -60,24 +63,30 @@ public void start(Map<String, String> map) {
 
   @Override
   public List<SourceRecord> poll() throws InterruptedException {
-    long timeout = MILLIS.between(Instant.now(), requestGenerator.getTimeOfNextRequest());
-    if (timeout > 0) {
-      logger.info("Waiting {} milliseconds for next available request", timeout);
-      Thread.sleep(timeout);
-    }
-
     LongAdder requestsGenerated = new LongAdder();
+    List<SourceRecord> requests;
+
+    do {
+      long timeout = MILLIS.between(Instant.now(), requestGenerator.getTimeOfNextRequest());
+      if (timeout > 0) {
+        logger.info("Waiting {} milliseconds for next available request", timeout);
+        Thread.sleep(timeout);
+      }
 
-    List<SourceRecord> requests = requestGenerator.requests()
-        .filter(RestRequest::isStillValid)
-        .peek(r -> {
-          logger.info("Requesting {}", r.getRequest().url());
-          requestsGenerated.increment();
-        })
-        .flatMap(tryOrNull(RestRequest::handleRequest,
-            (r, ex) -> logger.warn("Failed to make request: {}", ex.toString())))
-        .filter(Objects::nonNull)
-        .collect(Collectors.toList());
+      requests = requestGenerator.requests()
+          .filter(RestRequest::isStillValid)
+          .peek(r -> {
+            logger.info("Requesting {}", r.getRequest().url());
+            requestsGenerated.increment();
+          })
+          .map(tryOrNull(RestRequest::handleRequest,
+              (r, ex) -> logger.warn("Failed to make request: {}", ex.toString())))
+          .filter(Objects::nonNull)
+          .map(s -> s.collect(Collectors.toList()))
+          .filter(l -> !l.isEmpty())
+          .findAny()
+          .orElse(Collections.emptyList());
+    } while (requests.isEmpty());
 
     logger.info("Processed {} records from {} URLs", requests.size(), requestsGenerated.sum());