You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by govind-menon <gi...@git.apache.org> on 2017/06/02 21:56:12 UTC
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
GitHub user govind-menon opened a pull request:
https://github.com/apache/storm/pull/2148
Let Topology query if blobs have changed
WIP
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/govind-menon/storm YSTORM-4362
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/2148.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2148
----
commit 14a39a10a6faa4a88a4e67b72c7ce8d322067dab
Author: Govind Menon <go...@gmail.com>
Date: 2017-06-02T21:55:19Z
Let Topology query if blobs have changed
----
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120174827
--- Diff: storm-server/src/main/java/org/apache/storm/localizer/Localizer.java ---
@@ -226,9 +218,8 @@ public boolean accept(File dir, String name) {
protected void reconstructLocalizedResources() {
try {
LOG.info("Reconstruct localized resource: " + getUserCacheDir().getPath());
- File[] users = readDirContents(getUserCacheDir().getPath());
-
- if (users != null) {
+ Collection<File> users = ConfigUtils.readDirFiles(getUserCacheDir().getPath());
+ if (users == null || users.isEmpty()) {
--- End diff --
This can be reverted. The only need was to check for null so we don't NPE. This inverted the check.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by govind-menon <gi...@git.apache.org>.
Github user govind-menon commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120168623
--- Diff: storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java ---
@@ -246,7 +248,19 @@ public void start() throws Exception {
}
}
});
-
+
+ workerState.checkForUpdatedBlobsTimer.scheduleRecurring(0,
--- End diff --
I wanted the period for this check to be it's own value and not tied to other timers
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120134652
--- Diff: storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java ---
@@ -299,6 +313,33 @@ public void doExecutorHeartbeats() {
}
}
+ public Map<String, Long> getCurrentBlobVersions() throws IOException {
+ Map<String, Long> results = new HashMap();
+ Map<String, Map<String, Object>> blobstoreMap = (Map<String, Map<String, Object>>) workerState.getTopologyConf().get(Config.TOPOLOGY_BLOBSTORE_MAP);
+ if (blobstoreMap != null) {
+ String stormRoot = ConfigUtils.supervisorStormDistRoot(workerState.getTopologyConf(), workerState.getTopologyId());
+ for (Map.Entry<String, Map<String, Object>> entry : blobstoreMap.entrySet()) {
+ String keyName = entry.getKey();
+ String localFileName = keyName;
+ Map<String, Object> blobInfo = blobstoreMap.get(keyName);
+ if (blobInfo != null && blobInfo.containsKey("localname")) {
+ localFileName = (String) blobInfo.get("localname");
+ }
+ String blobPath = new File(stormRoot, localFileName).getCanonicalPath();
+ File versionFile = new File(blobPath.replaceAll(".[0-9]+$", DEFAULT_BLOB_VERSION_SUFFIX));
+ results.put(localFileName, Utils.getVersionFromBlobVersionFile(versionFile));
--- End diff --
Can we please do this slightly differently? The way it currently works `$whatever.version` can be updated before we officially commit the changes (I know that is not good and I am trying to change it), but for now we really should do it more like.
````
private static final Pattern BLOB_VERSION_EXTRACTION = Pattern.compile(".*\\.([0-9]+)$");
//...
String blobWithVersion = new File("./", localFileName).getCanonicalFile().getName();
Matcher m = BLOB_VERSION_EXTRACTION.matcher(blobWithVersion);
if (m.matches()) {
results.put(localFileName, Long.valueOf(m.group(1)));
}
````
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120174018
--- Diff: storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java ---
@@ -246,7 +248,19 @@ public void start() throws Exception {
}
}
});
-
+
+ workerState.checkForUpdatedBlobsTimer.scheduleRecurring(0,
--- End diff --
OK. We have so many now it might be good to switch it over to a pool in the future, but nothing you need to worry about.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120130253
--- Diff: storm-client/src/jvm/org/apache/storm/utils/ConfigUtils.java ---
@@ -27,12 +28,11 @@
import java.io.File;
import java.io.IOException;
import java.net.URLEncoder;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Random;
+import java.util.*;
--- End diff --
* in imports are generally discouraged.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: STORM-2543: Let Topology query if blobs have chang...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/storm/pull/2148
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120132118
--- Diff: storm-client/src/jvm/org/apache/storm/task/TopologyContext.java ---
@@ -57,13 +57,14 @@
private Map<String, Object> _executorData;
private Map<Integer,Map<Integer, Map<String, IMetric>>> _registeredMetrics;
private AtomicBoolean _openOrPrepareWasCalled;
-
+ private Map<String, Long> _blobToLastKnownVersion;
public TopologyContext(StormTopology topology,
Map<String, Object> topoConf,
Map<Integer, String> taskToComponent,
Map<String, List<Integer>> componentToSortedTasks,
Map<String, Map<String, Fields>> componentToStreamToFields,
+ Map<String, Long> blobToLastKnownVersion,
--- End diff --
So this is a little odd in that we send the versions to all of the children by passing in a variable that we expect to change behind the scenes. I think it would be good to call that out in some comments both in the constructor here, and on field too so it is obvious here in the code what is happening.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120129840
--- Diff: storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java ---
@@ -246,7 +248,19 @@ public void start() throws Exception {
}
}
});
-
+
+ workerState.checkForUpdatedBlobsTimer.scheduleRecurring(0,
--- End diff --
Is there another timer we could use here?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120129559
--- Diff: storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java ---
@@ -246,7 +248,19 @@ public void start() throws Exception {
}
}
});
-
+
+ workerState.checkForUpdatedBlobsTimer.scheduleRecurring(0,
+ (Integer) conf.getOrDefault(Config.WORKER_BLOB_UPDATE_POLL_INTERVAL_SECS, 10), new Runnable() {
+ @Override public void run() {
+ try {
+ LOG.info("Checking if blobs have updated");
+ updateBlobUpdates();
+ } catch (IOException e) {
+ e.printStackTrace();
--- End diff --
Can we log this instead of printing it to stderr? Also it might be nice to have a comment explaining why you are ignoring the exception.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120131449
--- Diff: storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java ---
@@ -299,6 +313,33 @@ public void doExecutorHeartbeats() {
}
}
+ public Map<String, Long> getCurrentBlobVersions() throws IOException {
+ Map<String, Long> results = new HashMap();
+ Map<String, Map<String, Object>> blobstoreMap = (Map<String, Map<String, Object>>) workerState.getTopologyConf().get(Config.TOPOLOGY_BLOBSTORE_MAP);
+ if (blobstoreMap != null) {
+ String stormRoot = ConfigUtils.supervisorStormDistRoot(workerState.getTopologyConf(), workerState.getTopologyId());
+ for (Map.Entry<String, Map<String, Object>> entry : blobstoreMap.entrySet()) {
+ String keyName = entry.getKey();
+ String localFileName = keyName;
+ Map<String, Object> blobInfo = blobstoreMap.get(keyName);
+ if (blobInfo != null && blobInfo.containsKey("localname")) {
+ localFileName = (String) blobInfo.get("localname");
+ }
+ String blobPath = new File(stormRoot, localFileName).getCanonicalPath();
+ File versionFile = new File(blobPath.replaceAll(".[0-9]+$", DEFAULT_BLOB_VERSION_SUFFIX));
+ results.put(localFileName, Utils.getVersionFromBlobVersionFile(versionFile));
+ }
+ }
+ return results;
+ }
+
+ public void updateBlobUpdates() throws IOException {
+ Map<String, Long> latestBlobVersions = getCurrentBlobVersions();
+ workerState.blobToLastKnownVersion.putAll(latestBlobVersions);
--- End diff --
We need some kind of synchronization here. Perhaps we can just make `blobToLastKnownVersion` a ConcurrentMap. That should be enough.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2148: Let Topology query if blobs have changed
Posted by govind-menon <gi...@git.apache.org>.
Github user govind-menon commented on the issue:
https://github.com/apache/storm/pull/2148
Tagging @satishd for another review.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120132248
--- Diff: storm-server/src/main/java/org/apache/storm/localizer/Localizer.java ---
@@ -41,10 +42,7 @@
import java.io.PrintWriter;
import java.nio.file.Files;
import java.nio.file.Paths;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
--- End diff --
Again `*` is discouraged here.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120130332
--- Diff: storm-client/src/jvm/org/apache/storm/utils/ConfigUtils.java ---
@@ -27,12 +28,11 @@
import java.io.File;
import java.io.IOException;
import java.net.URLEncoder;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Random;
+import java.util.*;
import java.util.concurrent.Callable;
+import java.util.stream.Collectors;
-public class ConfigUtils {
+public class ConfigUtils {
--- End diff --
can we remove the extra ' ' added?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120129279
--- Diff: storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java ---
@@ -246,7 +248,19 @@ public void start() throws Exception {
}
}
});
-
+
+ workerState.checkForUpdatedBlobsTimer.scheduleRecurring(0,
+ (Integer) conf.getOrDefault(Config.WORKER_BLOB_UPDATE_POLL_INTERVAL_SECS, 10), new Runnable() {
+ @Override public void run() {
+ try {
+ LOG.info("Checking if blobs have updated");
--- End diff --
This should probably be a debug log.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120130165
--- Diff: storm-client/src/jvm/org/apache/storm/task/TopologyContext.java ---
@@ -57,13 +57,14 @@
private Map<String, Object> _executorData;
private Map<Integer,Map<Integer, Map<String, IMetric>>> _registeredMetrics;
private AtomicBoolean _openOrPrepareWasCalled;
-
+ private Map<String, Long> _blobToLastKnownVersion;
--- End diff --
Can we remove the `_` at the beginning. I know it matches the rest of the file but we are moving away from this style.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120134821
--- Diff: storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java ---
@@ -299,6 +313,33 @@ public void doExecutorHeartbeats() {
}
}
+ public Map<String, Long> getCurrentBlobVersions() throws IOException {
+ Map<String, Long> results = new HashMap();
+ Map<String, Map<String, Object>> blobstoreMap = (Map<String, Map<String, Object>>) workerState.getTopologyConf().get(Config.TOPOLOGY_BLOBSTORE_MAP);
+ if (blobstoreMap != null) {
+ String stormRoot = ConfigUtils.supervisorStormDistRoot(workerState.getTopologyConf(), workerState.getTopologyId());
+ for (Map.Entry<String, Map<String, Object>> entry : blobstoreMap.entrySet()) {
+ String keyName = entry.getKey();
+ String localFileName = keyName;
+ Map<String, Object> blobInfo = blobstoreMap.get(keyName);
--- End diff --
Lets use `entry.getValue()` instead. It is slightly more efficient.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120131569
--- Diff: storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java ---
@@ -299,6 +313,33 @@ public void doExecutorHeartbeats() {
}
}
+ public Map<String, Long> getCurrentBlobVersions() throws IOException {
+ Map<String, Long> results = new HashMap();
+ Map<String, Map<String, Object>> blobstoreMap = (Map<String, Map<String, Object>>) workerState.getTopologyConf().get(Config.TOPOLOGY_BLOBSTORE_MAP);
+ if (blobstoreMap != null) {
+ String stormRoot = ConfigUtils.supervisorStormDistRoot(workerState.getTopologyConf(), workerState.getTopologyId());
+ for (Map.Entry<String, Map<String, Object>> entry : blobstoreMap.entrySet()) {
+ String keyName = entry.getKey();
+ String localFileName = keyName;
+ Map<String, Object> blobInfo = blobstoreMap.get(keyName);
+ if (blobInfo != null && blobInfo.containsKey("localname")) {
+ localFileName = (String) blobInfo.get("localname");
+ }
+ String blobPath = new File(stormRoot, localFileName).getCanonicalPath();
+ File versionFile = new File(blobPath.replaceAll(".[0-9]+$", DEFAULT_BLOB_VERSION_SUFFIX));
+ results.put(localFileName, Utils.getVersionFromBlobVersionFile(versionFile));
+ }
+ }
+ return results;
+ }
+
+ public void updateBlobUpdates() throws IOException {
+ Map<String, Long> latestBlobVersions = getCurrentBlobVersions();
+ workerState.blobToLastKnownVersion.putAll(latestBlobVersions);
+ LOG.debug("Latest versions for blobs " + latestBlobVersions.toString());
--- End diff --
```
LOG.debug("Latest versions for blobs {}", latestBlobVersions);
```
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120130996
--- Diff: storm-client/src/jvm/org/apache/storm/Constants.java ---
@@ -56,5 +56,7 @@
public static final String STORM_ACTIVE_ATOM = "storm-active-atom";
public static final String COMPONENT_TO_DEBUG_ATOM = "storm-component->debug-atom";
public static final Object LOAD_MAPPING = "load-mapping";
+ public static final String USERCACHE = "usercache";
+ public static final String DEFAULT_BLOB_VERSION_SUFFIX = ".version";
--- End diff --
I'm not sure these needed to be moved. They look like they were moved for one possible way of doing this, and then not reverted when they were not used by the client.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120129659
--- Diff: storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java ---
@@ -299,6 +313,33 @@ public void doExecutorHeartbeats() {
}
}
+ public Map<String, Long> getCurrentBlobVersions() throws IOException {
+ Map<String, Long> results = new HashMap();
--- End diff --
The new HashMap should have the generics on it `new HashMap<>();`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2148: Let Topology query if blobs have changed
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2148#discussion_r120135498
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2369,7 +2369,7 @@ public void launchServer() throws Exception {
for (ClusterMetricsConsumerExecutor exec: clusterConsumerExceutors) {
exec.prepare();
}
-
+ assertIsLeader();
if (isLeader()) {
--- End diff --
This is a little odd. Why do we need to be sure that we are the leader here? Especially with code right below it checking if we are the leader or not.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---