[FLINK-36039][autoscaler] Support clean historical event handler records in JDBC event handler #865

RocMarshal · 2024-08-14T02:27:38Z

What is the purpose of the change

Support clean historical event handler records in JDBC event handler

Brief change log

Verifying this change

Tested locally.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changes to the CustomResourceDescriptors: (yes / no)
Core observer or reconciler logic that is regularly executed: (yes / no)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

...alone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutor.java

...src/main/java/org/apache/flink/autoscaler/standalone/config/AutoscalerStandaloneOptions.java

...lone/src/main/java/org/apache/flink/autoscaler/standalone/AutoscalerEventHandlerFactory.java

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

1996fanrui · 2024-08-14T10:07:08Z

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

+                Executors.newSingleThreadScheduledExecutor(
+                        new ThreadFactoryBuilder()
+                                .setNameFormat("jdbc-autoscaler-events-cleaner")
+                                .setDaemon(false)


Why it's false here?

thanks for the comments.
Please let me know what are you referring is to the explicit call to ".setDaemon(false)" due the default value is false here, or are you suggesting that it should be set to true here?

I'm curious what you think about this? In which cases should we use true and in which cases should we use false.

And in the current case, why do you use false? (I don't care if false is set manually or is the default value).

It is necessary for developers to know the meaning of every line of code they write and why.

thx for the clarify.

IIUC, There is no explicit exit operation in the entire service, so the default value of false can be used here. It is not necessary to set it to true to automatically exit after waiting for other threads to finish.

Explicit call for readability

There is no explicit exit operation in the entire service

Jdbc event handler doesn't know who call itself, it doesn't know Autoscaler Standalone, and doesn't know k8s operator.

1996fanrui · 2024-08-14T10:10:03Z

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

+            String message,
+            String eventKey,
+            Timestamp createTime)
+            throws Exception {


Don't need introduce new method for testing, you could setClock to control the timestamp.

Check AbstractJdbcAutoscalerEventHandlerITCase#testEventIntervalWithoutMessageKey

1996fanrui · 2024-08-14T10:26:22Z

...st/java/org/apache/flink/autoscaler/jdbc/event/AbstractJdbcAutoscalerEventHandlerITCase.java

+                        quickQuery("select create_time from t_flink_autoscaler_event_handler")) {
+            assertThat(countResultSet).isNotNull();
+            assertThat(countResultSet.next()).isTrue();
+            assertThat(countResultSet.getInt(1)).isOne();


Could we extract the unexpired record number into one parameter? It could cover more cases.

1996fanrui · 2024-08-14T10:26:54Z

...st/java/org/apache/flink/autoscaler/jdbc/event/AbstractJdbcAutoscalerEventHandlerITCase.java

+            assertThat(createTimeResultSet.next()).isTrue();
+            assertThat(createTimeResultSet.getTimestamp(1)).isEqualTo(createTimeOfUnExpiredRecord);
+        } catch (SQLException e) {
+            throw new RuntimeException(e);


Don't understand why need this catch?

1996fanrui · 2024-08-14T10:27:30Z

...st/java/org/apache/flink/autoscaler/jdbc/event/AbstractJdbcAutoscalerEventHandlerITCase.java

+                AutoScalerEventHandler.Type.Normal,
+                "message",
+                "eventKey",
+                createTimeOfUnExpiredRecord);


unexpired is a word, right?

1996fanrui · 2024-08-14T10:33:41Z

...st/java/org/apache/flink/autoscaler/jdbc/event/AbstractJdbcAutoscalerEventHandlerITCase.java

+        final Clock clock = Clock.systemDefaultZone();
+        initExpiredEventHandlerRecords(eventHandlerTtl, expiredRecordsNum, clock);
+        final Timestamp createTimeOfUnExpiredRecord = Timestamp.from(clock.instant());
+        jdbcEventInteractor.createEvent(


Why don't insert event via event handler? It's a little wired that creating event hanldler after inserting data. In addion, It could cover more cases.

1996fanrui · 2024-08-14T10:39:58Z

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

+                return null;
+            }
+            var result = new ExpiredEventsResult(resultSet.getInt(1), resultSet.getLong(2));
+            resultSet.close();


Don't need it, copy from the comment of java.sql.ResultSet#close. Let us keep all queries are same style.

Note: A ResultSet object is automatically closed by the Statement object that generated it when that Statement object is closed, re-executed, or is used to retrieve the next result from a sequence of multiple results.

is this comment not addressed?

1996fanrui · 2024-08-14T10:42:48Z

CI is failed, please take a look as well, thanks~

RocMarshal

Thanks @1996fanrui for the quick review.
I updated it based on the comments. PTAL~

RocMarshal · 2024-08-16T01:49:10Z

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

+            var startId = expiredResult.minId;
+            var endId = Math.min(startId + batch, expiredResult.maxId);
+            while (startId <= expiredResult.maxId) {
+                jdbcEventInteractor.deleteExpiredEventsByIdRange(startId, endId);


The limit is not supported in postgres delete clause.
So, I try to use the current way

1996fanrui · 2024-08-15T09:54:29Z

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

+                Thread.sleep(restInterval);
+            }
+            log.info(
+                    "Deleted expired {} event handler records successfully whose id is smaller than {}.",


smaller than or equal to ?

1996fanrui · 2024-08-20T01:54:47Z

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

+        long maxId;
+
+        public ExpiredEventsResult(int expiredRecords, long minId, long maxId) {
+            this.expiredRecords = expiredRecords;


After introducing minId, the expiredRecords is not needed anymore, right?

Also, please update the comment of ExpiredEventsResult as well.

1996fanrui · 2024-08-20T01:58:12Z

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

+                return null;
+            }
+            var result = new ExpiredEventsResult(resultSet.getInt(1), resultSet.getLong(2));
+            resultSet.close();


is this comment not addressed?

1996fanrui · 2024-08-20T01:58:58Z

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

+    @Nullable
+    ExpiredEventsResult queryExpiredEventsAndMaxId(Duration eventHandlerTtl) throws Exception {
+        var query =
+                "SELECT COUNT(1) records_num,  min(id) min_existed_id, max(id) max_target_id "


could you unify the name of min_existed_id and max_target_id?

1996fanrui · 2024-08-20T02:03:29Z

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

+            var batch = 4098;
+            var startId = expiredResult.minId;
+            var endId = Math.min(startId + batch, expiredResult.maxId);
+            while (startId <= expiredResult.maxId) {
+                jdbcEventInteractor.deleteExpiredEventsByIdRange(startId, endId);
+                startId += batch;
+                endId = Math.min(endId + batch, expiredResult.maxId);
+                log.debug("Deleted expired event handler records by batch successfully.");
+                Thread.sleep(10L);
+            }


It seems the loop is pretty easier than while loop, and we don't need to introduce a series of variables.

Suggested change

var batch = 4098;

var startId = expiredResult.minId;

var endId = Math.min(startId + batch, expiredResult.maxId);

while (startId <= expiredResult.maxId) {

jdbcEventInteractor.deleteExpiredEventsByIdRange(startId, endId);

startId += batch;

endId = Math.min(endId + batch, expiredResult.maxId);

log.debug("Deleted expired event handler records by batch successfully.");

Thread.sleep(10L);

}

var batch = 4098;

for (var startId = expiredResult.minId; startId <= expiredResult.maxId; startId += batch) {

jdbcEventInteractor.deleteExpiredEventsByIdRange(startId, Math.min(start + batch, expiredResult.maxId));

log.debug("Deleted expired event handler records by batch successfully.");

Thread.sleep(10L);

}

1996fanrui · 2024-08-20T02:06:21Z

...st/java/org/apache/flink/autoscaler/jdbc/event/AbstractJdbcAutoscalerEventHandlerITCase.java

+                Arguments.of(1024, Duration.ofMinutes(2), 1),
+                Arguments.of(1024 * 5, Duration.ofMinutes(2), 2),
+                Arguments.of(1024, Duration.ofMinutes(100), 3),
+                Arguments.of(1024 * 5, Duration.ofMinutes(100), 100));


This test don't cover the case that unexpiredRecordsNum is 0.

1996fanrui · 2024-08-20T02:10:50Z

...st/java/org/apache/flink/autoscaler/jdbc/event/AbstractJdbcAutoscalerEventHandlerITCase.java

+        try (Connection con = getConnection();
+                PreparedStatement ps =
+                        con.prepareStatement("delete from t_flink_autoscaler_event_handler")) {
+            ps.execute();
+        }


Did you find the t_flink_autoscaler_event_handler table has data?

IIUC, Here would be some data resulted by the parameterized test running due to the shared database instance.

Actually, PostgreSQLExtension and MySQLExtension have afterEach, it cleans all data.

But they missed t_flink_autoscaler_event_handler table. I submitted #866 to fix it.

1996fanrui · 2024-08-20T02:26:08Z

...st/java/org/apache/flink/autoscaler/jdbc/event/AbstractJdbcAutoscalerEventHandlerITCase.java

+    @MethodSource("getExpiredEventHandlersCaseMatrix")
+    @ParameterizedTest
+    void testCleanExpiredEvents(
+            int expiredRecordsNum, Duration eventHandlerTtl, int unexpiredRecordsNum)


It's a little wired that JdbcAutoScalerEventHandler#eventHandlerTtl is always 0, but it could clean expired records.

Thanks for comments.
IIUC, Maybe it's accepted, because in the current case, we just want to test the method cleanExpiredEvents of JdbcAutoScalerEventHandler instead of the whole logic of the JdbcAutoScalerEventHandler.
If needed, I'd like to do some change for the stable test cases.

1996fanrui · 2024-08-20T06:52:40Z

...st/java/org/apache/flink/autoscaler/jdbc/event/AbstractJdbcAutoscalerEventHandlerITCase.java

+                Arguments.of(1024, Duration.ofMinutes(100), 3),
+                Arguments.of(1024, Duration.ofMinutes(100), 0),
+                Arguments.of(1024 * 5, Duration.ofMinutes(100), 100),
+                Arguments.of(1024 * 5, Duration.ofMinutes(100), 0));


You could try run unexpiredRecordsNum = 0 locally, all records cannot be cleaned.

In addition, it's needed to add the case that expiredRecordsNum = 0.

1996fanrui · 2024-08-21T07:02:03Z

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

                        + "   SELECT id FROM t_flink_autoscaler_event_handler "
-                        + "   WHERE create_time >= ? ORDER BY id ASC LIMIT 1)";
+                        + "   WHERE create_time < ? ORDER BY id DESC LIMIT 1)";


As I understand, desc will scan most of data.

thx for the comments.
Made some change for it. PTAL~ :)

1996fanrui · 2024-08-23T05:07:30Z

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

+            int deleted = 0, batch = 4098;
+            var count = 0L;
+            boolean cleanable = true;
+            for (long startId = minId; deleted == batch || cleanable; startId += batch) {


IIUC, deleted == batch isn't needed. If they are not equal, you updated the cleanable, so we check cleanable is enough.

If 1 is acceptable, we don't need the for loop. while(true) loop is enough, and we can break the loop if cleanable is false.

If 1 is acceptable, deleted can be defined inside of loop.

If 2 is acceptable, we don't need cleanable. We can check jdbcEventInteractor.queryMinEventIdByCreateTime(date) == null, if it's true, break directly.

Thank you very much for comments.

IIUC, deleted == batch isn't needed. If they are not equal, you updated the cleanable, so we check cleanable is enough.

Yes, it does overall. But here's a small difference in this condition.
The deleted==batch would reduce the sql query from the DB,

jdbcEventInteractor.queryMinEventIdByCreateTime(date) This query function is more like searching for the starting point of an interval composed of consecutive IDs that meet the deletion criteria. the deleted==batch would drive the deletion action in the consecutive IDs.

Please correct me if I'm wrong.

From the code, jdbcEventInteractor.queryMinEventIdByCreateTime(date) != null; is only called when deleted!=batch.

It means jdbcEventInteractor.queryMinEventIdByCreateTime(date) != null; won't be called when deleted==batch. And for loop will delete next round.

I don't understand why updating deleted == batch || cleanable to cleanable will query more times from Database.

1996fanrui · 2024-08-23T05:07:49Z

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

+                return;
+            }
+            int deleted = 0, batch = 4098;
+            var count = 0L;


Suggested change

var count = 0L;

var deletedTotalCount = 0L;

1996fanrui · 2024-08-23T05:08:21Z

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

+                log.warn("No expired event handlers queried at {}", new Date());
+                return;
+            }
+            int deleted = 0, batch = 4098;


How about renaming it to batchSize?

1996fanrui · 2024-08-23T05:09:42Z

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

+                }
+                log.debug(
+                        "Deleted expired {} event handler records by batch successfully.", deleted);
+                Thread.sleep(10L);


How about sleep 1s?

One minute could clean 4096 * 60 = 240 k, it's totally enough.

1996fanrui · 2024-08-23T05:12:45Z

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

+        var sql =
+                "SELECT id from t_flink_autoscaler_event_handler "
+                        + "           where id = (SELECT id FROM t_flink_autoscaler_event_handler order by id asc limit 1) "
+                        + "           and create_time < ?";


SELECT id, create_time FROM t_flink_autoscaler_event_handler order by id asc limit 1 is enough, we can check create_time in memory.

thanks for the comment.

I conducted some tests based on the above logic, and this SQL statement has these two characteristics:

It would run a few times per triggering..

Running on a large amount of data takes very little time. Even equivalent to this sentence SELECT id FROM t_flink_autoscaler_event_handler order by id asc limit 1

How about keeping it as now? of course, I'd like to updated it if needed.

How about keeping it as now? of course, I'd like to updated it if needed.

It's up to you, both of them are fine for me.

This sql only query the first row, so the performance isn't my concern.

I leave this comment because sql looks weird at first glance.

1996fanrui · 2024-08-23T05:14:43Z

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

+                                startId, endId, date);
+                count += deleted;
+                if (deleted != batch) {
+                    cleanable = jdbcEventInteractor.queryMinEventIdByCreateTime(date) != null;


IIUC, if it's not null, we should update the startId = jdbcEventInteractor.queryMinEventIdByCreateTime(date).

And please add some tests to cover this case if possible.

1996fanrui · 2024-08-23T05:18:13Z

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

+                                .getCurrentInstant()
+                                .minusMillis(eventHandlerTtl.toMillis()));
+        try {
+            Long minId = jdbcEventInteractor.queryMinEventIdByCreateTime(date);


We could define a nested loops:

The outer loop check Long minId = jdbcEventInteractor.queryMinEventIdByCreateTime(date), and minId != null

The inner loop check deleted == batchSize.

1996fanrui

LGTM assuming the CI is green.

Hey @huyuanfeng2018 , as I know, your employer also needs this feature. Would you mind reviewing this PR as well? thank you very much.

…rds in JDBC event handler

Updated based on review comments Addressed comments Updated based on review comments Addressed review comments.

huyuanfeng2018

Thanks to @RocMarshal for contribution and @1996fanrui for review, the code looks good to me!

RocMarshal force-pushed the FLINK-36039 branch from 0dddc09 to ecabeee Compare August 14, 2024 03:07

1996fanrui requested changes Aug 14, 2024

View reviewed changes

...er-plugin-jdbc/src/main/java/org/apache/flink/autoscaler/jdbc/event/JdbcEventInteractor.java Outdated Show resolved Hide resolved

...er-plugin-jdbc/src/main/java/org/apache/flink/autoscaler/jdbc/event/JdbcEventInteractor.java Outdated Show resolved Hide resolved

1996fanrui self-assigned this Aug 14, 2024

1996fanrui reviewed Aug 14, 2024

View reviewed changes

RocMarshal force-pushed the FLINK-36039 branch 2 times, most recently from 9d058a4 to c57e30b Compare August 15, 2024 16:10

RocMarshal commented Aug 16, 2024

View reviewed changes

1996fanrui reviewed Aug 20, 2024

View reviewed changes

1996fanrui mentioned this pull request Aug 20, 2024

[FLINK-36106][autoscaler] Clean up the t_flink_autoscaler_event_handler table after each test #866

Merged

1996fanrui requested changes Aug 20, 2024

View reviewed changes

RocMarshal force-pushed the FLINK-36039 branch from 06212b7 to 8f23741 Compare August 21, 2024 02:29

1996fanrui reviewed Aug 21, 2024

View reviewed changes

RocMarshal force-pushed the FLINK-36039 branch 6 times, most recently from c1d80a0 to 3605359 Compare August 23, 2024 02:48

1996fanrui reviewed Aug 23, 2024

View reviewed changes

1996fanrui force-pushed the FLINK-36039 branch from 106c4b3 to e3defd6 Compare August 23, 2024 08:02

1996fanrui approved these changes Aug 23, 2024

View reviewed changes

RocMarshal force-pushed the FLINK-36039 branch 2 times, most recently from d443243 to 63edf33 Compare August 23, 2024 13:28

RocMarshal added 3 commits August 26, 2024 09:44

[FLINK-36039][autoscaler] Support clean historical event handler reco…

1d98faf

…rds in JDBC event handler

Squished old commits

fd60e30

Updated based on review comments Addressed comments Updated based on review comments Addressed review comments.

Addressed review comments.

84e486e

1996fanrui and others added 4 commits August 26, 2024 09:44

Simplify the cleanExpiredEvents

b55e3e2

Polish the cleanExpiredEvents

c46fec2

Polish the test

1fc7856

Add the test case for quick cleaning.

a9b42bd

1996fanrui force-pushed the FLINK-36039 branch from 63edf33 to a9b42bd Compare August 26, 2024 01:44

huyuanfeng2018 approved these changes Aug 26, 2024

View reviewed changes

Polish the testDeleteCounterWhenIdNotConsecutive

5b333e6

1996fanrui merged commit 6a426b2 into apache:main Aug 26, 2024
223 checks passed

[FLINK-36039][autoscaler] Support clean historical event handler records in JDBC event handler #865

[FLINK-36039][autoscaler] Support clean historical event handler records in JDBC event handler #865

Conversation

RocMarshal commented Aug 14, 2024

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

RocMarshal Aug 15, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

1996fanrui commented Aug 14, 2024

RocMarshal left a comment

Choose a reason for hiding this comment

RocMarshal Aug 16, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

RocMarshal Aug 20, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

1996fanrui Aug 23, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

1996fanrui left a comment

Choose a reason for hiding this comment

huyuanfeng2018 left a comment

Choose a reason for hiding this comment

RocMarshal Aug 15, 2024 •

edited

Loading

RocMarshal Aug 16, 2024 •

edited

Loading

RocMarshal Aug 20, 2024 •

edited

Loading

1996fanrui Aug 23, 2024 •

edited

Loading