Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/1642#discussion_r78997835
--- Diff: storm-core/src/jvm/org/apache/storm/localizer/AsyncLocalizer.java
---
@@ -0,0 +1,422 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.localizer;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.JarURLConnection;
+import java.net.URL;
+import java.net.URLDecoder;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.storm.Config;
+import org.apache.storm.blobstore.BlobStore;
+import org.apache.storm.blobstore.ClientBlobStore;
+import org.apache.storm.daemon.Shutdownable;
+import org.apache.storm.daemon.supervisor.AdvancedFSOps;
+import org.apache.storm.daemon.supervisor.SupervisorUtils;
+import org.apache.storm.generated.LocalAssignment;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.utils.ConfigUtils;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+/**
+ * This is a wrapper around the Localizer class that provides the desired
+ * async interface to Slot.
+ */
+public class AsyncLocalizer implements ILocalizer, Shutdownable {
+ /**
+ * A future that has already completed.
+ */
+ private static class AllDoneFuture implements Future<Void> {
+
+ @Override
+ public boolean cancel(boolean mayInterruptIfRunning) {
+ return false;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return false;
+ }
+
+ @Override
+ public boolean isDone() {
+ return true;
+ }
+
+ @Override
+ public Void get() {
+ return null;
+ }
+
+ @Override
+ public Void get(long timeout, TimeUnit unit) {
+ return null;
+ }
+
+ }
+
+ private static final Logger LOG =
LoggerFactory.getLogger(AsyncLocalizer.class);
+
+ private final Localizer _localizer;
+ private final ExecutorService _execService;
+ private final boolean _isLocalMode;
+ private final Map<String, Object> _conf;
+ private final Map<String, LocalDownloadedResource> _basicPending;
+ private final Map<String, LocalDownloadedResource> _blobPending;
+ private final AdvancedFSOps _fsOps;
+
+ private class DownloadBaseBlobsDistributed implements Callable<Void> {
+ protected final String _topologyId;
+ protected final File _stormRoot;
+
+ public DownloadBaseBlobsDistributed(String topologyId) throws
IOException {
+ _topologyId = topologyId;
+ _stormRoot = new
File(ConfigUtils.supervisorStormDistRoot(_conf, _topologyId));
+ }
+
+ protected void downloadBaseBlobs(File tmproot) throws Exception {
+ String stormJarKey =
ConfigUtils.masterStormJarKey(_topologyId);
+ String stormCodeKey =
ConfigUtils.masterStormCodeKey(_topologyId);
+ String stormConfKey =
ConfigUtils.masterStormConfKey(_topologyId);
+ String jarPath =
ConfigUtils.supervisorStormJarPath(tmproot.getAbsolutePath());
+ String codePath =
ConfigUtils.supervisorStormCodePath(tmproot.getAbsolutePath());
+ String confPath =
ConfigUtils.supervisorStormConfPath(tmproot.getAbsolutePath());
+ _fsOps.forceMkdir(tmproot);
+ _fsOps.restrictDirectoryPermissions(tmproot);
+ ClientBlobStore blobStore =
Utils.getClientBlobStoreForSupervisor(_conf);
+ try {
+ Utils.downloadResourcesAsSupervisor(stormJarKey, jarPath,
blobStore);
+ Utils.downloadResourcesAsSupervisor(stormCodeKey,
codePath, blobStore);
+ Utils.downloadResourcesAsSupervisor(stormConfKey,
confPath, blobStore);
+ } finally {
+ blobStore.shutdown();
+ }
+ Utils.extractDirFromJar(jarPath, ConfigUtils.RESOURCES_SUBDIR,
tmproot);
+ }
+
+ @Override
+ public Void call() throws Exception {
+ if (_fsOps.fileExists(_stormRoot)) {
+ if (!_fsOps.supportsAtomicDirectoryMove()) {
+ LOG.warn("{} may have partially downloaded blobs,
recovering", _topologyId);
+ _fsOps.deleteIfExists(_stormRoot);
+ } else {
+ LOG.warn("{} already downloaded blobs, skipping",
_topologyId);
+ return null;
+ }
+ }
+ boolean deleteAll = true;
+ String tmproot = ConfigUtils.supervisorTmpDir(_conf) +
Utils.FILE_PATH_SEPARATOR + Utils.uuid();
+ File tr = new File(tmproot);
+ try {
+ downloadBaseBlobs(tr);
+ _fsOps.moveDirectoryPreferAtomic(tr, _stormRoot);
+
_fsOps.setupStormCodeDir(ConfigUtils.readSupervisorStormConf(_conf,
_topologyId), _stormRoot);
+ deleteAll = false;
+ } finally {
+ if (deleteAll) {
+ LOG.warn("Failed to download basic resources for
topology-id {}", _topologyId);
+ _fsOps.deleteIfExists(tr);
+ _fsOps.deleteIfExists(_stormRoot);
+ }
+ }
+ return null;
+ }
+ }
+
+ private class DownloadBaseBlobsLocal extends
DownloadBaseBlobsDistributed {
+
+ public DownloadBaseBlobsLocal(String topologyId) throws
IOException {
+ super(topologyId);
+ }
+
+ @Override
+ protected void downloadBaseBlobs(File tmproot) throws Exception {
+ _fsOps.forceMkdir(tmproot);
+ String stormCodeKey =
ConfigUtils.masterStormCodeKey(_topologyId);
+ String stormConfKey =
ConfigUtils.masterStormConfKey(_topologyId);
+ File codePath = new
File(ConfigUtils.supervisorStormCodePath(tmproot.getAbsolutePath()));
+ File confPath = new
File(ConfigUtils.supervisorStormConfPath(tmproot.getAbsolutePath()));
+ BlobStore blobStore = Utils.getNimbusBlobStore(_conf, null);
+ try {
+ try (OutputStream codeOutStream =
_fsOps.getOutputStream(codePath)){
+ blobStore.readBlobTo(stormCodeKey, codeOutStream,
null);
+ }
+ try (OutputStream confOutStream =
_fsOps.getOutputStream(confPath)) {
+ blobStore.readBlobTo(stormConfKey, confOutStream,
null);
+ }
+ } finally {
+ blobStore.shutdown();
+ }
+
+ ClassLoader classloader =
Thread.currentThread().getContextClassLoader();
+ String resourcesJar = AsyncLocalizer.resourcesJar();
+ URL url =
classloader.getResource(ConfigUtils.RESOURCES_SUBDIR);
+
+ String targetDir = tmproot + Utils.FILE_PATH_SEPARATOR +
ConfigUtils.RESOURCES_SUBDIR;
+
+ if (resourcesJar != null) {
+ LOG.info("Extracting resources from jar at {} to {}",
resourcesJar, targetDir);
+ Utils.extractDirFromJar(resourcesJar,
ConfigUtils.RESOURCES_SUBDIR, _stormRoot);
+ } else if (url != null) {
+ LOG.info("Copying resources at {} to {} ", url.toString(),
targetDir);
+ if ("jar".equals(url.getProtocol())) {
+ JarURLConnection urlConnection = (JarURLConnection)
url.openConnection();
+
Utils.extractDirFromJar(urlConnection.getJarFileURL().getFile(),
ConfigUtils.RESOURCES_SUBDIR, _stormRoot);
+ } else {
+ _fsOps.copyDirectory(new File(url.getFile()), new
File(targetDir));
+ }
+ }
+ }
+ }
+
+ private class DownloadBlobs implements Callable<Void> {
+ private final String _topologyId;
+
+ public DownloadBlobs(String topologyId) {
+ _topologyId = topologyId;
+ }
+
+ @Override
+ public Void call() throws Exception {
+ String stormroot = ConfigUtils.supervisorStormDistRoot(_conf,
_topologyId);
--- End diff --
You should probably consider throwing a try-catch Exception or even
Throwable around these callables for logging, in case there are bugs. I think
ExecutorService swallows exceptions by default.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
