top of the hour cooldown strategy for 429

Author: yatharthranjan

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

@@ -259,6 +259,11 @@ public class FitbitRestSourceConnectorConfig extends RestSourceConnectorConfig {
   private static final String FITBIT_TOO_MANY_REQUESTS_COOLDOWN_DISPLAY = "Too many requests cooldown (s)";
   private static final int FITBIT_TOO_MANY_REQUESTS_COOLDOWN_DEFAULT = 3600; // 1 hour
 
+  public static final String FITBIT_COOLDOWN_STRATEGY_CONFIG = "fitbit.cooldown.strategy";
+  private static final String FITBIT_COOLDOWN_STRATEGY_DOC = "Strategy for handling too many requests cooldown. Options: ROLLING_WINDOW, TOP_OF_HOUR";
+  private static final String FITBIT_COOLDOWN_STRATEGY_DISPLAY = "Cooldown strategy";
+  private static final String FITBIT_COOLDOWN_STRATEGY_DEFAULT = "ROLLING_WINDOW";
+
   private UserRepository userRepository;
   private final Headers clientCredentials;
 
@@ -703,6 +708,16 @@ public String toString() {
             ++orderInGroup,
             Width.SHORT,
             FITBIT_TOO_MANY_REQUESTS_COOLDOWN_DISPLAY)
+
+        .define(FITBIT_COOLDOWN_STRATEGY_CONFIG,
+            Type.STRING,
+            FITBIT_COOLDOWN_STRATEGY_DEFAULT,
+            Importance.MEDIUM,
+            FITBIT_COOLDOWN_STRATEGY_DOC,
+            group,
+            ++orderInGroup,
+            Width.SHORT,
+            FITBIT_COOLDOWN_STRATEGY_DISPLAY)
         ;
   }
 
@@ -910,4 +925,12 @@ public int getMaxForbidden() {
   public int getForbiddenBackoff() {
     return getInt(FITBIT_FORBIDDEN_BACKOFF_CONFIG);
   }
+
+  public String getCooldownStrategy() {
+    String value = getString(FITBIT_COOLDOWN_STRATEGY_CONFIG);
+    if (!value.equalsIgnoreCase("ROLLING_WINDOW") && !value.equalsIgnoreCase("TOP_OF_HOUR")) {
+      throw new ConfigException(FITBIT_COOLDOWN_STRATEGY_CONFIG, value, "Invalid cooldown strategy. Must be either ROLLING_WINDOW or TOP_OF_HOUR.");
+    }
+    return value;
+  }
 }

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

@@ -23,6 +23,7 @@
 import static java.time.ZoneOffset.UTC;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
+import java.time.temporal.ChronoUnit;
 import static java.time.temporal.ChronoUnit.DAYS;
 import static java.time.temporal.ChronoUnit.MINUTES;
 import static java.time.temporal.ChronoUnit.NANOS;
@@ -120,6 +121,7 @@ public abstract class FitbitPollingRoute implements PollingRequestRoute {
   private final Map<String, Integer> forbidden403Counter;
   private int maxForbiddenResponses;
   private Duration forbidden403Cooldown;
+  private String cooldownStrategy;
 
   public FitbitPollingRoute(
       FitbitRequestGenerator generator,
@@ -147,6 +149,7 @@ public void initialize(RestSourceConnectorConfig config) {
     this.maxForbiddenResponses = fitbitConfig.getMaxForbidden();
     this.converter().initialize(fitbitConfig);
     this.forbidden403Cooldown = Duration.ofSeconds(fitbitConfig.getForbiddenBackoff());
+    this.cooldownStrategy = fitbitConfig.getCooldownStrategy();
   }
 
   @Override
@@ -175,19 +178,19 @@ public void requestFailed(RestRequest request, Response response) {
     if (response != null && response.code() == 429) {
       User user = ((FitbitRestRequest) request).getUser();
       tooManyRequestsForUser.add(user);
-      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;
+      if ("TOP_OF_HOUR".equalsIgnoreCase(cooldownStrategy)) {
+        backOff = calculateTopOfHourBackoff();
+        logger.info("Too many requests for user {}. Using TOP_OF_HOUR strategy, backing off until: {}",
+            user, backOff);
+      } else {
+        backOff = calculateRollingWindowBackoff(response);
+        logger.info("Too many requests for user {}. Using ROLLING_WINDOW strategy, backing off until: {}",
+            user, backOff.plus(getPollIntervalPerUser()));
       }
-      Instant backOff = lastPoll.plus(cooldown);
+      
       lastPollPerUser.put(user.getId(), backOff);
-      logger.info("Too many requests for user {}. Backing off until {}",
-          user, backOff.plus(getPollIntervalPerUser()));
     } else if (response != null && response.code() == 403) {
       User user = ((FitbitRestRequest) request).getUser();
       String userId = user.getId();
@@ -206,6 +209,46 @@ public void requestFailed(RestRequest request, Response response) {
     }
   }
 
+  /**
+   * Calculate backoff using top-of-hour strategy.
+   * Waits until the top of the next hour to align with Fitbit's rate limit reset.
+   */
+  private Instant calculateTopOfHourBackoff() {
+    Instant now = Instant.now();
+    Instant topOfNextHour = calculateTopOfNextHour(now).plus(ONE_SECOND);
+    return topOfNextHour;
+  }
+
+  /**
+   * Calculate backoff using rolling window strategy.
+   * Uses configured cooldown duration from when the error occurred.
+   */
+  private Instant calculateRollingWindowBackoff(Response response) {
+    String cooldownString = response.header("Retry-After");
+    Duration cooldown = getTooManyRequestsCooldown();
+    
+    if (cooldownString != null) {
+      try {
+        cooldown = Duration.ofSeconds(Long.parseLong(cooldownString));
+      } catch (NumberFormatException ex) {
+        cooldown = getTooManyRequestsCooldown();
+      }
+    }
+    
+    return lastPoll.plus(cooldown);
+  }
+
+  /**
+   * Calculate the top of the next hour from the given instant.
+   */
+  private Instant calculateTopOfNextHour(Instant instant) {
+    ZonedDateTime zonedDateTime = instant.atZone(UTC);
+    ZonedDateTime topOfNextHour = zonedDateTime
+        .plusHours(1)
+        .truncatedTo(ChronoUnit.HOURS);
+    return topOfNextHour.toInstant();
+  }
+
   /**
    * Actually construct requests, based on the current offset
    * @param user Fitbit user