You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by zd-project <gi...@git.apache.org> on 2018/06/27 18:50:34 UTC
[GitHub] storm pull request #2739: [STORM-3125/3126/3127]: Refactoring methods in com...
GitHub user zd-project opened a pull request:
https://github.com/apache/storm/pull/2739
[STORM-3125/3126/3127]: Refactoring methods in components for Supervisor and DRPC
This is a supplement issue page to STORM-3099, separating out refactoring work from newly implemented metrics.
A few misc bug discovered during refactoring have been incorporate in this issue as well, including STORM-3126/3127.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/zd-project/storm STORM-3125
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/2739.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 #2739
----
commit 12ac608253ca96391b971568c3d08ed923e84d34
Author: Zhengdai Hu <hu...@...>
Date: 2018-06-18T15:44:19Z
STORM-3125: Refactored and commented code for Slot and ReadClusterState
(cherry picked from commit 5338491)
commit 71401e296be95a9cd12b6c7d546fec9255b4d7b2
Author: Zhengdai Hu <hu...@...>
Date: 2018-06-18T15:44:19Z
STORM-3126: Refactored methods for kill_workers to avoid
unnecessary force kill.
STORM-3125: Refactored methods of state machine for killing workers and
improved tests and metrics.
(cherry picked from commit 5338491)
commit e54dcd00d523897241b2d0c6627d3bd4a1ce0b5a
Author: Zhengdai Hu <hu...@...>
Date: 2018-06-21T19:25:05Z
STORM-3125: Refactored methods for blob localization
(cherry picked from commit d40397c)
commit 25e2249540891cc28172ff81ceb68f63758c7daf
Author: Zhengdai Hu <hu...@...>
Date: 2018-06-26T14:58:57Z
STORM-3127: Refactor AsyncLocalizer to avoid potential race condition
(cherry picked from commit 706029a)
commit 7503ff7bb58d31e76c9e71284facd3113a67742c
Author: Zhengdai Hu <hu...@...>
Date: 2018-06-26T18:58:33Z
STORM-3125: Refactored code for DRPC
(cherry picked from commit 3152e6f)
----
---
[GitHub] storm issue #2739: [STORM-3125/3126/3127]: Refactoring components for Superv...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on the issue:
https://github.com/apache/storm/pull/2739
@zd-project No it is good to merge. Just merged.
---
[GitHub] storm pull request #2739: [STORM-3125/3126/3127]: Refactoring components for...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2739#discussion_r198980565
--- Diff: storm-server/src/main/java/org/apache/storm/localizer/LocallyCachedBlob.java ---
@@ -59,57 +59,47 @@ protected LocallyCachedBlob(String blobDescription, String blobKey) {
this.blobKey = blobKey;
}
- protected static long downloadToTempLocation(ClientBlobStore store, String key, long currentVersion, IAdvancedFSOps fsOps,
- Function<Long, Path> getTempPath)
- throws KeyNotFoundException, AuthorizationException, IOException {
+ /**
+ * Helper function to download blob from blob store.
+ * @param store Blob store to fetch blobs from
+ * @param key Key to retrieve blobs
+ * @param pathSupplier A function that supplies the download destination of a blob. It guarantees the validity
+ * of path or throws {@link IOException}
+ * @param outStreamSupplier A function that supplies the {@link OutputStream} object
+ * @return The metadata of the download session, including blob's version and download destination
+ * @throws KeyNotFoundException Thrown if key to retrieve blob is invalid
+ * @throws AuthorizationException Thrown if the retrieval is not under security authorization
+ * @throws IOException Thrown if any IO error occurs
+ */
+ protected DownloadMeta fetch(ClientBlobStore store, String key,
+ IOFunction<Long, Path> pathSupplier,
+ IOFunction<File, OutputStream> outStreamSupplier)
+ throws KeyNotFoundException, AuthorizationException, IOException {
+
try (InputStreamWithMeta in = store.getBlob(key)) {
long newVersion = in.getVersion();
- if (newVersion == currentVersion) {
- LOG.warn("The version did not change, but going to download again {} {}", currentVersion, key);
+ if (newVersion == getLocalVersion()) {
--- End diff --
I agree with the caching, but could this be a premature optimization?
---
[GitHub] storm issue #2739: [STORM-3125/3126/3127]: Refactoring components for Superv...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on the issue:
https://github.com/apache/storm/pull/2739
All my manual tests passed looks good +1. Would like to see the nits addressed, but they are very minor.
---
[GitHub] storm pull request #2739: [STORM-3125/3126/3127]: Refactoring components for...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2739#discussion_r198900160
--- Diff: storm-server/src/main/java/org/apache/storm/utils/EnumUtil.java ---
@@ -0,0 +1,40 @@
+/**
+ * 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.utils;
+
+import java.util.EnumMap;
+import java.util.function.Function;
+
+public class EnumUtil {
+ public static String toString(Enum type) {
--- End diff --
You might call it toMetricName instead.
---
[GitHub] storm pull request #2739: [STORM-3125/3126/3127]: Refactoring components for...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2739#discussion_r198899945
--- Diff: storm-server/src/main/java/org/apache/storm/localizer/LocallyCachedBlob.java ---
@@ -59,57 +59,47 @@ protected LocallyCachedBlob(String blobDescription, String blobKey) {
this.blobKey = blobKey;
}
- protected static long downloadToTempLocation(ClientBlobStore store, String key, long currentVersion, IAdvancedFSOps fsOps,
- Function<Long, Path> getTempPath)
- throws KeyNotFoundException, AuthorizationException, IOException {
+ /**
+ * Helper function to download blob from blob store.
+ * @param store Blob store to fetch blobs from
+ * @param key Key to retrieve blobs
+ * @param pathSupplier A function that supplies the download destination of a blob. It guarantees the validity
+ * of path or throws {@link IOException}
+ * @param outStreamSupplier A function that supplies the {@link OutputStream} object
+ * @return The metadata of the download session, including blob's version and download destination
+ * @throws KeyNotFoundException Thrown if key to retrieve blob is invalid
+ * @throws AuthorizationException Thrown if the retrieval is not under security authorization
+ * @throws IOException Thrown if any IO error occurs
+ */
+ protected DownloadMeta fetch(ClientBlobStore store, String key,
+ IOFunction<Long, Path> pathSupplier,
+ IOFunction<File, OutputStream> outStreamSupplier)
+ throws KeyNotFoundException, AuthorizationException, IOException {
+
try (InputStreamWithMeta in = store.getBlob(key)) {
long newVersion = in.getVersion();
- if (newVersion == currentVersion) {
- LOG.warn("The version did not change, but going to download again {} {}", currentVersion, key);
+ if (newVersion == getLocalVersion()) {
--- End diff --
The code change is fine, I just don't want to call the API twice if we don't have to.
---
[GitHub] storm issue #2739: [STORM-3125/3126/3127]: Refactoring components for Superv...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on the issue:
https://github.com/apache/storm/pull/2739
Should I also rebase it onto current master branch?
---
[GitHub] storm pull request #2739: [STORM-3125/3126/3127]: Refactoring components for...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2739#discussion_r198900030
--- Diff: storm-server/src/main/java/org/apache/storm/localizer/LocallyCachedBlob.java ---
@@ -162,6 +152,31 @@ protected static long getSizeOnDisk(Path p) throws IOException {
*/
public abstract long getSizeOnDisk();
+ /**
+ * Get the size of p in bytes.
+ * @param p the path to read.
+ * @return the size of p in bytes.
+ */
+ protected static long getSizeOnDisk(Path p) throws IOException {
+ if (!Files.exists(p)) {
+ return 0;
+ } else if (Files.isRegularFile(p)) {
+ return Files.size(p);
+ } else {
+ //We will not follow sym links
+ return Files.walk(p)
+ .filter((subp) -> Files.isRegularFile(subp, LinkOption.NOFOLLOW_LINKS))
+ .mapToLong((subp) -> {
+ try {
+ return Files.size(subp);
+ } catch (IOException e) {
+ LOG.warn("Could not get the size of ");
--- End diff --
That may be, but if you could fix it that would be great.
---
[GitHub] storm pull request #2739: [STORM-3125/3126/3127]: Refactoring components for...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2739#discussion_r198977745
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/supervisor/Slot.java ---
@@ -241,64 +255,54 @@ static DynamicState prepareForNewAssignmentNoWorkersRunning(DynamicState dynamic
.withState(MachineState.WAITING_FOR_BLOB_LOCALIZATION);
}
- /**
- * Kill the current container and start downloading what the new assignment needs, if there is a new assignment.
- * PRECONDITION: container != null
- * @param dynamicState current state
- * @param staticState static data
- * @return the next state
- * @throws Exception
- */
- static DynamicState killContainerForChangedAssignment(DynamicState dynamicState, StaticState staticState) throws Exception {
- assert (dynamicState.container != null);
+ private static DynamicState killContainerFor(KillReason reason, DynamicState dynamicState, StaticState staticState)
+ throws Exception {
+ assert dynamicState.container != null;
- staticState.iSupervisor.killedWorker(staticState.port);
- dynamicState.container.kill();
- Future<Void> pendingDownload = null;
- if (dynamicState.newAssignment != null) {
- pendingDownload = staticState.localizer.requestDownloadTopologyBlobs(dynamicState.newAssignment, staticState.port,
- staticState.changingCallback);
+ //Skip special case if `storm kill_workers` is already invoked
+ Boolean isDead = dynamicState.container.areAllProcessesDead();
+ if (!isDead) {
+ //TODO: Is this necessary?
--- End diff --
Okay. I just want to make sure that this is indeed the desired behavior.
---
[GitHub] storm pull request #2739: [STORM-3125/3126/3127]: Refactoring components for...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/2739#discussion_r198983226
--- Diff: storm-server/src/main/java/org/apache/storm/localizer/LocallyCachedBlob.java ---
@@ -59,57 +59,47 @@ protected LocallyCachedBlob(String blobDescription, String blobKey) {
this.blobKey = blobKey;
}
- protected static long downloadToTempLocation(ClientBlobStore store, String key, long currentVersion, IAdvancedFSOps fsOps,
- Function<Long, Path> getTempPath)
- throws KeyNotFoundException, AuthorizationException, IOException {
+ /**
+ * Helper function to download blob from blob store.
+ * @param store Blob store to fetch blobs from
+ * @param key Key to retrieve blobs
+ * @param pathSupplier A function that supplies the download destination of a blob. It guarantees the validity
+ * of path or throws {@link IOException}
+ * @param outStreamSupplier A function that supplies the {@link OutputStream} object
+ * @return The metadata of the download session, including blob's version and download destination
+ * @throws KeyNotFoundException Thrown if key to retrieve blob is invalid
+ * @throws AuthorizationException Thrown if the retrieval is not under security authorization
+ * @throws IOException Thrown if any IO error occurs
+ */
+ protected DownloadMeta fetch(ClientBlobStore store, String key,
+ IOFunction<Long, Path> pathSupplier,
+ IOFunction<File, OutputStream> outStreamSupplier)
+ throws KeyNotFoundException, AuthorizationException, IOException {
+
try (InputStreamWithMeta in = store.getBlob(key)) {
long newVersion = in.getVersion();
- if (newVersion == currentVersion) {
- LOG.warn("The version did not change, but going to download again {} {}", currentVersion, key);
+ if (newVersion == getLocalVersion()) {
+ LOG.warn("The version did not change, but going to download again {} {}", getLocalVersion(), key);
}
- Path tmpLocation = getTempPath.apply(newVersion);
- long totalRead = 0;
+
//Make sure the parent directory is there and ready to go
- fsOps.forceMkdir(tmpLocation.getParent());
- try (OutputStream outStream = fsOps.getOutputStream(tmpLocation.toFile())) {
+ Path downloadPath = pathSupplier.apply(newVersion);
+ LOG.debug("Downloading {} to {}", key, downloadPath);
+
+ long totalRead = 0;
+ try (OutputStream out = outStreamSupplier.apply(downloadPath.toFile())) {
byte[] buffer = new byte[4096];
- int read = 0;
- while ((read = in.read(buffer)) > 0) {
- outStream.write(buffer, 0, read);
+ int read;
+ while ((read = in.read(buffer)) >= 0) {
+ out.write(buffer, 0, read);
totalRead += read;
}
}
long expectedSize = in.getFileLength();
if (totalRead != expectedSize) {
--- End diff --
It was mostly about where we were seeing errors at the time.
---
[GitHub] storm pull request #2739: [STORM-3125/3126/3127]: Refactoring components for...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/storm/pull/2739
---