Fix initial user poll

Author: blootsvoets

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

@@ -25,6 +25,7 @@
 import static org.radarbase.connect.rest.converter.PayloadToSourceRecordConverter.MIN_INSTANT;
 import static org.radarbase.connect.rest.converter.PayloadToSourceRecordConverter.TIMESTAMP_OFFSET_KEY;
 import static org.radarbase.connect.rest.converter.PayloadToSourceRecordConverter.nearFuture;
+import static org.radarbase.connect.rest.request.PollingRequestRoute.max;
 
 import java.io.IOException;
 import java.time.Duration;
@@ -316,7 +317,7 @@ protected Instant nextPoll(User user) {
     } else {
       Instant nextPoll = lastPollPerUser.getOrDefault(user.getId(), MIN_INSTANT)
           .plus(getPollIntervalPerUser());
-      return PollingRequestRoute.max(offset.plus(getLookbackTime()), nextPoll);
+      return max(offset.plus(getLookbackTime()), nextPoll);
     }
   }
 

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

@@ -113,6 +113,13 @@ public void initialize(RestSourceConnectorConfig config) {
 
   @Override
   public Stream<? extends User> stream() {
+    if (nextFetch.get().equals(Instant.MIN)) {
+      try {
+        applyPendingUpdates();
+      } catch (IOException ex) {
+        logger.error("Failed to initially get users from repository");
+      }
+    }
     return this.timedCachedUsers.stream();
   }
 

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

@@ -86,7 +86,7 @@ public class YamlUserRepository implements UserRepository {
 
   private Set<String> configuredUsers;
   private Headers headers;
-  private ConcurrentMap<String, LockedUser> users = new ConcurrentHashMap<>();
+  private final ConcurrentMap<String, LockedUser> users = new ConcurrentHashMap<>();
   private final AtomicReference<Instant> nextFetch = new AtomicReference<>(MIN_INSTANT);
   private Path credentialsDir;
 

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/user/firebase/CovidCollabFirebaseUserRepository.java

@@ -37,7 +37,7 @@ public class CovidCollabFirebaseUserRepository extends FirebaseUserRepository {
   private static final Logger logger =
       LoggerFactory.getLogger(CovidCollabFirebaseUserRepository.class);
 
-  private Map<String, FirebaseUser> cachedUsers = new HashMap<>();
+  private final Map<String, FirebaseUser> cachedUsers = new HashMap<>();
   private CollectionReference userCollection;
   private CollectionReference fitbitCollection;
   private FitbitTokenService fitbitTokenService;