STORM-3504: changed AsyncLocalizerTest to not mock AdvancedFSOps (#3126)
* STORM-3504: changed AsyncLocalizerTest to not mock AdvancedFSOps
diff --git a/storm-server/src/test/java/org/apache/storm/localizer/AsyncLocalizerTest.java b/storm-server/src/test/java/org/apache/storm/localizer/AsyncLocalizerTest.java
index 550c808..d3c1d74 100644
--- a/storm-server/src/test/java/org/apache/storm/localizer/AsyncLocalizerTest.java
+++ b/storm-server/src/test/java/org/apache/storm/localizer/AsyncLocalizerTest.java
@@ -18,7 +18,6 @@
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
-import java.io.OutputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
@@ -28,13 +27,12 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
-import org.apache.commons.io.FileUtils;
+
import org.apache.commons.io.IOUtils;
import org.apache.storm.Config;
import org.apache.storm.DaemonConfig;
@@ -53,13 +51,12 @@
import org.apache.storm.generated.SettableBlobMeta;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.security.auth.DefaultPrincipalToLocal;
+import org.apache.storm.testing.TmpPath;
import org.apache.storm.utils.ConfigUtils;
import org.apache.storm.utils.ReflectionUtils;
import org.apache.storm.utils.ServerUtils;
import org.apache.storm.utils.Time;
import org.apache.storm.utils.Utils;
-import org.junit.After;
-import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
import org.slf4j.Logger;
@@ -86,68 +83,66 @@
public class AsyncLocalizerTest {
private static final Logger LOG = LoggerFactory.getLogger(AsyncLocalizerTest.class);
+
private final String user1 = "user1";
private final String user2 = "user2";
private final String user3 = "user3";
- //From LocalizerTest
- private File baseDir;
- private ClientBlobStore mockblobstore = mock(ClientBlobStore.class);
- private static String getTestLocalizerRoot() {
- File f = new File("./target/" + Thread.currentThread().getStackTrace()[2].getMethodName() + "/localizer/");
- f.deleteOnExit();
- return f.getPath();
- }
+ private ClientBlobStore mockBlobStore = mock(ClientBlobStore.class);
@Test
public void testRequestDownloadBaseTopologyBlobs() throws Exception {
- final String topoId = "TOPO";
- final String user = "user";
- LocalAssignment la = new LocalAssignment();
- la.set_topology_id(topoId);
- la.set_owner(user);
- ExecutorInfo ei = new ExecutorInfo();
- ei.set_task_start(1);
- ei.set_task_end(1);
- la.add_to_executors(ei);
- final int port = 8080;
- final String stormLocal = "./target/DOWNLOAD-TEST/storm-local/";
- ClientBlobStore blobStore = mock(ClientBlobStore.class);
- Map<String, Object> conf = new HashMap<>();
- conf.put(DaemonConfig.SUPERVISOR_BLOBSTORE, ClientBlobStore.class.getName());
- conf.put(Config.STORM_PRINCIPAL_TO_LOCAL_PLUGIN, DefaultPrincipalToLocal.class.getName());
- conf.put(Config.STORM_CLUSTER_MODE, "distributed");
- conf.put(Config.STORM_LOCAL_DIR, stormLocal);
- AdvancedFSOps ops = mock(AdvancedFSOps.class);
- ReflectionUtils mockedRU = mock(ReflectionUtils.class);
- ServerUtils mockedU = mock(ServerUtils.class);
+ ReflectionUtils mockedReflectionUtils = mock(ReflectionUtils.class);
+ ServerUtils mockedServerUtils = mock(ServerUtils.class);
- AsyncLocalizer bl = spy(new AsyncLocalizer(conf, ops, getTestLocalizerRoot(), new StormMetricsRegistry()));
- LocallyCachedTopologyBlob jarBlob = mock(LocallyCachedTopologyBlob.class);
- doReturn(jarBlob).when(bl).getTopoJar(topoId, la.get_owner());
- when(jarBlob.getLocalVersion()).thenReturn(-1L);
- when(jarBlob.getRemoteVersion(any())).thenReturn(100L);
- when(jarBlob.fetchUnzipToTemp(any())).thenReturn(100L);
+ ReflectionUtils previousReflectionUtils = ReflectionUtils.setInstance(mockedReflectionUtils);
+ ServerUtils previousServerUtils = ServerUtils.setInstance(mockedServerUtils);
- LocallyCachedTopologyBlob codeBlob = mock(LocallyCachedTopologyBlob.class);
- doReturn(codeBlob).when(bl).getTopoCode(topoId, la.get_owner());
- when(codeBlob.getLocalVersion()).thenReturn(-1L);
- when(codeBlob.getRemoteVersion(any())).thenReturn(200L);
- when(codeBlob.fetchUnzipToTemp(any())).thenReturn(200L);
+ // cannot use automatic resource management here in this try because the AsyncLocalizer depends on a config map,
+ // which should take the storm local dir, and that storm local dir is declared in the try-with-resources.
+ AsyncLocalizer victim = null;
- LocallyCachedTopologyBlob confBlob = mock(LocallyCachedTopologyBlob.class);
- doReturn(confBlob).when(bl).getTopoConf(topoId, la.get_owner());
- when(confBlob.getLocalVersion()).thenReturn(-1L);
- when(confBlob.getRemoteVersion(any())).thenReturn(300L);
- when(confBlob.fetchUnzipToTemp(any())).thenReturn(300L);
+ try (TmpPath stormRoot = new TmpPath(); TmpPath localizerRoot = new TmpPath()) {
- ReflectionUtils origRU = ReflectionUtils.setInstance(mockedRU);
- ServerUtils origUtils = ServerUtils.setInstance(mockedU);
- try {
- when(mockedRU.newInstanceImpl(ClientBlobStore.class)).thenReturn(blobStore);
+ Map<String, Object> conf = new HashMap<>();
+ conf.put(DaemonConfig.SUPERVISOR_BLOBSTORE, ClientBlobStore.class.getName());
+ conf.put(Config.STORM_PRINCIPAL_TO_LOCAL_PLUGIN, DefaultPrincipalToLocal.class.getName());
+ conf.put(Config.STORM_CLUSTER_MODE, "distributed");
+ conf.put(Config.STORM_LOCAL_DIR, stormRoot.getPath());
- PortAndAssignment pna = new PortAndAssignmentImpl(port, la);
- Future<Void> f = bl.requestDownloadBaseTopologyBlobs(pna, null);
+ AdvancedFSOps ops = AdvancedFSOps.make(conf);
+
+ victim = spy(new AsyncLocalizer(conf, ops, localizerRoot.getPath(), new StormMetricsRegistry()));
+
+ final String topoId = "TOPO";
+
+ final LocalAssignment localAssignment = constructLocalAssignment(topoId, "user");
+ final int port = 8080;
+
+ ClientBlobStore blobStore = mock(ClientBlobStore.class);
+
+ LocallyCachedTopologyBlob jarBlob = mock(LocallyCachedTopologyBlob.class);
+ doReturn(jarBlob).when(victim).getTopoJar(topoId, localAssignment.get_owner());
+ when(jarBlob.getLocalVersion()).thenReturn(-1L);
+ when(jarBlob.getRemoteVersion(any())).thenReturn(100L);
+ when(jarBlob.fetchUnzipToTemp(any())).thenReturn(100L);
+
+ LocallyCachedTopologyBlob codeBlob = mock(LocallyCachedTopologyBlob.class);
+ doReturn(codeBlob).when(victim).getTopoCode(topoId, localAssignment.get_owner());
+ when(codeBlob.getLocalVersion()).thenReturn(-1L);
+ when(codeBlob.getRemoteVersion(any())).thenReturn(200L);
+ when(codeBlob.fetchUnzipToTemp(any())).thenReturn(200L);
+
+ LocallyCachedTopologyBlob confBlob = mock(LocallyCachedTopologyBlob.class);
+ doReturn(confBlob).when(victim).getTopoConf(topoId, localAssignment.get_owner());
+ when(confBlob.getLocalVersion()).thenReturn(-1L);
+ when(confBlob.getRemoteVersion(any())).thenReturn(300L);
+ when(confBlob.fetchUnzipToTemp(any())).thenReturn(300L);
+
+ when(mockedReflectionUtils.newInstanceImpl(ClientBlobStore.class)).thenReturn(blobStore);
+
+ PortAndAssignment pna = new PortAndAssignmentImpl(port, localAssignment);
+ Future<Void> f = victim.requestDownloadBaseTopologyBlobs(pna, null);
f.get(20, TimeUnit.SECONDS);
verify(jarBlob).fetchUnzipToTemp(any());
@@ -161,92 +156,88 @@
verify(confBlob).fetchUnzipToTemp(any());
verify(confBlob).informReferencesAndCommitNewVersion(300L);
verify(confBlob).cleanupOrphanedData();
+
} finally {
- bl.close();
- ReflectionUtils.setInstance(origRU);
- ServerUtils.setInstance(origUtils);
+ ReflectionUtils.setInstance(previousReflectionUtils);
+ ServerUtils.setInstance(previousServerUtils);
+
+ if (victim != null) {
+ victim.close();
+ }
}
}
@Test
public void testRequestDownloadTopologyBlobs() throws Exception {
- final String topoId = "TOPO-12345";
- final String user = "user";
- LocalAssignment la = new LocalAssignment();
- la.set_topology_id(topoId);
- la.set_owner(user);
- ExecutorInfo ei = new ExecutorInfo();
- ei.set_task_start(1);
- ei.set_task_end(1);
- la.add_to_executors(ei);
- final String topoName = "TOPO";
- final int port = 8080;
- final String simpleLocalName = "simple.txt";
- final String simpleKey = "simple";
+ ConfigUtils mockedConfigUtils = mock(ConfigUtils.class);
+ ConfigUtils previousConfigUtils = ConfigUtils.setInstance(mockedConfigUtils);
- final String stormLocal = "/tmp/storm-local/";
- final File userDir = new File(stormLocal, user);
- final String stormRoot = stormLocal + topoId + "/";
+ AsyncLocalizer victim = null;
- final String localizerRoot = getTestLocalizerRoot();
- final String simpleCurrentLocalFile = localizerRoot + "/usercache/" + user + "/filecache/files/simple.current";
+ try (TmpPath stormLocal = new TmpPath(); TmpPath localizerRoot = new TmpPath()) {
- final StormTopology st = new StormTopology();
- st.set_spouts(new HashMap<>());
- st.set_bolts(new HashMap<>());
- st.set_state_spouts(new HashMap<>());
+ Map<String, Object> conf = new HashMap<>();
+ conf.put(Config.STORM_LOCAL_DIR, stormLocal.getPath());
- Map<String, Map<String, Object>> topoBlobMap = new HashMap<>();
- Map<String, Object> simple = new HashMap<>();
- simple.put("localname", simpleLocalName);
- simple.put("uncompress", false);
- topoBlobMap.put(simpleKey, simple);
+ AdvancedFSOps ops = AdvancedFSOps.make(conf);
+ StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
- Map<String, Object> conf = new HashMap<>();
- conf.put(Config.STORM_LOCAL_DIR, stormLocal);
- AdvancedFSOps ops = mock(AdvancedFSOps.class);
- ConfigUtils mockedCU = mock(ConfigUtils.class);
+ victim = spy(new AsyncLocalizer(conf, ops, localizerRoot.getPath(), metricsRegistry));
- Map<String, Object> topoConf = new HashMap<>(conf);
- topoConf.put(Config.TOPOLOGY_BLOBSTORE_MAP, topoBlobMap);
- topoConf.put(Config.TOPOLOGY_NAME, topoName);
+ final String topoId = "TOPO-12345";
+ final String user = "user";
- List<LocalizedResource> localizedList = new ArrayList<>();
- StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
- LocalizedResource simpleLocal = new LocalizedResource(simpleKey, Paths.get(localizerRoot), false, ops, conf, user, metricsRegistry);
- localizedList.add(simpleLocal);
+ final Path userDir = Paths.get(stormLocal.getPath(), user);
+ final Path topologyDirRoot = Paths.get(stormLocal.getPath(), topoId);
- AsyncLocalizer bl = spy(new AsyncLocalizer(conf, ops, localizerRoot, metricsRegistry));
- ConfigUtils orig = ConfigUtils.setInstance(mockedCU);
- try {
- when(mockedCU.supervisorStormDistRootImpl(conf, topoId)).thenReturn(stormRoot);
- when(mockedCU.readSupervisorStormConfImpl(conf, topoId)).thenReturn(topoConf);
- when(mockedCU.readSupervisorTopologyImpl(conf, topoId, ops)).thenReturn(st);
+ final String simpleLocalName = "simple.txt";
+ final String simpleKey = "simple";
+ Map<String, Map<String, Object>> topoBlobMap = new HashMap<>();
+ Map<String, Object> simple = new HashMap<>();
+ simple.put("localname", simpleLocalName);
+ simple.put("uncompress", false);
+ topoBlobMap.put(simpleKey, simple);
+
+ final int port = 8080;
+
+ Map<String, Object> topoConf = new HashMap<>(conf);
+ topoConf.put(Config.TOPOLOGY_BLOBSTORE_MAP, topoBlobMap);
+ topoConf.put(Config.TOPOLOGY_NAME, "TOPO");
+
+ List<LocalizedResource> localizedList = new ArrayList<>();
+ LocalizedResource simpleLocal = new LocalizedResource(simpleKey, localizerRoot.getFile().toPath(), false, ops, conf, user, metricsRegistry);
+ localizedList.add(simpleLocal);
+
+ when(mockedConfigUtils.supervisorStormDistRootImpl(conf, topoId)).thenReturn(topologyDirRoot.toString());
+ when(mockedConfigUtils.readSupervisorStormConfImpl(conf, topoId)).thenReturn(topoConf);
+ when(mockedConfigUtils.readSupervisorTopologyImpl(conf, topoId, ops)).thenReturn(constructEmptyStormTopology());
//Write the mocking backwards so the actual method is not called on the spy object
- doReturn(CompletableFuture.supplyAsync(() -> null)).when(bl)
- .requestDownloadBaseTopologyBlobs(any(), eq(null));
- doReturn(userDir).when(bl).getLocalUserFileCacheDir(user);
- doReturn(localizedList).when(bl).getBlobs(any(List.class), any(), any());
+ doReturn(CompletableFuture.supplyAsync(() -> null)).when(victim)
+ .requestDownloadBaseTopologyBlobs(any(), eq(null));
- Future<Void> f = bl.requestDownloadTopologyBlobs(la, port, null);
+ Files.createDirectories(topologyDirRoot);
+
+ doReturn(userDir.toFile()).when(victim).getLocalUserFileCacheDir(user);
+ doReturn(localizedList).when(victim).getBlobs(any(List.class), any(), any());
+
+ Future<Void> f = victim.requestDownloadTopologyBlobs(constructLocalAssignment(topoId, user), port, null);
f.get(20, TimeUnit.SECONDS);
// We should be done now...
- verify(bl).getLocalUserFileCacheDir(user);
+ verify(victim).getLocalUserFileCacheDir(user);
- verify(ops).fileExists(userDir);
- verify(ops).forceMkdir(userDir);
+ assertTrue(ops.fileExists(userDir));
- verify(bl).getBlobs(any(List.class), any(), any());
+ verify(victim).getBlobs(any(List.class), any(), any());
- verify(ops).createSymlink(new File(stormRoot, simpleLocalName), new File(simpleCurrentLocalFile));
+ // symlink was created
+ assertTrue(Files.isSymbolicLink(topologyDirRoot.resolve(simpleLocalName)));
+
} finally {
- try {
- ConfigUtils.setInstance(orig);
- bl.close();
- } catch (Throwable e) {
- LOG.error("ERROR trying to close an object", e);
+ ConfigUtils.setInstance(previousConfigUtils);
+ if (victim != null) {
+ victim.close();
}
}
}
@@ -255,226 +246,220 @@
@Test
public void testRequestDownloadTopologyBlobsLocalMode() throws Exception {
// tests download of topology blobs in local mode on a topology without resources folder
- final String topoId = "TOPO-12345";
- final String user = "user";
- LocalAssignment la = new LocalAssignment();
- la.set_topology_id(topoId);
- la.set_owner(user);
- ExecutorInfo ei = new ExecutorInfo();
- ei.set_task_start(1);
- ei.set_task_end(1);
- la.add_to_executors(ei);
- final String topoName = "TOPO";
- final int port = 8080;
- final String simpleLocalName = "simple.txt";
- final String simpleKey = "simple";
- final String stormLocal = "/tmp/storm-local/";
- final File userDir = new File(stormLocal, user);
- final String stormRoot = stormLocal + topoId + "/";
+ ConfigUtils mockedConfigUtils = mock(ConfigUtils.class);
+ ServerUtils mockedServerUtils = mock(ServerUtils.class);
- final String localizerRoot = getTestLocalizerRoot();
+ ConfigUtils previousConfigUtils = ConfigUtils.setInstance(mockedConfigUtils);
+ ServerUtils previousServerUtils = ServerUtils.setInstance(mockedServerUtils);
- final StormTopology st = new StormTopology();
- st.set_spouts(new HashMap<>());
- st.set_bolts(new HashMap<>());
- st.set_state_spouts(new HashMap<>());
+ AsyncLocalizer victim = null;
- Map<String, Map<String, Object>> topoBlobMap = new HashMap<>();
- Map<String, Object> simple = new HashMap<>();
- simple.put("localname", simpleLocalName);
- simple.put("uncompress", false);
- topoBlobMap.put(simpleKey, simple);
+ try (TmpPath stormLocal = new TmpPath(); TmpPath localizerRoot = new TmpPath()) {
- Map<String, Object> conf = new HashMap<>();
- conf.put(Config.STORM_LOCAL_DIR, stormLocal);
- conf.put(Config.STORM_CLUSTER_MODE, "local");
- AdvancedFSOps ops = mock(AdvancedFSOps.class);
- ConfigUtils mockedCU = mock(ConfigUtils.class);
- ServerUtils mockedSU = mock(ServerUtils.class);
+ Map<String, Object> conf = new HashMap<>();
+ conf.put(Config.STORM_LOCAL_DIR, stormLocal.getPath());
+ conf.put(Config.STORM_CLUSTER_MODE, "local");
- Map<String, Object> topoConf = new HashMap<>(conf);
- topoConf.put(Config.TOPOLOGY_BLOBSTORE_MAP, topoBlobMap);
- topoConf.put(Config.TOPOLOGY_NAME, topoName);
+ StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
- List<LocalizedResource> localizedList = new ArrayList<>();
- StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
- LocalizedResource simpleLocal = new LocalizedResource(simpleKey, Paths.get(localizerRoot), false, ops, conf, user, metricsRegistry);
- localizedList.add(simpleLocal);
+ AdvancedFSOps ops = AdvancedFSOps.make(conf);
- AsyncLocalizer bl = spy(new AsyncLocalizer(conf, ops, localizerRoot, metricsRegistry));
- ConfigUtils orig = ConfigUtils.setInstance(mockedCU);
- ServerUtils origSU = ServerUtils.setInstance(mockedSU);
+ victim = spy(new AsyncLocalizer(conf, ops, localizerRoot.getPath(), metricsRegistry));
- try {
- when(mockedCU.supervisorStormDistRootImpl(conf, topoId)).thenReturn(stormRoot);
- when(mockedCU.readSupervisorStormConfImpl(conf, topoId)).thenReturn(topoConf);
- when(mockedCU.readSupervisorTopologyImpl(conf, topoId, ops)).thenReturn(st);
+ final String topoId = "TOPO-12345";
+ final String user = "user";
- doReturn(mockblobstore).when(bl).getClientBlobStore();
- doReturn(userDir).when(bl).getLocalUserFileCacheDir(user);
- doReturn(localizedList).when(bl).getBlobs(any(List.class), any(), any());
- doReturn(mock(OutputStream.class)).when(ops).getOutputStream(any());
+ final int port = 8080;
+
+ final Path userDir = Paths.get(stormLocal.getPath(), user);
+ final Path stormRoot = Paths.get(stormLocal.getPath(), topoId);
+
+ final String simpleLocalName = "simple.txt";
+ final String simpleKey = "simple";
+ Map<String, Map<String, Object>> topoBlobMap = new HashMap<>();
+ Map<String, Object> simple = new HashMap<>();
+ simple.put("localname", simpleLocalName);
+ simple.put("uncompress", false);
+ topoBlobMap.put(simpleKey, simple);
+
+
+ Map<String, Object> topoConf = new HashMap<>(conf);
+ topoConf.put(Config.TOPOLOGY_BLOBSTORE_MAP, topoBlobMap);
+ topoConf.put(Config.TOPOLOGY_NAME, "TOPO");
+
+ List<LocalizedResource> localizedList = new ArrayList<>();
+ LocalizedResource simpleLocal = new LocalizedResource(simpleKey, localizerRoot.getFile().toPath(), false, ops, conf, user, metricsRegistry);
+ localizedList.add(simpleLocal);
+
+ when(mockedConfigUtils.supervisorStormDistRootImpl(conf, topoId)).thenReturn(stormRoot.toString());
+ when(mockedConfigUtils.readSupervisorStormConfImpl(conf, topoId)).thenReturn(topoConf);
+ when(mockedConfigUtils.readSupervisorTopologyImpl(conf, topoId, ops)).thenReturn(constructEmptyStormTopology());
+
+ doReturn(mockBlobStore).when(victim).getClientBlobStore();
+ doReturn(userDir.toFile()).when(victim).getLocalUserFileCacheDir(user);
+ doReturn(localizedList).when(victim).getBlobs(any(List.class), any(), any());
ReadableBlobMeta blobMeta = new ReadableBlobMeta();
blobMeta.set_version(1);
- doReturn(blobMeta).when(mockblobstore).getBlobMeta(any());
- when(mockblobstore.getBlob(any())).thenAnswer(invocation -> new TestInputStreamWithMeta(LOCAL_MODE_JAR_VERSION));
+ doReturn(blobMeta).when(mockBlobStore).getBlobMeta(any());
+ when(mockBlobStore.getBlob(any())).thenAnswer(invocation -> new TestInputStreamWithMeta(LOCAL_MODE_JAR_VERSION));
- Future<Void> f = bl.requestDownloadTopologyBlobs(la, port, null);
+ Future<Void> f = victim.requestDownloadTopologyBlobs(constructLocalAssignment(topoId, user), port, null);
f.get(20, TimeUnit.SECONDS);
- verify(bl).getLocalUserFileCacheDir(user);
+ verify(victim).getLocalUserFileCacheDir(user);
- verify(ops).fileExists(userDir);
- verify(ops).forceMkdir(userDir);
+ assertTrue(ops.fileExists(userDir));
- verify(bl).getBlobs(any(List.class), any(), any());
+ verify(victim).getBlobs(any(List.class), any(), any());
- Path extractionDir = Paths.get(stormRoot,
- LocallyCachedTopologyBlob.TopologyBlobType.TOPO_JAR.getTempExtractionDir(LOCAL_MODE_JAR_VERSION));
-
- // make sure resources dir is created.
- verify(ops).forceMkdir(extractionDir);
+ // make sure resources directory after blob version commit is created.
+ Path extractionDir = stormRoot.resolve(LocallyCachedTopologyBlob.TopologyBlobType.TOPO_JAR.getExtractionDir());
+ assertTrue(ops.fileExists(extractionDir));
} finally {
- try {
- ConfigUtils.setInstance(orig);
- ServerUtils.setInstance(origSU);
- bl.close();
- } catch (Throwable e) {
- LOG.error("ERROR trying to close an object", e);
+ ConfigUtils.setInstance(previousConfigUtils);
+ ServerUtils.setInstance(previousServerUtils);
+
+ if (victim != null) {
+ victim.close();
}
}
}
- @Before
- public void setUp() throws Exception {
- baseDir = new File(System.getProperty("java.io.tmpdir") + "/blob-store-localizer-test-" + UUID.randomUUID());
- if (!baseDir.mkdir()) {
- throw new IOException("failed to create base directory");
- }
+ private LocalAssignment constructLocalAssignment(String topoId, String owner) {
+ return constructLocalAssignment(topoId, owner,
+ Collections.singletonList(new ExecutorInfo(1, 1))
+ );
}
- @After
- public void tearDown() throws Exception {
- try {
- FileUtils.deleteDirectory(baseDir);
- } catch (IOException ignore) {
- }
+ private LocalAssignment constructLocalAssignment(String topoId, String owner, List<ExecutorInfo> executorInfos) {
+ LocalAssignment assignment = new LocalAssignment(topoId, executorInfos);
+ assignment.set_owner(owner);
+ return assignment;
}
- ;
+ private StormTopology constructEmptyStormTopology() {
+ StormTopology topology = new StormTopology();
+ topology.set_spouts(new HashMap<>());
+ topology.set_bolts(new HashMap<>());
+ topology.set_state_spouts(new HashMap<>());
+ return topology;
+ }
- protected String joinPath(String... pathList) {
+ private String joinPath(String... pathList) {
return Joiner.on(File.separator).join(pathList);
}
- public String constructUserCacheDir(String base, String user) {
+ private String constructUserCacheDir(String base, String user) {
return joinPath(base, USERCACHE, user);
}
- public String constructExpectedFilesDir(String base, String user) {
+ private String constructExpectedFilesDir(String base, String user) {
return joinPath(constructUserCacheDir(base, user), LocalizedResource.FILECACHE, LocalizedResource.FILESDIR);
}
- public String constructExpectedArchivesDir(String base, String user) {
+ private String constructExpectedArchivesDir(String base, String user) {
return joinPath(constructUserCacheDir(base, user), LocalizedResource.FILECACHE, LocalizedResource.ARCHIVESDIR);
}
@Test
public void testDirPaths() throws Exception {
- Map<String, Object> conf = new HashMap();
- AsyncLocalizer localizer = new TestLocalizer(conf, baseDir.toString());
+ try (TmpPath tmp = new TmpPath()) {
+ Map<String, Object> conf = new HashMap();
+ AsyncLocalizer localizer = new TestLocalizer(conf, tmp.getPath());
- String expectedDir = constructUserCacheDir(baseDir.toString(), user1);
- assertEquals("get local user dir doesn't return right value",
- expectedDir, localizer.getLocalUserDir(user1).toString());
+ String expectedDir = constructUserCacheDir(tmp.getPath(), user1);
+ assertEquals("get local user dir doesn't return right value",
+ expectedDir, localizer.getLocalUserDir(user1).toString());
- String expectedFileDir = joinPath(expectedDir, LocalizedResource.FILECACHE);
- assertEquals("get local user file dir doesn't return right value",
- expectedFileDir, localizer.getLocalUserFileCacheDir(user1).toString());
+ String expectedFileDir = joinPath(expectedDir, LocalizedResource.FILECACHE);
+ assertEquals("get local user file dir doesn't return right value",
+ expectedFileDir, localizer.getLocalUserFileCacheDir(user1).toString());
+ }
}
@Test
public void testReconstruct() throws Exception {
- Map<String, Object> conf = new HashMap<>();
+ try (TmpPath tmp = new TmpPath()){
+ Map<String, Object> conf = new HashMap<>();
- String expectedFileDir1 = constructExpectedFilesDir(baseDir.toString(), user1);
- String expectedArchiveDir1 = constructExpectedArchivesDir(baseDir.toString(), user1);
- String expectedFileDir2 = constructExpectedFilesDir(baseDir.toString(), user2);
- String expectedArchiveDir2 = constructExpectedArchivesDir(baseDir.toString(), user2);
+ String expectedFileDir1 = constructExpectedFilesDir(tmp.getPath(), user1);
+ String expectedArchiveDir1 = constructExpectedArchivesDir(tmp.getPath(), user1);
+ String expectedFileDir2 = constructExpectedFilesDir(tmp.getPath(), user2);
+ String expectedArchiveDir2 = constructExpectedArchivesDir(tmp.getPath(), user2);
- String key1 = "testfile1.txt";
- String key2 = "testfile2.txt";
- String key3 = "testfile3.txt";
- String key4 = "testfile4.txt";
+ String key1 = "testfile1.txt";
+ String key2 = "testfile2.txt";
+ String key3 = "testfile3.txt";
+ String key4 = "testfile4.txt";
- String archive1 = "archive1";
- String archive2 = "archive2";
+ String archive1 = "archive1";
+ String archive2 = "archive2";
- File user1file1 = new File(expectedFileDir1, key1 + LocalizedResource.CURRENT_BLOB_SUFFIX);
- File user1file2 = new File(expectedFileDir1, key2 + LocalizedResource.CURRENT_BLOB_SUFFIX);
- File user2file3 = new File(expectedFileDir2, key3 + LocalizedResource.CURRENT_BLOB_SUFFIX);
- File user2file4 = new File(expectedFileDir2, key4 + LocalizedResource.CURRENT_BLOB_SUFFIX);
+ File user1file1 = new File(expectedFileDir1, key1 + LocalizedResource.CURRENT_BLOB_SUFFIX);
+ File user1file2 = new File(expectedFileDir1, key2 + LocalizedResource.CURRENT_BLOB_SUFFIX);
+ File user2file3 = new File(expectedFileDir2, key3 + LocalizedResource.CURRENT_BLOB_SUFFIX);
+ File user2file4 = new File(expectedFileDir2, key4 + LocalizedResource.CURRENT_BLOB_SUFFIX);
- File user1archive1 = new File(expectedArchiveDir1, archive1 + LocalizedResource.CURRENT_BLOB_SUFFIX);
- File user2archive2 = new File(expectedArchiveDir2, archive2 + LocalizedResource.CURRENT_BLOB_SUFFIX);
- File user1archive1file = new File(user1archive1, "file1");
- File user2archive2file = new File(user2archive2, "file2");
+ File user1archive1 = new File(expectedArchiveDir1, archive1 + LocalizedResource.CURRENT_BLOB_SUFFIX);
+ File user2archive2 = new File(expectedArchiveDir2, archive2 + LocalizedResource.CURRENT_BLOB_SUFFIX);
+ File user1archive1file = new File(user1archive1, "file1");
+ File user2archive2file = new File(user2archive2, "file2");
- // setup some files/dirs to emulate supervisor restart
- assertTrue("Failed setup filecache dir1", new File(expectedFileDir1).mkdirs());
- assertTrue("Failed setup filecache dir2", new File(expectedFileDir2).mkdirs());
- assertTrue("Failed setup file1", user1file1.createNewFile());
- assertTrue("Failed setup file2", user1file2.createNewFile());
- assertTrue("Failed setup file3", user2file3.createNewFile());
- assertTrue("Failed setup file4", user2file4.createNewFile());
- assertTrue("Failed setup archive dir1", user1archive1.mkdirs());
- assertTrue("Failed setup archive dir2", user2archive2.mkdirs());
- assertTrue("Failed setup file in archivedir1", user1archive1file.createNewFile());
- assertTrue("Failed setup file in archivedir2", user2archive2file.createNewFile());
+ // setup some files/dirs to emulate supervisor restart
+ assertTrue("Failed setup filecache dir1", new File(expectedFileDir1).mkdirs());
+ assertTrue("Failed setup filecache dir2", new File(expectedFileDir2).mkdirs());
+ assertTrue("Failed setup file1", user1file1.createNewFile());
+ assertTrue("Failed setup file2", user1file2.createNewFile());
+ assertTrue("Failed setup file3", user2file3.createNewFile());
+ assertTrue("Failed setup file4", user2file4.createNewFile());
+ assertTrue("Failed setup archive dir1", user1archive1.mkdirs());
+ assertTrue("Failed setup archive dir2", user2archive2.mkdirs());
+ assertTrue("Failed setup file in archivedir1", user1archive1file.createNewFile());
+ assertTrue("Failed setup file in archivedir2", user2archive2file.createNewFile());
- TestLocalizer localizer = new TestLocalizer(conf, baseDir.toString());
+ TestLocalizer localizer = new TestLocalizer(conf, tmp.getPath());
- ArrayList<LocalResource> arrUser1Keys = new ArrayList<>();
- arrUser1Keys.add(new LocalResource(key1, false, false));
- arrUser1Keys.add(new LocalResource(archive1, true, false));
- LocalAssignment topo1 = new LocalAssignment("topo1", Collections.emptyList());
- topo1.set_owner(user1);
- localizer.addReferences(arrUser1Keys, new PortAndAssignmentImpl(1, topo1), null);
+ ArrayList<LocalResource> arrUser1Keys = new ArrayList<>();
+ arrUser1Keys.add(new LocalResource(key1, false, false));
+ arrUser1Keys.add(new LocalResource(archive1, true, false));
+ LocalAssignment topo1 = constructLocalAssignment("topo1", user1, Collections.emptyList());
+ localizer.addReferences(arrUser1Keys, new PortAndAssignmentImpl(1, topo1), null);
- ConcurrentMap<String, LocalizedResource> lrsrcFiles = localizer.getUserFiles().get(user1);
- ConcurrentMap<String, LocalizedResource> lrsrcArchives = localizer.getUserArchives().get(user1);
- assertEquals("local resource set size wrong", 3, lrsrcFiles.size() + lrsrcArchives.size());
- LocalizedResource key1rsrc = lrsrcFiles.get(key1);
- assertNotNull("Local resource doesn't exist but should", key1rsrc);
- assertEquals("key doesn't match", key1, key1rsrc.getKey());
- assertEquals("references doesn't match " + key1rsrc.getDependencies(), true, key1rsrc.isUsed());
- LocalizedResource key2rsrc = lrsrcFiles.get(key2);
- assertNotNull("Local resource doesn't exist but should", key2rsrc);
- assertEquals("key doesn't match", key2, key2rsrc.getKey());
- assertEquals("refcount doesn't match " + key2rsrc.getDependencies(), false, key2rsrc.isUsed());
- LocalizedResource archive1rsrc = lrsrcArchives.get(archive1);
- assertNotNull("Local resource doesn't exist but should", archive1rsrc);
- assertEquals("key doesn't match", archive1, archive1rsrc.getKey());
- assertEquals("refcount doesn't match " + archive1rsrc.getDependencies(), true, archive1rsrc.isUsed());
+ ConcurrentMap<String, LocalizedResource> lrsrcFiles = localizer.getUserFiles().get(user1);
+ ConcurrentMap<String, LocalizedResource> lrsrcArchives = localizer.getUserArchives().get(user1);
+ assertEquals("local resource set size wrong", 3, lrsrcFiles.size() + lrsrcArchives.size());
+ LocalizedResource key1rsrc = lrsrcFiles.get(key1);
+ assertNotNull("Local resource doesn't exist but should", key1rsrc);
+ assertEquals("key doesn't match", key1, key1rsrc.getKey());
+ assertEquals("references doesn't match " + key1rsrc.getDependencies(), true, key1rsrc.isUsed());
+ LocalizedResource key2rsrc = lrsrcFiles.get(key2);
+ assertNotNull("Local resource doesn't exist but should", key2rsrc);
+ assertEquals("key doesn't match", key2, key2rsrc.getKey());
+ assertEquals("refcount doesn't match " + key2rsrc.getDependencies(), false, key2rsrc.isUsed());
+ LocalizedResource archive1rsrc = lrsrcArchives.get(archive1);
+ assertNotNull("Local resource doesn't exist but should", archive1rsrc);
+ assertEquals("key doesn't match", archive1, archive1rsrc.getKey());
+ assertEquals("refcount doesn't match " + archive1rsrc.getDependencies(), true, archive1rsrc.isUsed());
- ConcurrentMap<String, LocalizedResource> lrsrcFiles2 = localizer.getUserFiles().get(user2);
- ConcurrentMap<String, LocalizedResource> lrsrcArchives2 = localizer.getUserArchives().get(user2);
- assertEquals("local resource set size wrong", 3, lrsrcFiles2.size() + lrsrcArchives2.size());
- LocalizedResource key3rsrc = lrsrcFiles2.get(key3);
- assertNotNull("Local resource doesn't exist but should", key3rsrc);
- assertEquals("key doesn't match", key3, key3rsrc.getKey());
- assertEquals("refcount doesn't match " + key3rsrc.getDependencies(), false, key3rsrc.isUsed());
- LocalizedResource key4rsrc = lrsrcFiles2.get(key4);
- assertNotNull("Local resource doesn't exist but should", key4rsrc);
- assertEquals("key doesn't match", key4, key4rsrc.getKey());
- assertEquals("refcount doesn't match " + key4rsrc.getDependencies(), false, key4rsrc.isUsed());
- LocalizedResource archive2rsrc = lrsrcArchives2.get(archive2);
- assertNotNull("Local resource doesn't exist but should", archive2rsrc);
- assertEquals("key doesn't match", archive2, archive2rsrc.getKey());
- assertEquals("refcount doesn't match " + archive2rsrc.getDependencies(), false, archive2rsrc.isUsed());
+ ConcurrentMap<String, LocalizedResource> lrsrcFiles2 = localizer.getUserFiles().get(user2);
+ ConcurrentMap<String, LocalizedResource> lrsrcArchives2 = localizer.getUserArchives().get(user2);
+ assertEquals("local resource set size wrong", 3, lrsrcFiles2.size() + lrsrcArchives2.size());
+ LocalizedResource key3rsrc = lrsrcFiles2.get(key3);
+ assertNotNull("Local resource doesn't exist but should", key3rsrc);
+ assertEquals("key doesn't match", key3, key3rsrc.getKey());
+ assertEquals("refcount doesn't match " + key3rsrc.getDependencies(), false, key3rsrc.isUsed());
+ LocalizedResource key4rsrc = lrsrcFiles2.get(key4);
+ assertNotNull("Local resource doesn't exist but should", key4rsrc);
+ assertEquals("key doesn't match", key4, key4rsrc.getKey());
+ assertEquals("refcount doesn't match " + key4rsrc.getDependencies(), false, key4rsrc.isUsed());
+ LocalizedResource archive2rsrc = lrsrcArchives2.get(archive2);
+ assertNotNull("Local resource doesn't exist but should", archive2rsrc);
+ assertEquals("key doesn't match", archive2, archive2rsrc.getKey());
+ assertEquals("refcount doesn't match " + archive2rsrc.getDependencies(), false, archive2rsrc.isUsed());
+ }
}
@Test
@@ -513,7 +498,7 @@
// Windows should set this to false cause symlink in compressed file doesn't work properly.
supportSymlinks = false;
}
- try (Time.SimulatedTime st = new Time.SimulatedTime()) {
+ try (Time.SimulatedTime st = new Time.SimulatedTime(); TmpPath tmp = new TmpPath()) {
Map<String, Object> conf = new HashMap<>();
// set clean time really high so doesn't kick in
@@ -522,16 +507,16 @@
String key1 = archiveFile.getName();
String topo1 = "topo1";
LOG.info("About to create new AsyncLocalizer...");
- TestLocalizer localizer = new TestLocalizer(conf, baseDir.toString());
+ TestLocalizer localizer = new TestLocalizer(conf, tmp.getPath());
// set really small so will do cleanup
localizer.setTargetCacheSize(1);
LOG.info("created AsyncLocalizer...");
ReadableBlobMeta rbm = new ReadableBlobMeta();
rbm.set_settable(new SettableBlobMeta(WORLD_EVERYTHING));
- when(mockblobstore.getBlobMeta(key1)).thenReturn(rbm);
+ when(mockBlobStore.getBlobMeta(key1)).thenReturn(rbm);
- when(mockblobstore.getBlob(key1)).thenReturn(new TestInputStreamWithMeta(new
+ when(mockBlobStore.getBlob(key1)).thenReturn(new TestInputStreamWithMeta(new
FileInputStream(archiveFile.getAbsolutePath()),
0, archiveFile.length()));
@@ -539,15 +524,14 @@
Time.advanceTime(10);
File user1Dir = localizer.getLocalUserFileCacheDir(user1);
assertTrue("failed to create user dir", user1Dir.mkdirs());
- LocalAssignment topo1Assignment = new LocalAssignment(topo1, Collections.emptyList());
- topo1Assignment.set_owner(user1);
+ LocalAssignment topo1Assignment = constructLocalAssignment(topo1, user1, Collections.emptyList());
PortAndAssignment topo1Pna = new PortAndAssignmentImpl(1, topo1Assignment);
LocalizedResource lrsrc = localizer.getBlob(new LocalResource(key1, true, false), topo1Pna, null);
Time.advanceTime(10);
long timeAfter = Time.currentTimeMillis();
Time.advanceTime(10);
- String expectedUserDir = joinPath(baseDir.toString(), USERCACHE, user1);
+ String expectedUserDir = joinPath(tmp.getPath(), USERCACHE, user1);
String expectedFileDir = joinPath(expectedUserDir, LocalizedResource.FILECACHE, LocalizedResource.ARCHIVESDIR);
assertTrue("user filecache dir not created", new File(expectedFileDir).exists());
File keyFile = new File(expectedFileDir, key1 + ".0");
@@ -602,36 +586,35 @@
@Test
public void testBasic() throws Exception {
- try (Time.SimulatedTime st = new Time.SimulatedTime()) {
+ try (Time.SimulatedTime st = new Time.SimulatedTime(); TmpPath tmp = new TmpPath()) {
Map<String, Object> conf = new HashMap();
// set clean time really high so doesn't kick in
conf.put(DaemonConfig.SUPERVISOR_LOCALIZER_CACHE_CLEANUP_INTERVAL_MS, 60 * 60 * 1000);
String key1 = "key1";
String topo1 = "topo1";
- TestLocalizer localizer = new TestLocalizer(conf, baseDir.toString());
+ TestLocalizer localizer = new TestLocalizer(conf, tmp.getPath());
// set really small so will do cleanup
localizer.setTargetCacheSize(1);
ReadableBlobMeta rbm = new ReadableBlobMeta();
rbm.set_settable(new SettableBlobMeta(WORLD_EVERYTHING));
- when(mockblobstore.getBlobMeta(key1)).thenReturn(rbm);
+ when(mockBlobStore.getBlobMeta(key1)).thenReturn(rbm);
- when(mockblobstore.getBlob(key1)).thenReturn(new TestInputStreamWithMeta(1));
+ when(mockBlobStore.getBlob(key1)).thenReturn(new TestInputStreamWithMeta(1));
long timeBefore = Time.currentTimeMillis();
Time.advanceTime(10);
File user1Dir = localizer.getLocalUserFileCacheDir(user1);
assertTrue("failed to create user dir", user1Dir.mkdirs());
Time.advanceTime(10);
- LocalAssignment topo1Assignment = new LocalAssignment(topo1, Collections.emptyList());
- topo1Assignment.set_owner(user1);
+ LocalAssignment topo1Assignment = constructLocalAssignment(topo1, user1, Collections.emptyList());
PortAndAssignment topo1Pna = new PortAndAssignmentImpl(1, topo1Assignment);
LocalizedResource lrsrc = localizer.getBlob(new LocalResource(key1, false, false), topo1Pna, null);
long timeAfter = Time.currentTimeMillis();
Time.advanceTime(10);
- String expectedUserDir = joinPath(baseDir.toString(), USERCACHE, user1);
+ String expectedUserDir = joinPath(tmp.getPath(), USERCACHE, user1);
String expectedFileDir = joinPath(expectedUserDir, LocalizedResource.FILECACHE, LocalizedResource.FILESDIR);
assertTrue("user filecache dir not created", new File(expectedFileDir).exists());
File keyFile = new File(expectedFileDir, key1 + ".current");
@@ -678,7 +661,7 @@
@Test
public void testMultipleKeysOneUser() throws Exception {
- try (Time.SimulatedTime st = new Time.SimulatedTime()) {
+ try (Time.SimulatedTime st = new Time.SimulatedTime(); TmpPath tmp = new TmpPath()) {
Map<String, Object> conf = new HashMap<>();
// set clean time really high so doesn't kick in
conf.put(DaemonConfig.SUPERVISOR_LOCALIZER_CACHE_CLEANUP_INTERVAL_MS, 60 * 60 * 1_000);
@@ -687,32 +670,31 @@
String topo1 = "topo1";
String key2 = "key2";
String key3 = "key3";
- TestLocalizer localizer = new TestLocalizer(conf, baseDir.toString());
+ TestLocalizer localizer = new TestLocalizer(conf, tmp.getPath());
// set to keep 2 blobs (each of size 34)
localizer.setTargetCacheSize(68);
ReadableBlobMeta rbm = new ReadableBlobMeta();
rbm.set_settable(new SettableBlobMeta(WORLD_EVERYTHING));
- when(mockblobstore.getBlobMeta(anyString())).thenReturn(rbm);
- when(mockblobstore.isRemoteBlobExists(anyString())).thenReturn(true);
- when(mockblobstore.getBlob(key1)).thenReturn(new TestInputStreamWithMeta(0));
- when(mockblobstore.getBlob(key2)).thenReturn(new TestInputStreamWithMeta(0));
- when(mockblobstore.getBlob(key3)).thenReturn(new TestInputStreamWithMeta(0));
+ when(mockBlobStore.getBlobMeta(anyString())).thenReturn(rbm);
+ when(mockBlobStore.isRemoteBlobExists(anyString())).thenReturn(true);
+ when(mockBlobStore.getBlob(key1)).thenReturn(new TestInputStreamWithMeta(0));
+ when(mockBlobStore.getBlob(key2)).thenReturn(new TestInputStreamWithMeta(0));
+ when(mockBlobStore.getBlob(key3)).thenReturn(new TestInputStreamWithMeta(0));
List<LocalResource> keys = Arrays.asList(new LocalResource(key1, false, false),
new LocalResource(key2, false, false), new LocalResource(key3, false, false));
File user1Dir = localizer.getLocalUserFileCacheDir(user1);
assertTrue("failed to create user dir", user1Dir.mkdirs());
- LocalAssignment topo1Assignment = new LocalAssignment(topo1, Collections.emptyList());
- topo1Assignment.set_owner(user1);
+ LocalAssignment topo1Assignment = constructLocalAssignment(topo1, user1, Collections.emptyList());
PortAndAssignment topo1Pna = new PortAndAssignmentImpl(1, topo1Assignment);
List<LocalizedResource> lrsrcs = localizer.getBlobs(keys, topo1Pna, null);
LocalizedResource lrsrc = lrsrcs.get(0);
LocalizedResource lrsrc2 = lrsrcs.get(1);
LocalizedResource lrsrc3 = lrsrcs.get(2);
- String expectedFileDir = joinPath(baseDir.toString(), USERCACHE, user1,
+ String expectedFileDir = joinPath(tmp.getPath(), USERCACHE, user1,
LocalizedResource.FILECACHE, LocalizedResource.FILESDIR);
assertTrue("user filecache dir not created", new File(expectedFileDir).exists());
File keyFile = new File(expectedFileDir, key1 + LocalizedResource.CURRENT_BLOB_SUFFIX);
@@ -782,31 +764,32 @@
@Test(expected = AuthorizationException.class)
public void testFailAcls() throws Exception {
- Map<String, Object> conf = new HashMap();
- // set clean time really high so doesn't kick in
- conf.put(DaemonConfig.SUPERVISOR_LOCALIZER_CACHE_CLEANUP_INTERVAL_MS, 60 * 60 * 1000);
- // enable blobstore acl validation
- conf.put(Config.STORM_BLOBSTORE_ACL_VALIDATION_ENABLED, true);
+ try (TmpPath tmp = new TmpPath()) {
+ Map<String, Object> conf = new HashMap();
+ // set clean time really high so doesn't kick in
+ conf.put(DaemonConfig.SUPERVISOR_LOCALIZER_CACHE_CLEANUP_INTERVAL_MS, 60 * 60 * 1000);
+ // enable blobstore acl validation
+ conf.put(Config.STORM_BLOBSTORE_ACL_VALIDATION_ENABLED, true);
- String topo1 = "topo1";
- String key1 = "key1";
- TestLocalizer localizer = new TestLocalizer(conf, baseDir.toString());
+ String topo1 = "topo1";
+ String key1 = "key1";
+ TestLocalizer localizer = new TestLocalizer(conf, tmp.getPath());
- ReadableBlobMeta rbm = new ReadableBlobMeta();
- // set acl so user doesn't have read access
- AccessControl acl = new AccessControl(AccessControlType.USER, BlobStoreAclHandler.ADMIN);
- acl.set_name(user1);
- rbm.set_settable(new SettableBlobMeta(Arrays.asList(acl)));
- when(mockblobstore.getBlobMeta(anyString())).thenReturn(rbm);
- when(mockblobstore.getBlob(key1)).thenReturn(new TestInputStreamWithMeta(1));
- File user1Dir = localizer.getLocalUserFileCacheDir(user1);
- assertTrue("failed to create user dir", user1Dir.mkdirs());
+ ReadableBlobMeta rbm = new ReadableBlobMeta();
+ // set acl so user doesn't have read access
+ AccessControl acl = new AccessControl(AccessControlType.USER, BlobStoreAclHandler.ADMIN);
+ acl.set_name(user1);
+ rbm.set_settable(new SettableBlobMeta(Arrays.asList(acl)));
+ when(mockBlobStore.getBlobMeta(anyString())).thenReturn(rbm);
+ when(mockBlobStore.getBlob(key1)).thenReturn(new TestInputStreamWithMeta(1));
+ File user1Dir = localizer.getLocalUserFileCacheDir(user1);
+ assertTrue("failed to create user dir", user1Dir.mkdirs());
- LocalAssignment topo1Assignment = new LocalAssignment(topo1, Collections.emptyList());
- topo1Assignment.set_owner(user1);
- PortAndAssignment topo1Pna = new PortAndAssignmentImpl(1, topo1Assignment);
- // This should throw AuthorizationException because auth failed
- localizer.getBlob(new LocalResource(key1, false, false), topo1Pna, null);
+ LocalAssignment topo1Assignment = constructLocalAssignment(topo1, user1, Collections.emptyList());
+ PortAndAssignment topo1Pna = new PortAndAssignmentImpl(1, topo1Assignment);
+ // This should throw AuthorizationException because auth failed
+ localizer.getBlob(new LocalResource(key1, false, false), topo1Pna, null);
+ }
}
@Test(expected = KeyNotFoundException.class)
@@ -824,163 +807,162 @@
@Test
public void testMultipleUsers() throws Exception {
- Map<String, Object> conf = new HashMap<>();
- // set clean time really high so doesn't kick in
- conf.put(DaemonConfig.SUPERVISOR_LOCALIZER_CACHE_CLEANUP_INTERVAL_MS, 60 * 60 * 1000);
+ try (TmpPath tmp = new TmpPath()){
+ Map<String, Object> conf = new HashMap<>();
+ // set clean time really high so doesn't kick in
+ conf.put(DaemonConfig.SUPERVISOR_LOCALIZER_CACHE_CLEANUP_INTERVAL_MS, 60 * 60 * 1000);
- String topo1 = "topo1";
- String topo2 = "topo2";
- String topo3 = "topo3";
- String key1 = "key1";
- String key2 = "key2";
- String key3 = "key3";
- TestLocalizer localizer = new TestLocalizer(conf, baseDir.toString());
- // set to keep 2 blobs (each of size 34)
- localizer.setTargetCacheSize(68);
+ String topo1 = "topo1";
+ String topo2 = "topo2";
+ String topo3 = "topo3";
+ String key1 = "key1";
+ String key2 = "key2";
+ String key3 = "key3";
+ TestLocalizer localizer = new TestLocalizer(conf, tmp.getPath());
+ // set to keep 2 blobs (each of size 34)
+ localizer.setTargetCacheSize(68);
- ReadableBlobMeta rbm = new ReadableBlobMeta();
- rbm.set_settable(new SettableBlobMeta(WORLD_EVERYTHING));
- when(mockblobstore.getBlobMeta(anyString())).thenReturn(rbm);
- //thenReturn always returns the same object, which is already consumed by the time User3 tries to getBlob!
- when(mockblobstore.getBlob(key1)).thenAnswer((i) -> new TestInputStreamWithMeta(1));
- when(mockblobstore.getBlob(key2)).thenReturn(new TestInputStreamWithMeta(1));
- when(mockblobstore.getBlob(key3)).thenReturn(new TestInputStreamWithMeta(1));
+ ReadableBlobMeta rbm = new ReadableBlobMeta();
+ rbm.set_settable(new SettableBlobMeta(WORLD_EVERYTHING));
+ when(mockBlobStore.getBlobMeta(anyString())).thenReturn(rbm);
+ //thenReturn always returns the same object, which is already consumed by the time User3 tries to getBlob!
+ when(mockBlobStore.getBlob(key1)).thenAnswer((i) -> new TestInputStreamWithMeta(1));
+ when(mockBlobStore.getBlob(key2)).thenReturn(new TestInputStreamWithMeta(1));
+ when(mockBlobStore.getBlob(key3)).thenReturn(new TestInputStreamWithMeta(1));
- File user1Dir = localizer.getLocalUserFileCacheDir(user1);
- assertTrue("failed to create user dir", user1Dir.mkdirs());
- File user2Dir = localizer.getLocalUserFileCacheDir(user2);
- assertTrue("failed to create user dir", user2Dir.mkdirs());
- File user3Dir = localizer.getLocalUserFileCacheDir(user3);
- assertTrue("failed to create user dir", user3Dir.mkdirs());
+ File user1Dir = localizer.getLocalUserFileCacheDir(user1);
+ assertTrue("failed to create user dir", user1Dir.mkdirs());
+ File user2Dir = localizer.getLocalUserFileCacheDir(user2);
+ assertTrue("failed to create user dir", user2Dir.mkdirs());
+ File user3Dir = localizer.getLocalUserFileCacheDir(user3);
+ assertTrue("failed to create user dir", user3Dir.mkdirs());
- LocalAssignment topo1Assignment = new LocalAssignment(topo1, Collections.emptyList());
- topo1Assignment.set_owner(user1);
- PortAndAssignment topo1Pna = new PortAndAssignmentImpl(1, topo1Assignment);
- LocalizedResource lrsrc = localizer.getBlob(new LocalResource(key1, false, false), topo1Pna, null);
+ LocalAssignment topo1Assignment = constructLocalAssignment(topo1, user1, Collections.emptyList());
+ PortAndAssignment topo1Pna = new PortAndAssignmentImpl(1, topo1Assignment);
+ LocalizedResource lrsrc = localizer.getBlob(new LocalResource(key1, false, false), topo1Pna, null);
- LocalAssignment topo2Assignment = new LocalAssignment(topo2, Collections.emptyList());
- topo2Assignment.set_owner(user2);
- PortAndAssignment topo2Pna = new PortAndAssignmentImpl(2, topo2Assignment);
- LocalizedResource lrsrc2 = localizer.getBlob(new LocalResource(key2, false, false), topo2Pna, null);
+ LocalAssignment topo2Assignment = constructLocalAssignment(topo2, user2, Collections.emptyList());
+ PortAndAssignment topo2Pna = new PortAndAssignmentImpl(2, topo2Assignment);
+ LocalizedResource lrsrc2 = localizer.getBlob(new LocalResource(key2, false, false), topo2Pna, null);
- LocalAssignment topo3Assignment = new LocalAssignment(topo3, Collections.emptyList());
- topo3Assignment.set_owner(user3);
- PortAndAssignment topo3Pna = new PortAndAssignmentImpl(3, topo3Assignment);
- LocalizedResource lrsrc3 = localizer.getBlob(new LocalResource(key3, false, false), topo3Pna, null);
+ LocalAssignment topo3Assignment = constructLocalAssignment(topo3, user3, Collections.emptyList());
+ PortAndAssignment topo3Pna = new PortAndAssignmentImpl(3, topo3Assignment);
+ LocalizedResource lrsrc3 = localizer.getBlob(new LocalResource(key3, false, false), topo3Pna, null);
- // make sure we support different user reading same blob
- LocalizedResource lrsrc1_user3 = localizer.getBlob(new LocalResource(key1, false, false), topo3Pna, null);
+ // make sure we support different user reading same blob
+ LocalizedResource lrsrc1_user3 = localizer.getBlob(new LocalResource(key1, false, false), topo3Pna, null);
- String expectedUserDir1 = joinPath(baseDir.toString(), USERCACHE, user1);
- String expectedFileDirUser1 = joinPath(expectedUserDir1, LocalizedResource.FILECACHE, LocalizedResource.FILESDIR);
- String expectedFileDirUser2 = joinPath(baseDir.toString(), USERCACHE, user2,
- LocalizedResource.FILECACHE, LocalizedResource.FILESDIR);
- String expectedFileDirUser3 = joinPath(baseDir.toString(), USERCACHE, user3,
- LocalizedResource.FILECACHE, LocalizedResource.FILESDIR);
- assertTrue("user filecache dir user1 not created", new File(expectedFileDirUser1).exists());
- assertTrue("user filecache dir user2 not created", new File(expectedFileDirUser2).exists());
- assertTrue("user filecache dir user3 not created", new File(expectedFileDirUser3).exists());
+ String expectedUserDir1 = joinPath(tmp.getPath(), USERCACHE, user1);
+ String expectedFileDirUser1 = joinPath(expectedUserDir1, LocalizedResource.FILECACHE, LocalizedResource.FILESDIR);
+ String expectedFileDirUser2 = joinPath(tmp.getPath(), USERCACHE, user2,
+ LocalizedResource.FILECACHE, LocalizedResource.FILESDIR);
+ String expectedFileDirUser3 = joinPath(tmp.getPath(), USERCACHE, user3,
+ LocalizedResource.FILECACHE, LocalizedResource.FILESDIR);
+ assertTrue("user filecache dir user1 not created", new File(expectedFileDirUser1).exists());
+ assertTrue("user filecache dir user2 not created", new File(expectedFileDirUser2).exists());
+ assertTrue("user filecache dir user3 not created", new File(expectedFileDirUser3).exists());
- File keyFile = new File(expectedFileDirUser1, key1 + LocalizedResource.CURRENT_BLOB_SUFFIX);
- File keyFile2 = new File(expectedFileDirUser2, key2 + LocalizedResource.CURRENT_BLOB_SUFFIX);
- File keyFile3 = new File(expectedFileDirUser3, key3 + LocalizedResource.CURRENT_BLOB_SUFFIX);
- File keyFile1user3 = new File(expectedFileDirUser3, key1 + LocalizedResource.CURRENT_BLOB_SUFFIX);
+ File keyFile = new File(expectedFileDirUser1, key1 + LocalizedResource.CURRENT_BLOB_SUFFIX);
+ File keyFile2 = new File(expectedFileDirUser2, key2 + LocalizedResource.CURRENT_BLOB_SUFFIX);
+ File keyFile3 = new File(expectedFileDirUser3, key3 + LocalizedResource.CURRENT_BLOB_SUFFIX);
+ File keyFile1user3 = new File(expectedFileDirUser3, key1 + LocalizedResource.CURRENT_BLOB_SUFFIX);
- assertTrue("blob not created", keyFile.exists());
- assertTrue("blob not created", keyFile2.exists());
- assertTrue("blob not created", keyFile3.exists());
- assertTrue("blob not created", keyFile1user3.exists());
+ assertTrue("blob not created", keyFile.exists());
+ assertTrue("blob not created", keyFile2.exists());
+ assertTrue("blob not created", keyFile3.exists());
+ assertTrue("blob not created", keyFile1user3.exists());
- //Should assert file size
- assertEquals("size doesn't match", 34, lrsrc.getSizeOnDisk());
- assertEquals("size doesn't match", 34, lrsrc2.getSizeOnDisk());
- assertEquals("size doesn't match", 34, lrsrc3.getSizeOnDisk());
- //This was 0 byte in test
- assertEquals("size doesn't match", 34, lrsrc1_user3.getSizeOnDisk());
+ //Should assert file size
+ assertEquals("size doesn't match", 34, lrsrc.getSizeOnDisk());
+ assertEquals("size doesn't match", 34, lrsrc2.getSizeOnDisk());
+ assertEquals("size doesn't match", 34, lrsrc3.getSizeOnDisk());
+ //This was 0 byte in test
+ assertEquals("size doesn't match", 34, lrsrc1_user3.getSizeOnDisk());
- ConcurrentMap<String, LocalizedResource> lrsrcSet = localizer.getUserFiles().get(user1);
- assertEquals("local resource set size wrong", 1, lrsrcSet.size());
- ConcurrentMap<String, LocalizedResource> lrsrcSet2 = localizer.getUserFiles().get(user2);
- assertEquals("local resource set size wrong", 1, lrsrcSet2.size());
- ConcurrentMap<String, LocalizedResource> lrsrcSet3 = localizer.getUserFiles().get(user3);
- assertEquals("local resource set size wrong", 2, lrsrcSet3.size());
+ ConcurrentMap<String, LocalizedResource> lrsrcSet = localizer.getUserFiles().get(user1);
+ assertEquals("local resource set size wrong", 1, lrsrcSet.size());
+ ConcurrentMap<String, LocalizedResource> lrsrcSet2 = localizer.getUserFiles().get(user2);
+ assertEquals("local resource set size wrong", 1, lrsrcSet2.size());
+ ConcurrentMap<String, LocalizedResource> lrsrcSet3 = localizer.getUserFiles().get(user3);
+ assertEquals("local resource set size wrong", 2, lrsrcSet3.size());
- localizer.removeBlobReference(lrsrc.getKey(), topo1Pna, false);
- // should remove key1
- localizer.cleanup();
+ localizer.removeBlobReference(lrsrc.getKey(), topo1Pna, false);
+ // should remove key1
+ localizer.cleanup();
- lrsrcSet = localizer.getUserFiles().get(user1);
- lrsrcSet3 = localizer.getUserFiles().get(user3);
- assertNull("user set should be null", lrsrcSet);
- assertFalse("blob dir not deleted", new File(expectedFileDirUser1).exists());
- assertFalse("blob dir not deleted", new File(expectedUserDir1).exists());
- assertEquals("local resource set size wrong", 2, lrsrcSet3.size());
+ lrsrcSet = localizer.getUserFiles().get(user1);
+ lrsrcSet3 = localizer.getUserFiles().get(user3);
+ assertNull("user set should be null", lrsrcSet);
+ assertFalse("blob dir not deleted", new File(expectedFileDirUser1).exists());
+ assertFalse("blob dir not deleted", new File(expectedUserDir1).exists());
+ assertEquals("local resource set size wrong", 2, lrsrcSet3.size());
- assertTrue("blob deleted", keyFile2.exists());
- assertFalse("blob not deleted", keyFile.exists());
- assertTrue("blob deleted", keyFile3.exists());
- assertTrue("blob deleted", keyFile1user3.exists());
+ assertTrue("blob deleted", keyFile2.exists());
+ assertFalse("blob not deleted", keyFile.exists());
+ assertTrue("blob deleted", keyFile3.exists());
+ assertTrue("blob deleted", keyFile1user3.exists());
+ }
}
@Test
public void testUpdate() throws Exception {
- Map<String, Object> conf = new HashMap<>();
- // set clean time really high so doesn't kick in
- conf.put(DaemonConfig.SUPERVISOR_LOCALIZER_CACHE_CLEANUP_INTERVAL_MS, 60 * 60 * 1000);
+ try (TmpPath tmp = new TmpPath()) {
+ Map<String, Object> conf = new HashMap<>();
+ // set clean time really high so doesn't kick in
+ conf.put(DaemonConfig.SUPERVISOR_LOCALIZER_CACHE_CLEANUP_INTERVAL_MS, 60 * 60 * 1000);
- String key1 = "key1";
- String topo1 = "topo1";
- String topo2 = "topo2";
- TestLocalizer localizer = new TestLocalizer(conf, baseDir.toString());
+ String key1 = "key1";
+ String topo1 = "topo1";
+ String topo2 = "topo2";
+ TestLocalizer localizer = new TestLocalizer(conf, tmp.getPath());
- ReadableBlobMeta rbm = new ReadableBlobMeta();
- rbm.set_version(1);
- rbm.set_settable(new SettableBlobMeta(WORLD_EVERYTHING));
- when(mockblobstore.getBlobMeta(key1)).thenReturn(rbm);
- when(mockblobstore.getBlob(key1)).thenReturn(new TestInputStreamWithMeta(1));
+ ReadableBlobMeta rbm = new ReadableBlobMeta();
+ rbm.set_version(1);
+ rbm.set_settable(new SettableBlobMeta(WORLD_EVERYTHING));
+ when(mockBlobStore.getBlobMeta(key1)).thenReturn(rbm);
+ when(mockBlobStore.getBlob(key1)).thenReturn(new TestInputStreamWithMeta(1));
- File user1Dir = localizer.getLocalUserFileCacheDir(user1);
- assertTrue("failed to create user dir", user1Dir.mkdirs());
- LocalAssignment topo1Assignment = new LocalAssignment(topo1, Collections.emptyList());
- topo1Assignment.set_owner(user1);
- PortAndAssignment topo1Pna = new PortAndAssignmentImpl(1, topo1Assignment);
- LocalizedResource lrsrc = localizer.getBlob(new LocalResource(key1, false, false), topo1Pna, null);
+ File user1Dir = localizer.getLocalUserFileCacheDir(user1);
+ assertTrue("failed to create user dir", user1Dir.mkdirs());
+ LocalAssignment topo1Assignment = constructLocalAssignment(topo1, user1, Collections.emptyList());
+ PortAndAssignment topo1Pna = new PortAndAssignmentImpl(1, topo1Assignment);
+ LocalizedResource lrsrc = localizer.getBlob(new LocalResource(key1, false, false), topo1Pna, null);
- String expectedUserDir = joinPath(baseDir.toString(), USERCACHE, user1);
- String expectedFileDir = joinPath(expectedUserDir, LocalizedResource.FILECACHE, LocalizedResource.FILESDIR);
- assertTrue("user filecache dir not created", new File(expectedFileDir).exists());
- Path keyVersionFile = Paths.get(expectedFileDir, key1 + ".version");
- File keyFileCurrentSymlink = new File(expectedFileDir, key1 + LocalizedResource.CURRENT_BLOB_SUFFIX);
- assertTrue("blob not created", keyFileCurrentSymlink.exists());
- File versionFile = new File(expectedFileDir, key1 + LocalizedResource.BLOB_VERSION_SUFFIX);
- assertTrue("blob version file not created", versionFile.exists());
- assertEquals("blob version not correct", 1, LocalizedResource.localVersionOfBlob(keyVersionFile));
+ String expectedUserDir = joinPath(tmp.getPath(), USERCACHE, user1);
+ String expectedFileDir = joinPath(expectedUserDir, LocalizedResource.FILECACHE, LocalizedResource.FILESDIR);
+ assertTrue("user filecache dir not created", new File(expectedFileDir).exists());
+ Path keyVersionFile = Paths.get(expectedFileDir, key1 + ".version");
+ File keyFileCurrentSymlink = new File(expectedFileDir, key1 + LocalizedResource.CURRENT_BLOB_SUFFIX);
+ assertTrue("blob not created", keyFileCurrentSymlink.exists());
+ File versionFile = new File(expectedFileDir, key1 + LocalizedResource.BLOB_VERSION_SUFFIX);
+ assertTrue("blob version file not created", versionFile.exists());
+ assertEquals("blob version not correct", 1, LocalizedResource.localVersionOfBlob(keyVersionFile));
- ConcurrentMap<String, LocalizedResource> lrsrcSet = localizer.getUserFiles().get(user1);
- assertEquals("local resource set size wrong", 1, lrsrcSet.size());
+ ConcurrentMap<String, LocalizedResource> lrsrcSet = localizer.getUserFiles().get(user1);
+ assertEquals("local resource set size wrong", 1, lrsrcSet.size());
- // test another topology getting blob with updated version - it should update version now
- rbm.set_version(2);
- when(mockblobstore.getBlob(key1)).thenReturn(new TestInputStreamWithMeta(2));
+ // test another topology getting blob with updated version - it should update version now
+ rbm.set_version(2);
+ when(mockBlobStore.getBlob(key1)).thenReturn(new TestInputStreamWithMeta(2));
- LocalAssignment topo2Assignment = new LocalAssignment(topo2, Collections.emptyList());
- topo2Assignment.set_owner(user1);
- PortAndAssignment topo2Pna = new PortAndAssignmentImpl(1, topo2Assignment);
- localizer.getBlob(new LocalResource(key1, false, false), topo2Pna, null);
- assertTrue("blob version file not created", versionFile.exists());
- assertEquals("blob version not correct", 2, LocalizedResource.localVersionOfBlob(keyVersionFile));
- assertTrue("blob file with version 2 not created", new File(expectedFileDir, key1 + ".2").exists());
+ LocalAssignment topo2Assignment = constructLocalAssignment(topo2, user1, Collections.emptyList());
+ PortAndAssignment topo2Pna = new PortAndAssignmentImpl(1, topo2Assignment);
+ localizer.getBlob(new LocalResource(key1, false, false), topo2Pna, null);
+ assertTrue("blob version file not created", versionFile.exists());
+ assertEquals("blob version not correct", 2, LocalizedResource.localVersionOfBlob(keyVersionFile));
+ assertTrue("blob file with version 2 not created", new File(expectedFileDir, key1 + ".2").exists());
- // now test regular updateBlob
- rbm.set_version(3);
- when(mockblobstore.getBlob(key1)).thenReturn(new TestInputStreamWithMeta(3));
+ // now test regular updateBlob
+ rbm.set_version(3);
+ when(mockBlobStore.getBlob(key1)).thenReturn(new TestInputStreamWithMeta(3));
- ArrayList<LocalResource> arr = new ArrayList<>();
- arr.add(new LocalResource(key1, false, false));
- localizer.updateBlobs();
- assertTrue("blob version file not created", versionFile.exists());
- assertEquals("blob version not correct", 3, LocalizedResource.localVersionOfBlob(keyVersionFile));
- assertTrue("blob file with version 3 not created", new File(expectedFileDir, key1 + ".3").exists());
+ ArrayList<LocalResource> arr = new ArrayList<>();
+ arr.add(new LocalResource(key1, false, false));
+ localizer.updateBlobs();
+ assertTrue("blob version file not created", versionFile.exists());
+ assertEquals("blob version not correct", 3, LocalizedResource.localVersionOfBlob(keyVersionFile));
+ assertTrue("blob file with version 3 not created", new File(expectedFileDir, key1 + ".3").exists());
+ }
}
@Test
@@ -989,9 +971,9 @@
PortAndAssignment pna = new PortAndAssignmentImpl(1, la);
PortAndAssignment tpna = new TimePortAndAssignment(pna, new Timer());
- assertTrue(pna.equals(tpna));
- assertTrue(tpna.equals(pna));
- assertTrue(pna.hashCode() == tpna.hashCode());
+ assertEquals(pna, tpna);
+ assertEquals(tpna, pna);
+ assertEquals(pna.hashCode(), tpna.hashCode());
}
class TestLocalizer extends AsyncLocalizer {
@@ -1002,7 +984,7 @@
@Override
protected ClientBlobStore getClientBlobStore() {
- return mockblobstore;
+ return mockBlobStore;
}
synchronized void addReferences(List<LocalResource> localresource, PortAndAssignment pna, BlobChangingCallback cb) {
