From cfcfe21263a39a577d30bcc9e223bb113025075a Mon Sep 17 00:00:00 2001 From: Sachin Kale Date: Tue, 3 Sep 2024 06:32:54 +0530 Subject: [PATCH] Change pinned timestamp file format stored in remote store (#15526) --------- Signed-off-by: Sachin Kale Co-authored-by: Sachin Kale --- .../RemoteStorePinnedTimestampsIT.java | 91 ++++++- .../remote/model/RemotePinnedTimestamps.java | 144 ---------- .../RemoteStorePinnedTimestampsBlobStore.java | 43 --- .../index/remote/RemoteStoreUtils.java | 7 +- .../RemoteStorePinnedTimestampService.java | 250 ++++++++---------- .../model/RemotePinnedTimestampsTests.java | 101 ------- ...toreDirectoryWithPinnedTimestampTests.java | 38 +-- ...oteFsTranslogWithPinnedTimestampTests.java | 67 ++--- 8 files changed, 233 insertions(+), 508 deletions(-) delete mode 100644 server/src/main/java/org/opensearch/gateway/remote/model/RemotePinnedTimestamps.java delete mode 100644 server/src/main/java/org/opensearch/gateway/remote/model/RemoteStorePinnedTimestampsBlobStore.java delete mode 100644 server/src/test/java/org/opensearch/gateway/remote/model/RemotePinnedTimestampsTests.java diff --git a/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStorePinnedTimestampsIT.java b/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStorePinnedTimestampsIT.java index cb91c63e17245..2fcda8c2d2f27 100644 --- a/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStorePinnedTimestampsIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStorePinnedTimestampsIT.java @@ -8,6 +8,7 @@ package org.opensearch.remotestore; +import org.opensearch.action.LatchedActionListener; import org.opensearch.common.collect.Tuple; import org.opensearch.common.settings.Settings; import org.opensearch.common.unit.TimeValue; @@ -17,6 +18,7 @@ import org.opensearch.test.OpenSearchIntegTestCase; import java.util.Set; +import java.util.concurrent.CountDownLatch; @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0) public class RemoteStorePinnedTimestampsIT extends RemoteStoreBaseIntegTestCase { @@ -75,10 +77,25 @@ public void testTimestampPinUnpin() throws Exception { remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueMinutes(3)); - // This should be a no-op as pinning entity is different - remoteStorePinnedTimestampService.unpinTimestamp(timestamp1, "no-snapshot", noOpActionListener); // Unpinning already pinned entity remoteStorePinnedTimestampService.unpinTimestamp(timestamp2, "ss3", noOpActionListener); + + // This should fail as timestamp is not pinned by pinning entity + CountDownLatch latch = new CountDownLatch(1); + remoteStorePinnedTimestampService.unpinTimestamp(timestamp1, "no-snapshot", new LatchedActionListener<>(new ActionListener() { + @Override + public void onResponse(Void unused) { + // onResponse should not get called. + fail(); + } + + @Override + public void onFailure(Exception e) { + assertTrue(e instanceof IllegalArgumentException); + } + }, latch)); + latch.await(); + // Adding different entity to already pinned timestamp remoteStorePinnedTimestampService.pinTimestamp(timestamp3, "ss5", noOpActionListener); @@ -93,4 +110,74 @@ public void testTimestampPinUnpin() throws Exception { remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueMinutes(3)); } + + public void testPinnedTimestampClone() throws Exception { + prepareCluster(1, 1, INDEX_NAME, 0, 2); + ensureGreen(INDEX_NAME); + + RemoteStorePinnedTimestampService remoteStorePinnedTimestampService = internalCluster().getInstance( + RemoteStorePinnedTimestampService.class, + primaryNodeName(INDEX_NAME) + ); + + long timestamp1 = System.currentTimeMillis() + 30000L; + long timestamp2 = System.currentTimeMillis() + 60000L; + long timestamp3 = System.currentTimeMillis() + 900000L; + remoteStorePinnedTimestampService.pinTimestamp(timestamp1, "ss2", noOpActionListener); + remoteStorePinnedTimestampService.pinTimestamp(timestamp2, "ss3", noOpActionListener); + remoteStorePinnedTimestampService.pinTimestamp(timestamp3, "ss4", noOpActionListener); + + // Clone timestamp1 + remoteStorePinnedTimestampService.cloneTimestamp(timestamp1, "ss2", "ss2-2", noOpActionListener); + + // With clone, set of pinned timestamp will not change + remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueSeconds(1)); + assertBusy( + () -> assertEquals(Set.of(timestamp1, timestamp2, timestamp3), RemoteStorePinnedTimestampService.getPinnedTimestamps().v2()) + ); + remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueMinutes(3)); + + // Clone timestamp1 but provide invalid existing entity + CountDownLatch latch = new CountDownLatch(1); + remoteStorePinnedTimestampService.cloneTimestamp( + timestamp1, + "ss3", + "ss2-3", + new LatchedActionListener<>(new ActionListener() { + @Override + public void onResponse(Void unused) { + // onResponse should not get called. + fail(); + } + + @Override + public void onFailure(Exception e) { + assertTrue(e instanceof IllegalArgumentException); + } + }, latch) + ); + latch.await(); + + remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueSeconds(1)); + assertBusy( + () -> assertEquals(Set.of(timestamp1, timestamp2, timestamp3), RemoteStorePinnedTimestampService.getPinnedTimestamps().v2()) + ); + remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueMinutes(3)); + + // Now we have timestamp1 pinned by 2 entities, unpin 1, this should not change set of pinned timestamps + remoteStorePinnedTimestampService.unpinTimestamp(timestamp1, "ss2", noOpActionListener); + + remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueSeconds(1)); + assertBusy( + () -> assertEquals(Set.of(timestamp1, timestamp2, timestamp3), RemoteStorePinnedTimestampService.getPinnedTimestamps().v2()) + ); + remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueMinutes(3)); + + // Now unpin second entity as well, set of pinned timestamp should be reduced by 1 + remoteStorePinnedTimestampService.unpinTimestamp(timestamp1, "ss2-2", noOpActionListener); + + remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueSeconds(1)); + assertBusy(() -> assertEquals(Set.of(timestamp2, timestamp3), RemoteStorePinnedTimestampService.getPinnedTimestamps().v2())); + remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueMinutes(3)); + } } diff --git a/server/src/main/java/org/opensearch/gateway/remote/model/RemotePinnedTimestamps.java b/server/src/main/java/org/opensearch/gateway/remote/model/RemotePinnedTimestamps.java deleted file mode 100644 index 030491cf8b7b9..0000000000000 --- a/server/src/main/java/org/opensearch/gateway/remote/model/RemotePinnedTimestamps.java +++ /dev/null @@ -1,144 +0,0 @@ -/* - * SPDX-License-Identifier: Apache-2.0 - * - * The OpenSearch Contributors require contributions made to - * this file be licensed under the Apache-2.0 license or a - * compatible open source license. - */ - -package org.opensearch.gateway.remote.model; - -import org.apache.logging.log4j.LogManager; -import org.apache.logging.log4j.Logger; -import org.opensearch.common.io.Streams; -import org.opensearch.common.remote.BlobPathParameters; -import org.opensearch.common.remote.RemoteWriteableBlobEntity; -import org.opensearch.core.common.io.stream.StreamInput; -import org.opensearch.core.common.io.stream.StreamOutput; -import org.opensearch.core.common.io.stream.Writeable; -import org.opensearch.core.compress.Compressor; -import org.opensearch.index.remote.RemoteStoreUtils; -import org.opensearch.repositories.blobstore.ChecksumWritableBlobStoreFormat; - -import java.io.IOException; -import java.io.InputStream; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER; - -/** - * Wrapper class for uploading/downloading {@link RemotePinnedTimestamps} to/from remote blob store - * - * @opensearch.internal - */ -public class RemotePinnedTimestamps extends RemoteWriteableBlobEntity { - private static final Logger logger = LogManager.getLogger(RemotePinnedTimestamps.class); - - /** - * Represents a collection of pinned timestamps and their associated pinning entities. - * This class is thread-safe and implements the Writeable interface for serialization. - */ - public static class PinnedTimestamps implements Writeable { - private final Map> pinnedTimestampPinningEntityMap; - - public PinnedTimestamps(Map> pinnedTimestampPinningEntityMap) { - this.pinnedTimestampPinningEntityMap = new ConcurrentHashMap<>(pinnedTimestampPinningEntityMap); - } - - @Override - public void writeTo(StreamOutput out) throws IOException { - out.writeMap(pinnedTimestampPinningEntityMap, StreamOutput::writeLong, StreamOutput::writeStringCollection); - } - - public static PinnedTimestamps readFrom(StreamInput in) throws IOException { - return new PinnedTimestamps(in.readMap(StreamInput::readLong, StreamInput::readStringList)); - } - - /** - * Pins a timestamp against a pinning entity. - * - * @param timestamp The timestamp to pin. - * @param pinningEntity The entity pinning the timestamp. - */ - public void pin(Long timestamp, String pinningEntity) { - logger.debug("Pinning timestamp = {} against entity = {}", timestamp, pinningEntity); - pinnedTimestampPinningEntityMap.computeIfAbsent(timestamp, k -> new ArrayList<>()).add(pinningEntity); - } - - /** - * Unpins a timestamp for a specific pinning entity. - * - * @param timestamp The timestamp to unpin. - * @param pinningEntity The entity unpinning the timestamp. - */ - public void unpin(Long timestamp, String pinningEntity) { - logger.debug("Unpinning timestamp = {} against entity = {}", timestamp, pinningEntity); - if (pinnedTimestampPinningEntityMap.containsKey(timestamp) == false - || pinnedTimestampPinningEntityMap.get(timestamp).contains(pinningEntity) == false) { - logger.warn("Timestamp: {} is not pinned by entity: {}", timestamp, pinningEntity); - } - pinnedTimestampPinningEntityMap.compute(timestamp, (k, v) -> { - v.remove(pinningEntity); - return v.isEmpty() ? null : v; - }); - } - - public Map> getPinnedTimestampPinningEntityMap() { - return new HashMap<>(pinnedTimestampPinningEntityMap); - } - } - - public static final String PINNED_TIMESTAMPS = "pinned_timestamps"; - public static final ChecksumWritableBlobStoreFormat PINNED_TIMESTAMPS_FORMAT = new ChecksumWritableBlobStoreFormat<>( - PINNED_TIMESTAMPS, - PinnedTimestamps::readFrom - ); - - private PinnedTimestamps pinnedTimestamps; - - public RemotePinnedTimestamps(String clusterUUID, Compressor compressor) { - super(clusterUUID, compressor); - pinnedTimestamps = new PinnedTimestamps(new HashMap<>()); - } - - @Override - public BlobPathParameters getBlobPathParameters() { - return new BlobPathParameters(List.of(PINNED_TIMESTAMPS), PINNED_TIMESTAMPS); - } - - @Override - public String getType() { - return PINNED_TIMESTAMPS; - } - - @Override - public String generateBlobFileName() { - return this.blobFileName = String.join(DELIMITER, PINNED_TIMESTAMPS, RemoteStoreUtils.invertLong(System.currentTimeMillis())); - } - - @Override - public InputStream serialize() throws IOException { - return PINNED_TIMESTAMPS_FORMAT.serialize(pinnedTimestamps, generateBlobFileName(), getCompressor()).streamInput(); - } - - @Override - public PinnedTimestamps deserialize(InputStream inputStream) throws IOException { - return PINNED_TIMESTAMPS_FORMAT.deserialize(blobName, Streams.readFully(inputStream)); - } - - public void setBlobFileName(String blobFileName) { - this.blobFileName = blobFileName; - } - - public void setPinnedTimestamps(PinnedTimestamps pinnedTimestamps) { - this.pinnedTimestamps = pinnedTimestamps; - } - - public PinnedTimestamps getPinnedTimestamps() { - return pinnedTimestamps; - } -} diff --git a/server/src/main/java/org/opensearch/gateway/remote/model/RemoteStorePinnedTimestampsBlobStore.java b/server/src/main/java/org/opensearch/gateway/remote/model/RemoteStorePinnedTimestampsBlobStore.java deleted file mode 100644 index 2a65dd993d0af..0000000000000 --- a/server/src/main/java/org/opensearch/gateway/remote/model/RemoteStorePinnedTimestampsBlobStore.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * SPDX-License-Identifier: Apache-2.0 - * - * The OpenSearch Contributors require contributions made to - * this file be licensed under the Apache-2.0 license or a - * compatible open source license. - */ - -package org.opensearch.gateway.remote.model; - -import org.opensearch.common.blobstore.BlobPath; -import org.opensearch.common.remote.RemoteWriteableBlobEntity; -import org.opensearch.common.remote.RemoteWriteableEntityBlobStore; -import org.opensearch.index.translog.transfer.BlobStoreTransferService; -import org.opensearch.repositories.blobstore.BlobStoreRepository; -import org.opensearch.threadpool.ThreadPool; - -/** - * Extends the RemoteClusterStateBlobStore to support {@link RemotePinnedTimestamps} - */ -public class RemoteStorePinnedTimestampsBlobStore extends RemoteWriteableEntityBlobStore< - RemotePinnedTimestamps.PinnedTimestamps, - RemotePinnedTimestamps> { - - public static final String PINNED_TIMESTAMPS_PATH_TOKEN = "pinned_timestamps"; - private final BlobStoreRepository blobStoreRepository; - - public RemoteStorePinnedTimestampsBlobStore( - BlobStoreTransferService blobStoreTransferService, - BlobStoreRepository blobStoreRepository, - String clusterName, - ThreadPool threadPool, - String executor - ) { - super(blobStoreTransferService, blobStoreRepository, clusterName, threadPool, executor, PINNED_TIMESTAMPS_PATH_TOKEN); - this.blobStoreRepository = blobStoreRepository; - } - - @Override - public BlobPath getBlobPathForUpload(final RemoteWriteableBlobEntity obj) { - return blobStoreRepository.basePath().add(PINNED_TIMESTAMPS_PATH_TOKEN); - } -} diff --git a/server/src/main/java/org/opensearch/index/remote/RemoteStoreUtils.java b/server/src/main/java/org/opensearch/index/remote/RemoteStoreUtils.java index 871e2eb3ce47f..edcfe17d31fb8 100644 --- a/server/src/main/java/org/opensearch/index/remote/RemoteStoreUtils.java +++ b/server/src/main/java/org/opensearch/index/remote/RemoteStoreUtils.java @@ -542,8 +542,11 @@ public static List filterOutMetadataFilesBasedOnAge( if (RemoteStoreSettings.isPinnedTimestampsEnabled() == false) { return new ArrayList<>(metadataFiles); } - long maximumAllowedTimestamp = lastSuccessfulFetchOfPinnedTimestamps - RemoteStoreSettings.getPinnedTimestampsLookbackInterval() - .getMillis(); + // We allow now() - loopback interval to be pinned. Also, the actual pinning can take at most loopback interval + // This means the pinned timestamp can be available for read after at most (2 * loopback interval) + long maximumAllowedTimestamp = lastSuccessfulFetchOfPinnedTimestamps - (2 * RemoteStoreSettings + .getPinnedTimestampsLookbackInterval() + .getMillis()); List metadataFilesWithMinAge = new ArrayList<>(); for (String metadataFileName : metadataFiles) { long metadataTimestamp = getTimestampFunction.apply(metadataFileName); diff --git a/server/src/main/java/org/opensearch/node/remotestore/RemoteStorePinnedTimestampService.java b/server/src/main/java/org/opensearch/node/remotestore/RemoteStorePinnedTimestampService.java index f7b262664d147..782ba5e9a6540 100644 --- a/server/src/main/java/org/opensearch/node/remotestore/RemoteStorePinnedTimestampService.java +++ b/server/src/main/java/org/opensearch/node/remotestore/RemoteStorePinnedTimestampService.java @@ -10,19 +10,15 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; -import org.opensearch.cluster.ClusterName; +import org.apache.logging.log4j.message.ParameterizedMessage; import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.blobstore.BlobContainer; import org.opensearch.common.blobstore.BlobMetadata; -import org.opensearch.common.blobstore.BlobPath; import org.opensearch.common.collect.Tuple; import org.opensearch.common.settings.Settings; import org.opensearch.common.unit.TimeValue; import org.opensearch.common.util.concurrent.AbstractAsyncTask; import org.opensearch.core.action.ActionListener; -import org.opensearch.gateway.remote.model.RemotePinnedTimestamps; -import org.opensearch.gateway.remote.model.RemotePinnedTimestamps.PinnedTimestamps; -import org.opensearch.gateway.remote.model.RemoteStorePinnedTimestampsBlobStore; -import org.opensearch.index.translog.transfer.BlobStoreTransferService; import org.opensearch.indices.RemoteStoreSettings; import org.opensearch.node.Node; import org.opensearch.repositories.RepositoriesService; @@ -30,15 +26,13 @@ import org.opensearch.repositories.blobstore.BlobStoreRepository; import org.opensearch.threadpool.ThreadPool; +import java.io.ByteArrayInputStream; import java.io.Closeable; import java.io.IOException; -import java.util.HashMap; import java.util.List; +import java.util.Locale; +import java.util.Map; import java.util.Set; -import java.util.concurrent.Semaphore; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; -import java.util.function.Consumer; import java.util.function.Supplier; import java.util.stream.Collectors; @@ -51,17 +45,15 @@ public class RemoteStorePinnedTimestampService implements Closeable { private static final Logger logger = LogManager.getLogger(RemoteStorePinnedTimestampService.class); private static Tuple> pinnedTimestampsSet = new Tuple<>(-1L, Set.of()); - public static final int PINNED_TIMESTAMP_FILES_TO_KEEP = 5; + public static final String PINNED_TIMESTAMPS_PATH_TOKEN = "pinned_timestamps"; + public static final String PINNED_TIMESTAMPS_FILENAME_SEPARATOR = "__"; private final Supplier repositoriesService; private final Settings settings; private final ThreadPool threadPool; private final ClusterService clusterService; - private BlobStoreRepository blobStoreRepository; - private BlobStoreTransferService blobStoreTransferService; - private RemoteStorePinnedTimestampsBlobStore pinnedTimestampsBlobStore; + private BlobContainer blobContainer; private AsyncUpdatePinnedTimestampTask asyncUpdatePinnedTimestampTask; - private final Semaphore updateTimetampPinningSemaphore = new Semaphore(1); public RemoteStorePinnedTimestampService( Supplier repositoriesService, @@ -82,7 +74,6 @@ public RemoteStorePinnedTimestampService( */ public void start() { validateRemoteStoreConfiguration(); - initializeComponents(); startAsyncUpdateTask(RemoteStoreSettings.getPinnedTimestampsSchedulerInterval()); } @@ -93,19 +84,8 @@ private void validateRemoteStoreConfiguration() { assert remoteStoreRepo != null : "Remote Segment Store repository is not configured"; final Repository repository = repositoriesService.get().repository(remoteStoreRepo); assert repository instanceof BlobStoreRepository : "Repository should be instance of BlobStoreRepository"; - blobStoreRepository = (BlobStoreRepository) repository; - } - - private void initializeComponents() { - String clusterName = ClusterName.CLUSTER_NAME_SETTING.get(settings).value(); - blobStoreTransferService = new BlobStoreTransferService(blobStoreRepository.blobStore(), this.threadPool); - pinnedTimestampsBlobStore = new RemoteStorePinnedTimestampsBlobStore( - blobStoreTransferService, - blobStoreRepository, - clusterName, - this.threadPool, - ThreadPool.Names.REMOTE_STATE_READ - ); + BlobStoreRepository blobStoreRepository = (BlobStoreRepository) repository; + blobContainer = blobStoreRepository.blobStore().blobContainer(blobStoreRepository.basePath().add(PINNED_TIMESTAMPS_PATH_TOKEN)); } private void startAsyncUpdateTask(TimeValue pinnedTimestampsSchedulerInterval) { @@ -129,101 +109,109 @@ public void pinTimestamp(long timestamp, String pinningEntity, ActionListener pinnedTimestamps.pin(timestamp, pinningEntity), listener); + long startTime = System.nanoTime(); + try { + logger.debug("Pinning timestamp = {} against entity = {}", timestamp, pinningEntity); + blobContainer.writeBlob(getBlobName(timestamp, pinningEntity), new ByteArrayInputStream(new byte[0]), 0, true); + long elapsedTime = System.nanoTime() - startTime; + if (elapsedTime > RemoteStoreSettings.getPinnedTimestampsLookbackInterval().nanos()) { + String errorMessage = String.format( + Locale.ROOT, + "Timestamp pinning took %s nanoseconds which is more than limit of %s nanoseconds, failing the operation", + elapsedTime, + RemoteStoreSettings.getPinnedTimestampsLookbackInterval().nanos() + ); + logger.error(errorMessage); + unpinTimestamp(timestamp, pinningEntity, ActionListener.wrap(() -> listener.onFailure(new RuntimeException(errorMessage)))); + } else { + listener.onResponse(null); + } + } catch (IOException e) { + listener.onFailure(e); + } } /** - * Unpins a timestamp from the remote store. + * Clones a timestamp by creating a new pinning entity for an existing timestamp. * - * @param timestamp The timestamp to be unpinned - * @param pinningEntity The entity responsible for unpinning the timestamp - * @param listener A listener to be notified when the unpinning operation completes + * This method attempts to create a new pinning entity for a given timestamp that is already + * associated with an existing pinning entity. If the timestamp exists for the existing entity, + * a new blob is created for the new pinning entity. If the timestamp doesn't exist for the + * existing entity, the operation fails with an IllegalArgumentException. + * + * @param timestamp The timestamp to be cloned. + * @param existingPinningEntity The name of the existing entity that has pinned the timestamp. + * @param newPinningEntity The name of the new entity to pin the timestamp to. + * @param listener An ActionListener that will be notified of the operation's success or failure. + * On success, onResponse will be called with null. On failure, onFailure will + * be called with the appropriate exception. */ - public void unpinTimestamp(long timestamp, String pinningEntity, ActionListener listener) { - updatePinning(pinnedTimestamps -> pinnedTimestamps.unpin(timestamp, pinningEntity), listener); - } - - private void updatePinning(Consumer updateConsumer, ActionListener listener) { - RemotePinnedTimestamps remotePinnedTimestamps = new RemotePinnedTimestamps( - clusterService.state().metadata().clusterUUID(), - blobStoreRepository.getCompressor() - ); - BlobPath path = pinnedTimestampsBlobStore.getBlobPathForUpload(remotePinnedTimestamps); + public void cloneTimestamp(long timestamp, String existingPinningEntity, String newPinningEntity, ActionListener listener) { try { - if (updateTimetampPinningSemaphore.tryAcquire(10, TimeUnit.MINUTES)) { - ActionListener semaphoreAwareListener = ActionListener.runBefore(listener, updateTimetampPinningSemaphore::release); - ActionListener> listCallResponseListener = getListenerForListCallResponse( - remotePinnedTimestamps, - updateConsumer, - semaphoreAwareListener - ); - blobStoreTransferService.listAllInSortedOrder( - path, - remotePinnedTimestamps.getType(), - Integer.MAX_VALUE, - listCallResponseListener - ); + logger.debug( + "cloning timestamp = {} with existing pinningEntity = {} with new pinningEntity = {}", + timestamp, + existingPinningEntity, + newPinningEntity + ); + String blobName = getBlobName(timestamp, existingPinningEntity); + if (blobContainer.blobExists(blobName)) { + logger.debug("Pinning timestamp = {} against entity = {}", timestamp, newPinningEntity); + blobContainer.writeBlob(getBlobName(timestamp, newPinningEntity), new ByteArrayInputStream(new byte[0]), 0, true); + listener.onResponse(null); } else { - throw new TimeoutException("Timed out while waiting to acquire lock in updatePinning"); + String errorMessage = String.format( + Locale.ROOT, + "Timestamp: %s is not pinned by existing entity: %s", + timestamp, + existingPinningEntity + ); + logger.error(errorMessage); + listener.onFailure(new IllegalArgumentException(errorMessage)); } - } catch (InterruptedException | TimeoutException e) { + } catch (IOException e) { listener.onFailure(e); } } - private ActionListener> getListenerForListCallResponse( - RemotePinnedTimestamps remotePinnedTimestamps, - Consumer updateConsumer, - ActionListener listener - ) { - return ActionListener.wrap(blobMetadata -> { - PinnedTimestamps pinnedTimestamps = new PinnedTimestamps(new HashMap<>()); - if (blobMetadata.isEmpty() == false) { - pinnedTimestamps = readExistingPinnedTimestamps(blobMetadata.get(0).name(), remotePinnedTimestamps); - } - updateConsumer.accept(pinnedTimestamps); - remotePinnedTimestamps.setPinnedTimestamps(pinnedTimestamps); - ActionListener writeCallResponseListener = getListenerForWriteCallResponse( - remotePinnedTimestamps, - blobMetadata, - listener - ); - pinnedTimestampsBlobStore.writeAsync(remotePinnedTimestamps, writeCallResponseListener); - }, listener::onFailure); + private String getBlobName(long timestamp, String pinningEntity) { + return String.join(PINNED_TIMESTAMPS_FILENAME_SEPARATOR, pinningEntity, String.valueOf(timestamp)); } - private ActionListener getListenerForWriteCallResponse( - RemotePinnedTimestamps remotePinnedTimestamps, - List blobMetadata, - ActionListener listener - ) { - return ActionListener.wrap(unused -> { - // Delete older pinnedTimestamp files - if (blobMetadata.size() > PINNED_TIMESTAMP_FILES_TO_KEEP) { - List oldFilesToBeDeleted = blobMetadata.subList(PINNED_TIMESTAMP_FILES_TO_KEEP, blobMetadata.size()) - .stream() - .map(BlobMetadata::name) - .collect(Collectors.toList()); - try { - blobStoreTransferService.deleteBlobs( - pinnedTimestampsBlobStore.getBlobPathForUpload(remotePinnedTimestamps), - oldFilesToBeDeleted - ); - } catch (IOException e) { - logger.error("Exception while deleting stale pinned timestamps", e); - } - } - listener.onResponse(null); - }, listener::onFailure); + private long getTimestampFromBlobName(String blobName) { + String[] blobNameTokens = blobName.split(PINNED_TIMESTAMPS_FILENAME_SEPARATOR); + if (blobNameTokens.length < 2) { + logger.error("Pinned timestamps blob name contains invalid format: {}", blobName); + } + try { + return Long.parseLong(blobNameTokens[blobNameTokens.length - 1]); + } catch (NumberFormatException e) { + logger.error(() -> new ParameterizedMessage("Pinned timestamps blob name contains invalid format: {}", blobName), e); + } + return -1; } - private PinnedTimestamps readExistingPinnedTimestamps(String blobFilename, RemotePinnedTimestamps remotePinnedTimestamps) { - remotePinnedTimestamps.setBlobFileName(blobFilename); - remotePinnedTimestamps.setFullBlobName(pinnedTimestampsBlobStore().getBlobPathForUpload(remotePinnedTimestamps)); + /** + * Unpins a timestamp from the remote store. + * + * @param timestamp The timestamp to be unpinned + * @param pinningEntity The entity responsible for unpinning the timestamp + * @param listener A listener to be notified when the unpinning operation completes + */ + public void unpinTimestamp(long timestamp, String pinningEntity, ActionListener listener) { try { - return pinnedTimestampsBlobStore().read(remotePinnedTimestamps); + logger.debug("Unpinning timestamp = {} against entity = {}", timestamp, pinningEntity); + String blobName = getBlobName(timestamp, pinningEntity); + if (blobContainer.blobExists(blobName)) { + blobContainer.deleteBlobsIgnoringIfNotExists(List.of(blobName)); + listener.onResponse(null); + } else { + String errorMessage = String.format(Locale.ROOT, "Timestamp: %s is not pinned by entity: %s", timestamp, pinningEntity); + logger.error(errorMessage); + listener.onFailure(new IllegalArgumentException(errorMessage)); + } } catch (IOException e) { - throw new RuntimeException("Failed to read existing pinned timestamps", e); + listener.onFailure(e); } } @@ -245,14 +233,6 @@ public static Tuple> getPinnedTimestamps() { return pinnedTimestampsSet; } - public RemoteStorePinnedTimestampsBlobStore pinnedTimestampsBlobStore() { - return pinnedTimestampsBlobStore; - } - - public BlobStoreTransferService blobStoreTransferService() { - return blobStoreTransferService; - } - /** * Inner class for asynchronously updating the pinned timestamp set. */ @@ -270,32 +250,22 @@ protected boolean mustReschedule() { @Override protected void runInternal() { long triggerTimestamp = System.currentTimeMillis(); - RemotePinnedTimestamps remotePinnedTimestamps = new RemotePinnedTimestamps( - clusterService.state().metadata().clusterUUID(), - blobStoreRepository.getCompressor() - ); - BlobPath path = pinnedTimestampsBlobStore().getBlobPathForUpload(remotePinnedTimestamps); - blobStoreTransferService().listAllInSortedOrder(path, remotePinnedTimestamps.getType(), 1, new ActionListener<>() { - @Override - public void onResponse(List blobMetadata) { - if (blobMetadata.isEmpty()) { - pinnedTimestampsSet = new Tuple<>(triggerTimestamp, Set.of()); - return; - } - PinnedTimestamps pinnedTimestamps = readExistingPinnedTimestamps(blobMetadata.get(0).name(), remotePinnedTimestamps); - logger.debug( - "Fetched pinned timestamps from remote store: {} - {}", - triggerTimestamp, - pinnedTimestamps.getPinnedTimestampPinningEntityMap().keySet() - ); - pinnedTimestampsSet = new Tuple<>(triggerTimestamp, pinnedTimestamps.getPinnedTimestampPinningEntityMap().keySet()); - } - - @Override - public void onFailure(Exception e) { - logger.error("Exception while listing pinned timestamp files", e); + try { + Map pinnedTimestampList = blobContainer.listBlobs(); + if (pinnedTimestampList.isEmpty()) { + pinnedTimestampsSet = new Tuple<>(triggerTimestamp, Set.of()); + return; } - }); + Set pinnedTimestamps = pinnedTimestampList.keySet() + .stream() + .map(RemoteStorePinnedTimestampService.this::getTimestampFromBlobName) + .filter(timestamp -> timestamp != -1) + .collect(Collectors.toSet()); + logger.debug("Fetched pinned timestamps from remote store: {} - {}", triggerTimestamp, pinnedTimestamps); + pinnedTimestampsSet = new Tuple<>(triggerTimestamp, pinnedTimestamps); + } catch (Throwable t) { + logger.error("Exception while fetching pinned timestamp details", t); + } } } } diff --git a/server/src/test/java/org/opensearch/gateway/remote/model/RemotePinnedTimestampsTests.java b/server/src/test/java/org/opensearch/gateway/remote/model/RemotePinnedTimestampsTests.java deleted file mode 100644 index 309263a634265..0000000000000 --- a/server/src/test/java/org/opensearch/gateway/remote/model/RemotePinnedTimestampsTests.java +++ /dev/null @@ -1,101 +0,0 @@ -/* - * SPDX-License-Identifier: Apache-2.0 - * - * The OpenSearch Contributors require contributions made to - * this file be licensed under the Apache-2.0 license or a - * compatible open source license. - */ - -package org.opensearch.gateway.remote.model; - -import org.opensearch.common.compress.DeflateCompressor; -import org.opensearch.common.io.stream.BytesStreamOutput; -import org.opensearch.core.common.io.stream.BytesStreamInput; -import org.opensearch.core.common.io.stream.StreamInput; -import org.opensearch.core.compress.Compressor; -import org.opensearch.test.OpenSearchTestCase; -import org.junit.Before; - -import java.io.IOException; -import java.io.InputStream; -import java.util.Arrays; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -public class RemotePinnedTimestampsTests extends OpenSearchTestCase { - - private RemotePinnedTimestamps remotePinnedTimestamps; - - @Before - public void setup() { - Compressor compressor = new DeflateCompressor(); - remotePinnedTimestamps = new RemotePinnedTimestamps("testClusterUUID", compressor); - } - - public void testGenerateBlobFileName() { - String fileName = remotePinnedTimestamps.generateBlobFileName(); - assertTrue(fileName.startsWith(RemotePinnedTimestamps.PINNED_TIMESTAMPS)); - assertEquals(fileName, remotePinnedTimestamps.getBlobFileName()); - } - - public void testSerializeAndDeserialize() throws IOException { - RemotePinnedTimestamps.PinnedTimestamps pinnedTimestamps = new RemotePinnedTimestamps.PinnedTimestamps(new HashMap<>()); - pinnedTimestamps.pin(1000L, "entity1"); - pinnedTimestamps.pin(2000L, "entity2"); - remotePinnedTimestamps.setPinnedTimestamps(pinnedTimestamps); - - InputStream serialized = remotePinnedTimestamps.serialize(); - RemotePinnedTimestamps.PinnedTimestamps deserialized = remotePinnedTimestamps.deserialize(serialized); - - assertEquals(pinnedTimestamps.getPinnedTimestampPinningEntityMap(), deserialized.getPinnedTimestampPinningEntityMap()); - } - - public void testSetAndGetPinnedTimestamps() { - RemotePinnedTimestamps.PinnedTimestamps pinnedTimestamps = new RemotePinnedTimestamps.PinnedTimestamps(new HashMap<>()); - remotePinnedTimestamps.setPinnedTimestamps(pinnedTimestamps); - assertEquals(pinnedTimestamps, remotePinnedTimestamps.getPinnedTimestamps()); - } - - public void testPinnedTimestampsPin() { - RemotePinnedTimestamps.PinnedTimestamps pinnedTimestamps = new RemotePinnedTimestamps.PinnedTimestamps(new HashMap<>()); - pinnedTimestamps.pin(1000L, "entity1"); - pinnedTimestamps.pin(1000L, "entity2"); - pinnedTimestamps.pin(2000L, "entity3"); - - Map> expected = new HashMap<>(); - expected.put(1000L, Arrays.asList("entity1", "entity2")); - expected.put(2000L, List.of("entity3")); - - assertEquals(expected, pinnedTimestamps.getPinnedTimestampPinningEntityMap()); - } - - public void testPinnedTimestampsUnpin() { - RemotePinnedTimestamps.PinnedTimestamps pinnedTimestamps = new RemotePinnedTimestamps.PinnedTimestamps(new HashMap<>()); - pinnedTimestamps.pin(1000L, "entity1"); - pinnedTimestamps.pin(1000L, "entity2"); - pinnedTimestamps.pin(2000L, "entity3"); - - pinnedTimestamps.unpin(1000L, "entity1"); - pinnedTimestamps.unpin(2000L, "entity3"); - - Map> expected = new HashMap<>(); - expected.put(1000L, List.of("entity2")); - - assertEquals(expected, pinnedTimestamps.getPinnedTimestampPinningEntityMap()); - } - - public void testPinnedTimestampsReadFromAndWriteTo() throws IOException { - RemotePinnedTimestamps.PinnedTimestamps original = new RemotePinnedTimestamps.PinnedTimestamps(new HashMap<>()); - original.pin(1000L, "entity1"); - original.pin(2000L, "entity2"); - - BytesStreamOutput out = new BytesStreamOutput(); - original.writeTo(out); - - StreamInput in = new BytesStreamInput(out.bytes().toBytesRef().bytes); - RemotePinnedTimestamps.PinnedTimestamps deserialized = RemotePinnedTimestamps.PinnedTimestamps.readFrom(in); - - assertEquals(original.getPinnedTimestampPinningEntityMap(), deserialized.getPinnedTimestampPinningEntityMap()); - } -} diff --git a/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryWithPinnedTimestampTests.java b/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryWithPinnedTimestampTests.java index 107d59aa97549..e71023125d4cd 100644 --- a/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryWithPinnedTimestampTests.java +++ b/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryWithPinnedTimestampTests.java @@ -8,15 +8,12 @@ package org.opensearch.index.store; -import org.opensearch.common.blobstore.BlobMetadata; +import org.opensearch.common.blobstore.BlobContainer; import org.opensearch.common.blobstore.BlobPath; +import org.opensearch.common.blobstore.BlobStore; import org.opensearch.common.blobstore.support.PlainBlobMetadata; import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; -import org.opensearch.core.action.ActionListener; -import org.opensearch.gateway.remote.model.RemotePinnedTimestamps; -import org.opensearch.gateway.remote.model.RemoteStorePinnedTimestampsBlobStore; -import org.opensearch.index.translog.transfer.BlobStoreTransferService; import org.opensearch.indices.RemoteStoreSettings; import org.opensearch.node.Node; import org.opensearch.node.remotestore.RemoteStoreNodeAttribute; @@ -26,7 +23,7 @@ import org.junit.Before; import java.io.IOException; -import java.util.ArrayList; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.function.Supplier; @@ -38,7 +35,6 @@ import static org.hamcrest.CoreMatchers.is; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.anyInt; -import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; @@ -48,8 +44,7 @@ public class RemoteSegmentStoreDirectoryWithPinnedTimestampTests extends RemoteSegmentStoreDirectoryTests { Runnable updatePinnedTimstampTask; - BlobStoreTransferService blobStoreTransferService; - RemoteStorePinnedTimestampsBlobStore remoteStorePinnedTimestampsBlobStore; + BlobContainer blobContainer; RemoteStorePinnedTimestampService remoteStorePinnedTimestampServiceSpy; @Before @@ -82,16 +77,13 @@ public void setupPinnedTimestamp() throws IOException { ); remoteStorePinnedTimestampServiceSpy = Mockito.spy(remoteStorePinnedTimestampService); - remoteStorePinnedTimestampsBlobStore = mock(RemoteStorePinnedTimestampsBlobStore.class); - blobStoreTransferService = mock(BlobStoreTransferService.class); - when(remoteStorePinnedTimestampServiceSpy.pinnedTimestampsBlobStore()).thenReturn(remoteStorePinnedTimestampsBlobStore); - when(remoteStorePinnedTimestampServiceSpy.blobStoreTransferService()).thenReturn(blobStoreTransferService); + BlobStore blobStore = mock(BlobStore.class); + when(blobStoreRepository.blobStore()).thenReturn(blobStore); + when(blobStoreRepository.basePath()).thenReturn(new BlobPath()); + blobContainer = mock(BlobContainer.class); + when(blobStore.blobContainer(any())).thenReturn(blobContainer); - doAnswer(invocationOnMock -> { - ActionListener> actionListener = invocationOnMock.getArgument(3); - actionListener.onResponse(new ArrayList<>()); - return null; - }).when(blobStoreTransferService).listAllInSortedOrder(any(), any(), eq(1), any()); + when(blobContainer.listBlobs()).thenReturn(new HashMap<>()); remoteStorePinnedTimestampServiceSpy.start(); @@ -195,15 +187,9 @@ public void testDeleteStaleCommitsPinnedTimestampMdFile() throws Exception { ) ).thenReturn(List.of(metadataFilename, metadataFilename2, metadataFilename3)); - doAnswer(invocationOnMock -> { - ActionListener> actionListener = invocationOnMock.getArgument(3); - actionListener.onResponse(List.of(new PlainBlobMetadata("pinned_timestamp_123", 1000))); - return null; - }).when(blobStoreTransferService).listAllInSortedOrder(any(), any(), eq(1), any()); - long pinnedTimestampMatchingMetadataFilename2 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getTimestamp(metadataFilename2) + 10; - when(remoteStorePinnedTimestampsBlobStore.read(any())).thenReturn(new RemotePinnedTimestamps.PinnedTimestamps(Map.of(pinnedTimestampMatchingMetadataFilename2, List.of("xyz")))); - when(remoteStorePinnedTimestampsBlobStore.getBlobPathForUpload(any())).thenReturn(new BlobPath()); + String blobName = "snapshot1__" + pinnedTimestampMatchingMetadataFilename2; + when(blobContainer.listBlobs()).thenReturn(Map.of(blobName, new PlainBlobMetadata(blobName, 100))); final Map> metadataFilenameContentMapping = populateMetadata(); final List filesToBeDeleted = metadataFilenameContentMapping.get(metadataFilename3) diff --git a/server/src/test/java/org/opensearch/index/translog/RemoteFsTranslogWithPinnedTimestampTests.java b/server/src/test/java/org/opensearch/index/translog/RemoteFsTranslogWithPinnedTimestampTests.java index e35ab6bcd14c3..386dde4dffc48 100644 --- a/server/src/test/java/org/opensearch/index/translog/RemoteFsTranslogWithPinnedTimestampTests.java +++ b/server/src/test/java/org/opensearch/index/translog/RemoteFsTranslogWithPinnedTimestampTests.java @@ -12,8 +12,10 @@ import org.opensearch.action.LatchedActionListener; import org.opensearch.cluster.metadata.RepositoryMetadata; import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.blobstore.BlobContainer; import org.opensearch.common.blobstore.BlobMetadata; import org.opensearch.common.blobstore.BlobPath; +import org.opensearch.common.blobstore.BlobStore; import org.opensearch.common.blobstore.support.PlainBlobMetadata; import org.opensearch.common.collect.Tuple; import org.opensearch.common.lease.Releasable; @@ -22,12 +24,9 @@ import org.opensearch.common.settings.Settings; import org.opensearch.common.unit.TimeValue; import org.opensearch.core.action.ActionListener; -import org.opensearch.gateway.remote.model.RemotePinnedTimestamps; -import org.opensearch.gateway.remote.model.RemoteStorePinnedTimestampsBlobStore; import org.opensearch.index.remote.RemoteStoreUtils; import org.opensearch.index.remote.RemoteTranslogTransferTracker; import org.opensearch.index.seqno.SequenceNumbers; -import org.opensearch.index.translog.transfer.BlobStoreTransferService; import org.opensearch.index.translog.transfer.TranslogTransferManager; import org.opensearch.index.translog.transfer.TranslogTransferMetadata; import org.opensearch.indices.DefaultRemoteStoreSettings; @@ -63,8 +62,6 @@ import static org.opensearch.indices.RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_ENABLED; import static org.junit.Assert.fail; import static org.mockito.ArgumentMatchers.any; -import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -73,8 +70,7 @@ public class RemoteFsTranslogWithPinnedTimestampTests extends RemoteFsTranslogTests { Runnable updatePinnedTimstampTask; - BlobStoreTransferService pinnedTimestampBlobStoreTransferService; - RemoteStorePinnedTimestampsBlobStore remoteStorePinnedTimestampsBlobStore; + BlobContainer blobContainer; RemoteStorePinnedTimestampService remoteStorePinnedTimestampServiceSpy; @Before @@ -118,16 +114,13 @@ public void setUp() throws Exception { ); remoteStorePinnedTimestampServiceSpy = Mockito.spy(remoteStorePinnedTimestampService); - remoteStorePinnedTimestampsBlobStore = mock(RemoteStorePinnedTimestampsBlobStore.class); - pinnedTimestampBlobStoreTransferService = mock(BlobStoreTransferService.class); - when(remoteStorePinnedTimestampServiceSpy.pinnedTimestampsBlobStore()).thenReturn(remoteStorePinnedTimestampsBlobStore); - when(remoteStorePinnedTimestampServiceSpy.blobStoreTransferService()).thenReturn(pinnedTimestampBlobStoreTransferService); + BlobStore blobStore = mock(BlobStore.class); + when(blobStoreRepository.blobStore()).thenReturn(blobStore); + when(blobStoreRepository.basePath()).thenReturn(new BlobPath()); + blobContainer = mock(BlobContainer.class); + when(blobStore.blobContainer(any())).thenReturn(blobContainer); - doAnswer(invocationOnMock -> { - ActionListener> actionListener = invocationOnMock.getArgument(3); - actionListener.onResponse(new ArrayList<>()); - return null; - }).when(pinnedTimestampBlobStoreTransferService).listAllInSortedOrder(any(), any(), eq(1), any()); + when(blobContainer.listBlobs()).thenReturn(new HashMap<>()); remoteStorePinnedTimestampServiceSpy.start(); } @@ -362,20 +355,12 @@ public void onResponse(List blobMetadataList) { } } - doAnswer(invocationOnMock -> { - ActionListener> actionListener = invocationOnMock.getArgument(3); - actionListener.onResponse(List.of(new PlainBlobMetadata("pinned_timestamp_123", 1000))); - return null; - }).when(pinnedTimestampBlobStoreTransferService).listAllInSortedOrder(any(), any(), eq(1), any()); - - Map> pinnedTimestampsMap = new HashMap<>(); - pinnedTimestamps.forEach(ts -> pinnedTimestampsMap.put(ts, new ArrayList<>())); + Map pinnedTimestampsMap = new HashMap<>(); + pinnedTimestamps.forEach(ts -> pinnedTimestampsMap.put(randomInt(1000) + "__" + ts, new PlainBlobMetadata("x", 100))); try { - when(remoteStorePinnedTimestampsBlobStore.read(any())).thenReturn( - new RemotePinnedTimestamps.PinnedTimestamps(pinnedTimestampsMap) - ); - when(remoteStorePinnedTimestampsBlobStore.getBlobPathForUpload(any())).thenReturn(new BlobPath()); + + when(blobContainer.listBlobs()).thenReturn(pinnedTimestampsMap); Set dataFilesBeforeTrim = blobStoreTransferService.listAll( getTranslogDirectory().add(DATA_DIR).add(String.valueOf(primaryTerm.get())) @@ -642,7 +627,7 @@ public void testGetMetadataFilesToBeDeletedNoExclusion() { public void testGetMetadataFilesToBeDeletedExclusionBasedOnAgeOnly() { updatePinnedTimstampTask.run(); long currentTimeInMillis = System.currentTimeMillis(); - String md1Timestamp = RemoteStoreUtils.invertLong(currentTimeInMillis - 100000); + String md1Timestamp = RemoteStoreUtils.invertLong(currentTimeInMillis - 200000); String md2Timestamp = RemoteStoreUtils.invertLong(currentTimeInMillis + 30000); String md3Timestamp = RemoteStoreUtils.invertLong(currentTimeInMillis + 60000); @@ -659,21 +644,12 @@ public void testGetMetadataFilesToBeDeletedExclusionBasedOnAgeOnly() { public void testGetMetadataFilesToBeDeletedExclusionBasedOnPinningOnly() throws IOException { long currentTimeInMillis = System.currentTimeMillis(); - String md1Timestamp = RemoteStoreUtils.invertLong(currentTimeInMillis - 100000); + String md1Timestamp = RemoteStoreUtils.invertLong(currentTimeInMillis - 200000); String md2Timestamp = RemoteStoreUtils.invertLong(currentTimeInMillis - 300000); String md3Timestamp = RemoteStoreUtils.invertLong(currentTimeInMillis - 600000); - doAnswer(invocationOnMock -> { - ActionListener> actionListener = invocationOnMock.getArgument(3); - actionListener.onResponse(List.of(new PlainBlobMetadata("pinned_timestamp_123", 1000))); - return null; - }).when(pinnedTimestampBlobStoreTransferService).listAllInSortedOrder(any(), any(), eq(1), any()); - long pinnedTimestamp = RemoteStoreUtils.invertLong(md2Timestamp) + 10000; - when(remoteStorePinnedTimestampsBlobStore.read(any())).thenReturn( - new RemotePinnedTimestamps.PinnedTimestamps(Map.of(pinnedTimestamp, List.of("xyz"))) - ); - when(remoteStorePinnedTimestampsBlobStore.getBlobPathForUpload(any())).thenReturn(new BlobPath()); + when(blobContainer.listBlobs()).thenReturn(Map.of(randomInt(100) + "__" + pinnedTimestamp, new PlainBlobMetadata("xyz", 100))); updatePinnedTimstampTask.run(); @@ -695,17 +671,8 @@ public void testGetMetadataFilesToBeDeletedExclusionBasedOnAgeAndPinning() throw String md2Timestamp = RemoteStoreUtils.invertLong(currentTimeInMillis - 300000); String md3Timestamp = RemoteStoreUtils.invertLong(currentTimeInMillis - 600000); - doAnswer(invocationOnMock -> { - ActionListener> actionListener = invocationOnMock.getArgument(3); - actionListener.onResponse(List.of(new PlainBlobMetadata("pinned_timestamp_123", 1000))); - return null; - }).when(pinnedTimestampBlobStoreTransferService).listAllInSortedOrder(any(), any(), eq(1), any()); - long pinnedTimestamp = RemoteStoreUtils.invertLong(md2Timestamp) + 10000; - when(remoteStorePinnedTimestampsBlobStore.read(any())).thenReturn( - new RemotePinnedTimestamps.PinnedTimestamps(Map.of(pinnedTimestamp, List.of("xyz"))) - ); - when(remoteStorePinnedTimestampsBlobStore.getBlobPathForUpload(any())).thenReturn(new BlobPath()); + when(blobContainer.listBlobs()).thenReturn(Map.of(randomInt(100) + "__" + pinnedTimestamp, new PlainBlobMetadata("xyz", 100))); updatePinnedTimstampTask.run();