diff --git a/src/main/java/org/springframework/data/elasticsearch/cache/CacheDocument.java b/src/main/java/org/springframework/data/elasticsearch/cache/CacheDocument.java
new file mode 100644
index 000000000..96507bc38
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/cache/CacheDocument.java
@@ -0,0 +1,96 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.elasticsearch.cache;
+
+import org.jspecify.annotations.Nullable;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.elasticsearch.annotations.Document;
+import org.springframework.data.elasticsearch.annotations.Field;
+import org.springframework.data.elasticsearch.annotations.FieldType;
+import org.springframework.data.elasticsearch.annotations.WriteTypeHint;
+
+/**
+ * Internal representation of a persisted cache entry.
+ *
+ * @author Anıl Şenocak
+ * @since 6.2
+ */
+@Document(indexName = ElasticsearchCacheManager.DEFAULT_INDEX_NAME, createIndex = false,
+		writeTypeHint = WriteTypeHint.FALSE, storeIdInSource = false)
+class CacheDocument {
+
+	@Id private @Nullable String id;
+	@Field(type = FieldType.Keyword) private @Nullable String cacheName;
+	@Field(type = FieldType.Keyword) private @Nullable String cacheKey;
+	@Field(type = FieldType.Keyword, index = false) private @Nullable String valueType;
+	@Field(type = FieldType.Text, index = false) private @Nullable String valueJson;
+	@Field(type = FieldType.Boolean) private boolean scalarValue;
+	@Field(type = FieldType.Boolean) private boolean nullValue;
+	@Field(type = FieldType.Long) private @Nullable Long expiresAt;
+
+	CacheDocument() {}
+
+	CacheDocument(String id, String cacheName, String cacheKey, @Nullable String valueType, String valueJson,
+			boolean scalarValue, boolean nullValue, @Nullable Long expiresAt) {
+		this.id = id;
+		this.cacheName = cacheName;
+		this.cacheKey = cacheKey;
+		this.valueType = valueType;
+		this.valueJson = valueJson;
+		this.scalarValue = scalarValue;
+		this.nullValue = nullValue;
+		this.expiresAt = expiresAt;
+	}
+
+	String getRequiredId() {
+		return required(id, "id");
+	}
+
+	String getRequiredCacheName() {
+		return required(cacheName, "cacheName");
+	}
+
+	String getRequiredCacheKey() {
+		return required(cacheKey, "cacheKey");
+	}
+
+	@Nullable String getValueType() {
+		return valueType;
+	}
+
+	String getRequiredValueJson() {
+		return required(valueJson, "valueJson");
+	}
+
+	boolean isScalarValue() {
+		return scalarValue;
+	}
+
+	boolean isNullValue() {
+		return nullValue;
+	}
+
+	@Nullable Long getExpiresAt() {
+		return expiresAt;
+	}
+
+	private static String required(@Nullable String value, String property) {
+		if (value == null) {
+			throw new IllegalStateException("Cache document property '%s' must not be null".formatted(property));
+		}
+		return value;
+	}
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/cache/CacheDocumentStore.java b/src/main/java/org/springframework/data/elasticsearch/cache/CacheDocumentStore.java
new file mode 100644
index 000000000..5f7a6b1ed
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/cache/CacheDocumentStore.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.elasticsearch.cache;
+
+import org.jspecify.annotations.Nullable;
+
+import java.util.List;
+
+/**
+ * Persistence operations for cache documents.
+ *
+ * @author Anıl Şenocak
+ * @since 6.2
+ */
+interface CacheDocumentStore {
+
+	@Nullable CacheDocument get(String cacheName, String cacheKey);
+
+	void put(CacheDocument document);
+
+	@Nullable CacheDocument delete(String cacheName, String cacheKey);
+
+	List<CacheDocument> findAll(String cacheName);
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/cache/ElasticsearchCache.java b/src/main/java/org/springframework/data/elasticsearch/cache/ElasticsearchCache.java
new file mode 100644
index 000000000..6556a49c0
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/cache/ElasticsearchCache.java
@@ -0,0 +1,326 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.elasticsearch.cache;
+
+import org.jspecify.annotations.Nullable;
+import org.springframework.cache.Cache;
+import org.springframework.cache.support.SimpleValueWrapper;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.data.elasticsearch.cache.event.CacheEvictedEvent;
+import org.springframework.data.elasticsearch.cache.event.CacheInsertedEvent;
+import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
+import org.springframework.data.elasticsearch.core.document.Document;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * A Spring {@link Cache} backed by Elasticsearch.
+ *
+ * @author Anıl Şenocak
+ * @since 6.2
+ */
+public final class ElasticsearchCache implements Cache {
+
+	private static final String SCALAR_VALUE_FIELD = "value";
+
+	private final String name;
+	private final CacheDocumentStore store;
+	private final ElasticsearchConverter converter;
+	private final ElasticsearchCacheConfiguration configuration;
+	private final @Nullable ApplicationEventPublisher eventPublisher;
+	private final ReentrantLock operationLock = new ReentrantLock();
+
+	ElasticsearchCache(String name, CacheDocumentStore store, ElasticsearchConverter converter,
+			ElasticsearchCacheConfiguration configuration, @Nullable ApplicationEventPublisher eventPublisher) {
+
+		Assert.hasText(name, "Cache name must not be empty");
+		Assert.notNull(store, "CacheDocumentStore must not be null");
+		Assert.notNull(converter, "ElasticsearchConverter must not be null");
+		Assert.notNull(configuration, "ElasticsearchCacheConfiguration must not be null");
+		this.name = name;
+		this.store = store;
+		this.converter = converter;
+		this.configuration = configuration;
+		this.eventPublisher = eventPublisher;
+	}
+
+	@Override
+	public String getName() {
+		return name;
+	}
+
+	@Override
+	public Object getNativeCache() {
+		return store;
+	}
+
+	@Override
+	public @Nullable ValueWrapper get(Object key) {
+
+		CacheDocument document = getDocument(key);
+		return document != null ? new SimpleValueWrapper(readValue(document)) : null;
+	}
+
+	@Override
+	public <T> @Nullable T get(Object key, @Nullable Class<T> type) {
+
+		CacheDocument document = getDocument(key);
+		if (document == null) {
+			return null;
+		}
+		Object value = readValue(document);
+		if (value == null) {
+			return null;
+		}
+		if (type != null && !type.isInstance(value)) {
+			throw new IllegalStateException("Cached value is not of required type [%s]: %s".formatted(type.getName(), value));
+		}
+		@SuppressWarnings("unchecked")
+		T result = (T) value;
+		return result;
+	}
+
+	@Override
+	public <T> @Nullable T get(Object key, Callable<T> valueLoader) {
+
+		Assert.notNull(valueLoader, "ValueLoader must not be null");
+		CacheDocument document = getDocument(key);
+		if (document != null) {
+			@SuppressWarnings("unchecked")
+			T cached = (T) readValue(document);
+			return cached;
+		}
+		try {
+			T value = valueLoader.call();
+			put(key, value);
+			return value;
+		}
+		catch (Exception exception) {
+			throw new ValueRetrievalException(key, valueLoader, exception);
+		}
+	}
+
+	@Override
+	public void put(Object key, @Nullable Object value) {
+
+		Assert.notNull(key, "Key must not be null");
+		if (value == null && !configuration.getAllowCacheNullValues()) {
+			throw new IllegalArgumentException("Cache '%s' does not allow null values".formatted(name));
+		}
+		CacheInsertedEvent event;
+		operationLock.lock();
+		try {
+			String cacheKey = configuration.getKey(key);
+			CacheDocument previous = store.get(name, cacheKey);
+			Object previousValue = previous != null && !isExpired(previous, nowMillis()) ? readValue(previous) : null;
+			store.put(createDocument(cacheKey, value));
+			event = new CacheInsertedEvent(name, key, value, previousValue);
+		}
+		finally {
+			operationLock.unlock();
+		}
+		publish(event);
+	}
+
+	@Override
+	public @Nullable ValueWrapper putIfAbsent(Object key, @Nullable Object value) {
+
+		operationLock.lock();
+		try {
+			CacheDocument existing = getDocument(key);
+			if (existing != null) {
+				return new SimpleValueWrapper(readValue(existing));
+			}
+			put(key, value);
+			return null;
+		}
+		finally {
+			operationLock.unlock();
+		}
+	}
+
+	@Override
+	public void evict(Object key) {
+		evictIfPresent(key);
+	}
+
+	@Override
+	public boolean evictIfPresent(Object key) {
+
+		Assert.notNull(key, "Key must not be null");
+		CacheDocument removed;
+		operationLock.lock();
+		try {
+			removed = store.delete(name, configuration.getKey(key));
+		}
+		finally {
+			operationLock.unlock();
+		}
+		if (removed == null) {
+			return false;
+		}
+		publish(new CacheEvictedEvent(name, key, readValue(removed)));
+		return true;
+	}
+
+	@Override
+	public void clear() {
+		clearAndCount();
+	}
+
+	@Override
+	public boolean invalidate() {
+		return clearAndCount() > 0;
+	}
+
+	/**
+	 * Remove expired entries and publish a {@link CacheEvictedEvent} for each removed entry.
+	 *
+	 * @return the number of removed entries.
+	 */
+	public int evictExpired() {
+
+		List<CacheDocument> removed = new ArrayList<>();
+		operationLock.lock();
+		try {
+			long nowMillis = nowMillis();
+			for (CacheDocument document : store.findAll(name)) {
+				if (isExpired(document, nowMillis)) {
+					remove(document, removed);
+				}
+			}
+		}
+		finally {
+			operationLock.unlock();
+		}
+		publishEvictions(removed);
+		return removed.size();
+	}
+
+	private int clearAndCount() {
+
+		List<CacheDocument> removed = new ArrayList<>();
+		operationLock.lock();
+		try {
+			for (CacheDocument document : store.findAll(name)) {
+				remove(document, removed);
+			}
+		}
+		finally {
+			operationLock.unlock();
+		}
+		publishEvictions(removed);
+		return removed.size();
+	}
+
+	private void remove(CacheDocument document, List<CacheDocument> removedDocuments) {
+
+		CacheDocument removed = store.delete(name, document.getRequiredCacheKey());
+		if (removed != null) {
+			removedDocuments.add(removed);
+		}
+	}
+
+	private void publishEvictions(List<CacheDocument> removedDocuments) {
+		removedDocuments.forEach(document ->
+				publish(new CacheEvictedEvent(name, document.getRequiredCacheKey(), readValue(document))));
+	}
+
+	private @Nullable CacheDocument getDocument(Object key) {
+
+		Assert.notNull(key, "Key must not be null");
+		operationLock.lock();
+		try {
+			CacheDocument document = store.get(name, configuration.getKey(key));
+			return document == null || isExpired(document, nowMillis()) ? null : document;
+		}
+		finally {
+			operationLock.unlock();
+		}
+	}
+
+	private CacheDocument createDocument(String cacheKey, @Nullable Object value) {
+
+		Long expiresAt = expiresAt(nowMillis());
+		String documentId = ElasticsearchCacheDocumentStore.documentId(name, cacheKey);
+		if (value == null) {
+			return new CacheDocument(documentId, name, cacheKey, null, "{}", false, true, expiresAt);
+		}
+
+		Class<?> valueType = value.getClass();
+		ConversionService conversionService = converter.getConversionService();
+		if (conversionService.canConvert(valueType, String.class)
+				&& conversionService.canConvert(String.class, valueType)) {
+			Document scalar = Document.create();
+			scalar.put(SCALAR_VALUE_FIELD, conversionService.convert(value, String.class));
+			return new CacheDocument(documentId, name, cacheKey, valueType.getName(), scalar.toJson(), true, false,
+					expiresAt);
+		}
+
+		return new CacheDocument(documentId, name, cacheKey, valueType.getName(), converter.mapObject(value).toJson(), false,
+				false, expiresAt);
+	}
+
+	private @Nullable Object readValue(CacheDocument document) {
+
+		if (document.isNullValue()) {
+			return null;
+		}
+		String valueTypeName = document.getValueType();
+		if (valueTypeName == null) {
+			throw new IllegalStateException("Cache document value type must not be null");
+		}
+		Class<?> valueType = ClassUtils.resolveClassName(valueTypeName, ClassUtils.getDefaultClassLoader());
+		Document valueDocument = Document.parse(document.getRequiredValueJson());
+		if (!document.isScalarValue()) {
+			return converter.read(valueType, valueDocument);
+		}
+
+		Object scalar = valueDocument.get(SCALAR_VALUE_FIELD);
+		if (scalar == null || valueType.isInstance(scalar)) {
+			return scalar;
+		}
+		return converter.getConversionService().convert(scalar, valueType);
+	}
+
+	private @Nullable Long expiresAt(long nowMillis) {
+
+		Duration ttl = configuration.getEntryTtl();
+		return ttl.isZero() ? null : nowMillis + Math.max(ttl.toMillis(), 1);
+	}
+
+	private boolean isExpired(CacheDocument document, long nowMillis) {
+		Long expiresAt = document.getExpiresAt();
+		return expiresAt != null && expiresAt <= nowMillis;
+	}
+
+	private long nowMillis() {
+		return System.currentTimeMillis();
+	}
+
+	private void publish(Object event) {
+		if (eventPublisher != null) {
+			eventPublisher.publishEvent(event);
+		}
+	}
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheConfiguration.java b/src/main/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheConfiguration.java
new file mode 100644
index 000000000..2a6db200c
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheConfiguration.java
@@ -0,0 +1,89 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.elasticsearch.cache;
+
+import org.springframework.util.Assert;
+
+import java.time.Duration;
+import java.util.function.Function;
+
+/**
+ * Configuration used by an {@link ElasticsearchCache}.
+ *
+ * @author Anıl Şenocak
+ * @since 6.2
+ */
+public final class ElasticsearchCacheConfiguration {
+
+	private final Duration entryTtl;
+	private final Function<Object, String> keyConverter;
+	private final boolean cacheNullValues;
+
+	private ElasticsearchCacheConfiguration(Duration entryTtl, Function<Object, String> keyConverter,
+			boolean cacheNullValues) {
+		this.entryTtl = entryTtl;
+		this.keyConverter = keyConverter;
+		this.cacheNullValues = cacheNullValues;
+	}
+
+	/**
+	 * Create a default configuration without expiration and using {@link Object#toString()} for cache keys.
+	 */
+	public static ElasticsearchCacheConfiguration defaultCacheConfig() {
+		return new ElasticsearchCacheConfiguration(Duration.ZERO, Object::toString, true);
+	}
+
+	/**
+	 * Return a copy configured with the given time-to-live. {@link Duration#ZERO} disables expiration.
+	 */
+	public ElasticsearchCacheConfiguration entryTtl(Duration entryTtl) {
+
+		Assert.notNull(entryTtl, "Entry TTL must not be null");
+		Assert.isTrue(!entryTtl.isNegative(), "Entry TTL must not be negative");
+		return new ElasticsearchCacheConfiguration(entryTtl, keyConverter, cacheNullValues);
+	}
+
+	/**
+	 * Return a copy using the given converter to create the persisted cache key.
+	 */
+	public ElasticsearchCacheConfiguration serializeKeysWith(Function<Object, String> keyConverter) {
+
+		Assert.notNull(keyConverter, "Key converter must not be null");
+		return new ElasticsearchCacheConfiguration(entryTtl, keyConverter, cacheNullValues);
+	}
+
+	/**
+	 * Return a copy that rejects {@literal null} cache values.
+	 */
+	public ElasticsearchCacheConfiguration disableCachingNullValues() {
+		return new ElasticsearchCacheConfiguration(entryTtl, keyConverter, false);
+	}
+
+	Duration getEntryTtl() {
+		return entryTtl;
+	}
+
+	boolean getAllowCacheNullValues() {
+		return cacheNullValues;
+	}
+
+	String getKey(Object key) {
+
+		String converted = keyConverter.apply(key);
+		Assert.notNull(converted, "Key converter must not return null");
+		return converted;
+	}
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheDocumentStore.java b/src/main/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheDocumentStore.java
new file mode 100644
index 000000000..3662489b0
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheDocumentStore.java
@@ -0,0 +1,139 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.elasticsearch.cache;
+
+import org.jspecify.annotations.Nullable;
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
+import org.springframework.data.elasticsearch.core.IndexOperations;
+import org.springframework.data.elasticsearch.core.RefreshPolicy;
+import org.springframework.data.elasticsearch.core.SearchHit;
+import org.springframework.data.elasticsearch.core.SearchHitsIterator;
+import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
+import org.springframework.data.elasticsearch.core.query.Criteria;
+import org.springframework.data.elasticsearch.core.query.CriteriaQuery;
+import org.springframework.util.Assert;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * Persists cache documents using the store-independent {@link ElasticsearchOperations} API.
+ *
+ * @author Anıl Şenocak
+ * @since 6.2
+ */
+final class ElasticsearchCacheDocumentStore implements CacheDocumentStore {
+
+	private final ElasticsearchOperations operations;
+	private final IndexCoordinates index;
+	private final ReentrantLock indexCreationLock = new ReentrantLock();
+
+	ElasticsearchCacheDocumentStore(ElasticsearchOperations operations, String indexName) {
+
+		Assert.notNull(operations, "ElasticsearchOperations must not be null");
+		Assert.hasText(indexName, "Index name must not be empty");
+		this.operations = operations.withRefreshPolicy(RefreshPolicy.IMMEDIATE);
+		this.index = IndexCoordinates.of(indexName);
+	}
+
+	@Override
+	public @Nullable CacheDocument get(String cacheName, String cacheKey) {
+
+		if (!indexExists()) {
+			return null;
+		}
+		return operations.get(documentId(cacheName, cacheKey), CacheDocument.class, index);
+	}
+
+	@Override
+	public void put(CacheDocument document) {
+		ensureIndexExists();
+		operations.save(document, index);
+	}
+
+	@Override
+	public @Nullable CacheDocument delete(String cacheName, String cacheKey) {
+
+		CacheDocument existing = get(cacheName, cacheKey);
+		if (existing != null) {
+			operations.delete(documentId(cacheName, cacheKey), index);
+		}
+		return existing;
+	}
+
+	@Override
+	public List<CacheDocument> findAll(String cacheName) {
+
+		if (!indexExists()) {
+			return List.of();
+		}
+		CriteriaQuery query = new CriteriaQuery(Criteria.where("cacheName").is(cacheName));
+		List<CacheDocument> documents = new ArrayList<>();
+		try (SearchHitsIterator<CacheDocument> iterator = operations.searchForStream(query, CacheDocument.class, index)) {
+			while (iterator.hasNext()) {
+				SearchHit<CacheDocument> hit = iterator.next();
+				documents.add(hit.getContent());
+			}
+		}
+		return documents;
+	}
+
+	private void ensureIndexExists() {
+
+		if (indexExists()) {
+			return;
+		}
+		indexCreationLock.lock();
+		try {
+			if (indexExists()) {
+				return;
+			}
+			IndexOperations indexOperations = operations.indexOps(index);
+			try {
+				indexOperations.create(indexOperations.createSettings(CacheDocument.class),
+						indexOperations.createMapping(CacheDocument.class));
+			}
+			catch (DataAccessException exception) {
+				if (!indexExists()) {
+					throw exception;
+				}
+			}
+		}
+		finally {
+			indexCreationLock.unlock();
+		}
+	}
+
+	private boolean indexExists() {
+		return operations.indexOps(index).exists();
+	}
+
+	static String documentId(String cacheName, String cacheKey) {
+		try {
+			byte[] digest = MessageDigest.getInstance("SHA-256").digest(cacheKey.getBytes(StandardCharsets.UTF_8));
+			return cacheName + ':' + HexFormat.of().formatHex(digest);
+		}
+		catch (NoSuchAlgorithmException exception) {
+			throw new IllegalStateException("SHA-256 digest is unavailable", exception);
+		}
+	}
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheManager.java b/src/main/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheManager.java
new file mode 100644
index 000000000..16a9769b2
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheManager.java
@@ -0,0 +1,295 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.elasticsearch.cache;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.jspecify.annotations.Nullable;
+import org.springframework.cache.CacheManager;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
+import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
+import org.springframework.util.Assert;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A {@link CacheManager} that stores cache entries in Elasticsearch.
+ *
+ * @author Anıl Şenocak
+ * @since 6.2
+ */
+public final class ElasticsearchCacheManager implements CacheManager, AutoCloseable {
+
+	/** The default Elasticsearch index used to persist cache entries. */
+	public static final String DEFAULT_INDEX_NAME = "spring_cache_entries";
+
+	private static final Log LOGGER = LogFactory.getLog(ElasticsearchCacheManager.class);
+
+	private final CacheDocumentStore store;
+	private final ElasticsearchConverter converter;
+	private final ElasticsearchCacheConfiguration defaultCacheConfiguration;
+	private final Map<String, ElasticsearchCacheConfiguration> initialCacheConfigurations;
+	private final @Nullable ApplicationEventPublisher eventPublisher;
+	private final ConcurrentMap<String, ElasticsearchCache> caches = new ConcurrentHashMap<>();
+	private final @Nullable ScheduledExecutorService evictionExecutor;
+
+	/**
+	 * Create a cache manager that uses {@value #DEFAULT_INDEX_NAME} and the default cache configuration.
+	 *
+	 * @param operations the operations used to access Elasticsearch.
+	 */
+	public ElasticsearchCacheManager(ElasticsearchOperations operations) {
+		this(operations, DEFAULT_INDEX_NAME, ElasticsearchCacheConfiguration.defaultCacheConfig(), Map.of(), null, null);
+	}
+
+	/**
+	 * Create a cache manager that uses {@value #DEFAULT_INDEX_NAME}.
+	 *
+	 * @param operations the operations used to access Elasticsearch.
+	 * @param defaultCacheConfiguration the configuration applied to caches without a specific configuration.
+	 */
+	public ElasticsearchCacheManager(ElasticsearchOperations operations,
+			ElasticsearchCacheConfiguration defaultCacheConfiguration) {
+		this(operations, DEFAULT_INDEX_NAME, defaultCacheConfiguration, Map.of(), null, null);
+	}
+
+	private ElasticsearchCacheManager(ElasticsearchOperations operations, String indexName,
+			ElasticsearchCacheConfiguration defaultCacheConfiguration,
+			Map<String, ElasticsearchCacheConfiguration> initialCacheConfigurations,
+			@Nullable ApplicationEventPublisher eventPublisher, @Nullable Duration evictionInterval) {
+
+		Assert.notNull(operations, "ElasticsearchOperations must not be null");
+		Assert.hasText(indexName, "Index name must not be empty");
+		Assert.notNull(defaultCacheConfiguration, "Default cache configuration must not be null");
+		Assert.notNull(initialCacheConfigurations, "Initial cache configurations must not be null");
+		this.store = new ElasticsearchCacheDocumentStore(operations, indexName);
+		this.converter = operations.getElasticsearchConverter();
+		this.defaultCacheConfiguration = defaultCacheConfiguration;
+		this.initialCacheConfigurations = Map.copyOf(initialCacheConfigurations);
+		this.eventPublisher = eventPublisher;
+		this.evictionExecutor = scheduleEviction(evictionInterval);
+	}
+
+	ElasticsearchCacheManager(CacheDocumentStore store, ElasticsearchConverter converter,
+			ElasticsearchCacheConfiguration defaultCacheConfiguration,
+			Map<String, ElasticsearchCacheConfiguration> initialCacheConfigurations,
+			@Nullable ApplicationEventPublisher eventPublisher, @Nullable Duration evictionInterval) {
+
+		Assert.notNull(store, "CacheDocumentStore must not be null");
+		Assert.notNull(converter, "ElasticsearchConverter must not be null");
+		Assert.notNull(defaultCacheConfiguration, "Default cache configuration must not be null");
+		Assert.notNull(initialCacheConfigurations, "Initial cache configurations must not be null");
+		this.store = store;
+		this.converter = converter;
+		this.defaultCacheConfiguration = defaultCacheConfiguration;
+		this.initialCacheConfigurations = Map.copyOf(initialCacheConfigurations);
+		this.eventPublisher = eventPublisher;
+		this.evictionExecutor = scheduleEviction(evictionInterval);
+	}
+
+	/**
+	 * Create a builder for an {@link ElasticsearchCacheManager}.
+	 *
+	 * @param operations the operations used to access Elasticsearch.
+	 * @return a new builder.
+	 */
+	public static Builder builder(ElasticsearchOperations operations) {
+		return new Builder(operations);
+	}
+
+	@Override
+	public ElasticsearchCache getCache(String name) {
+
+		Assert.hasText(name, "Cache name must not be empty");
+		return caches.computeIfAbsent(name, this::createCache);
+	}
+
+	@Override
+	public Collection<String> getCacheNames() {
+
+		LinkedHashSet<String> names = new LinkedHashSet<>(initialCacheConfigurations.keySet());
+		names.addAll(caches.keySet());
+		return List.copyOf(names);
+	}
+
+	/**
+	 * Remove every entry from all cache names known to this manager.
+	 */
+	public void clearAll() {
+		getCacheNames().forEach(cacheName -> getCache(cacheName).clear());
+	}
+
+	/**
+	 * Remove expired entries from all cache names known to this manager.
+	 *
+	 * @return the number of removed entries.
+	 */
+	public int evictExpired() {
+		return getCacheNames().stream().mapToInt(cacheName -> getCache(cacheName).evictExpired()).sum();
+	}
+
+	/**
+	 * Return whether this manager is configured to remove expired entries periodically.
+	 */
+	public boolean isEvictionScheduled() {
+		return evictionExecutor != null;
+	}
+
+	@Override
+	public void close() {
+
+		if (evictionExecutor != null) {
+			evictionExecutor.shutdownNow();
+		}
+	}
+
+	private ElasticsearchCache createCache(String name) {
+
+		ElasticsearchCacheConfiguration configuration = initialCacheConfigurations.getOrDefault(name,
+				defaultCacheConfiguration);
+		return new ElasticsearchCache(name, store, converter, configuration, eventPublisher);
+	}
+
+	private @Nullable ScheduledExecutorService scheduleEviction(@Nullable Duration interval) {
+
+		if (interval == null || interval.isZero()) {
+			return null;
+		}
+		Assert.isTrue(!interval.isNegative(), "Eviction interval must not be negative");
+
+		ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(runnable -> {
+			Thread thread = new Thread(runnable, "elasticsearch-cache-expirer");
+			thread.setDaemon(true);
+			return thread;
+		});
+		long delay = Math.max(interval.toMillis(), 1L);
+		executor.scheduleWithFixedDelay(this::evictExpiredSafely, delay, delay, TimeUnit.MILLISECONDS);
+		return executor;
+	}
+
+	private void evictExpiredSafely() {
+
+		try {
+			evictExpired();
+		}
+		catch (RuntimeException exception) {
+			LOGGER.warn("Cannot evict expired Elasticsearch cache entries", exception);
+		}
+	}
+
+	/**
+	 * Builder for {@link ElasticsearchCacheManager} instances.
+	 *
+	 * @author Anıl Şenocak
+	 * @since 6.2
+	 */
+	public static final class Builder {
+
+		private final ElasticsearchOperations operations;
+		private String indexName = DEFAULT_INDEX_NAME;
+		private ElasticsearchCacheConfiguration defaultCacheConfiguration = ElasticsearchCacheConfiguration.defaultCacheConfig();
+		private final Map<String, ElasticsearchCacheConfiguration> initialCacheConfigurations = new LinkedHashMap<>();
+		private @Nullable ApplicationEventPublisher eventPublisher;
+		private @Nullable Duration evictionInterval;
+
+		private Builder(ElasticsearchOperations operations) {
+			Assert.notNull(operations, "ElasticsearchOperations must not be null");
+			this.operations = operations;
+		}
+
+		/**
+		 * Set the Elasticsearch index used to persist cache entries.
+		 */
+		public Builder indexName(String indexName) {
+
+			Assert.hasText(indexName, "Index name must not be empty");
+			this.indexName = indexName;
+			return this;
+		}
+
+		/**
+		 * Set the configuration applied to caches without a specific configuration.
+		 */
+		public Builder cacheDefaults(ElasticsearchCacheConfiguration defaultCacheConfiguration) {
+
+			Assert.notNull(defaultCacheConfiguration, "Default cache configuration must not be null");
+			this.defaultCacheConfiguration = defaultCacheConfiguration;
+			return this;
+		}
+
+		/**
+		 * Configure a named cache before it is first requested.
+		 */
+		public Builder withCacheConfiguration(String cacheName, ElasticsearchCacheConfiguration cacheConfiguration) {
+
+			Assert.hasText(cacheName, "Cache name must not be empty");
+			Assert.notNull(cacheConfiguration, "Cache configuration must not be null");
+			initialCacheConfigurations.put(cacheName, cacheConfiguration);
+			return this;
+		}
+
+		/**
+		 * Configure named caches before they are first requested.
+		 */
+		public Builder withInitialCacheConfigurations(
+				Map<String, ElasticsearchCacheConfiguration> initialCacheConfigurations) {
+
+			Assert.notNull(initialCacheConfigurations, "Initial cache configurations must not be null");
+			initialCacheConfigurations.forEach(this::withCacheConfiguration);
+			return this;
+		}
+
+		/**
+		 * Set the publisher that receives cache insertion and eviction events.
+		 */
+		public Builder applicationEventPublisher(ApplicationEventPublisher eventPublisher) {
+
+			Assert.notNull(eventPublisher, "ApplicationEventPublisher must not be null");
+			this.eventPublisher = eventPublisher;
+			return this;
+		}
+
+		/**
+		 * Remove expired entries at the given interval. A zero duration disables scheduled eviction.
+		 */
+		public Builder evictionInterval(Duration evictionInterval) {
+
+			Assert.notNull(evictionInterval, "Eviction interval must not be null");
+			Assert.isTrue(!evictionInterval.isNegative(), "Eviction interval must not be negative");
+			this.evictionInterval = evictionInterval;
+			return this;
+		}
+
+		/**
+		 * Build the cache manager.
+		 */
+		public ElasticsearchCacheManager build() {
+			return new ElasticsearchCacheManager(operations, indexName, defaultCacheConfiguration,
+					initialCacheConfigurations, eventPublisher, evictionInterval);
+		}
+	}
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/cache/event/CacheEvent.java b/src/main/java/org/springframework/data/elasticsearch/cache/event/CacheEvent.java
new file mode 100644
index 000000000..dc194a76f
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/cache/event/CacheEvent.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.elasticsearch.cache.event;
+
+import org.jspecify.annotations.Nullable;
+
+import java.time.Instant;
+
+/**
+ * A change to an Elasticsearch-backed cache entry.
+ *
+ * @author Anıl Şenocak
+ * @since 6.2
+ */
+public sealed interface CacheEvent permits CacheInsertedEvent, CacheEvictedEvent {
+
+	String getCacheName();
+
+	Object getKey();
+
+	@Nullable Object getValue();
+
+	Instant getOccurredAt();
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/cache/event/CacheEvictedEvent.java b/src/main/java/org/springframework/data/elasticsearch/cache/event/CacheEvictedEvent.java
new file mode 100644
index 000000000..f5e43637d
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/cache/event/CacheEvictedEvent.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.elasticsearch.cache.event;
+
+import org.jspecify.annotations.Nullable;
+import org.springframework.util.Assert;
+
+import java.time.Instant;
+
+/**
+ * Event published after an entry has been removed from an Elasticsearch-backed cache.
+ *
+ * @author Anıl Şenocak
+ * @since 6.2
+ */
+public final class CacheEvictedEvent implements CacheEvent {
+
+	private final String cacheName;
+	private final Object key;
+	private final @Nullable Object value;
+	private final Instant occurredAt;
+
+	public CacheEvictedEvent(String cacheName, Object key, @Nullable Object value) {
+		this(cacheName, key, value, Instant.now());
+	}
+
+	public CacheEvictedEvent(String cacheName, Object key, @Nullable Object value, Instant occurredAt) {
+
+		Assert.hasText(cacheName, "Cache name must not be empty");
+		Assert.notNull(key, "Key must not be null");
+		Assert.notNull(occurredAt, "OccurredAt must not be null");
+		this.cacheName = cacheName;
+		this.key = key;
+		this.value = value;
+		this.occurredAt = occurredAt;
+	}
+
+	@Override
+	public String getCacheName() {
+		return cacheName;
+	}
+
+	@Override
+	public Object getKey() {
+		return key;
+	}
+
+	@Override
+	public @Nullable Object getValue() {
+		return value;
+	}
+
+	@Override
+	public Instant getOccurredAt() {
+		return occurredAt;
+	}
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/cache/event/CacheInsertedEvent.java b/src/main/java/org/springframework/data/elasticsearch/cache/event/CacheInsertedEvent.java
new file mode 100644
index 000000000..a13c4e994
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/cache/event/CacheInsertedEvent.java
@@ -0,0 +1,77 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.elasticsearch.cache.event;
+
+import org.jspecify.annotations.Nullable;
+import org.springframework.util.Assert;
+
+import java.time.Instant;
+
+/**
+ * Event published after a value has been written to an Elasticsearch-backed cache.
+ *
+ * @author Anıl Şenocak
+ * @since 6.2
+ */
+public final class CacheInsertedEvent implements CacheEvent {
+
+	private final String cacheName;
+	private final Object key;
+	private final @Nullable Object value;
+	private final @Nullable Object previousValue;
+	private final Instant occurredAt;
+
+	public CacheInsertedEvent(String cacheName, Object key, @Nullable Object value, @Nullable Object previousValue) {
+		this(cacheName, key, value, previousValue, Instant.now());
+	}
+
+	public CacheInsertedEvent(String cacheName, Object key, @Nullable Object value, @Nullable Object previousValue,
+			Instant occurredAt) {
+
+		Assert.hasText(cacheName, "Cache name must not be empty");
+		Assert.notNull(key, "Key must not be null");
+		Assert.notNull(occurredAt, "OccurredAt must not be null");
+		this.cacheName = cacheName;
+		this.key = key;
+		this.value = value;
+		this.previousValue = previousValue;
+		this.occurredAt = occurredAt;
+	}
+
+	@Override
+	public String getCacheName() {
+		return cacheName;
+	}
+
+	@Override
+	public Object getKey() {
+		return key;
+	}
+
+	@Override
+	public @Nullable Object getValue() {
+		return value;
+	}
+
+	public @Nullable Object getPreviousValue() {
+		return previousValue;
+	}
+
+	@Override
+	public Instant getOccurredAt() {
+		return occurredAt;
+	}
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/cache/event/package-info.java b/src/main/java/org/springframework/data/elasticsearch/cache/event/package-info.java
new file mode 100644
index 000000000..b822d098b
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/cache/event/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.
+ */
+
+/**
+ * Events published by Elasticsearch-backed caches.
+ */
+@org.jspecify.annotations.NullMarked
+package org.springframework.data.elasticsearch.cache.event;
diff --git a/src/main/java/org/springframework/data/elasticsearch/cache/package-info.java b/src/main/java/org/springframework/data/elasticsearch/cache/package-info.java
new file mode 100644
index 000000000..7de67484c
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/cache/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.
+ */
+
+/**
+ * Spring Cache support backed by Elasticsearch.
+ */
+@org.jspecify.annotations.NullMarked
+package org.springframework.data.elasticsearch.cache;
diff --git a/src/test/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheELCIntegrationTests.java b/src/test/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheELCIntegrationTests.java
new file mode 100644
index 000000000..b225afbe5
--- /dev/null
+++ b/src/test/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheELCIntegrationTests.java
@@ -0,0 +1,136 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.elasticsearch.cache;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
+import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
+import org.springframework.data.elasticsearch.junit.jupiter.ElasticsearchTemplateConfiguration;
+import org.springframework.data.elasticsearch.junit.jupiter.SpringIntegrationTest;
+import org.springframework.test.context.ContextConfiguration;
+
+import java.time.Duration;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Anıl Şenocak
+ * @since 6.2
+ */
+@SpringIntegrationTest
+@ContextConfiguration(classes = ElasticsearchCacheELCIntegrationTests.Config.class)
+class ElasticsearchCacheELCIntegrationTests {
+
+	@Autowired private ElasticsearchOperations operations;
+
+	private String indexName;
+	private ElasticsearchCacheManager cacheManager;
+
+	@BeforeEach
+	void setUp() {
+		indexName = "spring-cache-it-" + UUID.randomUUID();
+		cacheManager = newCacheManager(ElasticsearchCacheConfiguration.defaultCacheConfig());
+	}
+
+	@AfterEach
+	void tearDown() {
+		cacheManager.close();
+		var indexOperations = operations.indexOps(IndexCoordinates.of(indexName));
+		if (indexOperations.exists()) {
+			indexOperations.delete();
+		}
+	}
+
+	@Test
+	void persistsEntriesAndReloadsThemWithANewCacheManager() {
+
+		CachedUser expected = new CachedUser("42", "Ada");
+		cacheManager.getCache("users").put(expected.id(), expected);
+
+		assertThat(cacheManager.getCache("users").get(expected.id(), CachedUser.class)).isEqualTo(expected);
+		try (ElasticsearchCacheManager reloadedCacheManager =
+				newCacheManager(ElasticsearchCacheConfiguration.defaultCacheConfig())) {
+			assertThat(reloadedCacheManager.getCache("users").get(expected.id(), CachedUser.class)).isEqualTo(expected);
+		}
+	}
+
+	@Test
+	void updatesExistingEntriesInElasticsearch() {
+
+		CachedUser original = new CachedUser("42", "Ada");
+		CachedUser updated = new CachedUser("42", "Grace");
+		ElasticsearchCache cache = cacheManager.getCache("users");
+		cache.put(original.id(), original);
+		cache.put(updated.id(), updated);
+
+		try (ElasticsearchCacheManager reloadedCacheManager =
+				newCacheManager(ElasticsearchCacheConfiguration.defaultCacheConfig())) {
+			assertThat(reloadedCacheManager.getCache("users").get(updated.id(), CachedUser.class)).isEqualTo(updated);
+		}
+	}
+
+	@Test
+	void evictsEntriesFromElasticsearch() {
+
+		CachedUser expected = new CachedUser("42", "Ada");
+		ElasticsearchCache cache = cacheManager.getCache("users");
+		cache.put(expected.id(), expected);
+
+		assertThat(cache.evictIfPresent(expected.id())).isTrue();
+		try (ElasticsearchCacheManager reloadedCacheManager =
+				newCacheManager(ElasticsearchCacheConfiguration.defaultCacheConfig())) {
+			assertThat(reloadedCacheManager.getCache("users").get(expected.id())).isNull();
+		}
+	}
+
+	@Test
+	void expiresEntriesAndRemovesThemFromElasticsearch() throws InterruptedException {
+
+		ElasticsearchCacheManager expiringCacheManager = newCacheManager(
+				ElasticsearchCacheConfiguration.defaultCacheConfig().entryTtl(Duration.ofMillis(10)));
+		try {
+			CachedUser expected = new CachedUser("42", "Ada");
+			expiringCacheManager.getCache("users").put(expected.id(), expected);
+			Thread.sleep(25);
+
+			assertThat(expiringCacheManager.getCache("users").get(expected.id())).isNull();
+			assertThat(expiringCacheManager.evictExpired()).isOne();
+			try (ElasticsearchCacheManager reloadedCacheManager =
+					newCacheManager(ElasticsearchCacheConfiguration.defaultCacheConfig())) {
+				assertThat(reloadedCacheManager.getCache("users").get(expected.id())).isNull();
+			}
+		}
+		finally {
+			expiringCacheManager.close();
+		}
+	}
+
+	private ElasticsearchCacheManager newCacheManager(ElasticsearchCacheConfiguration configuration) {
+		return ElasticsearchCacheManager.builder(operations).indexName(indexName).cacheDefaults(configuration).build();
+	}
+
+	@Configuration
+	@Import(ElasticsearchTemplateConfiguration.class)
+	static class Config {}
+
+	private record CachedUser(String id, String username) {}
+}
diff --git a/src/test/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheManagerTests.java b/src/test/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheManagerTests.java
new file mode 100644
index 000000000..7c0f96173
--- /dev/null
+++ b/src/test/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheManagerTests.java
@@ -0,0 +1,132 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.elasticsearch.cache;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.data.elasticsearch.cache.event.CacheEvictedEvent;
+import org.springframework.data.elasticsearch.cache.event.CacheInsertedEvent;
+
+import java.time.Duration;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Anıl Şenocak
+ */
+class ElasticsearchCacheManagerTests extends ElasticsearchCacheTestSupport {
+
+	@Test
+	void reusesCachesAndIncludesConfiguredCacheNames() {
+
+		ElasticsearchCacheManager manager = newManager(Map.of("expiring",
+				ElasticsearchCacheConfiguration.defaultCacheConfig().entryTtl(Duration.ofMinutes(1))));
+
+		ElasticsearchCache users = manager.getCache("users");
+
+		assertThat(manager.getCache("users")).isSameAs(users);
+		assertThat(manager.getCacheNames()).containsExactlyInAnyOrder("users", "expiring");
+	}
+
+	@Test
+	void clearsAllKnownCachesAndPublishesEvents() {
+
+		RecordingApplicationEventPublisher publisher = new RecordingApplicationEventPublisher();
+		ElasticsearchCacheManager manager = new ElasticsearchCacheManager(new InMemoryCacheDocumentStore(), newConverter(),
+				ElasticsearchCacheConfiguration.defaultCacheConfig(), Map.of(), publisher, null);
+		manager.getCache("users").put("42", new CachedUser("42", "Ada"));
+		manager.getCache("products").put("7", new CachedUser("7", "Grace"));
+
+		manager.clearAll();
+
+		assertThat(manager.getCache("users").get("42")).isNull();
+		assertThat(manager.getCache("products").get("7")).isNull();
+		assertThat(publisher.events).hasSize(4);
+		assertThat(publisher.events.subList(0, 2)).allMatch(CacheInsertedEvent.class::isInstance);
+		assertThat(publisher.events.subList(2, 4)).allMatch(CacheEvictedEvent.class::isInstance);
+	}
+
+	@Test
+	void appliesNamedCacheConfiguration() throws InterruptedException {
+
+		ElasticsearchCacheManager manager = newManager(Map.of("short-lived",
+				ElasticsearchCacheConfiguration.defaultCacheConfig().entryTtl(Duration.ofMillis(10))));
+		manager.getCache("short-lived").put("42", new CachedUser("42", "Ada"));
+		manager.getCache("default").put("7", new CachedUser("7", "Grace"));
+		Thread.sleep(25);
+
+		assertThat(manager.evictExpired()).isOne();
+		assertThat(manager.getCache("short-lived").get("42")).isNull();
+		assertThat(manager.getCache("default").get("7", CachedUser.class)).isEqualTo(new CachedUser("7", "Grace"));
+	}
+
+	@Test
+	void removesExpiredEntriesAcrossKnownCaches() throws InterruptedException {
+
+		ElasticsearchCacheConfiguration expiring = ElasticsearchCacheConfiguration.defaultCacheConfig()
+				.entryTtl(Duration.ofMillis(10));
+		ElasticsearchCacheManager manager = new ElasticsearchCacheManager(new InMemoryCacheDocumentStore(), newConverter(),
+				expiring, Map.of(), null, null);
+		manager.getCache("users").put("42", new CachedUser("42", "Ada"));
+		manager.getCache("products").put("7", new CachedUser("7", "Grace"));
+		Thread.sleep(25);
+
+		assertThat(manager.evictExpired()).isEqualTo(2);
+		assertThat(manager.getCache("users").get("42")).isNull();
+		assertThat(manager.getCache("products").get("7")).isNull();
+	}
+
+	@Test
+	void periodicallyRemovesExpiredEntries() throws InterruptedException {
+
+		ElasticsearchCacheManager manager = new ElasticsearchCacheManager(new InMemoryCacheDocumentStore(), newConverter(),
+				ElasticsearchCacheConfiguration.defaultCacheConfig().entryTtl(Duration.ofMillis(10)), Map.of(), null,
+				Duration.ofMillis(5));
+		try {
+			manager.getCache("users").put("42", new CachedUser("42", "Ada"));
+
+			assertThat(manager.isEvictionScheduled()).isTrue();
+			assertEventually(Duration.ofSeconds(1), () -> manager.getCache("users").get("42") == null);
+		}
+		finally {
+			manager.close();
+		}
+	}
+
+	private ElasticsearchCacheManager newManager(Map<String, ElasticsearchCacheConfiguration> configurations) {
+		return new ElasticsearchCacheManager(new InMemoryCacheDocumentStore(), newConverter(),
+				ElasticsearchCacheConfiguration.defaultCacheConfig(), configurations, null, null);
+	}
+
+	private void assertEventually(Duration timeout, Condition condition) throws InterruptedException {
+
+		long deadline = System.nanoTime() + timeout.toNanos();
+		while (System.nanoTime() < deadline) {
+			if (condition.evaluate()) {
+				return;
+			}
+			Thread.sleep(5);
+		}
+		assertThat(condition.evaluate()).isTrue();
+	}
+
+	@FunctionalInterface
+	private interface Condition {
+		boolean evaluate();
+	}
+
+	private record CachedUser(String id, String name) {}
+}
diff --git a/src/test/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheTestSupport.java b/src/test/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheTestSupport.java
new file mode 100644
index 000000000..8b71211cf
--- /dev/null
+++ b/src/test/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheTestSupport.java
@@ -0,0 +1,82 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.elasticsearch.cache;
+
+import org.jspecify.annotations.Nullable;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
+import org.springframework.data.elasticsearch.core.convert.MappingElasticsearchConverter;
+import org.springframework.data.elasticsearch.core.mapping.SimpleElasticsearchMappingContext;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author Anıl Şenocak
+ */
+abstract class ElasticsearchCacheTestSupport {
+
+	static ElasticsearchConverter newConverter() {
+
+		SimpleElasticsearchMappingContext mappingContext = new SimpleElasticsearchMappingContext();
+		mappingContext.afterPropertiesSet();
+		MappingElasticsearchConverter converter = new MappingElasticsearchConverter(mappingContext);
+		converter.afterPropertiesSet();
+		return converter;
+	}
+
+	static final class InMemoryCacheDocumentStore implements CacheDocumentStore {
+
+		private final Map<String, CacheDocument> documents = new LinkedHashMap<>();
+
+		@Override
+		public @Nullable CacheDocument get(String cacheName, String cacheKey) {
+			return documents.get(documentKey(cacheName, cacheKey));
+		}
+
+		@Override
+		public void put(CacheDocument document) {
+			documents.put(documentKey(document.getRequiredCacheName(), document.getRequiredCacheKey()), document);
+		}
+
+		@Override
+		public @Nullable CacheDocument delete(String cacheName, String cacheKey) {
+			return documents.remove(documentKey(cacheName, cacheKey));
+		}
+
+		@Override
+		public List<CacheDocument> findAll(String cacheName) {
+			return documents.values().stream()
+					.filter(document -> document.getRequiredCacheName().equals(cacheName)).toList();
+		}
+
+		private String documentKey(String cacheName, String cacheKey) {
+			return cacheName + ':' + cacheKey;
+		}
+	}
+
+	static final class RecordingApplicationEventPublisher implements ApplicationEventPublisher {
+
+		final List<Object> events = new ArrayList<>();
+
+		@Override
+		public void publishEvent(Object event) {
+			events.add(event);
+		}
+	}
+}
diff --git a/src/test/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheTests.java b/src/test/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheTests.java
new file mode 100644
index 000000000..6c0fbc6e6
--- /dev/null
+++ b/src/test/java/org/springframework/data/elasticsearch/cache/ElasticsearchCacheTests.java
@@ -0,0 +1,132 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.elasticsearch.cache;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.cache.Cache;
+import org.springframework.data.elasticsearch.cache.event.CacheEvictedEvent;
+import org.springframework.data.elasticsearch.cache.event.CacheInsertedEvent;
+
+import java.time.Duration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
+
+/**
+ * @author Anıl Şenocak
+ */
+class ElasticsearchCacheTests extends ElasticsearchCacheTestSupport {
+
+	@Test
+	void persistsAndReadsValuesWithTheConfiguredConverter() {
+
+		InMemoryCacheDocumentStore store = new InMemoryCacheDocumentStore();
+		ElasticsearchCache cache = newCache("users", store);
+		CachedUser user = new CachedUser("42", "Ada");
+
+		cache.put("42", user);
+
+		ElasticsearchCache reloadedCache = newCache("users", store);
+		assertThat(reloadedCache.get("42", CachedUser.class)).isEqualTo(user);
+		assertThat(reloadedCache.get("42").get()).isEqualTo(user);
+		assertThatIllegalStateException().isThrownBy(() -> reloadedCache.get("42", String.class));
+		assertThat(reloadedCache.getNativeCache()).isSameAs(store);
+	}
+
+	@Test
+	void storesAndReadsScalarValues() {
+
+		ElasticsearchCache cache = newCache("numbers", new InMemoryCacheDocumentStore());
+
+		cache.put("answer", 42);
+
+		assertThat(cache.get("answer", Integer.class)).isEqualTo(42);
+	}
+
+	@Test
+	void cachesNullValuesByDefaultAndCanRejectThem() {
+
+		ElasticsearchCache acceptingCache = newCache("accepting", new InMemoryCacheDocumentStore());
+		acceptingCache.put("nullable", null);
+
+		assertThat(acceptingCache.get("nullable")).isNotNull();
+		assertThat(acceptingCache.get("nullable").get()).isNull();
+
+		ElasticsearchCache rejectingCache = new ElasticsearchCache("rejecting", new InMemoryCacheDocumentStore(),
+				newConverter(), ElasticsearchCacheConfiguration.defaultCacheConfig().disableCachingNullValues(), null);
+		assertThatIllegalArgumentException().isThrownBy(() -> rejectingCache.put("nullable", null));
+	}
+
+	@Test
+	void wrapsValueLoaderFailures() {
+
+		ElasticsearchCache cache = newCache("users", new InMemoryCacheDocumentStore());
+
+		assertThatExceptionOfType(Cache.ValueRetrievalException.class)
+				.isThrownBy(() -> cache.get("42", () -> {
+					throw new IllegalStateException("boom");
+				}))
+				.withCauseInstanceOf(IllegalStateException.class)
+				.withMessageContaining("42");
+	}
+
+	@Test
+	void evictsExpiredEntriesAndPublishesEntryEvents() throws InterruptedException {
+
+		RecordingApplicationEventPublisher publisher = new RecordingApplicationEventPublisher();
+		ElasticsearchCache cache = new ElasticsearchCache("users", new InMemoryCacheDocumentStore(), newConverter(),
+				ElasticsearchCacheConfiguration.defaultCacheConfig().entryTtl(Duration.ofMillis(10)), publisher);
+		CachedUser user = new CachedUser("42", "Ada");
+
+		cache.put("42", user);
+		cache.put("42", new CachedUser("42", "Grace"));
+		Thread.sleep(25);
+
+		assertThat(cache.get("42")).isNull();
+		assertThat(cache.evictExpired()).isOne();
+		assertThat(cache.evictExpired()).isZero();
+		assertThat(publisher.events).hasSize(3);
+		CacheInsertedEvent inserted = (CacheInsertedEvent) publisher.events.get(0);
+		CacheInsertedEvent updated = (CacheInsertedEvent) publisher.events.get(1);
+		CacheEvictedEvent evicted = (CacheEvictedEvent) publisher.events.get(2);
+		assertThat(inserted.getValue()).isEqualTo(user);
+		assertThat(updated.getPreviousValue()).isEqualTo(user);
+		assertThat(evicted.getValue()).isEqualTo(new CachedUser("42", "Grace"));
+	}
+
+	@Test
+	void convertsKeysBeforePersistingThem() {
+
+		ElasticsearchCacheConfiguration configuration = ElasticsearchCacheConfiguration.defaultCacheConfig()
+				.serializeKeysWith(key -> ((CacheKey) key).value());
+		ElasticsearchCache cache = new ElasticsearchCache("users", new InMemoryCacheDocumentStore(), newConverter(),
+				configuration, null);
+
+		cache.put(new CacheKey("42"), new CachedUser("42", "Ada"));
+
+		assertThat(cache.get(new CacheKey("42"), CachedUser.class)).isEqualTo(new CachedUser("42", "Ada"));
+	}
+
+	private ElasticsearchCache newCache(String name, CacheDocumentStore store) {
+		return new ElasticsearchCache(name, store, newConverter(), ElasticsearchCacheConfiguration.defaultCacheConfig(), null);
+	}
+
+	private record CacheKey(String value) {}
+
+	private record CachedUser(String id, String name) {}
+}