[FIP-28] Version-gate historical partition lookup: bump ApiKeys.LOOKUP and add the client-side gate

[ss666]

Search before asking

• I searched in the issues and found nothing similar.

Description

Parent issue: #3631. Follow-up to #3632.

LOOKUP carries original_partition_name but, unlike PUT_KV and PRODUCE_LOG, has neither a version bump nor a gate: ApiKeys.LOOKUP is still (1017, 0, 1, PUBLIC), validateVersionCompatibility has no LOOKUP branch, so the field is sent whatever version was negotiated.

The bump is the prerequisite for making this capability detectable, as discussed in #4001. The discussion in #3630 on LOOKUP expected an old server to return null; what happens instead is an NPE: dispatch is keyed on the echoed name, so an un-echoed response matches no in-flight batch and is dereferenced unchecked, failing the whole request as an UnknownServerException. Reproduced on main with a gateway stub that answers normally but does not echo the field:

org.apache.fluss.exception.UnknownServerException: java.lang.NullPointerException
        at org.apache.fluss.client.lookup.LookupSender.handleLookupResponse(LookupSender.java:438)
        at org.apache.fluss.client.lookup.LookupSender.lambda$sendLookupRequestAndHandleResponse$10(LookupSender.java:349)
        ...

Proposed change

1. Raise ApiKeys.LOOKUP to 2. The maximum is still 1 from [lake] Fix lake encoder causing prefix lookup to miss data when bucket key is a subset of primary key #2464, so 1 cannot be given that meaning retroactively; new/old then negotiates 1 and the gate fires. The only server-side LOOKUP version check, ReplicaManager#validateClientVersionForPkTable, is >= 1, so it is unaffected. The server must also keep accepting original_partition_name at v1, since clients from [lookup] Support historical partition lookup for lake tables #3630 onward already send it there.
2. Add HISTORICAL_LOOKUP_MIN_VERSION = 2 and a LOOKUP branch in validateVersionCompatibility using hasHistoricalLookup, throwing UnsupportedVersionException.

Since release-1.0 advertises the same LOOKUP maximum as 0.9, a bump that lands only on main would make the gate reject 1.0 servers that do support historical lookup, so it may be worth considering for release-1.0.

Willingness to contribute

• I'm willing to submit a PR!

[luoyuxia]

@ss666 One compatibility concern: some of our clusters are already running a version that includes #4001. Those TabletServers support historical partition lookup, but still advertise LOOKUP max version 1.

I would like the following combination to remain supported:

fixed client → pre-fix server that supports historical lookup but still advertises v1

If the fixed client rejects every historical lookup whose negotiated version is lower than 2, it will reject these already deployed and otherwise capable servers.

One possible solution is to keep a transitional compatibility path:

• v2 means historical lookup is explicitly supported.
• For v1, treat the capability as unknown and allow the historical lookup request to be sent.
• Require every successful historical lookup response to echo original_partition_name and match it against the in-flight batch.
• If a successful response does not echo the field, fail with a clear UnsupportedVersionException instead of returning an empty result or triggering an NPE.
• Error responses without the echoed field may need separate handling, since existing capable servers can omit it for some routing-validation errors.

Backporting the v2 advertisement to release-1.0 would help patched servers, but it would not cover clusters that have already been deployed.

I hope we can get this PR merged soon. If you do not have enough time to continue working on it, I am happy to take it over.

[ss666]

Thanks so much for your comments! Reworked along the transitional path:

• ApiKeys.LOOKUP stays at 2.
• The LOOKUP branch of ServerConnection#validateVersionCompatibility is gone.
• Detection moved into the response dispatch in LookupSender. A bucket that echoesoriginal_partition_name is matched on (table bucket, original partition name) exactly as before; a historical bucket that does not echo it takes a separate path. That path splits on the error code:
  • Without an error code the server answered as if this had been a normal lookup, so the capability is missing, and it fails with anUnsupportedVersionException.
  • With an error code the server's own error is propagated unchanged.