add the test

flink-autoscaler-plugin-jdbc/src/main/java/org/apache/flink/autoscaler/jdbc/event/JdbcAutoScalerEventHandler.java

@@ -28,7 +28,6 @@
 import javax.annotation.Nullable;
 
 import java.time.Duration;
-import java.time.Instant;
 import java.util.Map;
 import java.util.Objects;
 
@@ -109,7 +108,8 @@ public void handleScalingEvent(
      * @return True means the existing event is still in the interval duration we can update it.
      *     Otherwise, it's too early, we should create a new one instead of updating it.
      */
-    private static boolean intervalCheck(AutoScalerEvent existing, Duration interval) {
-        return existing.getCreateTime().isAfter(Instant.now().minusMillis(interval.toMillis()));
+    private boolean intervalCheck(AutoScalerEvent existing, Duration interval) {
+        return existing.getCreateTime()
+                .isAfter(jdbcEventInteractor.getCurrentInstant().minusMillis(interval.toMillis()));
     }
 }

flink-autoscaler-plugin-jdbc/src/main/java/org/apache/flink/autoscaler/jdbc/event/JdbcEventInteractor.java

@@ -28,6 +28,9 @@
 import java.sql.SQLException;
 import java.sql.Timestamp;
 import java.time.Clock;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Optional;
 
 import static org.apache.flink.util.Preconditions.checkState;
@@ -122,6 +125,29 @@ public void updateEvent(long id, String message, int eventCount) throws Exceptio
         }
     }
 
+    public Instant getCurrentInstant() {
+        return clock.instant();
+    }
+
+    @VisibleForTesting
+    protected List<AutoScalerEvent> queryEvents(String jobKey, String reason) throws Exception {
+        var query =
+                "select * from t_flink_autoscaler_event_handler "
+                        + "where job_key = ? and reason = ? ";
+
+        try (var pstmt = conn.prepareStatement(query)) {
+            pstmt.setString(1, jobKey);
+            pstmt.setString(2, reason);
+
+            var rs = pstmt.executeQuery();
+            var events = new ArrayList<AutoScalerEvent>();
+            while (rs.next()) {
+                events.add(generateEvent(rs));
+            }
+            return events;
+        }
+    }
+
     @VisibleForTesting
     void setClock(@Nonnull Clock clock) {
         this.clock = Preconditions.checkNotNull(clock);