diff --git a/storm-server/src/main/java/org/apache/storm/blobstore/BlobStoreUtils.java b/storm-server/src/main/java/org/apache/storm/blobstore/BlobStoreUtils.java index cc1d5455067..f42b0c7e03e 100644 --- a/storm-server/src/main/java/org/apache/storm/blobstore/BlobStoreUtils.java +++ b/storm-server/src/main/java/org/apache/storm/blobstore/BlobStoreUtils.java @@ -150,6 +150,7 @@ public static boolean downloadMissingBlob(Map conf, BlobStore bl throw new RuntimeException(exception); } catch (KeyAlreadyExistsException kae) { LOG.info("KeyAlreadyExistsException Key: {} {}", key, kae); + isSuccess = true; } catch (KeyNotFoundException knf) { // Catching and logging KeyNotFoundException because, if // there is a subsequent update and delete, the non-leader diff --git a/storm-server/src/test/java/org/apache/storm/blobstore/BlobStoreUtilsTest.java b/storm-server/src/test/java/org/apache/storm/blobstore/BlobStoreUtilsTest.java index fa435269196..24fb1b59bab 100644 --- a/storm-server/src/test/java/org/apache/storm/blobstore/BlobStoreUtilsTest.java +++ b/storm-server/src/test/java/org/apache/storm/blobstore/BlobStoreUtilsTest.java @@ -12,17 +12,35 @@ package org.apache.storm.blobstore; +import java.io.IOException; +import java.io.InputStream; +import java.util.Collections; import java.util.List; import java.util.Map; + +import org.apache.storm.generated.*; import org.apache.storm.nimbus.NimbusInfo; +import org.apache.storm.thrift.TException; +import org.apache.storm.utils.NimbusClient; import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import javax.security.auth.Subject; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyMap; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.when; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.atLeastOnce; + + public class BlobStoreUtilsTest { @@ -126,4 +144,33 @@ public void testUpdateKeyForBlobStore_noMatch() { verify(nimbusDetails).getHost(); verify(conf, atLeastOnce()).get(anyString()); } + + @Test + public void testDownloadMissingBlob_KeyAkreadyExists() throws TException, IOException { + + NimbusClient.Builder builder1 = mock(NimbusClient.Builder.class); + NimbusClient.Builder builder2 = mock(NimbusClient.Builder.class); + NimbusClient client = mock(NimbusClient.class); + ReadableBlobMeta readableBlobMeta = mock(ReadableBlobMeta.class); + Nimbus.Iface iface = mock(Nimbus.Iface.class); + + + try (MockedStatic mockedNimbusClient = Mockito.mockStatic(NimbusClient.Builder.class)) { + + mockedNimbusClient.when(() ->NimbusClient.Builder.withConf(anyMap())).thenReturn(builder1); + when(builder1.forDaemon()).thenReturn(builder2); + when(builder2.buildWithNimbusHostPort(anyString(),anyInt())).thenReturn(client); + when(client.getClient()).thenReturn(iface); + when(iface.getBlobMeta(anyString())).thenReturn(readableBlobMeta); + when(readableBlobMeta.get_settable()).thenReturn(new SettableBlobMeta()); + when(iface.beginBlobDownload(anyString())).thenReturn(new BeginDownloadResult()); + when(nimbusDetails.getHost()).thenReturn("localhost"); + when(nimbusDetails.getPort()).thenReturn(1234); + + doThrow(new KeyAlreadyExistsException()).when(blobStore).createBlob(anyString(),any(InputStream.class),any(SettableBlobMeta.class),any(Subject.class)); + + assertTrue((BlobStoreUtils.downloadMissingBlob(conf, blobStore, "testKey", Collections.singleton(nimbusDetails)))); + } + + } }