fix: report remaining, not elapsed, time from LinearRateLimiter

`RateLimiter.isLimited` is documented to return the "minimal duration
until a permission could be acquired again", but `LinearRateLimiter`
returned the time *elapsed* since the current period started:

    Duration.between(actualState.getLastRefreshTime(), LocalDateTime.now())

The two are inverted. Measured with a 1000ms refresh period:

    moment                  reported    correct
    right after limit hit      19ms     ~1000ms
    800ms into the period     804ms      ~200ms

`EventProcessor.handleRateLimitedSubmission` feeds this value straight
into `TimerEventSource.scheduleOnce`, so a rate-limited resource is
rescheduled almost immediately after the limit is reached (floored at
MINIMAL_RATE_LIMIT_RESCHEDULE_DURATION), gets rate-limited again, and
repeats — producing a burst of pointless timer events. Conversely, a
resource limited near the end of a period waits roughly a full extra
period after a permission was already available.

Now returns the time until the current period ends, clamped at zero.

`returnsMinimalDurationToAcquirePermission` only asserted
`isLessThan(REFRESH_PERIOD)`, which held for both the correct and the
inverted value; it now also asserts the reported wait is close to the
full period. A second test asserts the reported duration shrinks as the
period elapses, which is what distinguishes remaining from elapsed. Both
fail without this change.

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/rate/LinearRateLimiter.java

@@ -64,7 +64,10 @@ public Optional<Duration> isLimited(RateLimitState rateLimitState) {
       actualState.increaseCount();
       return Optional.empty();
     } else {
-      return Optional.of(Duration.between(actualState.getLastRefreshTime(), LocalDateTime.now()));
+      var remaining =
+          Duration.between(
+              LocalDateTime.now(), actualState.getLastRefreshTime().plus(refreshPeriod));
+      return Optional.of(remaining.isNegative() ? Duration.ZERO : remaining);
     }
   }
 

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/rate/LinearRateLimiterTest.java

@@ -53,7 +53,22 @@ void returnsMinimalDurationToAcquirePermission() {
     res = rl.isLimited(state);
 
     assertThat(res).isPresent();
-    assertThat(res.get()).isLessThan(REFRESH_PERIOD);
+    assertThat(res.get()).isLessThanOrEqualTo(REFRESH_PERIOD);
+    // the whole period is still ahead of us, so the reported wait must be close to it
+    assertThat(res.get()).isGreaterThan(REFRESH_PERIOD.dividedBy(2));
+  }
+
+  @Test
+  void reportedDurationIsTheTimeRemainingNotTheTimeElapsed() throws InterruptedException {
+    var rl = new LinearRateLimiter(REFRESH_PERIOD, 1);
+    assertThat(rl.isLimited(state)).isEmpty();
+
+    var justAfterLimit = rl.isLimited(state).orElseThrow();
+    Thread.sleep(REFRESH_PERIOD.toMillis() / 2);
+    var halfWayThroughPeriod = rl.isLimited(state).orElseThrow();
+
+    // as the refresh period elapses, less time remains until a permission is available again
+    assertThat(halfWayThroughPeriod).isLessThan(justAfterLimit);
   }
 
   @Test