1996fanrui commented on code in PR #865:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/865#discussion_r1728486535
##########
flink-autoscaler-plugin-jdbc/src/main/java/org/apache/flink/autoscaler/jdbc/event/JdbcAutoScalerEventHandler.java:
##########
@@ -104,6 +135,49 @@ public void handleScalingEvent(
}
}
+ @Override
+ public void close() {
+ if (Objects.nonNull(scheduledEventHandlerCleaner)
+ && !scheduledEventHandlerCleaner.isShutdown()) {
+ scheduledEventHandlerCleaner.shutdownNow();
+ }
+ }
+
+ @VisibleForTesting
+ void cleanExpiredEvents() {
+ var date =
+ Timestamp.from(
+ jdbcEventInteractor
+ .getCurrentInstant()
+ .minusMillis(eventHandlerTtl.toMillis()));
+ try {
+ Long minId = jdbcEventInteractor.queryMinEventIdByCreateTime(date);
+ if (Objects.isNull(minId)) {
+ log.warn("No expired event handlers queried at {}", new
Date());
+ return;
+ }
+ int deleted = 0, batch = 4098;
+ var count = 0L;
+ boolean cleanable = true;
+ for (long startId = minId; deleted == batch || cleanable; startId
+= batch) {
+ var endId = startId + batch;
+ deleted =
+
jdbcEventInteractor.deleteExpiredEventsByIdRangeAndDate(
+ startId, endId, date);
+ count += deleted;
+ if (deleted != batch) {
+ cleanable =
jdbcEventInteractor.queryMinEventIdByCreateTime(date) != null;
Review Comment:
And please add some tests to cover this case if possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
