From 587c566bdf5c6c5d627d6b6883b3d732a6a8ac06 Mon Sep 17 00:00:00 2001 From: Shivansh Arora Date: Mon, 17 Jun 2024 13:52:21 +0530 Subject: [PATCH 1/4] Add tests for RemoteGlobalMetadataManager Signed-off-by: Shivansh Arora --- .../RemoteGlobalMetadataManagerTests.java | 552 +++++++++++++++++- .../RemoteCoordinationMetadataTests.java | 2 +- .../model/RemoteCustomMetadataTests.java | 2 +- .../model/RemoteGlobalMetadataTests.java | 2 +- ...RemoteHashesOfConsistentSettingsTests.java | 2 +- ...RemotePersistentSettingsMetadataTests.java | 2 +- .../model/RemoteTemplatesMetadataTests.java | 2 +- 7 files changed, 554 insertions(+), 10 deletions(-) diff --git a/server/src/test/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManagerTests.java b/server/src/test/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManagerTests.java index bd01bc1ab0cdb..61baf93d63bc9 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManagerTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManagerTests.java @@ -9,19 +9,36 @@ package org.opensearch.gateway.remote; import org.opensearch.Version; +import org.opensearch.action.LatchedActionListener; import org.opensearch.cluster.ClusterModule; import org.opensearch.cluster.ClusterName; import org.opensearch.cluster.ClusterState; import org.opensearch.cluster.DiffableUtils; +import org.opensearch.cluster.coordination.CoordinationMetadata; +import org.opensearch.cluster.metadata.DiffableStringMap; import org.opensearch.cluster.metadata.IndexGraveyard; import org.opensearch.cluster.metadata.Metadata; import org.opensearch.cluster.metadata.Metadata.XContentContext; +import org.opensearch.cluster.metadata.TemplatesMetadata; +import org.opensearch.common.CheckedRunnable; +import org.opensearch.common.blobstore.BlobPath; import org.opensearch.common.network.NetworkModule; import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; +import org.opensearch.core.action.ActionListener; +import org.opensearch.core.common.io.stream.NamedWriteableRegistry; import org.opensearch.core.compress.Compressor; import org.opensearch.core.compress.NoneCompressor; import org.opensearch.core.xcontent.NamedXContentRegistry; +import org.opensearch.gateway.remote.model.RemoteCoordinationMetadata; +import org.opensearch.gateway.remote.model.RemoteCustomMetadata; +import org.opensearch.gateway.remote.model.RemoteGlobalMetadata; +import org.opensearch.gateway.remote.model.RemoteHashesOfConsistentSettings; +import org.opensearch.gateway.remote.model.RemotePersistentSettingsMetadata; +import org.opensearch.gateway.remote.model.RemoteReadResult; +import org.opensearch.gateway.remote.model.RemoteTemplatesMetadata; +import org.opensearch.gateway.remote.model.RemoteTransientSettingsMetadata; +import org.opensearch.index.remote.RemoteStoreUtils; import org.opensearch.index.translog.transfer.BlobStoreTransferService; import org.opensearch.indices.IndicesModule; import org.opensearch.repositories.blobstore.BlobStoreRepository; @@ -32,41 +49,87 @@ import org.junit.After; import org.junit.Before; +import java.io.IOException; +import java.io.InputStream; import java.util.EnumSet; import java.util.List; import java.util.Map; import java.util.function.Function; import java.util.stream.Stream; +import org.mockito.ArgumentCaptor; + import static java.util.stream.Collectors.toList; +import static org.opensearch.cluster.metadata.Metadata.isGlobalStateEquals; +import static org.opensearch.common.blobstore.stream.write.WritePriority.URGENT; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.CUSTOM_DELIMITER; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.FORMAT_PARAMS; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.GLOBAL_METADATA_CURRENT_CODEC_VERSION; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.GLOBAL_METADATA_PATH_TOKEN; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.PATH_DELIMITER; +import static org.opensearch.gateway.remote.model.RemoteCoordinationMetadata.COORDINATION_METADATA; +import static org.opensearch.gateway.remote.model.RemoteCoordinationMetadata.COORDINATION_METADATA_FORMAT; +import static org.opensearch.gateway.remote.model.RemoteCoordinationMetadataTests.getCoordinationMetadata; +import static org.opensearch.gateway.remote.model.RemoteCustomMetadata.CUSTOM_METADATA; +import static org.opensearch.gateway.remote.model.RemoteCustomMetadataTests.getCustomMetadata; +import static org.opensearch.gateway.remote.model.RemoteGlobalMetadata.GLOBAL_METADATA; +import static org.opensearch.gateway.remote.model.RemoteGlobalMetadata.GLOBAL_METADATA_FORMAT; +import static org.opensearch.gateway.remote.model.RemoteGlobalMetadataTests.getGlobalMetadata; +import static org.opensearch.gateway.remote.model.RemoteHashesOfConsistentSettings.HASHES_OF_CONSISTENT_SETTINGS; +import static org.opensearch.gateway.remote.model.RemoteHashesOfConsistentSettings.HASHES_OF_CONSISTENT_SETTINGS_FORMAT; +import static org.opensearch.gateway.remote.model.RemoteHashesOfConsistentSettingsTests.getHashesOfConsistentSettings; +import static org.opensearch.gateway.remote.model.RemotePersistentSettingsMetadata.SETTING_METADATA; +import static org.opensearch.gateway.remote.model.RemotePersistentSettingsMetadataTests.getSettings; +import static org.opensearch.gateway.remote.model.RemoteTemplatesMetadata.TEMPLATES_METADATA; +import static org.opensearch.gateway.remote.model.RemoteTemplatesMetadata.TEMPLATES_METADATA_FORMAT; +import static org.opensearch.gateway.remote.model.RemoteTemplatesMetadataTests.getTemplatesMetadata; +import static org.opensearch.gateway.remote.model.RemoteTransientSettingsMetadata.TRANSIENT_SETTING_METADATA; import static org.hamcrest.Matchers.is; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyIterable; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; public class RemoteGlobalMetadataManagerTests extends OpenSearchTestCase { private RemoteGlobalMetadataManager remoteGlobalMetadataManager; private ClusterSettings clusterSettings; private BlobStoreRepository blobStoreRepository; + private BlobStoreTransferService blobStoreTransferService; + private Compressor compressor; + private NamedXContentRegistry xContentRegistry; + private NamedWriteableRegistry namedWriteableRegistry; private final ThreadPool threadPool = new TestThreadPool(getClass().getName()); + private final long METADATA_VERSION = 7331L; + private final String CLUSTER_NAME = "test-cluster"; + private final String CLUSTER_UUID = "test-cluster-uuid"; @Before public void setup() { clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); blobStoreRepository = mock(BlobStoreRepository.class); - BlobStoreTransferService blobStoreTransferService = mock(BlobStoreTransferService.class); - NamedXContentRegistry xContentRegistry = new NamedXContentRegistry( + blobStoreTransferService = mock(BlobStoreTransferService.class); + compressor = new NoneCompressor(); + xContentRegistry = new NamedXContentRegistry( Stream.of( NetworkModule.getNamedXContents().stream(), IndicesModule.getNamedXContents().stream(), ClusterModule.getNamedXWriteables().stream() ).flatMap(Function.identity()).collect(toList()) ); - Compressor compressor = new NoneCompressor(); + namedWriteableRegistry = writableRegistry(); + BlobPath blobPath = new BlobPath(); when(blobStoreRepository.getCompressor()).thenReturn(compressor); when(blobStoreRepository.getNamedXContentRegistry()).thenReturn(xContentRegistry); + when(blobStoreRepository.basePath()).thenReturn(blobPath); remoteGlobalMetadataManager = new RemoteGlobalMetadataManager( clusterSettings, - "test-cluster", + CLUSTER_NAME, blobStoreRepository, blobStoreTransferService, writableRegistry(), @@ -96,6 +159,487 @@ public void testGlobalMetadataUploadWaitTimeSetting() { assertEquals(globalMetadataUploadTimeout, remoteGlobalMetadataManager.getGlobalMetadataUploadTimeout().seconds()); } + public void testGetReadMetadataAsyncAction_CoordinationMetadata() throws Exception { + CoordinationMetadata coordinationMetadata = getCoordinationMetadata(); + String fileName = randomAlphaOfLength(10); + RemoteCoordinationMetadata coordinationMetadataForDownload = new RemoteCoordinationMetadata( + fileName, + CLUSTER_UUID, + compressor, + xContentRegistry + ); + when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( + COORDINATION_METADATA_FORMAT.serialize(coordinationMetadata, fileName, compressor, FORMAT_PARAMS).streamInput() + ); + LatchedActionListener listener = mock(LatchedActionListener.class); + + CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataReadAction( + coordinationMetadataForDownload, + COORDINATION_METADATA, + listener + ); + runnable.run(); + ArgumentCaptor savedResult = ArgumentCaptor.forClass(RemoteReadResult.class); + assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); + assertEquals(coordinationMetadata, savedResult.getValue().getObj()); + assertEquals(COORDINATION_METADATA, savedResult.getValue().getComponent()); + assertEquals(COORDINATION_METADATA, savedResult.getValue().getComponentName()); + } + + public void testGetAsyncMetadataWriteAction_CoordinationMetadata() throws Exception { + CoordinationMetadata coordinationMetadata = getCoordinationMetadata(); + RemoteCoordinationMetadata remoteCoordinationMetadata = new RemoteCoordinationMetadata( + coordinationMetadata, + METADATA_VERSION, + CLUSTER_UUID, + compressor, + xContentRegistry + ); + doAnswer(invocationOnMock -> { + invocationOnMock.getArgument(4, ActionListener.class).onResponse(null); + return null; + }).when(blobStoreTransferService) + .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); + LatchedActionListener listener = mock(LatchedActionListener.class); + + CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataWriteAction( + remoteCoordinationMetadata, + listener + ); + runnable.run(); + + ArgumentCaptor savedResult = ArgumentCaptor.forClass( + ClusterMetadataManifest.UploadedMetadata.class + ); + assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); + + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = savedResult.getValue(); + assertNotNull(uploadedMetadata); + assertEquals(COORDINATION_METADATA, uploadedMetadata.getComponent()); + String uploadedFileName = uploadedMetadata.getUploadedFilename(); + String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); + assertEquals(5, pathTokens.length); + assertEquals(RemoteClusterStateUtils.encodeString(CLUSTER_NAME), pathTokens[0]); + assertEquals("cluster-state", pathTokens[1]); + assertEquals(CLUSTER_UUID, pathTokens[2]); + assertEquals(GLOBAL_METADATA_PATH_TOKEN, pathTokens[3]); + String[] splitFileName = pathTokens[4].split(DELIMITER); + assertEquals(4, splitFileName.length); + assertEquals(COORDINATION_METADATA, splitFileName[0]); + assertEquals(RemoteStoreUtils.invertLong(METADATA_VERSION), splitFileName[1]); + assertEquals(GLOBAL_METADATA_CURRENT_CODEC_VERSION, Integer.parseInt(splitFileName[3])); + } + + public void testGetReadMetadataAsyncAction_PersistentSettings() throws Exception { + Settings settingsMetadata = getSettings(); + String fileName = randomAlphaOfLength(10); + RemotePersistentSettingsMetadata persistentSettings = new RemotePersistentSettingsMetadata( + fileName, + CLUSTER_UUID, + compressor, + xContentRegistry + ); + when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( + RemotePersistentSettingsMetadata.SETTINGS_METADATA_FORMAT.serialize(settingsMetadata, fileName, compressor, FORMAT_PARAMS) + .streamInput() + ); + LatchedActionListener listener = mock(LatchedActionListener.class); + + CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataReadAction( + persistentSettings, + SETTING_METADATA, + listener + ); + runnable.run(); + ArgumentCaptor savedResult = ArgumentCaptor.forClass(RemoteReadResult.class); + assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); + assertEquals(settingsMetadata, savedResult.getValue().getObj()); + assertEquals(SETTING_METADATA, savedResult.getValue().getComponent()); + assertEquals(SETTING_METADATA, savedResult.getValue().getComponentName()); + } + + public void testGetAsyncMetadataWriteAction_PersistentSettings() throws Exception { + Settings settingsMetadata = getSettings(); + RemotePersistentSettingsMetadata persistentSettings = new RemotePersistentSettingsMetadata( + settingsMetadata, + METADATA_VERSION, + CLUSTER_UUID, + compressor, + xContentRegistry + ); + doAnswer(invocationOnMock -> { + invocationOnMock.getArgument(4, ActionListener.class).onResponse(null); + return null; + }).when(blobStoreTransferService) + .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); + LatchedActionListener listener = mock(LatchedActionListener.class); + CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataWriteAction(persistentSettings, listener); + runnable.run(); + + ArgumentCaptor savedResult = ArgumentCaptor.forClass( + ClusterMetadataManifest.UploadedMetadata.class + ); + assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); + + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = savedResult.getValue(); + assertNotNull(uploadedMetadata); + assertEquals(SETTING_METADATA, uploadedMetadata.getComponent()); + String uploadedFileName = uploadedMetadata.getUploadedFilename(); + String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); + assertEquals(5, pathTokens.length); + assertEquals(RemoteClusterStateUtils.encodeString(CLUSTER_NAME), pathTokens[0]); + assertEquals("cluster-state", pathTokens[1]); + assertEquals(CLUSTER_UUID, pathTokens[2]); + assertEquals(GLOBAL_METADATA_PATH_TOKEN, pathTokens[3]); + String[] splitFileName = pathTokens[4].split(DELIMITER); + assertEquals(4, splitFileName.length); + assertEquals(SETTING_METADATA, splitFileName[0]); + assertEquals(RemoteStoreUtils.invertLong(METADATA_VERSION), splitFileName[1]); + assertEquals(GLOBAL_METADATA_CURRENT_CODEC_VERSION, Integer.parseInt(splitFileName[3])); + } + + public void testGetReadMetadataAsyncAction_TransientSettings() throws Exception { + Settings settingsMetadata = getSettings(); + String fileName = randomAlphaOfLength(10); + RemoteTransientSettingsMetadata transientSettings = new RemoteTransientSettingsMetadata( + fileName, + CLUSTER_UUID, + compressor, + xContentRegistry + ); + when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( + RemoteTransientSettingsMetadata.SETTINGS_METADATA_FORMAT.serialize(settingsMetadata, fileName, compressor, FORMAT_PARAMS) + .streamInput() + ); + LatchedActionListener listener = mock(LatchedActionListener.class); + + CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataReadAction( + transientSettings, + TRANSIENT_SETTING_METADATA, + listener + ); + runnable.run(); + ArgumentCaptor savedResult = ArgumentCaptor.forClass(RemoteReadResult.class); + assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); + assertEquals(settingsMetadata, savedResult.getValue().getObj()); + assertEquals(TRANSIENT_SETTING_METADATA, savedResult.getValue().getComponent()); + assertEquals(TRANSIENT_SETTING_METADATA, savedResult.getValue().getComponentName()); + } + + public void testGetAsyncMetadataWriteAction_TransientSettings() throws Exception { + Settings settingsMetadata = getSettings(); + RemoteTransientSettingsMetadata transientSettings = new RemoteTransientSettingsMetadata( + settingsMetadata, + METADATA_VERSION, + CLUSTER_UUID, + compressor, + xContentRegistry + ); + doAnswer(invocationOnMock -> { + invocationOnMock.getArgument(4, ActionListener.class).onResponse(null); + return null; + }).when(blobStoreTransferService) + .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); + LatchedActionListener listener = mock(LatchedActionListener.class); + CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataWriteAction(transientSettings, listener); + runnable.run(); + + ArgumentCaptor savedResult = ArgumentCaptor.forClass( + ClusterMetadataManifest.UploadedMetadata.class + ); + assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = savedResult.getValue(); + assertNotNull(uploadedMetadata); + assertEquals(TRANSIENT_SETTING_METADATA, uploadedMetadata.getComponent()); + String uploadedFileName = uploadedMetadata.getUploadedFilename(); + String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); + assertEquals(5, pathTokens.length); + assertEquals(RemoteClusterStateUtils.encodeString(CLUSTER_NAME), pathTokens[0]); + assertEquals("cluster-state", pathTokens[1]); + assertEquals(CLUSTER_UUID, pathTokens[2]); + assertEquals(GLOBAL_METADATA_PATH_TOKEN, pathTokens[3]); + String[] splitFileName = pathTokens[4].split(DELIMITER); + assertEquals(4, splitFileName.length); + assertEquals(TRANSIENT_SETTING_METADATA, splitFileName[0]); + assertEquals(RemoteStoreUtils.invertLong(METADATA_VERSION), splitFileName[1]); + assertEquals(GLOBAL_METADATA_CURRENT_CODEC_VERSION, Integer.parseInt(splitFileName[3])); + } + + public void testGetReadMetadataAsyncAction_HashesOfConsistentSettings() throws Exception { + DiffableStringMap hashesOfConsistentSettings = getHashesOfConsistentSettings(); + String fileName = randomAlphaOfLength(10); + RemoteHashesOfConsistentSettings hashesOfConsistentSettingsForDownload = new RemoteHashesOfConsistentSettings( + fileName, + CLUSTER_UUID, + compressor + ); + when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( + HASHES_OF_CONSISTENT_SETTINGS_FORMAT.serialize(hashesOfConsistentSettings, fileName, compressor).streamInput() + ); + LatchedActionListener listener = mock(LatchedActionListener.class); + + CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataReadAction( + hashesOfConsistentSettingsForDownload, + HASHES_OF_CONSISTENT_SETTINGS, + listener + ); + runnable.run(); + ArgumentCaptor savedResult = ArgumentCaptor.forClass(RemoteReadResult.class); + assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); + assertEquals(hashesOfConsistentSettings, savedResult.getValue().getObj()); + assertEquals(HASHES_OF_CONSISTENT_SETTINGS, savedResult.getValue().getComponent()); + assertEquals(HASHES_OF_CONSISTENT_SETTINGS, savedResult.getValue().getComponentName()); + } + + public void testGetAsyncMetadataWriteAction_HashesOfConsistentSettings() throws Exception { + DiffableStringMap hashesOfConsistentSettings = getHashesOfConsistentSettings(); + RemoteHashesOfConsistentSettings hashesOfConsistentSettingsForUpload = new RemoteHashesOfConsistentSettings( + hashesOfConsistentSettings, + METADATA_VERSION, + CLUSTER_UUID, + compressor + ); + doAnswer(invocationOnMock -> { + invocationOnMock.getArgument(4, ActionListener.class).onResponse(null); + return null; + }).when(blobStoreTransferService) + .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); + LatchedActionListener listener = mock(LatchedActionListener.class); + CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataWriteAction( + hashesOfConsistentSettingsForUpload, + listener + ); + runnable.run(); + + ArgumentCaptor savedResult = ArgumentCaptor.forClass( + ClusterMetadataManifest.UploadedMetadata.class + ); + assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = savedResult.getValue(); + assertNotNull(uploadedMetadata); + assertEquals(HASHES_OF_CONSISTENT_SETTINGS, uploadedMetadata.getComponent()); + String uploadedFileName = uploadedMetadata.getUploadedFilename(); + String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); + assertEquals(5, pathTokens.length); + assertEquals(RemoteClusterStateUtils.encodeString(CLUSTER_NAME), pathTokens[0]); + assertEquals("cluster-state", pathTokens[1]); + assertEquals(CLUSTER_UUID, pathTokens[2]); + assertEquals(GLOBAL_METADATA_PATH_TOKEN, pathTokens[3]); + String[] splitFileName = pathTokens[4].split(DELIMITER); + assertEquals(4, splitFileName.length); + assertEquals(HASHES_OF_CONSISTENT_SETTINGS, splitFileName[0]); + assertEquals(RemoteStoreUtils.invertLong(METADATA_VERSION), splitFileName[1]); + assertEquals(GLOBAL_METADATA_CURRENT_CODEC_VERSION, Integer.parseInt(splitFileName[3])); + } + + public void testGetReadMetadataAsyncAction_TemplatesMetadata() throws Exception { + TemplatesMetadata templatesMetadata = getTemplatesMetadata(); + String fileName = randomAlphaOfLength(10); + RemoteTemplatesMetadata templatesMetadataForDownload = new RemoteTemplatesMetadata( + fileName, + CLUSTER_UUID, + compressor, + xContentRegistry + ); + when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( + TEMPLATES_METADATA_FORMAT.serialize(templatesMetadata, fileName, compressor, FORMAT_PARAMS).streamInput() + ); + LatchedActionListener listener = mock(LatchedActionListener.class); + CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataReadAction( + templatesMetadataForDownload, + TEMPLATES_METADATA, + listener + ); + runnable.run(); + + ArgumentCaptor savedResult = ArgumentCaptor.forClass(RemoteReadResult.class); + assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); + assertEquals(templatesMetadata, savedResult.getValue().getObj()); + assertEquals(TEMPLATES_METADATA, savedResult.getValue().getComponent()); + assertEquals(TEMPLATES_METADATA, savedResult.getValue().getComponentName()); + } + + public void testGetAsyncMetadataWriteAction_TemplatesMetadata() throws Exception { + TemplatesMetadata templatesMetadata = getTemplatesMetadata(); + RemoteTemplatesMetadata templateMetadataForUpload = new RemoteTemplatesMetadata( + templatesMetadata, + METADATA_VERSION, + CLUSTER_UUID, + compressor, + xContentRegistry + ); + doAnswer(invocationOnMock -> { + invocationOnMock.getArgument(4, ActionListener.class).onResponse(null); + return null; + }).when(blobStoreTransferService) + .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); + LatchedActionListener listener = mock(LatchedActionListener.class); + CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataWriteAction( + templateMetadataForUpload, + listener + ); + runnable.run(); + + ArgumentCaptor savedResult = ArgumentCaptor.forClass( + ClusterMetadataManifest.UploadedMetadata.class + ); + assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = savedResult.getValue(); + assertNotNull(uploadedMetadata); + assertEquals(TEMPLATES_METADATA, uploadedMetadata.getComponent()); + String uploadedFileName = uploadedMetadata.getUploadedFilename(); + String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); + assertEquals(5, pathTokens.length); + assertEquals(RemoteClusterStateUtils.encodeString(CLUSTER_NAME), pathTokens[0]); + assertEquals("cluster-state", pathTokens[1]); + assertEquals(CLUSTER_UUID, pathTokens[2]); + assertEquals(GLOBAL_METADATA_PATH_TOKEN, pathTokens[3]); + String[] splitFileName = pathTokens[4].split(DELIMITER); + assertEquals(4, splitFileName.length); + assertEquals(TEMPLATES_METADATA, splitFileName[0]); + assertEquals(RemoteStoreUtils.invertLong(METADATA_VERSION), splitFileName[1]); + assertEquals(GLOBAL_METADATA_CURRENT_CODEC_VERSION, Integer.parseInt(splitFileName[3])); + } + + public void testGetReadMetadataAsyncAction_CustomMetadata() throws Exception { + Metadata.Custom customMetadata = getCustomMetadata(); + String fileName = randomAlphaOfLength(10); + RemoteCustomMetadata customMetadataForDownload = new RemoteCustomMetadata( + fileName, + IndexGraveyard.TYPE, + CLUSTER_UUID, + compressor, + namedWriteableRegistry + ); + when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( + customMetadataForDownload.customBlobStoreFormat.serialize(customMetadata, fileName, compressor).streamInput() + ); + LatchedActionListener listener = mock(LatchedActionListener.class); + CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataReadAction( + customMetadataForDownload, + IndexGraveyard.TYPE, + listener + ); + runnable.run(); + + ArgumentCaptor savedResult = ArgumentCaptor.forClass(RemoteReadResult.class); + assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); + assertEquals(customMetadata, savedResult.getValue().getObj()); + assertEquals(CUSTOM_METADATA, savedResult.getValue().getComponent()); + assertEquals(IndexGraveyard.TYPE, savedResult.getValue().getComponentName()); + } + + public void testGetAsyncMetadataWriteAction_CustomMetadata() throws Exception { + Metadata.Custom customMetadata = getCustomMetadata(); + RemoteCustomMetadata customMetadataForUpload = new RemoteCustomMetadata( + customMetadata, + IndexGraveyard.TYPE, + METADATA_VERSION, + CLUSTER_UUID, + compressor, + namedWriteableRegistry + ); + doAnswer(invocationOnMock -> { + invocationOnMock.getArgument(4, ActionListener.class).onResponse(null); + return null; + }).when(blobStoreTransferService) + .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); + LatchedActionListener listener = mock(LatchedActionListener.class); + CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataWriteAction(customMetadataForUpload, listener); + runnable.run(); + + ArgumentCaptor savedResult = ArgumentCaptor.forClass( + ClusterMetadataManifest.UploadedMetadata.class + ); + assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = savedResult.getValue(); + assertNotNull(uploadedMetadata); + assertEquals(String.join(CUSTOM_DELIMITER, CUSTOM_METADATA, IndexGraveyard.TYPE), uploadedMetadata.getComponent()); + String uploadedFileName = uploadedMetadata.getUploadedFilename(); + String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); + assertEquals(5, pathTokens.length); + assertEquals(RemoteClusterStateUtils.encodeString(CLUSTER_NAME), pathTokens[0]); + assertEquals("cluster-state", pathTokens[1]); + assertEquals(CLUSTER_UUID, pathTokens[2]); + assertEquals(GLOBAL_METADATA_PATH_TOKEN, pathTokens[3]); + String[] splitFileName = pathTokens[4].split(DELIMITER); + assertEquals(4, splitFileName.length); + assertEquals(String.join(CUSTOM_DELIMITER, CUSTOM_METADATA, IndexGraveyard.TYPE), splitFileName[0]); + assertEquals(RemoteStoreUtils.invertLong(METADATA_VERSION), splitFileName[1]); + assertEquals(GLOBAL_METADATA_CURRENT_CODEC_VERSION, Integer.parseInt(splitFileName[3])); + } + + public void testGetReadMetadataAsyncAction_GlobalMetadata() throws Exception { + Metadata metadata = getGlobalMetadata(); + String fileName = randomAlphaOfLength(10); + RemoteGlobalMetadata globalMetadataForDownload = new RemoteGlobalMetadata(fileName, CLUSTER_UUID, compressor, xContentRegistry); + when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( + GLOBAL_METADATA_FORMAT.serialize(metadata, fileName, compressor, FORMAT_PARAMS).streamInput() + ); + LatchedActionListener listener = mock(LatchedActionListener.class); + CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataReadAction( + globalMetadataForDownload, + GLOBAL_METADATA, + listener + ); + runnable.run(); + + ArgumentCaptor savedResult = ArgumentCaptor.forClass(RemoteReadResult.class); + assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); + assertTrue(isGlobalStateEquals(metadata, (Metadata) savedResult.getValue().getObj())); + assertEquals(GLOBAL_METADATA, savedResult.getValue().getComponent()); + assertEquals(GLOBAL_METADATA, savedResult.getValue().getComponentName()); + } + + public void testGetReadMetadataAsyncAction_IOException() throws Exception { + String fileName = randomAlphaOfLength(10); + RemoteCoordinationMetadata coordinationMetadataForDownload = new RemoteCoordinationMetadata( + fileName, + CLUSTER_UUID, + compressor, + xContentRegistry + ); + IOException ioException = new IOException("mock test exception"); + when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenThrow(ioException); + LatchedActionListener listener = mock(LatchedActionListener.class); + CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataReadAction( + coordinationMetadataForDownload, + COORDINATION_METADATA, + listener + ); + runnable.run(); + ArgumentCaptor savedException = ArgumentCaptor.forClass(Exception.class); + assertBusy(() -> verify(listener, times(1)).onFailure(savedException.capture())); + assertEquals(ioException, savedException.getValue()); + } + + public void testGetAsyncMetadataWriteAction_IOException() throws Exception { + CoordinationMetadata coordinationMetadata = getCoordinationMetadata(); + RemoteCoordinationMetadata remoteCoordinationMetadata = new RemoteCoordinationMetadata( + coordinationMetadata, + METADATA_VERSION, + CLUSTER_UUID, + compressor, + xContentRegistry + ); + doAnswer(invocationOnMock -> { + invocationOnMock.getArgument(4, ActionListener.class).onFailure(new IOException("mock test exception")); + return null; + }).when(blobStoreTransferService) + .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); + + LatchedActionListener listener = mock(LatchedActionListener.class); + CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataWriteAction( + remoteCoordinationMetadata, + listener + ); + runnable.run(); + + ArgumentCaptor savedException = ArgumentCaptor.forClass(Exception.class); + assertBusy(() -> verify(listener, times(1)).onFailure(savedException.capture())); + assertTrue(savedException.getValue() instanceof RemoteStateTransferException); + } + public void testGetUpdatedCustoms() { Map previousCustoms = Map.of( CustomMetadata1.TYPE, diff --git a/server/src/test/java/org/opensearch/gateway/remote/model/RemoteCoordinationMetadataTests.java b/server/src/test/java/org/opensearch/gateway/remote/model/RemoteCoordinationMetadataTests.java index 9484afe6b7d6c..63d6de05a737c 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/model/RemoteCoordinationMetadataTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/model/RemoteCoordinationMetadataTests.java @@ -229,7 +229,7 @@ public void testSerDe() throws IOException { } } - private CoordinationMetadata getCoordinationMetadata() { + public static CoordinationMetadata getCoordinationMetadata() { return CoordinationMetadata.builder() .term(TERM) .lastAcceptedConfiguration(new VotingConfiguration(Set.of("node1"))) diff --git a/server/src/test/java/org/opensearch/gateway/remote/model/RemoteCustomMetadataTests.java b/server/src/test/java/org/opensearch/gateway/remote/model/RemoteCustomMetadataTests.java index 1bce176273270..1e28817be79f2 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/model/RemoteCustomMetadataTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/model/RemoteCustomMetadataTests.java @@ -232,7 +232,7 @@ public void testSerDe() throws IOException { } } - private Custom getCustomMetadata() { + public static Custom getCustomMetadata() { return IndexGraveyard.builder().addTombstone(new Index("test-index", "3q2423")).build(); } diff --git a/server/src/test/java/org/opensearch/gateway/remote/model/RemoteGlobalMetadataTests.java b/server/src/test/java/org/opensearch/gateway/remote/model/RemoteGlobalMetadataTests.java index 02ddc8ba93071..23de485357547 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/model/RemoteGlobalMetadataTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/model/RemoteGlobalMetadataTests.java @@ -180,7 +180,7 @@ public void testSerDe() throws IOException { } } - private Metadata getGlobalMetadata() { + public static Metadata getGlobalMetadata() { return Metadata.builder() .templates( TemplatesMetadata.builder() diff --git a/server/src/test/java/org/opensearch/gateway/remote/model/RemoteHashesOfConsistentSettingsTests.java b/server/src/test/java/org/opensearch/gateway/remote/model/RemoteHashesOfConsistentSettingsTests.java index d883eabf9fbc9..b931f24f98631 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/model/RemoteHashesOfConsistentSettingsTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/model/RemoteHashesOfConsistentSettingsTests.java @@ -185,7 +185,7 @@ public void testSerDe() throws IOException { } } - private DiffableStringMap getHashesOfConsistentSettings() { + public static DiffableStringMap getHashesOfConsistentSettings() { Map hashesOfConsistentSettings = new HashMap<>(); hashesOfConsistentSettings.put("secure-setting-key", "secure-setting-value"); return new DiffableStringMap(hashesOfConsistentSettings); diff --git a/server/src/test/java/org/opensearch/gateway/remote/model/RemotePersistentSettingsMetadataTests.java b/server/src/test/java/org/opensearch/gateway/remote/model/RemotePersistentSettingsMetadataTests.java index 850c18f03fa49..5e4d5d66ca1b7 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/model/RemotePersistentSettingsMetadataTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/model/RemotePersistentSettingsMetadataTests.java @@ -224,7 +224,7 @@ public void testSerDe() throws IOException { } } - private Settings getSettings() { + public static Settings getSettings() { return Settings.builder().put("random_index_setting_" + randomAlphaOfLength(3), randomAlphaOfLength(5)).build(); } } diff --git a/server/src/test/java/org/opensearch/gateway/remote/model/RemoteTemplatesMetadataTests.java b/server/src/test/java/org/opensearch/gateway/remote/model/RemoteTemplatesMetadataTests.java index b86044003aa55..d7ecd2ad3f44a 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/model/RemoteTemplatesMetadataTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/model/RemoteTemplatesMetadataTests.java @@ -227,7 +227,7 @@ public void testSerDe() throws IOException { } } - private TemplatesMetadata getTemplatesMetadata() { + public static TemplatesMetadata getTemplatesMetadata() { return TemplatesMetadata.builder() .put( IndexTemplateMetadata.builder("template" + randomAlphaOfLength(3)) From ec042aabe1dcf5f68d7511d30c795501f3aec965 Mon Sep 17 00:00:00 2001 From: Shivansh Arora Date: Wed, 19 Jun 2024 00:10:47 +0530 Subject: [PATCH 2/4] Add TestCapturingListener Signed-off-by: Shivansh Arora --- .../RemoteGlobalMetadataManagerTests.java | 334 +++++++++--------- .../gateway/remote/RemoteStateTestUtil.java | 28 ++ 2 files changed, 188 insertions(+), 174 deletions(-) create mode 100644 server/src/test/java/org/opensearch/gateway/remote/RemoteStateTestUtil.java diff --git a/server/src/test/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManagerTests.java b/server/src/test/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManagerTests.java index 61baf93d63bc9..1b361e7818bf0 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManagerTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManagerTests.java @@ -20,7 +20,6 @@ import org.opensearch.cluster.metadata.Metadata; import org.opensearch.cluster.metadata.Metadata.XContentContext; import org.opensearch.cluster.metadata.TemplatesMetadata; -import org.opensearch.common.CheckedRunnable; import org.opensearch.common.blobstore.BlobPath; import org.opensearch.common.network.NetworkModule; import org.opensearch.common.settings.ClusterSettings; @@ -54,14 +53,14 @@ import java.util.EnumSet; import java.util.List; import java.util.Map; +import java.util.concurrent.CountDownLatch; import java.util.function.Function; import java.util.stream.Stream; -import org.mockito.ArgumentCaptor; - import static java.util.stream.Collectors.toList; import static org.opensearch.cluster.metadata.Metadata.isGlobalStateEquals; import static org.opensearch.common.blobstore.stream.write.WritePriority.URGENT; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN; import static org.opensearch.gateway.remote.RemoteClusterStateUtils.CUSTOM_DELIMITER; import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER; import static org.opensearch.gateway.remote.RemoteClusterStateUtils.FORMAT_PARAMS; @@ -92,8 +91,6 @@ import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; public class RemoteGlobalMetadataManagerTests extends OpenSearchTestCase { @@ -171,19 +168,20 @@ public void testGetReadMetadataAsyncAction_CoordinationMetadata() throws Excepti when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( COORDINATION_METADATA_FORMAT.serialize(coordinationMetadata, fileName, compressor, FORMAT_PARAMS).streamInput() ); - LatchedActionListener listener = mock(LatchedActionListener.class); + RemoteStateTestUtil.TestCapturingListener listener = new RemoteStateTestUtil.TestCapturingListener<>(); + CountDownLatch latch = new CountDownLatch(1); - CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataReadAction( + remoteGlobalMetadataManager.getAsyncMetadataReadAction( coordinationMetadataForDownload, COORDINATION_METADATA, - listener - ); - runnable.run(); - ArgumentCaptor savedResult = ArgumentCaptor.forClass(RemoteReadResult.class); - assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); - assertEquals(coordinationMetadata, savedResult.getValue().getObj()); - assertEquals(COORDINATION_METADATA, savedResult.getValue().getComponent()); - assertEquals(COORDINATION_METADATA, savedResult.getValue().getComponentName()); + new LatchedActionListener<>(listener, latch) + ).run(); + latch.await(); + assertNull(listener.failure); + assertNotNull(listener.result); + assertEquals(coordinationMetadata, listener.result.getObj()); + assertEquals(COORDINATION_METADATA, listener.result.getComponent()); + assertEquals(COORDINATION_METADATA, listener.result.getComponentName()); } public void testGetAsyncMetadataWriteAction_CoordinationMetadata() throws Exception { @@ -200,27 +198,22 @@ public void testGetAsyncMetadataWriteAction_CoordinationMetadata() throws Except return null; }).when(blobStoreTransferService) .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); - LatchedActionListener listener = mock(LatchedActionListener.class); - - CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataWriteAction( - remoteCoordinationMetadata, - listener - ); - runnable.run(); - - ArgumentCaptor savedResult = ArgumentCaptor.forClass( - ClusterMetadataManifest.UploadedMetadata.class - ); - assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); - - ClusterMetadataManifest.UploadedMetadata uploadedMetadata = savedResult.getValue(); - assertNotNull(uploadedMetadata); + RemoteStateTestUtil.TestCapturingListener listener = + new RemoteStateTestUtil.TestCapturingListener<>(); + CountDownLatch latch = new CountDownLatch(1); + + remoteGlobalMetadataManager.getAsyncMetadataWriteAction(remoteCoordinationMetadata, new LatchedActionListener<>(listener, latch)) + .run(); + latch.await(); + assertNull(listener.failure); + assertNotNull(listener.result); + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.result; assertEquals(COORDINATION_METADATA, uploadedMetadata.getComponent()); String uploadedFileName = uploadedMetadata.getUploadedFilename(); String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); assertEquals(5, pathTokens.length); assertEquals(RemoteClusterStateUtils.encodeString(CLUSTER_NAME), pathTokens[0]); - assertEquals("cluster-state", pathTokens[1]); + assertEquals(CLUSTER_STATE_PATH_TOKEN, pathTokens[1]); assertEquals(CLUSTER_UUID, pathTokens[2]); assertEquals(GLOBAL_METADATA_PATH_TOKEN, pathTokens[3]); String[] splitFileName = pathTokens[4].split(DELIMITER); @@ -243,19 +236,20 @@ public void testGetReadMetadataAsyncAction_PersistentSettings() throws Exception RemotePersistentSettingsMetadata.SETTINGS_METADATA_FORMAT.serialize(settingsMetadata, fileName, compressor, FORMAT_PARAMS) .streamInput() ); - LatchedActionListener listener = mock(LatchedActionListener.class); + RemoteStateTestUtil.TestCapturingListener listener = new RemoteStateTestUtil.TestCapturingListener<>(); + CountDownLatch latch = new CountDownLatch(1); - CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataReadAction( + remoteGlobalMetadataManager.getAsyncMetadataReadAction( persistentSettings, SETTING_METADATA, - listener - ); - runnable.run(); - ArgumentCaptor savedResult = ArgumentCaptor.forClass(RemoteReadResult.class); - assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); - assertEquals(settingsMetadata, savedResult.getValue().getObj()); - assertEquals(SETTING_METADATA, savedResult.getValue().getComponent()); - assertEquals(SETTING_METADATA, savedResult.getValue().getComponentName()); + new LatchedActionListener<>(listener, latch) + ).run(); + latch.await(); + assertNull(listener.failure); + assertNotNull(listener.result); + assertEquals(settingsMetadata, listener.result.getObj()); + assertEquals(SETTING_METADATA, listener.result.getComponent()); + assertEquals(SETTING_METADATA, listener.result.getComponentName()); } public void testGetAsyncMetadataWriteAction_PersistentSettings() throws Exception { @@ -272,23 +266,21 @@ public void testGetAsyncMetadataWriteAction_PersistentSettings() throws Exceptio return null; }).when(blobStoreTransferService) .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); - LatchedActionListener listener = mock(LatchedActionListener.class); - CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataWriteAction(persistentSettings, listener); - runnable.run(); - - ArgumentCaptor savedResult = ArgumentCaptor.forClass( - ClusterMetadataManifest.UploadedMetadata.class - ); - assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); - - ClusterMetadataManifest.UploadedMetadata uploadedMetadata = savedResult.getValue(); - assertNotNull(uploadedMetadata); + RemoteStateTestUtil.TestCapturingListener listener = + new RemoteStateTestUtil.TestCapturingListener<>(); + CountDownLatch latch = new CountDownLatch(1); + remoteGlobalMetadataManager.getAsyncMetadataWriteAction(persistentSettings, new LatchedActionListener<>(listener, latch)).run(); + + latch.await(); + assertNull(listener.failure); + assertNotNull(listener.result); + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.result; assertEquals(SETTING_METADATA, uploadedMetadata.getComponent()); String uploadedFileName = uploadedMetadata.getUploadedFilename(); String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); assertEquals(5, pathTokens.length); assertEquals(RemoteClusterStateUtils.encodeString(CLUSTER_NAME), pathTokens[0]); - assertEquals("cluster-state", pathTokens[1]); + assertEquals(CLUSTER_STATE_PATH_TOKEN, pathTokens[1]); assertEquals(CLUSTER_UUID, pathTokens[2]); assertEquals(GLOBAL_METADATA_PATH_TOKEN, pathTokens[3]); String[] splitFileName = pathTokens[4].split(DELIMITER); @@ -311,19 +303,20 @@ public void testGetReadMetadataAsyncAction_TransientSettings() throws Exception RemoteTransientSettingsMetadata.SETTINGS_METADATA_FORMAT.serialize(settingsMetadata, fileName, compressor, FORMAT_PARAMS) .streamInput() ); - LatchedActionListener listener = mock(LatchedActionListener.class); + RemoteStateTestUtil.TestCapturingListener listener = new RemoteStateTestUtil.TestCapturingListener<>(); + CountDownLatch latch = new CountDownLatch(1); - CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataReadAction( + remoteGlobalMetadataManager.getAsyncMetadataReadAction( transientSettings, TRANSIENT_SETTING_METADATA, - listener - ); - runnable.run(); - ArgumentCaptor savedResult = ArgumentCaptor.forClass(RemoteReadResult.class); - assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); - assertEquals(settingsMetadata, savedResult.getValue().getObj()); - assertEquals(TRANSIENT_SETTING_METADATA, savedResult.getValue().getComponent()); - assertEquals(TRANSIENT_SETTING_METADATA, savedResult.getValue().getComponentName()); + new LatchedActionListener<>(listener, latch) + ).run(); + latch.await(); + assertNull(listener.failure); + assertNotNull(listener.result); + assertEquals(settingsMetadata, listener.result.getObj()); + assertEquals(TRANSIENT_SETTING_METADATA, listener.result.getComponent()); + assertEquals(TRANSIENT_SETTING_METADATA, listener.result.getComponentName()); } public void testGetAsyncMetadataWriteAction_TransientSettings() throws Exception { @@ -340,22 +333,20 @@ public void testGetAsyncMetadataWriteAction_TransientSettings() throws Exception return null; }).when(blobStoreTransferService) .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); - LatchedActionListener listener = mock(LatchedActionListener.class); - CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataWriteAction(transientSettings, listener); - runnable.run(); - - ArgumentCaptor savedResult = ArgumentCaptor.forClass( - ClusterMetadataManifest.UploadedMetadata.class - ); - assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); - ClusterMetadataManifest.UploadedMetadata uploadedMetadata = savedResult.getValue(); - assertNotNull(uploadedMetadata); + RemoteStateTestUtil.TestCapturingListener listener = + new RemoteStateTestUtil.TestCapturingListener<>(); + CountDownLatch latch = new CountDownLatch(1); + remoteGlobalMetadataManager.getAsyncMetadataWriteAction(transientSettings, new LatchedActionListener<>(listener, latch)).run(); + latch.await(); + assertNull(listener.failure); + assertNotNull(listener.result); + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.result; assertEquals(TRANSIENT_SETTING_METADATA, uploadedMetadata.getComponent()); String uploadedFileName = uploadedMetadata.getUploadedFilename(); String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); assertEquals(5, pathTokens.length); assertEquals(RemoteClusterStateUtils.encodeString(CLUSTER_NAME), pathTokens[0]); - assertEquals("cluster-state", pathTokens[1]); + assertEquals(CLUSTER_STATE_PATH_TOKEN, pathTokens[1]); assertEquals(CLUSTER_UUID, pathTokens[2]); assertEquals(GLOBAL_METADATA_PATH_TOKEN, pathTokens[3]); String[] splitFileName = pathTokens[4].split(DELIMITER); @@ -376,19 +367,20 @@ public void testGetReadMetadataAsyncAction_HashesOfConsistentSettings() throws E when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( HASHES_OF_CONSISTENT_SETTINGS_FORMAT.serialize(hashesOfConsistentSettings, fileName, compressor).streamInput() ); - LatchedActionListener listener = mock(LatchedActionListener.class); + RemoteStateTestUtil.TestCapturingListener listener = new RemoteStateTestUtil.TestCapturingListener<>(); + CountDownLatch latch = new CountDownLatch(1); - CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataReadAction( + remoteGlobalMetadataManager.getAsyncMetadataReadAction( hashesOfConsistentSettingsForDownload, HASHES_OF_CONSISTENT_SETTINGS, - listener - ); - runnable.run(); - ArgumentCaptor savedResult = ArgumentCaptor.forClass(RemoteReadResult.class); - assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); - assertEquals(hashesOfConsistentSettings, savedResult.getValue().getObj()); - assertEquals(HASHES_OF_CONSISTENT_SETTINGS, savedResult.getValue().getComponent()); - assertEquals(HASHES_OF_CONSISTENT_SETTINGS, savedResult.getValue().getComponentName()); + new LatchedActionListener<>(listener, latch) + ).run(); + latch.await(); + assertNull(listener.failure); + assertNotNull(listener.result); + assertEquals(hashesOfConsistentSettings, listener.result.getObj()); + assertEquals(HASHES_OF_CONSISTENT_SETTINGS, listener.result.getComponent()); + assertEquals(HASHES_OF_CONSISTENT_SETTINGS, listener.result.getComponentName()); } public void testGetAsyncMetadataWriteAction_HashesOfConsistentSettings() throws Exception { @@ -404,25 +396,23 @@ public void testGetAsyncMetadataWriteAction_HashesOfConsistentSettings() throws return null; }).when(blobStoreTransferService) .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); - LatchedActionListener listener = mock(LatchedActionListener.class); - CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataWriteAction( + RemoteStateTestUtil.TestCapturingListener listener = + new RemoteStateTestUtil.TestCapturingListener<>(); + CountDownLatch latch = new CountDownLatch(1); + remoteGlobalMetadataManager.getAsyncMetadataWriteAction( hashesOfConsistentSettingsForUpload, - listener - ); - runnable.run(); - - ArgumentCaptor savedResult = ArgumentCaptor.forClass( - ClusterMetadataManifest.UploadedMetadata.class - ); - assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); - ClusterMetadataManifest.UploadedMetadata uploadedMetadata = savedResult.getValue(); - assertNotNull(uploadedMetadata); + new LatchedActionListener<>(listener, latch) + ).run(); + latch.await(); + assertNull(listener.failure); + assertNotNull(listener.result); + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.result; assertEquals(HASHES_OF_CONSISTENT_SETTINGS, uploadedMetadata.getComponent()); String uploadedFileName = uploadedMetadata.getUploadedFilename(); String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); assertEquals(5, pathTokens.length); assertEquals(RemoteClusterStateUtils.encodeString(CLUSTER_NAME), pathTokens[0]); - assertEquals("cluster-state", pathTokens[1]); + assertEquals(CLUSTER_STATE_PATH_TOKEN, pathTokens[1]); assertEquals(CLUSTER_UUID, pathTokens[2]); assertEquals(GLOBAL_METADATA_PATH_TOKEN, pathTokens[3]); String[] splitFileName = pathTokens[4].split(DELIMITER); @@ -444,19 +434,19 @@ public void testGetReadMetadataAsyncAction_TemplatesMetadata() throws Exception when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( TEMPLATES_METADATA_FORMAT.serialize(templatesMetadata, fileName, compressor, FORMAT_PARAMS).streamInput() ); - LatchedActionListener listener = mock(LatchedActionListener.class); - CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataReadAction( + RemoteStateTestUtil.TestCapturingListener listener = new RemoteStateTestUtil.TestCapturingListener<>(); + CountDownLatch latch = new CountDownLatch(1); + remoteGlobalMetadataManager.getAsyncMetadataReadAction( templatesMetadataForDownload, TEMPLATES_METADATA, - listener - ); - runnable.run(); - - ArgumentCaptor savedResult = ArgumentCaptor.forClass(RemoteReadResult.class); - assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); - assertEquals(templatesMetadata, savedResult.getValue().getObj()); - assertEquals(TEMPLATES_METADATA, savedResult.getValue().getComponent()); - assertEquals(TEMPLATES_METADATA, savedResult.getValue().getComponentName()); + new LatchedActionListener<>(listener, latch) + ).run(); + latch.await(); + assertNull(listener.failure); + assertNotNull(listener.result); + assertEquals(templatesMetadata, listener.result.getObj()); + assertEquals(TEMPLATES_METADATA, listener.result.getComponent()); + assertEquals(TEMPLATES_METADATA, listener.result.getComponentName()); } public void testGetAsyncMetadataWriteAction_TemplatesMetadata() throws Exception { @@ -473,25 +463,21 @@ public void testGetAsyncMetadataWriteAction_TemplatesMetadata() throws Exception return null; }).when(blobStoreTransferService) .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); - LatchedActionListener listener = mock(LatchedActionListener.class); - CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataWriteAction( - templateMetadataForUpload, - listener - ); - runnable.run(); - - ArgumentCaptor savedResult = ArgumentCaptor.forClass( - ClusterMetadataManifest.UploadedMetadata.class - ); - assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); - ClusterMetadataManifest.UploadedMetadata uploadedMetadata = savedResult.getValue(); - assertNotNull(uploadedMetadata); + RemoteStateTestUtil.TestCapturingListener listener = + new RemoteStateTestUtil.TestCapturingListener<>(); + CountDownLatch latch = new CountDownLatch(1); + remoteGlobalMetadataManager.getAsyncMetadataWriteAction(templateMetadataForUpload, new LatchedActionListener<>(listener, latch)) + .run(); + latch.await(); + assertNull(listener.failure); + assertNotNull(listener.result); + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.result; assertEquals(TEMPLATES_METADATA, uploadedMetadata.getComponent()); String uploadedFileName = uploadedMetadata.getUploadedFilename(); String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); assertEquals(5, pathTokens.length); assertEquals(RemoteClusterStateUtils.encodeString(CLUSTER_NAME), pathTokens[0]); - assertEquals("cluster-state", pathTokens[1]); + assertEquals(CLUSTER_STATE_PATH_TOKEN, pathTokens[1]); assertEquals(CLUSTER_UUID, pathTokens[2]); assertEquals(GLOBAL_METADATA_PATH_TOKEN, pathTokens[3]); String[] splitFileName = pathTokens[4].split(DELIMITER); @@ -514,19 +500,19 @@ public void testGetReadMetadataAsyncAction_CustomMetadata() throws Exception { when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( customMetadataForDownload.customBlobStoreFormat.serialize(customMetadata, fileName, compressor).streamInput() ); - LatchedActionListener listener = mock(LatchedActionListener.class); - CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataReadAction( + RemoteStateTestUtil.TestCapturingListener listener = new RemoteStateTestUtil.TestCapturingListener<>(); + CountDownLatch latch = new CountDownLatch(1); + remoteGlobalMetadataManager.getAsyncMetadataReadAction( customMetadataForDownload, IndexGraveyard.TYPE, - listener - ); - runnable.run(); - - ArgumentCaptor savedResult = ArgumentCaptor.forClass(RemoteReadResult.class); - assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); - assertEquals(customMetadata, savedResult.getValue().getObj()); - assertEquals(CUSTOM_METADATA, savedResult.getValue().getComponent()); - assertEquals(IndexGraveyard.TYPE, savedResult.getValue().getComponentName()); + new LatchedActionListener<>(listener, latch) + ).run(); + latch.await(); + assertNull(listener.failure); + assertNotNull(listener.result); + assertEquals(customMetadata, listener.result.getObj()); + assertEquals(CUSTOM_METADATA, listener.result.getComponent()); + assertEquals(IndexGraveyard.TYPE, listener.result.getComponentName()); } public void testGetAsyncMetadataWriteAction_CustomMetadata() throws Exception { @@ -544,22 +530,21 @@ public void testGetAsyncMetadataWriteAction_CustomMetadata() throws Exception { return null; }).when(blobStoreTransferService) .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); - LatchedActionListener listener = mock(LatchedActionListener.class); - CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataWriteAction(customMetadataForUpload, listener); - runnable.run(); - - ArgumentCaptor savedResult = ArgumentCaptor.forClass( - ClusterMetadataManifest.UploadedMetadata.class - ); - assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); - ClusterMetadataManifest.UploadedMetadata uploadedMetadata = savedResult.getValue(); - assertNotNull(uploadedMetadata); + RemoteStateTestUtil.TestCapturingListener listener = + new RemoteStateTestUtil.TestCapturingListener<>(); + CountDownLatch latch = new CountDownLatch(1); + remoteGlobalMetadataManager.getAsyncMetadataWriteAction(customMetadataForUpload, new LatchedActionListener<>(listener, latch)) + .run(); + latch.await(); + assertNull(listener.failure); + assertNotNull(listener.result); + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.result; assertEquals(String.join(CUSTOM_DELIMITER, CUSTOM_METADATA, IndexGraveyard.TYPE), uploadedMetadata.getComponent()); String uploadedFileName = uploadedMetadata.getUploadedFilename(); String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); assertEquals(5, pathTokens.length); assertEquals(RemoteClusterStateUtils.encodeString(CLUSTER_NAME), pathTokens[0]); - assertEquals("cluster-state", pathTokens[1]); + assertEquals(CLUSTER_STATE_PATH_TOKEN, pathTokens[1]); assertEquals(CLUSTER_UUID, pathTokens[2]); assertEquals(GLOBAL_METADATA_PATH_TOKEN, pathTokens[3]); String[] splitFileName = pathTokens[4].split(DELIMITER); @@ -576,19 +561,19 @@ public void testGetReadMetadataAsyncAction_GlobalMetadata() throws Exception { when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( GLOBAL_METADATA_FORMAT.serialize(metadata, fileName, compressor, FORMAT_PARAMS).streamInput() ); - LatchedActionListener listener = mock(LatchedActionListener.class); - CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataReadAction( + RemoteStateTestUtil.TestCapturingListener listener = new RemoteStateTestUtil.TestCapturingListener<>(); + CountDownLatch latch = new CountDownLatch(1); + remoteGlobalMetadataManager.getAsyncMetadataReadAction( globalMetadataForDownload, GLOBAL_METADATA, - listener - ); - runnable.run(); - - ArgumentCaptor savedResult = ArgumentCaptor.forClass(RemoteReadResult.class); - assertBusy(() -> verify(listener, times(1)).onResponse(savedResult.capture())); - assertTrue(isGlobalStateEquals(metadata, (Metadata) savedResult.getValue().getObj())); - assertEquals(GLOBAL_METADATA, savedResult.getValue().getComponent()); - assertEquals(GLOBAL_METADATA, savedResult.getValue().getComponentName()); + new LatchedActionListener<>(listener, latch) + ).run(); + latch.await(); + assertNull(listener.failure); + assertNotNull(listener.result); + assertTrue(isGlobalStateEquals(metadata, (Metadata) listener.result.getObj())); + assertEquals(GLOBAL_METADATA, listener.result.getComponent()); + assertEquals(GLOBAL_METADATA, listener.result.getComponentName()); } public void testGetReadMetadataAsyncAction_IOException() throws Exception { @@ -601,16 +586,17 @@ public void testGetReadMetadataAsyncAction_IOException() throws Exception { ); IOException ioException = new IOException("mock test exception"); when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenThrow(ioException); - LatchedActionListener listener = mock(LatchedActionListener.class); - CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataReadAction( + RemoteStateTestUtil.TestCapturingListener listener = new RemoteStateTestUtil.TestCapturingListener<>(); + CountDownLatch latch = new CountDownLatch(1); + remoteGlobalMetadataManager.getAsyncMetadataReadAction( coordinationMetadataForDownload, COORDINATION_METADATA, - listener - ); - runnable.run(); - ArgumentCaptor savedException = ArgumentCaptor.forClass(Exception.class); - assertBusy(() -> verify(listener, times(1)).onFailure(savedException.capture())); - assertEquals(ioException, savedException.getValue()); + new LatchedActionListener<>(listener, latch) + ).run(); + latch.await(); + assertNull(listener.result); + assertNotNull(listener.failure); + assertEquals(ioException, listener.failure); } public void testGetAsyncMetadataWriteAction_IOException() throws Exception { @@ -622,22 +608,22 @@ public void testGetAsyncMetadataWriteAction_IOException() throws Exception { compressor, xContentRegistry ); + IOException ioException = new IOException("mock test exception"); doAnswer(invocationOnMock -> { - invocationOnMock.getArgument(4, ActionListener.class).onFailure(new IOException("mock test exception")); + invocationOnMock.getArgument(4, ActionListener.class).onFailure(ioException); return null; }).when(blobStoreTransferService) .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); - LatchedActionListener listener = mock(LatchedActionListener.class); - CheckedRunnable runnable = remoteGlobalMetadataManager.getAsyncMetadataWriteAction( - remoteCoordinationMetadata, - listener - ); - runnable.run(); - - ArgumentCaptor savedException = ArgumentCaptor.forClass(Exception.class); - assertBusy(() -> verify(listener, times(1)).onFailure(savedException.capture())); - assertTrue(savedException.getValue() instanceof RemoteStateTransferException); + RemoteStateTestUtil.TestCapturingListener listener = + new RemoteStateTestUtil.TestCapturingListener<>(); + CountDownLatch latch = new CountDownLatch(1); + remoteGlobalMetadataManager.getAsyncMetadataWriteAction(remoteCoordinationMetadata, new LatchedActionListener<>(listener, latch)) + .run(); + assertNull(listener.result); + assertNotNull(listener.failure); + assertTrue(listener.failure instanceof RemoteStateTransferException); + assertEquals(ioException, listener.failure.getCause()); } public void testGetUpdatedCustoms() { diff --git a/server/src/test/java/org/opensearch/gateway/remote/RemoteStateTestUtil.java b/server/src/test/java/org/opensearch/gateway/remote/RemoteStateTestUtil.java new file mode 100644 index 0000000000000..50d17651602bf --- /dev/null +++ b/server/src/test/java/org/opensearch/gateway/remote/RemoteStateTestUtil.java @@ -0,0 +1,28 @@ +/* + * 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; + +import org.opensearch.core.action.ActionListener; + +public class RemoteStateTestUtil { + public static class TestCapturingListener implements ActionListener { + T result; + Exception failure; + + @Override + public void onResponse(T result) { + this.result = result; + } + + @Override + public void onFailure(Exception e) { + this.failure = e; + } + } +} From 5924b9228a634c2cde2d3274f696f6c16139ad6f Mon Sep 17 00:00:00 2001 From: Shivansh Arora Date: Wed, 19 Jun 2024 13:31:25 +0530 Subject: [PATCH 3/4] Move TestCapturingListener to test/framework Signed-off-by: Shivansh Arora --- .../RemoteGlobalMetadataManagerTests.java | 158 +++++++++--------- .../gateway/remote/RemoteStateTestUtil.java | 28 ---- .../common/util/TestCapturingListener.java | 34 ++++ 3 files changed, 110 insertions(+), 110 deletions(-) delete mode 100644 server/src/test/java/org/opensearch/gateway/remote/RemoteStateTestUtil.java create mode 100644 test/framework/src/main/java/org/opensearch/common/util/TestCapturingListener.java diff --git a/server/src/test/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManagerTests.java b/server/src/test/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManagerTests.java index 1b361e7818bf0..c543f986b3e86 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManagerTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManagerTests.java @@ -24,6 +24,7 @@ import org.opensearch.common.network.NetworkModule; import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; +import org.opensearch.common.util.TestCapturingListener; import org.opensearch.core.action.ActionListener; import org.opensearch.core.common.io.stream.NamedWriteableRegistry; import org.opensearch.core.compress.Compressor; @@ -168,7 +169,7 @@ public void testGetReadMetadataAsyncAction_CoordinationMetadata() throws Excepti when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( COORDINATION_METADATA_FORMAT.serialize(coordinationMetadata, fileName, compressor, FORMAT_PARAMS).streamInput() ); - RemoteStateTestUtil.TestCapturingListener listener = new RemoteStateTestUtil.TestCapturingListener<>(); + TestCapturingListener listener = new TestCapturingListener<>(); CountDownLatch latch = new CountDownLatch(1); remoteGlobalMetadataManager.getAsyncMetadataReadAction( @@ -177,11 +178,11 @@ public void testGetReadMetadataAsyncAction_CoordinationMetadata() throws Excepti new LatchedActionListener<>(listener, latch) ).run(); latch.await(); - assertNull(listener.failure); - assertNotNull(listener.result); - assertEquals(coordinationMetadata, listener.result.getObj()); - assertEquals(COORDINATION_METADATA, listener.result.getComponent()); - assertEquals(COORDINATION_METADATA, listener.result.getComponentName()); + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + assertEquals(coordinationMetadata, listener.getResult().getObj()); + assertEquals(COORDINATION_METADATA, listener.getResult().getComponent()); + assertEquals(COORDINATION_METADATA, listener.getResult().getComponentName()); } public void testGetAsyncMetadataWriteAction_CoordinationMetadata() throws Exception { @@ -198,16 +199,15 @@ public void testGetAsyncMetadataWriteAction_CoordinationMetadata() throws Except return null; }).when(blobStoreTransferService) .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); - RemoteStateTestUtil.TestCapturingListener listener = - new RemoteStateTestUtil.TestCapturingListener<>(); + TestCapturingListener listener = new TestCapturingListener<>(); CountDownLatch latch = new CountDownLatch(1); remoteGlobalMetadataManager.getAsyncMetadataWriteAction(remoteCoordinationMetadata, new LatchedActionListener<>(listener, latch)) .run(); latch.await(); - assertNull(listener.failure); - assertNotNull(listener.result); - ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.result; + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.getResult(); assertEquals(COORDINATION_METADATA, uploadedMetadata.getComponent()); String uploadedFileName = uploadedMetadata.getUploadedFilename(); String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); @@ -236,7 +236,7 @@ public void testGetReadMetadataAsyncAction_PersistentSettings() throws Exception RemotePersistentSettingsMetadata.SETTINGS_METADATA_FORMAT.serialize(settingsMetadata, fileName, compressor, FORMAT_PARAMS) .streamInput() ); - RemoteStateTestUtil.TestCapturingListener listener = new RemoteStateTestUtil.TestCapturingListener<>(); + TestCapturingListener listener = new TestCapturingListener<>(); CountDownLatch latch = new CountDownLatch(1); remoteGlobalMetadataManager.getAsyncMetadataReadAction( @@ -245,11 +245,11 @@ public void testGetReadMetadataAsyncAction_PersistentSettings() throws Exception new LatchedActionListener<>(listener, latch) ).run(); latch.await(); - assertNull(listener.failure); - assertNotNull(listener.result); - assertEquals(settingsMetadata, listener.result.getObj()); - assertEquals(SETTING_METADATA, listener.result.getComponent()); - assertEquals(SETTING_METADATA, listener.result.getComponentName()); + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + assertEquals(settingsMetadata, listener.getResult().getObj()); + assertEquals(SETTING_METADATA, listener.getResult().getComponent()); + assertEquals(SETTING_METADATA, listener.getResult().getComponentName()); } public void testGetAsyncMetadataWriteAction_PersistentSettings() throws Exception { @@ -266,15 +266,14 @@ public void testGetAsyncMetadataWriteAction_PersistentSettings() throws Exceptio return null; }).when(blobStoreTransferService) .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); - RemoteStateTestUtil.TestCapturingListener listener = - new RemoteStateTestUtil.TestCapturingListener<>(); + TestCapturingListener listener = new TestCapturingListener<>(); CountDownLatch latch = new CountDownLatch(1); remoteGlobalMetadataManager.getAsyncMetadataWriteAction(persistentSettings, new LatchedActionListener<>(listener, latch)).run(); latch.await(); - assertNull(listener.failure); - assertNotNull(listener.result); - ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.result; + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.getResult(); assertEquals(SETTING_METADATA, uploadedMetadata.getComponent()); String uploadedFileName = uploadedMetadata.getUploadedFilename(); String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); @@ -303,7 +302,7 @@ public void testGetReadMetadataAsyncAction_TransientSettings() throws Exception RemoteTransientSettingsMetadata.SETTINGS_METADATA_FORMAT.serialize(settingsMetadata, fileName, compressor, FORMAT_PARAMS) .streamInput() ); - RemoteStateTestUtil.TestCapturingListener listener = new RemoteStateTestUtil.TestCapturingListener<>(); + TestCapturingListener listener = new TestCapturingListener<>(); CountDownLatch latch = new CountDownLatch(1); remoteGlobalMetadataManager.getAsyncMetadataReadAction( @@ -312,11 +311,11 @@ public void testGetReadMetadataAsyncAction_TransientSettings() throws Exception new LatchedActionListener<>(listener, latch) ).run(); latch.await(); - assertNull(listener.failure); - assertNotNull(listener.result); - assertEquals(settingsMetadata, listener.result.getObj()); - assertEquals(TRANSIENT_SETTING_METADATA, listener.result.getComponent()); - assertEquals(TRANSIENT_SETTING_METADATA, listener.result.getComponentName()); + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + assertEquals(settingsMetadata, listener.getResult().getObj()); + assertEquals(TRANSIENT_SETTING_METADATA, listener.getResult().getComponent()); + assertEquals(TRANSIENT_SETTING_METADATA, listener.getResult().getComponentName()); } public void testGetAsyncMetadataWriteAction_TransientSettings() throws Exception { @@ -333,14 +332,13 @@ public void testGetAsyncMetadataWriteAction_TransientSettings() throws Exception return null; }).when(blobStoreTransferService) .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); - RemoteStateTestUtil.TestCapturingListener listener = - new RemoteStateTestUtil.TestCapturingListener<>(); + TestCapturingListener listener = new TestCapturingListener<>(); CountDownLatch latch = new CountDownLatch(1); remoteGlobalMetadataManager.getAsyncMetadataWriteAction(transientSettings, new LatchedActionListener<>(listener, latch)).run(); latch.await(); - assertNull(listener.failure); - assertNotNull(listener.result); - ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.result; + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.getResult(); assertEquals(TRANSIENT_SETTING_METADATA, uploadedMetadata.getComponent()); String uploadedFileName = uploadedMetadata.getUploadedFilename(); String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); @@ -367,7 +365,7 @@ public void testGetReadMetadataAsyncAction_HashesOfConsistentSettings() throws E when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( HASHES_OF_CONSISTENT_SETTINGS_FORMAT.serialize(hashesOfConsistentSettings, fileName, compressor).streamInput() ); - RemoteStateTestUtil.TestCapturingListener listener = new RemoteStateTestUtil.TestCapturingListener<>(); + TestCapturingListener listener = new TestCapturingListener<>(); CountDownLatch latch = new CountDownLatch(1); remoteGlobalMetadataManager.getAsyncMetadataReadAction( @@ -376,11 +374,11 @@ public void testGetReadMetadataAsyncAction_HashesOfConsistentSettings() throws E new LatchedActionListener<>(listener, latch) ).run(); latch.await(); - assertNull(listener.failure); - assertNotNull(listener.result); - assertEquals(hashesOfConsistentSettings, listener.result.getObj()); - assertEquals(HASHES_OF_CONSISTENT_SETTINGS, listener.result.getComponent()); - assertEquals(HASHES_OF_CONSISTENT_SETTINGS, listener.result.getComponentName()); + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + assertEquals(hashesOfConsistentSettings, listener.getResult().getObj()); + assertEquals(HASHES_OF_CONSISTENT_SETTINGS, listener.getResult().getComponent()); + assertEquals(HASHES_OF_CONSISTENT_SETTINGS, listener.getResult().getComponentName()); } public void testGetAsyncMetadataWriteAction_HashesOfConsistentSettings() throws Exception { @@ -396,17 +394,16 @@ public void testGetAsyncMetadataWriteAction_HashesOfConsistentSettings() throws return null; }).when(blobStoreTransferService) .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); - RemoteStateTestUtil.TestCapturingListener listener = - new RemoteStateTestUtil.TestCapturingListener<>(); + TestCapturingListener listener = new TestCapturingListener<>(); CountDownLatch latch = new CountDownLatch(1); remoteGlobalMetadataManager.getAsyncMetadataWriteAction( hashesOfConsistentSettingsForUpload, new LatchedActionListener<>(listener, latch) ).run(); latch.await(); - assertNull(listener.failure); - assertNotNull(listener.result); - ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.result; + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.getResult(); assertEquals(HASHES_OF_CONSISTENT_SETTINGS, uploadedMetadata.getComponent()); String uploadedFileName = uploadedMetadata.getUploadedFilename(); String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); @@ -434,7 +431,7 @@ public void testGetReadMetadataAsyncAction_TemplatesMetadata() throws Exception when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( TEMPLATES_METADATA_FORMAT.serialize(templatesMetadata, fileName, compressor, FORMAT_PARAMS).streamInput() ); - RemoteStateTestUtil.TestCapturingListener listener = new RemoteStateTestUtil.TestCapturingListener<>(); + TestCapturingListener listener = new TestCapturingListener<>(); CountDownLatch latch = new CountDownLatch(1); remoteGlobalMetadataManager.getAsyncMetadataReadAction( templatesMetadataForDownload, @@ -442,11 +439,11 @@ public void testGetReadMetadataAsyncAction_TemplatesMetadata() throws Exception new LatchedActionListener<>(listener, latch) ).run(); latch.await(); - assertNull(listener.failure); - assertNotNull(listener.result); - assertEquals(templatesMetadata, listener.result.getObj()); - assertEquals(TEMPLATES_METADATA, listener.result.getComponent()); - assertEquals(TEMPLATES_METADATA, listener.result.getComponentName()); + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + assertEquals(templatesMetadata, listener.getResult().getObj()); + assertEquals(TEMPLATES_METADATA, listener.getResult().getComponent()); + assertEquals(TEMPLATES_METADATA, listener.getResult().getComponentName()); } public void testGetAsyncMetadataWriteAction_TemplatesMetadata() throws Exception { @@ -463,15 +460,14 @@ public void testGetAsyncMetadataWriteAction_TemplatesMetadata() throws Exception return null; }).when(blobStoreTransferService) .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); - RemoteStateTestUtil.TestCapturingListener listener = - new RemoteStateTestUtil.TestCapturingListener<>(); + TestCapturingListener listener = new TestCapturingListener<>(); CountDownLatch latch = new CountDownLatch(1); remoteGlobalMetadataManager.getAsyncMetadataWriteAction(templateMetadataForUpload, new LatchedActionListener<>(listener, latch)) .run(); latch.await(); - assertNull(listener.failure); - assertNotNull(listener.result); - ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.result; + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.getResult(); assertEquals(TEMPLATES_METADATA, uploadedMetadata.getComponent()); String uploadedFileName = uploadedMetadata.getUploadedFilename(); String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); @@ -500,7 +496,7 @@ public void testGetReadMetadataAsyncAction_CustomMetadata() throws Exception { when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( customMetadataForDownload.customBlobStoreFormat.serialize(customMetadata, fileName, compressor).streamInput() ); - RemoteStateTestUtil.TestCapturingListener listener = new RemoteStateTestUtil.TestCapturingListener<>(); + TestCapturingListener listener = new TestCapturingListener<>(); CountDownLatch latch = new CountDownLatch(1); remoteGlobalMetadataManager.getAsyncMetadataReadAction( customMetadataForDownload, @@ -508,11 +504,11 @@ public void testGetReadMetadataAsyncAction_CustomMetadata() throws Exception { new LatchedActionListener<>(listener, latch) ).run(); latch.await(); - assertNull(listener.failure); - assertNotNull(listener.result); - assertEquals(customMetadata, listener.result.getObj()); - assertEquals(CUSTOM_METADATA, listener.result.getComponent()); - assertEquals(IndexGraveyard.TYPE, listener.result.getComponentName()); + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + assertEquals(customMetadata, listener.getResult().getObj()); + assertEquals(CUSTOM_METADATA, listener.getResult().getComponent()); + assertEquals(IndexGraveyard.TYPE, listener.getResult().getComponentName()); } public void testGetAsyncMetadataWriteAction_CustomMetadata() throws Exception { @@ -530,15 +526,14 @@ public void testGetAsyncMetadataWriteAction_CustomMetadata() throws Exception { return null; }).when(blobStoreTransferService) .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); - RemoteStateTestUtil.TestCapturingListener listener = - new RemoteStateTestUtil.TestCapturingListener<>(); + TestCapturingListener listener = new TestCapturingListener<>(); CountDownLatch latch = new CountDownLatch(1); remoteGlobalMetadataManager.getAsyncMetadataWriteAction(customMetadataForUpload, new LatchedActionListener<>(listener, latch)) .run(); latch.await(); - assertNull(listener.failure); - assertNotNull(listener.result); - ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.result; + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + ClusterMetadataManifest.UploadedMetadata uploadedMetadata = listener.getResult(); assertEquals(String.join(CUSTOM_DELIMITER, CUSTOM_METADATA, IndexGraveyard.TYPE), uploadedMetadata.getComponent()); String uploadedFileName = uploadedMetadata.getUploadedFilename(); String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); @@ -561,7 +556,7 @@ public void testGetReadMetadataAsyncAction_GlobalMetadata() throws Exception { when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( GLOBAL_METADATA_FORMAT.serialize(metadata, fileName, compressor, FORMAT_PARAMS).streamInput() ); - RemoteStateTestUtil.TestCapturingListener listener = new RemoteStateTestUtil.TestCapturingListener<>(); + TestCapturingListener listener = new TestCapturingListener<>(); CountDownLatch latch = new CountDownLatch(1); remoteGlobalMetadataManager.getAsyncMetadataReadAction( globalMetadataForDownload, @@ -569,11 +564,11 @@ public void testGetReadMetadataAsyncAction_GlobalMetadata() throws Exception { new LatchedActionListener<>(listener, latch) ).run(); latch.await(); - assertNull(listener.failure); - assertNotNull(listener.result); - assertTrue(isGlobalStateEquals(metadata, (Metadata) listener.result.getObj())); - assertEquals(GLOBAL_METADATA, listener.result.getComponent()); - assertEquals(GLOBAL_METADATA, listener.result.getComponentName()); + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + assertTrue(isGlobalStateEquals(metadata, (Metadata) listener.getResult().getObj())); + assertEquals(GLOBAL_METADATA, listener.getResult().getComponent()); + assertEquals(GLOBAL_METADATA, listener.getResult().getComponentName()); } public void testGetReadMetadataAsyncAction_IOException() throws Exception { @@ -586,7 +581,7 @@ public void testGetReadMetadataAsyncAction_IOException() throws Exception { ); IOException ioException = new IOException("mock test exception"); when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenThrow(ioException); - RemoteStateTestUtil.TestCapturingListener listener = new RemoteStateTestUtil.TestCapturingListener<>(); + TestCapturingListener listener = new TestCapturingListener<>(); CountDownLatch latch = new CountDownLatch(1); remoteGlobalMetadataManager.getAsyncMetadataReadAction( coordinationMetadataForDownload, @@ -594,9 +589,9 @@ public void testGetReadMetadataAsyncAction_IOException() throws Exception { new LatchedActionListener<>(listener, latch) ).run(); latch.await(); - assertNull(listener.result); - assertNotNull(listener.failure); - assertEquals(ioException, listener.failure); + assertNull(listener.getResult()); + assertNotNull(listener.getFailure()); + assertEquals(ioException, listener.getFailure()); } public void testGetAsyncMetadataWriteAction_IOException() throws Exception { @@ -615,15 +610,14 @@ public void testGetAsyncMetadataWriteAction_IOException() throws Exception { }).when(blobStoreTransferService) .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); - RemoteStateTestUtil.TestCapturingListener listener = - new RemoteStateTestUtil.TestCapturingListener<>(); + TestCapturingListener listener = new TestCapturingListener<>(); CountDownLatch latch = new CountDownLatch(1); remoteGlobalMetadataManager.getAsyncMetadataWriteAction(remoteCoordinationMetadata, new LatchedActionListener<>(listener, latch)) .run(); - assertNull(listener.result); - assertNotNull(listener.failure); - assertTrue(listener.failure instanceof RemoteStateTransferException); - assertEquals(ioException, listener.failure.getCause()); + assertNull(listener.getResult()); + assertNotNull(listener.getFailure()); + assertTrue(listener.getFailure() instanceof RemoteStateTransferException); + assertEquals(ioException, listener.getFailure().getCause()); } public void testGetUpdatedCustoms() { diff --git a/server/src/test/java/org/opensearch/gateway/remote/RemoteStateTestUtil.java b/server/src/test/java/org/opensearch/gateway/remote/RemoteStateTestUtil.java deleted file mode 100644 index 50d17651602bf..0000000000000 --- a/server/src/test/java/org/opensearch/gateway/remote/RemoteStateTestUtil.java +++ /dev/null @@ -1,28 +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; - -import org.opensearch.core.action.ActionListener; - -public class RemoteStateTestUtil { - public static class TestCapturingListener implements ActionListener { - T result; - Exception failure; - - @Override - public void onResponse(T result) { - this.result = result; - } - - @Override - public void onFailure(Exception e) { - this.failure = e; - } - } -} diff --git a/test/framework/src/main/java/org/opensearch/common/util/TestCapturingListener.java b/test/framework/src/main/java/org/opensearch/common/util/TestCapturingListener.java new file mode 100644 index 0000000000000..b970df77f7399 --- /dev/null +++ b/test/framework/src/main/java/org/opensearch/common/util/TestCapturingListener.java @@ -0,0 +1,34 @@ +/* + * 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.common.util; + +import org.opensearch.core.action.ActionListener; + +public class TestCapturingListener implements ActionListener { + T result; + Exception failure; + + @Override + public void onResponse(T result) { + this.result = result; + } + + @Override + public void onFailure(Exception e) { + this.failure = e; + } + + public T getResult() { + return result; + } + + public Exception getFailure() { + return failure; + } +} From b456243a5ee0c01084cb1e7c1aedd89900214f19 Mon Sep 17 00:00:00 2001 From: Shivansh Arora Date: Wed, 19 Jun 2024 16:16:30 +0530 Subject: [PATCH 4/4] Added javadoc Signed-off-by: Shivansh Arora --- .../opensearch/common/util/TestCapturingListener.java | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/test/framework/src/main/java/org/opensearch/common/util/TestCapturingListener.java b/test/framework/src/main/java/org/opensearch/common/util/TestCapturingListener.java index b970df77f7399..a3c8cc15de927 100644 --- a/test/framework/src/main/java/org/opensearch/common/util/TestCapturingListener.java +++ b/test/framework/src/main/java/org/opensearch/common/util/TestCapturingListener.java @@ -10,9 +10,14 @@ import org.opensearch.core.action.ActionListener; +/** + * A simple implementation of {@link ActionListener} that captures the response and failures used for testing purposes. + * + * @param the result type + */ public class TestCapturingListener implements ActionListener { - T result; - Exception failure; + private T result; + private Exception failure; @Override public void onResponse(T result) {