diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index 5f82c3c7b2e..2a0b415314f 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -25,7 +25,7 @@
files="(.*).java"/>
+ files="(AvroData|ConfigResource|DownloadSchemaRegistryMojo|KafkaSchemaRegistry|KafkaStore|KafkaStoreReaderThread|MessageDefinition|Schema|SchemaValue|SchemaDiff|MessageSchemaDiff|AbstractKafkaSchemaSerDe|AbstractKafkaAvroSerializer|AbstractKafkaAvroDeserializer|AbstractKafkaJsonSchemaDeserializer|AbstractKafkaProtobufDeserializer|ProtobufData|ProtobufSchemaUtils|JsonSchemaData|SchemaMessageFormatter|SchemaMessageReader|ContextFilter|QualifiedSubject|SubjectVersionsResource|Rule|WildcardMatcher|JsonSchemaComparator|LocalSchemaRegistryClient|DataEncryptionKeyId|FieldEncryptionExecutor|FieldRuleExecutor|SchemaRegistryCoordinator|DekRegistry|CachedSchemaRegistryClient).java"/>
diff --git a/client/src/main/java/io/confluent/kafka/schemaregistry/client/CachedSchemaRegistryClient.java b/client/src/main/java/io/confluent/kafka/schemaregistry/client/CachedSchemaRegistryClient.java
index fc2fb770862..81ce9d827cf 100644
--- a/client/src/main/java/io/confluent/kafka/schemaregistry/client/CachedSchemaRegistryClient.java
+++ b/client/src/main/java/io/confluent/kafka/schemaregistry/client/CachedSchemaRegistryClient.java
@@ -19,6 +19,7 @@
import com.google.common.base.Ticker;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
+import com.google.common.collect.ImmutableMap;
import io.confluent.kafka.schemaregistry.client.rest.entities.requests.RegisterSchemaRequest;
import io.confluent.kafka.schemaregistry.client.rest.entities.requests.RegisterSchemaResponse;
import io.confluent.kafka.schemaregistry.utils.QualifiedSubject;
@@ -72,6 +73,8 @@ public class CachedSchemaRegistryClient implements SchemaRegistryClient {
private final Map> idToSchemaCache;
private final Map> schemaToVersionCache;
private final Map> versionToSchemaCache;
+ private final Cache latestVersionCache;
+ private final Cache latestWithMetadataCache;
private final Cache missingSchemaCache;
private final Cache missingIdCache;
private final Cache missingVersionCache;
@@ -197,6 +200,25 @@ public CachedSchemaRegistryClient(
this.restService = restService;
this.ticker = ticker;
+ long latestTTL = SchemaRegistryClientConfig.getLatestTTL(configs);
+
+ CacheBuilder