You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2026/10/21 02:27:07 UTC
(pulsar) branch master updated: [improve][broker] Cache schema payloads to stop re-reading BookKeeper history on every explicit-version request (#26960)
This is an automated email from the ASF dual-hosted git repository.
codelipenghui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 9873806c54bf [improve][broker] Cache schema payloads to stop re-reading BookKeeper history on every explicit-version request (#26960)
9873806c54bf is described below
commit 9873806c54bf86bdd87de1cc14ece8a30895363b
Author: void-ptr974 <vo...@protonmail.com>
AuthorDate: Wed Oct 21 10:27:00 2026 +0800
[improve][broker] Cache schema payloads to stop re-reading BookKeeper history on every explicit-version request (#26960)
---
.../service/schema/BookkeeperSchemaStorage.java | 96 +++++-
.../service/schema/SchemaRegistryServiceImpl.java | 20 +-
.../BookkeeperSchemaStorageEntryCacheTest.java | 342 +++++++++++++++++++++
.../broker/service/schema/ClientGetSchemaTest.java | 6 +
.../schema/SchemaRegistryServiceImplTest.java | 81 +++++
5 files changed, 533 insertions(+), 12 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
index 3712dc8f99f2..bbbc7069c538 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
@@ -23,9 +23,13 @@ import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage.Functions.newSchemaEntry;
import static org.apache.pulsar.metadata.api.MetadataStoreException.AlreadyExistsException;
import static org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException;
+import com.github.benmanes.caffeine.cache.AsyncCache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Ticker;
import com.google.common.annotations.VisibleForTesting;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
@@ -77,8 +81,27 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
private final ConcurrentMap<String, CompletableFuture<StoredSchema>> readSchemaOperations =
new ConcurrentHashMap<>();
+ private static final int SCHEMA_ENTRY_CACHE_OVERHEAD_BYTES = 128;
+ private static final long SCHEMA_ENTRY_CACHE_MAX_WEIGHT_BYTES = 32L * 1024 * 1024;
+ private static final Duration SCHEMA_ENTRY_CACHE_EXPIRATION = Duration.ofHours(1);
+
+ // Cache only the versioned schema-read path. Latest and the ordinary getAll remain uncached
+ // because producer registration and schema-recovery paths also use them and must keep their
+ // original BookKeeper validation behavior. Each version is immutable and its (ledgerId, entryId)
+ // position permanently identifies the payload. Only payload bytes are cached; the embedded
+ // history index is still read directly when an older version must be located.
+ private final AsyncCache<SchemaEntryKey, byte[]> schemaEntryCache;
+
+ private record SchemaEntryKey(long ledgerId, long entryId) {
+ }
+
@VisibleForTesting
BookkeeperSchemaStorage(PulsarService pulsar) {
+ this(pulsar, Ticker.systemTicker());
+ }
+
+ @VisibleForTesting
+ BookkeeperSchemaStorage(PulsarService pulsar, Ticker ticker) {
this.pulsar = pulsar;
this.store = pulsar.getLocalMetadataStore();
this.config = pulsar.getConfiguration();
@@ -96,6 +119,18 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
return loc;
}
});
+ this.schemaEntryCache = buildSchemaEntryCache(ticker);
+ }
+
+ private AsyncCache<SchemaEntryKey, byte[]> buildSchemaEntryCache(Ticker ticker) {
+ Caffeine<SchemaEntryKey, byte[]> builder = Caffeine.newBuilder()
+ .maximumWeight(SCHEMA_ENTRY_CACHE_MAX_WEIGHT_BYTES)
+ .weigher((SchemaEntryKey key, byte[] schemaData) ->
+ (int) Math.min(Integer.MAX_VALUE,
+ (long) schemaData.length + SCHEMA_ENTRY_CACHE_OVERHEAD_BYTES))
+ .expireAfterWrite(SCHEMA_ENTRY_CACHE_EXPIRATION)
+ .ticker(ticker);
+ return builder.buildAsync();
}
@Override
@@ -167,6 +202,10 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
return getAllWithLocator(key).thenApply(Pair::getRight);
}
+ CompletableFuture<List<CompletableFuture<StoredSchema>>> getAllForVersionedRead(String key) {
+ return getAllWithLocatorForVersionedRead(key).thenApply(Pair::getRight);
+ }
+
private CompletableFuture<Pair<Optional<LocatorEntry>, List<CompletableFuture<StoredSchema>>>> getAllWithLocator(
String key) {
return getLocator(key).thenApply(locator -> {
@@ -191,6 +230,30 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
});
}
+ private CompletableFuture<Pair<Optional<LocatorEntry>, List<CompletableFuture<StoredSchema>>>>
+ getAllWithLocatorForVersionedRead(String key) {
+ return getLocator(key).thenApply(locator -> {
+ log.debug().attr("key", key).attr("locator", locator).log("Get all schemas for versioned read");
+
+ if (locator.isEmpty()) {
+ return Pair.of(locator, Collections.emptyList());
+ }
+
+ SchemaLocator schemaLocator = locator.get().locator;
+ List<CompletableFuture<StoredSchema>> list = new ArrayList<>();
+ for (int i = 0; i < schemaLocator.getIndexsCount(); i++) {
+ IndexEntry indexEntry = schemaLocator.getIndexAt(i);
+ list.add(readSchemaData(indexEntry.getPosition())
+ .thenApply(schemaData -> new StoredSchema(
+ schemaData,
+ new LongSchemaVersion(indexEntry.getVersion())
+ ))
+ );
+ }
+ return Pair.of(locator, list);
+ });
+ }
+
CompletableFuture<Optional<LocatorEntry>> getLocator(String key) {
return getSchemaLocator(getSchemaPath(key));
}
@@ -307,13 +370,13 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
for (int i = 0; i < schemaLocator.getIndexsCount(); i++) {
indexList.add(schemaLocator.getIndexAt(i));
}
- return findSchemaEntryByVersion(indexList, version)
- .thenApply(entry ->
- new StoredSchema(
- entry.getSchemaData(),
- new LongSchemaVersion(version)
- )
- );
+ return findSchemaPositionByVersion(indexList, version).thenCompose(position -> {
+ if (position == null) {
+ return completedFuture(null);
+ }
+ return readSchemaData(position).thenApply(schemaData ->
+ new StoredSchema(schemaData, new LongSchemaVersion(version)));
+ });
});
}
@@ -544,7 +607,7 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
}
@NonNull
- private CompletableFuture<SchemaEntry> findSchemaEntryByVersion(
+ private CompletableFuture<PositionInfo> findSchemaPositionByVersion(
List<IndexEntry> index,
long version
) {
@@ -555,19 +618,21 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
IndexEntry lowest = index.get(0);
if (version < lowest.getVersion()) {
+ // A stored entry embeds the index snapshot of all older versions, so read the full
+ // entry (uncached) and continue the lookup in that snapshot
return readSchemaEntry(lowest.getPosition())
.thenCompose(entry -> {
List<IndexEntry> entryIndex = new ArrayList<>(entry.getIndexsCount());
for (int i = 0; i < entry.getIndexsCount(); i++) {
entryIndex.add(entry.getIndexAt(i));
}
- return findSchemaEntryByVersion(entryIndex, version);
+ return findSchemaPositionByVersion(entryIndex, version);
});
}
for (IndexEntry entry : index) {
if (entry.getVersion() == version) {
- return readSchemaEntry(entry.getPosition());
+ return completedFuture(entry.getPosition());
} else if (entry.getVersion() > version) {
break;
}
@@ -576,6 +641,17 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
return completedFuture(null);
}
+ @NonNull
+ private CompletableFuture<byte[]> readSchemaData(PositionInfo position) {
+ long ledgerId = position.getLedgerId();
+ long entryId = position.getEntryId();
+ SchemaEntryKey key = new SchemaEntryKey(ledgerId, entryId);
+ return schemaEntryCache.get(key, (ignored, executor) ->
+ FutureUtil.supplySafely(() -> readSchemaEntry(position))
+ .thenApply(SchemaEntry::getSchemaData))
+ .thenApply(schemaData -> schemaData.clone());
+ }
+
@NonNull
private CompletableFuture<SchemaEntry> readSchemaEntry(
PositionInfo position
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java
index 52457eca6bb8..a797ea928e05 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java
@@ -35,6 +35,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
import java.util.stream.Collectors;
import lombok.CustomLog;
import org.apache.avro.Schema;
@@ -107,7 +108,9 @@ public class SchemaRegistryServiceImpl implements SchemaRegistryService {
} else {
long longVersion = ((LongSchemaVersion) version).getVersion();
//If the schema has been deleted, it cannot be obtained
- completableFuture = trimDeletedSchemaAndGetList(schemaId)
+ CompletableFuture<List<CompletableFuture<SchemaAndMetadata>>> schemaFutureList =
+ getAllSchemas(schemaId, this::getAllForVersionedRead);
+ completableFuture = trimDeletedSchemaAndGetList(schemaId, schemaFutureList)
.thenApply(metadataList -> metadataList.stream().filter(schemaAndMetadata ->
((LongSchemaVersion) schemaAndMetadata.version).getVersion() == longVersion)
.collect(Collectors.toList())
@@ -148,9 +151,22 @@ public class SchemaRegistryServiceImpl implements SchemaRegistryService {
@Override
public CompletableFuture<List<CompletableFuture<SchemaAndMetadata>>> getAllSchemas(String schemaId) {
+ return getAllSchemas(schemaId, schemaStorage::getAll);
+ }
+
+ private CompletableFuture<List<CompletableFuture<StoredSchema>>> getAllForVersionedRead(String schemaId) {
+ if (schemaStorage instanceof BookkeeperSchemaStorage bookkeeperSchemaStorage) {
+ return bookkeeperSchemaStorage.getAllForVersionedRead(schemaId);
+ }
+ return schemaStorage.getAll(schemaId);
+ }
+
+ private CompletableFuture<List<CompletableFuture<SchemaAndMetadata>>> getAllSchemas(
+ String schemaId,
+ Function<String, CompletableFuture<List<CompletableFuture<StoredSchema>>>> schemaReader) {
long start = this.clock.millis();
- return schemaStorage.getAll(schemaId)
+ return schemaReader.apply(schemaId)
.thenCompose(schemas -> convertToSchemaAndMetadata(schemaId, schemas))
.whenComplete((v, t) -> {
var latencyMs = this.clock.millis() - start;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorageEntryCacheTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorageEntryCacheTest.java
new file mode 100644
index 000000000000..5e0aa96f53a4
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorageEntryCacheTest.java
@@ -0,0 +1,342 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.schema;
+
+import static org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage.bkException;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotSame;
+import com.github.benmanes.caffeine.cache.Ticker;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.bookkeeper.client.AsyncCallback;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerEntry;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.pulsar.broker.BookKeeperClientFactory;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.protocol.schema.SchemaVersion;
+import org.apache.pulsar.common.protocol.schema.StoredSchema;
+import org.apache.pulsar.common.schema.LongSchemaVersion;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.metadata.api.MetadataStoreConfig;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker", timeOut = 30000)
+public class BookkeeperSchemaStorageEntryCacheTest {
+ private static final byte[] SchemaData = new byte[] {1, 2, 3};
+ private static final AtomicLong StoreIds = new AtomicLong();
+
+ private static byte[] dataForVersion(int version) {
+ return version == 0 ? SchemaData : ("schema-v" + version).getBytes(StandardCharsets.UTF_8);
+ }
+
+ private static final class Harness implements AutoCloseable {
+ private final MetadataStoreExtended store;
+ final BookKeeper bookKeeper = mock(BookKeeper.class);
+ final LedgerHandle handle = mock(LedgerHandle.class);
+ final AtomicInteger openCount = new AtomicInteger();
+ private final AtomicInteger nextOpenRc = new AtomicInteger(BKException.Code.OK);
+ private final LinkedBlockingQueue<PendingRead> pendingReads = new LinkedBlockingQueue<>();
+ private final Map<Long, byte[]> storedEntries = new ConcurrentHashMap<>();
+ private final Map<Long, LedgerHandle> readHandles = new ConcurrentHashMap<>();
+ final BookkeeperSchemaStorage storage;
+
+ Harness(int versions) throws Exception {
+ this(versions, false, Ticker.systemTicker());
+ }
+
+ Harness(int versions, boolean latestOnlyLocator) throws Exception {
+ this(versions, latestOnlyLocator, Ticker.systemTicker());
+ }
+
+ Harness(Ticker cacheTicker) throws Exception {
+ this(1, false, cacheTicker);
+ }
+
+ private Harness(int versions, boolean latestOnlyLocator, Ticker cacheTicker) throws Exception {
+ String metadataUrl = "memory:schema-entry-cache-" + StoreIds.incrementAndGet();
+ store = MetadataStoreExtended.create(metadataUrl, MetadataStoreConfig.builder().build());
+
+ when(handle.getId()).thenReturn(1L);
+ stubReadCapture(handle);
+ stubImmediateClose(handle);
+ readHandles.put(1L, handle);
+
+ List<IndexEntry> indexes = new ArrayList<>(versions);
+ for (int version = 0; version < versions; version++) {
+ SchemaEntry schemaEntry = new SchemaEntry().setSchemaData(dataForVersion(version));
+ for (IndexEntry previous : indexes) {
+ schemaEntry.addIndex().copyFrom(previous);
+ }
+ storedEntries.put(version + 1L, schemaEntry.toByteArray());
+
+ IndexEntry index = new IndexEntry();
+ index.setVersion(version);
+ index.setHash(new byte[0]);
+ index.setPosition().setLedgerId(version + 1L).setEntryId(0L);
+ indexes.add(index);
+ }
+
+ doAnswer(invocation -> {
+ long ledgerId = invocation.getArgument(0);
+ int rc = nextOpenRc.getAndSet(BKException.Code.OK);
+ LedgerHandle ledgerHandle = rc == BKException.Code.OK
+ ? readHandles.computeIfAbsent(ledgerId, id -> {
+ LedgerHandle newHandle = mock(LedgerHandle.class);
+ when(newHandle.getId()).thenReturn(id);
+ stubReadCapture(newHandle);
+ stubImmediateClose(newHandle);
+ return newHandle;
+ }) : null;
+ openCount.incrementAndGet();
+ AsyncCallback.OpenCallback callback = invocation.getArgument(3);
+ callback.openComplete(rc, ledgerHandle, invocation.getArgument(4));
+ return null;
+ }).when(bookKeeper).asyncOpenLedger(anyLong(), any(), any(), any(), isNull(), eq(true));
+
+ SchemaLocator locator = new SchemaLocator();
+ if (latestOnlyLocator) {
+ locator.addIndex().copyFrom(indexes.get(indexes.size() - 1));
+ } else {
+ for (IndexEntry index : indexes) {
+ locator.addIndex().copyFrom(index);
+ }
+ }
+ IndexEntry latest = indexes.get(indexes.size() - 1);
+ locator.setInfo().copyFrom(latest);
+ store.put("/schemas/test", locator.toByteArray(), Optional.empty()).get(5, TimeUnit.SECONDS);
+
+ BookKeeperClientFactory clientFactory = mock(BookKeeperClientFactory.class);
+ when(clientFactory.create(any(), eq(store), isNull(), eq(Optional.empty()), isNull()))
+ .thenReturn(CompletableFuture.completedFuture(bookKeeper));
+ PulsarService pulsar = mock(PulsarService.class);
+ when(pulsar.getLocalMetadataStore()).thenReturn(store);
+ when(pulsar.getConfiguration()).thenReturn(new ServiceConfiguration());
+ when(pulsar.getBookKeeperClientFactory()).thenReturn(clientFactory);
+ storage = new BookkeeperSchemaStorage(pulsar, cacheTicker);
+ storage.start();
+ }
+
+ void failNextOpen(int rc) {
+ nextOpenRc.set(rc);
+ }
+
+ private void stubReadCapture(LedgerHandle ledgerHandle) {
+ doAnswer(invocation -> {
+ pendingReads.add(new PendingRead(invocation.getArgument(2), ledgerHandle));
+ return null;
+ }).when(ledgerHandle).asyncReadEntries(eq(0L), eq(0L), any(), isNull());
+ }
+
+ private void stubImmediateClose(LedgerHandle ledgerHandle) {
+ doAnswer(invocation -> {
+ AsyncCallback.CloseCallback callback = invocation.getArgument(0);
+ callback.closeComplete(BKException.Code.OK, ledgerHandle, null);
+ return null;
+ }).when(ledgerHandle).asyncClose(any(), isNull());
+ }
+
+ void completePendingRead(int rc) {
+ PendingRead pendingRead;
+ try {
+ pendingRead = pendingReads.poll(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ if (pendingRead == null) {
+ throw new IllegalStateException("No pending BookKeeper read to complete");
+ }
+ byte[] storedEntry = storedEntries.get(pendingRead.handle().getId());
+ if (storedEntry == null) {
+ throw new IllegalStateException("No stored entry for ledger " + pendingRead.handle().getId());
+ }
+ LedgerEntry entry = mock(LedgerEntry.class);
+ when(entry.getEntry()).thenReturn(storedEntry);
+ pendingRead.callback().readComplete(rc, pendingRead.handle(),
+ Collections.enumeration(List.of(entry)), null);
+ }
+
+ @Override
+ public void close() throws Exception {
+ storage.close();
+ store.close();
+ }
+ }
+
+ private record PendingRead(AsyncCallback.ReadCallback callback, LedgerHandle handle) {
+ }
+
+ @Test
+ public void testVersionedReadCacheSharesHistoryAndTargetLoads() throws Exception {
+ try (Harness harness = new Harness(3)) {
+ List<CompletableFuture<StoredSchema>> firstHistory =
+ harness.storage.getAllForVersionedRead("test").get(5, TimeUnit.SECONDS);
+ CompletableFuture<StoredSchema> target =
+ harness.storage.get("test", new LongSchemaVersion(0));
+ List<CompletableFuture<StoredSchema>> secondHistory =
+ harness.storage.getAllForVersionedRead("test").get(5, TimeUnit.SECONDS);
+
+ assertFalse(firstHistory.get(0).isDone());
+ assertFalse(target.isDone());
+ assertFalse(secondHistory.get(0).isDone());
+
+ completeReadsAndAssertSchemas(harness, firstHistory);
+ assertSchemas(secondHistory);
+ StoredSchema targetSchema = target.get(5, TimeUnit.SECONDS);
+ StoredSchema historySchema = firstHistory.get(0).get(5, TimeUnit.SECONDS);
+ assertEquals(targetSchema.data, dataForVersion(0));
+ assertNotSame(historySchema.data, targetSchema.data);
+
+ historySchema.data[0] = 99;
+ assertEquals(targetSchema.data, dataForVersion(0));
+ for (int version = 0; version < 3; version++) {
+ assertEquals(harness.storage.get("test", new LongSchemaVersion(version))
+ .get(5, TimeUnit.SECONDS).data, dataForVersion(version));
+ }
+ assertEquals(harness.openCount.get(), 3);
+ }
+ }
+
+ @Test
+ public void testExpiredFailedVersionedReadIsRetried() throws Exception {
+ AtomicLong tickerNanos = new AtomicLong();
+ try (Harness harness = new Harness(tickerNanos::get)) {
+ CompletableFuture<StoredSchema> initial =
+ harness.storage.get("test", new LongSchemaVersion(0));
+ harness.completePendingRead(BKException.Code.OK);
+ assertEquals(initial.get(5, TimeUnit.SECONDS).data, SchemaData);
+
+ tickerNanos.addAndGet(TimeUnit.MINUTES.toNanos(59));
+ assertEquals(harness.storage.get("test", new LongSchemaVersion(0))
+ .get(5, TimeUnit.SECONDS).data, SchemaData);
+ assertEquals(harness.openCount.get(), 1);
+
+ tickerNanos.addAndGet(TimeUnit.MINUTES.toNanos(2));
+ harness.failNextOpen(BKException.Code.ReadException);
+ CompletableFuture<StoredSchema> failed =
+ harness.storage.get("test", new LongSchemaVersion(0));
+ Throwable error = failed.handle((__, ex) -> FutureUtil.unwrapCompletionException(ex))
+ .get(5, TimeUnit.SECONDS);
+ assertEquals(error.getMessage(), bkException("Failed to open ledger",
+ BKException.Code.ReadException, 1, -1, false).getMessage());
+
+ CompletableFuture<StoredSchema> retry =
+ harness.storage.get("test", new LongSchemaVersion(0));
+ harness.completePendingRead(BKException.Code.OK);
+ assertEquals(retry.get(5, TimeUnit.SECONDS).data, SchemaData);
+ assertEquals(harness.openCount.get(), 3);
+ }
+ }
+
+ @Test
+ public void testNestedHistoryLookupOnlyCachesFinalVersionPayload() throws Exception {
+ try (Harness harness = new Harness(3, true)) {
+ CompletableFuture<StoredSchema> first =
+ harness.storage.get("test", new LongSchemaVersion(0));
+ harness.completePendingRead(BKException.Code.OK);
+ harness.completePendingRead(BKException.Code.OK);
+ assertEquals(first.get(5, TimeUnit.SECONDS).data, dataForVersion(0));
+ assertEquals(harness.openCount.get(), 2);
+
+ CompletableFuture<StoredSchema> second =
+ harness.storage.get("test", new LongSchemaVersion(0));
+ // The checkpoint is read again, while the final version-0 payload is cached.
+ harness.completePendingRead(BKException.Code.OK);
+ assertEquals(second.get(5, TimeUnit.SECONDS).data, dataForVersion(0));
+ assertEquals(harness.openCount.get(), 3);
+ }
+ }
+
+ @Test
+ public void testLatestAndRawGetAllBypassVersionedReadCache() throws Exception {
+ try (Harness harness = new Harness(3)) {
+ CompletableFuture<StoredSchema> versioned =
+ harness.storage.get("test", new LongSchemaVersion(0));
+ harness.completePendingRead(BKException.Code.OK);
+ assertEquals(versioned.get(5, TimeUnit.SECONDS).data, dataForVersion(0));
+
+ CompletableFuture<StoredSchema> firstLatest =
+ harness.storage.get("test", SchemaVersion.Latest);
+ harness.completePendingRead(BKException.Code.OK);
+ assertEquals(firstLatest.get(5, TimeUnit.SECONDS).data, dataForVersion(2));
+
+ CompletableFuture<StoredSchema> secondLatest =
+ harness.storage.get("test", SchemaVersion.Latest);
+ harness.completePendingRead(BKException.Code.OK);
+ assertEquals(secondLatest.get(5, TimeUnit.SECONDS).data, dataForVersion(2));
+ assertEquals(harness.openCount.get(), 3);
+
+ assertRawGetAllReadsEveryVersionFromBookKeeper(harness);
+ assertEquals(harness.openCount.get(), 6);
+
+ assertRawGetAllReadsEveryVersionFromBookKeeper(harness);
+ assertEquals(harness.openCount.get(), 9);
+
+ assertEquals(harness.storage.get("test", new LongSchemaVersion(0))
+ .get(5, TimeUnit.SECONDS).data, dataForVersion(0));
+ assertEquals(harness.openCount.get(), 9);
+ }
+ }
+
+ private static void assertRawGetAllReadsEveryVersionFromBookKeeper(Harness harness) throws Exception {
+ List<CompletableFuture<StoredSchema>> schemas = harness.storage.getAll("test")
+ .get(5, TimeUnit.SECONDS);
+ completeReadsAndAssertSchemas(harness, schemas);
+ }
+
+ private static void completeReadsAndAssertSchemas(
+ Harness harness, List<CompletableFuture<StoredSchema>> schemas) throws Exception {
+ assertEquals(schemas.size(), 3);
+ for (int version = 0; version < 3; version++) {
+ harness.completePendingRead(BKException.Code.OK);
+ }
+ assertSchemas(schemas);
+ }
+
+ private static void assertSchemas(List<CompletableFuture<StoredSchema>> schemas) throws Exception {
+ assertEquals(schemas.size(), 3);
+ for (int version = 0; version < 3; version++) {
+ StoredSchema schema = schemas.get(version).get(5, TimeUnit.SECONDS);
+ assertEquals(((LongSchemaVersion) schema.version).getVersion(), version);
+ assertEquals(schema.data, dataForVersion(version));
+ }
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ClientGetSchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ClientGetSchemaTest.java
index c9540f060371..e8aa87aaef6d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ClientGetSchemaTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ClientGetSchemaTest.java
@@ -32,6 +32,7 @@ import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.schema.SchemaDefinition;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.schema.LongSchemaVersion;
import org.apache.pulsar.schema.Schemas;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -101,6 +102,11 @@ public class ClientGetSchemaTest extends SharedPulsarBaseTest {
BookkeeperSchemaStorage schemaStorage = (BookkeeperSchemaStorage) getSchemaStorage();
long schemaLedgerId = schemaStorage.getSchemaLedgerList(key).get(0);
+ // Warm the versioned-read cache. Producer registration must still read BookKeeper directly
+ // so a missing ledger reaches the existing recovery path.
+ assertNotNull(getPulsar().getSchemaRegistryService()
+ .getSchema(key, new LongSchemaVersion(0)).get());
+
// (2) break schema locator by deleting schema-ledger
schemaStorage.getBookKeeper().deleteLedger(schemaLedgerId);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImplTest.java
new file mode 100644
index 000000000000..8d424c00eca0
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImplTest.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.schema;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertNull;
+import java.time.Clock;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.common.protocol.schema.SchemaStorage;
+import org.apache.pulsar.common.schema.LongSchemaVersion;
+import org.mockito.MockedConstruction;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class SchemaRegistryServiceImplTest {
+
+ @Test
+ public void testExplicitVersionUsesBookkeeperSchemaReadPath() throws Exception {
+ String schemaId = "tenant/namespace/topic";
+ LongSchemaVersion version = new LongSchemaVersion(0);
+ BookkeeperSchemaStorage storage = mock(BookkeeperSchemaStorage.class);
+ when(storage.getAllForVersionedRead(schemaId))
+ .thenReturn(CompletableFuture.completedFuture(Collections.emptyList()));
+
+ try (MockedConstruction<SchemaRegistryStats> ignored = mockConstruction(SchemaRegistryStats.class)) {
+ SchemaRegistryServiceImpl service = new SchemaRegistryServiceImpl(
+ storage, Map.of(), Clock.systemUTC(), mock(PulsarService.class));
+
+ assertNull(service.getSchema(schemaId, version).get(5, TimeUnit.SECONDS));
+ verify(storage).getAllForVersionedRead(schemaId);
+ verify(storage, never()).getAll(schemaId);
+ verify(storage, never()).get(schemaId, version);
+
+ service.close();
+ }
+ }
+
+ @Test
+ public void testExplicitVersionFallsBackToSchemaStorageGetAll() throws Exception {
+ String schemaId = "tenant/namespace/topic";
+ LongSchemaVersion version = new LongSchemaVersion(0);
+ SchemaStorage storage = mock(SchemaStorage.class);
+ when(storage.getAll(schemaId))
+ .thenReturn(CompletableFuture.completedFuture(Collections.emptyList()));
+
+ try (MockedConstruction<SchemaRegistryStats> ignored = mockConstruction(SchemaRegistryStats.class)) {
+ SchemaRegistryServiceImpl service = new SchemaRegistryServiceImpl(
+ storage, Map.of(), Clock.systemUTC(), mock(PulsarService.class));
+
+ assertNull(service.getSchema(schemaId, version).get(5, TimeUnit.SECONDS));
+ verify(storage).getAll(schemaId);
+ verify(storage, never()).get(schemaId, version);
+
+ service.close();
+ }
+ }
+}