[core] Read expire_tags older_than as a wall clock, not as an instant

[PDGGK]

Purpose

expire_tags's older_than argument is shifted by the JVM's UTC offset, so on any non-UTC deployment the cutoff is not the time the user typed.

LocalDateTime olderThanTime =
        DateTimeUtils.parseTimestampData(olderThanStr, 3, TimeZone.getDefault())
                .toLocalDateTime();
tagTimeExpire.withOlderThanTime(olderThanTime);

The two halves disagree about what the value is:

• parseTimestampData(s, p, tz) is Timestamp.fromInstant(local.atZone(tz).toInstant()) (DateTimeUtils:547-552), so the resulting Timestamp holds a true epoch milli;
• Timestamp.toLocalDateTime() (Timestamp:92-103) divides that milli by MILLIS_PER_DAY with no zone at all — it renders the instant as a UTC wall clock. It is the exact inverse of fromLocalDateTime, not of fromInstant.

What it is compared against is a local wall clock: tags are created with LocalDateTime.now() (TagManager:168), the fallback create time is DateTimeUtils.toLocalDateTime(modificationTime) = atZone(systemDefault()) (TagTimeExpire:81), and the sibling retention check uses LocalDateTime.now() (:88). The comparison is olderThanTime.isAfter(createTime) at :89.

Measured, for the documented example older_than => '2024-09-06 11:00:00' (docs/flink/procedures.md):

JVM timezone	cutoff actually used
UTC	2024-09-06 11:00
America/New_York	2024-09-06 15:00
Asia/Shanghai	2024-09-06 03:00

It is wrong in both directions: west of UTC it deletes tags the user asked to keep — and once a tag's snapshot has expired, TagManager.deleteTag falls through to cleaning the data files, so that is not recoverable. East of UTC it keeps tags the user asked to expire, which is quieter but still not what was requested.

What changes

Three call sites — paimon-flink-common, paimon-flink-1.18 and paimon-spark-common — parse the argument as the wall clock it is:

LocalDateTime olderThanTime = DateTimeUtils.toLocalDateTime(olderThanStr, 3);

DateTimeUtils.toLocalDateTime(String, int) (:563) is fromTemporalAccessor(...) with no zone conversion, so the parsed wall clock is compared against wall clocks.

Deliberately not touched: the other parseTimestampData(..., TimeZone.getDefault()) call sites are correct, because they keep the instant rather than re-rendering it — ProcedureUtils:91 takes .getMillisecond() and subtracts it from System.currentTimeMillis(), and OrphanFilesClean:477 compares the Timestamp against Timestamp.fromEpochMillis(System.currentTimeMillis()). Both sides are epoch millis there. The defect is specifically parseTimestampData(..., tz) followed by .toLocalDateTime().

Why the existing tests did not catch it

ExpireTagsProcedureITCase and ExpireTagsActionTest do not pass the procedure a wall clock. They take a tag's create time and put it through a round-trip that happens to cancel the bug out:

LocalDateTime olderThanTime = table.tagManager().getOrThrow("tag-2").getTagCreateTime();
java.sql.Timestamp timestamp =
        new java.sql.Timestamp(Timestamp.fromLocalDateTime(olderThanTime).getMillisecond());

fromLocalDateTime(...).getMillisecond() encodes a local wall clock as if it were UTC, and java.sql.Timestamp.toString() then renders that instant back in the local zone — so the string handed to the procedure is already pre-shifted by exactly the offset the procedure re-applies. Measured, the composition returns the original 11:00 in UTC, America/New_York and Asia/Shanghai alike.

A user typing the documented form does not do that dance, so the tests were green while the feature was wrong. Both are updated here to pass the plain wall clock, which is what the procedure's argument is.

Test evidence

	UTC	Asia/Shanghai
this branch	5 tests, 0 failures	5 tests, 0 failures
master behaviour restored, tests as updated	—	1 failure, the wrong tags expired

The failure names the symptom directly — expire_tags returned tags that the assertion lists under "elements not expected".

The UTC column is why this is worth flagging rather than assuming CI would have caught it: under -Duser.timezone=UTC the change is a no-op and every test passes either way.

API and Format

No change to any option, on-disk format or public signature. The interpretation of older_than changes on non-UTC JVMs, which is the point; a caller that had empirically compensated for the shift would need to stop doing so.