From 246fbeb21b651a86a5a302660d3fb491077a8047 Mon Sep 17 00:00:00 2001 From: Shuai Liu <390105636@qq.com> Date: Fri, 18 Sep 2026 14:26:43 +0800 Subject: [PATCH 1/2] [rpc] version-gate historical partition lookup --- .../rpc/netty/client/ServerConnection.java | 17 ++++++ .../apache/fluss/rpc/protocol/ApiKeys.java | 3 +- .../netty/client/ServerConnectionTest.java | 61 +++++++++++++++++++ 3 files changed, 80 insertions(+), 1 deletion(-) diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java index 8eb9d805c2..5ed79ef827 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java @@ -32,6 +32,7 @@ import org.apache.fluss.rpc.messages.ApiVersionsResponse; import org.apache.fluss.rpc.messages.AuthenticateRequest; import org.apache.fluss.rpc.messages.AuthenticateResponse; +import org.apache.fluss.rpc.messages.LookupRequest; import org.apache.fluss.rpc.messages.ProduceLogRequest; import org.apache.fluss.rpc.messages.PutKvRequest; import org.apache.fluss.rpc.metrics.ClientMetricGroup; @@ -66,6 +67,7 @@ import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; +import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.hasHistoricalLookup; import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.hasHistoricalProduce; import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.hasHistoricalPut; import static org.apache.fluss.utils.IOUtils.closeQuietly; @@ -76,6 +78,7 @@ final class ServerConnection { private static final Logger LOG = LoggerFactory.getLogger(ServerConnection.class); private static final short HISTORICAL_PRODUCE_LOG_MIN_VERSION = 1; private static final short HISTORICAL_PUT_KV_MIN_VERSION = 3; + private static final short HISTORICAL_LOOKUP_MIN_VERSION = 2; private static final short ALTER_BUCKET_COUNT_MIN_VERSION = 1; private final ServerNode node; @@ -413,6 +416,20 @@ void validateVersionCompatibility(ApiKeys apiKey, short version, ApiMessage rawR } } + if (apiKey == ApiKeys.LOOKUP && version < HISTORICAL_LOOKUP_MIN_VERSION) { + LookupRequest lookupRequest = (LookupRequest) rawRequest; + if (hasHistoricalLookup(lookupRequest)) { + throw new UnsupportedVersionException( + "Historical partition lookups require LOOKUP version " + + HISTORICAL_LOOKUP_MIN_VERSION + + " or newer, but server " + + node + + " negotiated version " + + version + + '.'); + } + } + if (apiKey == ApiKeys.ALTER_TABLE && version < ALTER_BUCKET_COUNT_MIN_VERSION) { AlterTableRequest alterTableRequest = (AlterTableRequest) rawRequest; if (alterTableRequest.hasModifyBucketCount()) { diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java index fd08a540a4..797433b686 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java @@ -58,7 +58,8 @@ public enum ApiKeys { // Version 0: Uses lake's encoder for primary key encoding (legacy behavior). // Version 1: Uses CompactedKeyEncoder for primary key encoding when bucket key differs from // primary key, enabling prefix lookup support. - LOOKUP(1017, 0, 1, PUBLIC), + // Version 2: Supports original_partition_name in requests and responses for historical lookups. + LOOKUP(1017, 0, 2, PUBLIC), NOTIFY_LEADER_AND_ISR(1018, 0, 0, PRIVATE), STOP_REPLICA(1019, 0, 0, PRIVATE), diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/ServerConnectionTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/ServerConnectionTest.java index c5e10f3970..1ba880f28d 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/ServerConnectionTest.java +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/ServerConnectionTest.java @@ -40,6 +40,7 @@ import org.apache.fluss.rpc.messages.GetTableSchemaRequest; import org.apache.fluss.rpc.messages.ListDatabasesRequest; import org.apache.fluss.rpc.messages.LookupRequest; +import org.apache.fluss.rpc.messages.LookupResponse; import org.apache.fluss.rpc.messages.PbApiVersion; import org.apache.fluss.rpc.messages.PbLookupReqForBucket; import org.apache.fluss.rpc.messages.PbTablePath; @@ -368,6 +369,45 @@ void testRejectHistoricalWritesForOldServer() throws Exception { } } + @Test + void testRejectHistoricalLookupsForOldServer() throws Exception { + nettyServer.close(); + buildNettyServer(new OldLookupGatewayService()); + + ServerConnection connection = + new ServerConnection( + bootstrap, + serverNode, + TestingClientMetricGroup.newInstance(), + clientAuthenticator, + (con, ignore) -> {}); + try { + assertThat(connection.send(ApiKeys.LOOKUP, lookupRequest(null)).get()) + .isInstanceOf(LookupResponse.class); + + assertThatThrownBy( + () -> + connection + .send(ApiKeys.LOOKUP, lookupRequest("dt=20260823")) + .get()) + .rootCause() + .isInstanceOf(UnsupportedVersionException.class) + .hasMessageContaining("require LOOKUP version 2 or newer") + .hasMessageContaining("negotiated version 1"); + } finally { + connection.close().get(); + } + } + + private static LookupRequest lookupRequest(String originalPartitionName) { + LookupRequest request = new LookupRequest().setTableId(1L); + request.addBucketsReq().setBucketId(0).addKey(new byte[] {1}); + if (originalPartitionName != null) { + request.getBucketsReqAt(0).setOriginalPartitionName(originalPartitionName); + } + return request; + } + private static PutKvRequest putKvRequest(String originalPartitionName) { PutKvRequest request = new PutKvRequest().setTableId(1L).setAcks(1).setTimeoutMs(10_000); request.addBucketsReq().setBucketId(0).setRecords(new byte[0]); @@ -443,6 +483,27 @@ public CompletableFuture produceLog(ProduceLogRequest reques } } + private static class OldLookupGatewayService extends TestingTabletGatewayService { + @Override + public CompletableFuture apiVersions(ApiVersionsRequest request) { + return super.apiVersions(request) + .thenApply( + response -> { + for (PbApiVersion apiVersion : response.getApiVersionsList()) { + if (apiVersion.getApiKey() == ApiKeys.LOOKUP.id) { + apiVersion.setMaxVersion(1); + } + } + return response; + }); + } + + @Override + public CompletableFuture lookup(LookupRequest request) { + return CompletableFuture.completedFuture(new LookupResponse()); + } + } + private static class MockMetricRegistry extends NOPMetricRegistry { Map registeredMetrics = new HashMap<>(); From 64b9bcf803dbeac2b2a9af512b2881b2a126e18a Mon Sep 17 00:00:00 2001 From: Shuai Liu <390105636@qq.com> Date: Fri, 18 Sep 2026 19:00:34 +0800 Subject: [PATCH 2/2] add a transitional compatibility path for historical lookup --- .../fluss/client/lookup/LookupSender.java | 60 ++++++++-- .../fluss/client/lookup/LookupSenderTest.java | 109 ++++++++++++++++++ .../rpc/netty/client/ServerConnection.java | 17 --- .../netty/client/ServerConnectionTest.java | 24 +--- .../fluss/rpc/protocol/ApiKeysTest.java | 5 + 5 files changed, 173 insertions(+), 42 deletions(-) diff --git a/fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupSender.java b/fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupSender.java index 7823388c4d..14ec361c92 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupSender.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupSender.java @@ -28,6 +28,7 @@ import org.apache.fluss.exception.LeaderNotAvailableException; import org.apache.fluss.exception.PartitionNotExistException; import org.apache.fluss.exception.RetriableException; +import org.apache.fluss.exception.UnsupportedVersionException; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TablePartition; @@ -61,6 +62,7 @@ import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeLookupRequest; import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makePrefixLookupRequest; +import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.hasHistoricalLookup; /** * This background thread pool lookup operations from {@link #lookupQueue}, and send lookup requests @@ -342,12 +344,19 @@ private void sendLookupRequestAndHandleResponse( Thread.currentThread().interrupt(); throw new FlussRuntimeException("interrupted:", e); } + // Determined from the request rather than from the batches, so that the client and the + // server decide the lookup kind by the exact same predicate. + boolean historicalRequest = hasHistoricalLookup(lookupRequest); gateway.lookup(lookupRequest) .thenAccept( lookupResponse -> { try { handleLookupResponse( - tableId, destination, lookupResponse, lookupsByBatchKey); + tableId, + destination, + historicalRequest, + lookupResponse, + lookupsByBatchKey); } finally { maxInFlightReuqestsSemaphore.release(); } @@ -403,6 +412,7 @@ private void sendPrefixLookupRequestAndHandleResponse( private void handleLookupResponse( long tableId, int destination, + boolean historicalRequest, LookupResponse lookupResponse, Map lookupsByBatchKey) { for (PbLookupRespForBucket pbLookupRespForBucket : lookupResponse.getBucketsRespsList()) { @@ -413,12 +423,18 @@ private void handleLookupResponse( ? pbLookupRespForBucket.getPartitionId() : null, pbLookupRespForBucket.getBucketId()); - LookupBatchKey lookupBatchKey = - new LookupBatchKey( - tableBucket, - pbLookupRespForBucket.hasOriginalPartitionName() - ? pbLookupRespForBucket.getOriginalPartitionName() - : null); + String originalPartitionName = + pbLookupRespForBucket.hasOriginalPartitionName() + ? pbLookupRespForBucket.getOriginalPartitionName() + : null; + + if (historicalRequest && originalPartitionName == null) { + handleUnEchoedHistoricalResponse( + destination, tableBucket, pbLookupRespForBucket, lookupsByBatchKey); + continue; + } + + LookupBatchKey lookupBatchKey = new LookupBatchKey(tableBucket, originalPartitionName); LookupBatch lookupBatch = lookupsByBatchKey.get(lookupBatchKey); if (pbLookupRespForBucket.hasErrorCode()) { ApiError error = ApiError.fromErrorMessage(pbLookupRespForBucket); @@ -440,6 +456,36 @@ private void handleLookupResponse( } } + private void handleUnEchoedHistoricalResponse( + int destination, + TableBucket tableBucket, + PbLookupRespForBucket pbLookupRespForBucket, + Map lookupsByBatchKey) { + ApiError error; + if (pbLookupRespForBucket.hasErrorCode()) { + error = ApiError.fromErrorMessage(pbLookupRespForBucket); + } else { + error = + ApiError.fromThrowable( + new UnsupportedVersionException( + "Server " + + destination + + " answered a historical partition lookup on " + + tableBucket + + " without echoing the original partition name, so it" + + " does not support historical partition lookup." + + " Please upgrade the tablet server to a newer" + + " version.")); + } + List lookups = new ArrayList<>(); + for (LookupBatch lookupBatch : lookupsByBatchKey.values()) { + if (lookupBatch.tableBucket().equals(tableBucket)) { + lookups.addAll(lookupBatch.lookups()); + } + } + handleLookupError(tableBucket, destination, error, lookups, "lookup"); + } + private void handlePrefixLookupResponse( long tableId, int destination, diff --git a/fluss-client/src/test/java/org/apache/fluss/client/lookup/LookupSenderTest.java b/fluss-client/src/test/java/org/apache/fluss/client/lookup/LookupSenderTest.java index d6564769f8..cabc62f019 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/lookup/LookupSenderTest.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/lookup/LookupSenderTest.java @@ -27,6 +27,7 @@ import org.apache.fluss.exception.NotLeaderOrFollowerException; import org.apache.fluss.exception.TableNotExistException; import org.apache.fluss.exception.TimeoutException; +import org.apache.fluss.exception.UnsupportedVersionException; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableInfo; @@ -52,8 +53,10 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; @@ -172,6 +175,60 @@ void testHistoricalLookupsBatchDifferentPartitionsForSameBucket() throws Excepti .containsExactly("dt=20200101", "dt=20200102", "dt=20200103"); } + /** + * A server that does not support historical lookup answers the request as a normal lookup, so + * it succeeds without echoing the original partition name and its values belong to another + * keyspace. The negotiated version cannot report this, so the un-echoed response is what the + * capability is detected from, and it must not surface as an empty result or an NPE. + */ + @Test + void testHistoricalLookupFailsWhenPartitionNameNotEchoed() { + gateway.setLookupHandler(this::createNonEchoingResponse); + + LookupQuery query = + new LookupQuery( + DATA1_TABLE_PATH_PK, TABLE_BUCKET, bytes("key1"), false, "dt=20200101"); + + lookupSender.sendLookups(1, LookupType.LOOKUP, Collections.singletonList(query)); + + assertThatThrownBy(() -> query.future().get(5, TimeUnit.SECONDS)) + .isInstanceOf(ExecutionException.class) + .hasRootCauseInstanceOf(UnsupportedVersionException.class) + .hasMessageContaining("without echoing the original partition name") + .hasMessageContaining("does not support historical partition lookup"); + } + + /** + * Routing validation runs before the historical path on the server and reports once per bucket + * without the original partition name, so a server that fully supports historical lookup also + * produces un-echoed buckets. Those carry an error code, and the server's own error must reach + * every batch routed to the bucket rather than being reported as a missing capability. + */ + @Test + void testHistoricalLookupErrorWithoutEchoedNameFailsAllBatchesOnBucket() { + gateway.setLookupHandler( + request -> + createNonEchoingFailedResponse( + request, + new InvalidBucketRoutingException("invalid bucket routing"))); + + LookupQuery query1 = + new LookupQuery( + DATA1_TABLE_PATH_PK, TABLE_BUCKET, bytes("key1"), false, "dt=20200101"); + LookupQuery query2 = + new LookupQuery( + DATA1_TABLE_PATH_PK, TABLE_BUCKET, bytes("key2"), false, "dt=20200102"); + + lookupSender.sendLookups(1, LookupType.LOOKUP, Arrays.asList(query1, query2)); + + for (LookupQuery query : Arrays.asList(query1, query2)) { + assertThatThrownBy(() -> query.future().get(5, TimeUnit.SECONDS)) + .isInstanceOf(ExecutionException.class) + .hasRootCauseInstanceOf(InvalidBucketRoutingException.class); + } + assertThat(metadataUpdater.getBucketLocation(TABLE_BUCKET)).isEmpty(); + } + @Test void testNormalAndHistoricalLookupsSplitRequests() throws Exception { List receivedRequests = Collections.synchronizedList(new ArrayList<>()); @@ -715,6 +772,58 @@ private CompletableFuture createPartitionNameEchoResponse( return CompletableFuture.completedFuture(response); } + /** + * A response from a server that does not implement historical lookup: the unknown request field + * is skipped, the request is answered as a normal lookup, and the original partition name is + * not echoed. + */ + private CompletableFuture createNonEchoingResponse(LookupRequest request) { + LookupResponse response = new LookupResponse(); + for (PbLookupReqForBucket bucketRequest : request.getBucketsReqsList()) { + PbLookupRespForBucket bucketResponse = response.addBucketsResp(); + bucketResponse.setBucketId(bucketRequest.getBucketId()); + if (bucketRequest.hasPartitionId()) { + bucketResponse.setPartitionId(bucketRequest.getPartitionId()); + } + for (int i = 0; i < bucketRequest.getKeysCount(); i++) { + bucketResponse + .addValue() + .setValues( + responseValue( + "", + new String( + bucketRequest.getKeyAt(i), + StandardCharsets.UTF_8))); + } + } + return CompletableFuture.completedFuture(response); + } + + /** + * An error response without the echoed original partition name, as the server's routing + * validation produces it. Routing errors are collected once per bucket, so a bucket appears + * once even when several original partitions were batched onto it. + */ + private CompletableFuture createNonEchoingFailedResponse( + LookupRequest request, Exception exception) { + LookupResponse response = new LookupResponse(); + ApiError error = ApiError.fromThrowable(exception); + Set respondedBuckets = new HashSet<>(); + for (PbLookupReqForBucket bucketRequest : request.getBucketsReqsList()) { + if (!respondedBuckets.add(bucketRequest.getBucketId())) { + continue; + } + PbLookupRespForBucket bucketResponse = response.addBucketsResp(); + bucketResponse.setBucketId(bucketRequest.getBucketId()); + if (bucketRequest.hasPartitionId()) { + bucketResponse.setPartitionId(bucketRequest.getPartitionId()); + } + bucketResponse.setErrorCode(error.error().code()); + bucketResponse.setErrorMessage(error.formatErrMsg()); + } + return CompletableFuture.completedFuture(response); + } + private static byte[] bytes(String value) { return value.getBytes(StandardCharsets.UTF_8); } diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java index 5ed79ef827..8eb9d805c2 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java @@ -32,7 +32,6 @@ import org.apache.fluss.rpc.messages.ApiVersionsResponse; import org.apache.fluss.rpc.messages.AuthenticateRequest; import org.apache.fluss.rpc.messages.AuthenticateResponse; -import org.apache.fluss.rpc.messages.LookupRequest; import org.apache.fluss.rpc.messages.ProduceLogRequest; import org.apache.fluss.rpc.messages.PutKvRequest; import org.apache.fluss.rpc.metrics.ClientMetricGroup; @@ -67,7 +66,6 @@ import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; -import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.hasHistoricalLookup; import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.hasHistoricalProduce; import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.hasHistoricalPut; import static org.apache.fluss.utils.IOUtils.closeQuietly; @@ -78,7 +76,6 @@ final class ServerConnection { private static final Logger LOG = LoggerFactory.getLogger(ServerConnection.class); private static final short HISTORICAL_PRODUCE_LOG_MIN_VERSION = 1; private static final short HISTORICAL_PUT_KV_MIN_VERSION = 3; - private static final short HISTORICAL_LOOKUP_MIN_VERSION = 2; private static final short ALTER_BUCKET_COUNT_MIN_VERSION = 1; private final ServerNode node; @@ -416,20 +413,6 @@ void validateVersionCompatibility(ApiKeys apiKey, short version, ApiMessage rawR } } - if (apiKey == ApiKeys.LOOKUP && version < HISTORICAL_LOOKUP_MIN_VERSION) { - LookupRequest lookupRequest = (LookupRequest) rawRequest; - if (hasHistoricalLookup(lookupRequest)) { - throw new UnsupportedVersionException( - "Historical partition lookups require LOOKUP version " - + HISTORICAL_LOOKUP_MIN_VERSION - + " or newer, but server " - + node - + " negotiated version " - + version - + '.'); - } - } - if (apiKey == ApiKeys.ALTER_TABLE && version < ALTER_BUCKET_COUNT_MIN_VERSION) { AlterTableRequest alterTableRequest = (AlterTableRequest) rawRequest; if (alterTableRequest.hasModifyBucketCount()) { diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/ServerConnectionTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/ServerConnectionTest.java index 1ba880f28d..e4e32c8862 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/ServerConnectionTest.java +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/ServerConnectionTest.java @@ -370,9 +370,9 @@ void testRejectHistoricalWritesForOldServer() throws Exception { } @Test - void testRejectHistoricalLookupsForOldServer() throws Exception { + void testHistoricalLookupNotRejectedForServerAdvertisingLookupV1() throws Exception { nettyServer.close(); - buildNettyServer(new OldLookupGatewayService()); + buildNettyServer(new LookupV1GatewayService()); ServerConnection connection = new ServerConnection( @@ -384,16 +384,8 @@ void testRejectHistoricalLookupsForOldServer() throws Exception { try { assertThat(connection.send(ApiKeys.LOOKUP, lookupRequest(null)).get()) .isInstanceOf(LookupResponse.class); - - assertThatThrownBy( - () -> - connection - .send(ApiKeys.LOOKUP, lookupRequest("dt=20260823")) - .get()) - .rootCause() - .isInstanceOf(UnsupportedVersionException.class) - .hasMessageContaining("require LOOKUP version 2 or newer") - .hasMessageContaining("negotiated version 1"); + assertThat(connection.send(ApiKeys.LOOKUP, lookupRequest("dt=20260823")).get()) + .isInstanceOf(LookupResponse.class); } finally { connection.close().get(); } @@ -483,7 +475,8 @@ public CompletableFuture produceLog(ProduceLogRequest reques } } - private static class OldLookupGatewayService extends TestingTabletGatewayService { + /** A server that supports historical lookup but still advertises {@code LOOKUP} version 1. */ + private static class LookupV1GatewayService extends TestingTabletGatewayService { @Override public CompletableFuture apiVersions(ApiVersionsRequest request) { return super.apiVersions(request) @@ -497,11 +490,6 @@ public CompletableFuture apiVersions(ApiVersionsRequest req return response; }); } - - @Override - public CompletableFuture lookup(LookupRequest request) { - return CompletableFuture.completedFuture(new LookupResponse()); - } } private static class MockMetricRegistry extends NOPMetricRegistry { diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/ApiKeysTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/ApiKeysTest.java index 1ae5e55ad2..32c07a6065 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/ApiKeysTest.java +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/ApiKeysTest.java @@ -48,4 +48,9 @@ void testSnapshotMetadataSupportsLayoutAwareClient() { void testAlterTableSupportsBucketCountChange() { assertThat(ApiKeys.ALTER_TABLE.highestSupportedVersion).isEqualTo((short) 1); } + + @Test + void testLookupAdvertisesHistoricalPartitionSupport() { + assertThat(ApiKeys.LOOKUP.highestSupportedVersion).isEqualTo((short) 2); + } }