You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2018/04/30 21:23:02 UTC
[01/15] flink git commit: [FLINK-9214] [tests,
yarn] YarnClient should be stopped in
YARNSessionCapacitySchedulerITCase#testDetachedPerJobYarnClusterInternal
Repository: flink
Updated Branches:
refs/heads/master 07aa2d469 -> ecde6c328
[FLINK-9214] [tests, yarn] YarnClient should be stopped in YARNSessionCapacitySchedulerITCase#testDetachedPerJobYarnClusterInternal
This closes #5892
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/bd8cb86b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/bd8cb86b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/bd8cb86b
Branch: refs/heads/master
Commit: bd8cb86bf17bfcdd1ce2dd1f01e40ca07f1148ab
Parents: 07aa2d4
Author: yanghua <ya...@gmail.com>
Authored: Sun Apr 22 16:38:28 2018 +0800
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Apr 30 21:06:16 2018 +0200
----------------------------------------------------------------------
.../apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java | 6 ++++++
1 file changed, 6 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/bd8cb86b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
index 3c442b9..4bbd500 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
@@ -590,6 +590,12 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
LOG.warn("testDetachedPerJobYarnClusterInternal: Exception while deleting the JobManager address file", e);
}
+ try {
+ LOG.info("testDetachedPerJobYarnClusterInternal: Closing the yarn client");
+ yc.stop();
+ } catch (Exception e) {
+ LOG.warn("testDetachedPerJobYarnClusterInternal: Exception while close the yarn client", e);
+ }
}
}
[04/15] flink git commit: [hotfix] [mesos] Delete unused class
FlinkMesosSessionCli.
Posted by se...@apache.org.
[hotfix] [mesos] Delete unused class FlinkMesosSessionCli.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/17c68e55
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/17c68e55
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/17c68e55
Branch: refs/heads/master
Commit: 17c68e5592bd839e8c0ce9372ffcc46d0315c412
Parents: 4fffd60
Author: gyao <ga...@data-artisans.com>
Authored: Thu Apr 26 17:38:20 2018 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Apr 30 21:20:56 2018 +0200
----------------------------------------------------------------------
.../flink/mesos/cli/FlinkMesosSessionCli.java | 75 --------------------
1 file changed, 75 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/17c68e55/flink-mesos/src/main/java/org/apache/flink/mesos/cli/FlinkMesosSessionCli.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/cli/FlinkMesosSessionCli.java b/flink-mesos/src/main/java/org/apache/flink/mesos/cli/FlinkMesosSessionCli.java
deleted file mode 100644
index e0df1c9..0000000
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/cli/FlinkMesosSessionCli.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * 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.flink.mesos.cli;
-
-import org.apache.flink.configuration.Configuration;
-
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.type.TypeReference;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
-
-import java.io.IOException;
-import java.util.Map;
-
-/**
- * Class handling the command line interface to the Mesos session.
- */
-public class FlinkMesosSessionCli {
-
- private static final ObjectMapper mapper = new ObjectMapper();
-
- /**
- * Decode encoded dynamic properties.
- *
- * @param dynamicPropertiesEncoded encoded properties produced by the encoding method.
- * @return a configuration instance to be merged with the static configuration.
- */
- public static Configuration decodeDynamicProperties(String dynamicPropertiesEncoded) {
- try {
- Configuration configuration = new Configuration();
- if (dynamicPropertiesEncoded != null) {
- TypeReference<Map<String, String>> typeRef = new TypeReference<Map<String, String>>() {
- };
- Map<String, String> props = mapper.readValue(dynamicPropertiesEncoded, typeRef);
- for (Map.Entry<String, String> property : props.entrySet()) {
- configuration.setString(property.getKey(), property.getValue());
- }
- }
- return configuration;
- } catch (IOException ex) {
- throw new IllegalArgumentException("unreadable encoded properties", ex);
- }
- }
-
- /**
- * Encode dynamic properties as a string to be transported as an environment variable.
- *
- * @param configuration the dynamic properties to encode.
- * @return a string to be decoded later.
- */
- public static String encodeDynamicProperties(Configuration configuration) {
- try {
- String dynamicPropertiesEncoded = mapper.writeValueAsString(configuration.toMap());
- return dynamicPropertiesEncoded;
- }
- catch (JsonProcessingException ex) {
- throw new IllegalArgumentException("unwritable properties", ex);
- }
- }
-}
[10/15] flink git commit: [FLINK-9196] [flip6,
yarn] Cleanup application files when deregistering YARN AM
Posted by se...@apache.org.
[FLINK-9196] [flip6, yarn] Cleanup application files when deregistering YARN AM
Enable graceful cluster shut down via HTTP.
Remove Flink application files from remote file system when the
YarnResourceManager deregisters the YARN ApplicationMaster.
This closes #5938
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/37df3c87
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/37df3c87
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/37df3c87
Branch: refs/heads/master
Commit: 37df3c8720385da6393a7157654a963c03e9ebb3
Parents: 9c49dfc
Author: gyao <ga...@data-artisans.com>
Authored: Thu Apr 19 12:07:54 2018 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Apr 30 21:21:09 2018 +0200
----------------------------------------------------------------------
.../apache/flink/client/cli/CliFrontend.java | 4 +-
.../client/deployment/ClusterDescriptor.java | 2 +-
.../LegacyStandaloneClusterDescriptor.java | 2 +-
.../deployment/StandaloneClusterDescriptor.java | 2 +-
.../flink/client/program/ClusterClient.java | 3 +
.../client/program/rest/RestClusterClient.java | 16 +++++
.../client/cli/util/DummyClusterDescriptor.java | 2 +-
.../flink/runtime/dispatcher/Dispatcher.java | 6 ++
.../rest/handler/cluster/ShutdownHandler.java | 58 ++++++++++++++++
.../rest/messages/cluster/ShutdownHeaders.java | 69 ++++++++++++++++++++
.../runtime/webmonitor/RestfulGateway.java | 4 ++
.../runtime/webmonitor/WebMonitorEndpoint.java | 11 ++++
.../flink/yarn/YarnConfigurationITCase.java | 2 +-
.../util/NonDeployingYarnClusterDescriptor.java | 2 +-
.../yarn/AbstractYarnClusterDescriptor.java | 19 +++++-
.../main/java/org/apache/flink/yarn/Utils.java | 24 +++++++
.../apache/flink/yarn/YarnClusterClient.java | 15 +++++
.../apache/flink/yarn/YarnResourceManager.java | 2 +
.../flink/yarn/cli/FlinkYarnSessionCli.java | 10 +--
.../org/apache/flink/yarn/YarnJobManager.scala | 4 +-
.../java/org/apache/flink/yarn/UtilsTest.java | 52 +++++++++++++++
.../flink/yarn/YarnResourceManagerTest.java | 26 +++++++-
22 files changed, 312 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
index 7745ca0..95a9949 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
@@ -277,8 +277,8 @@ public class CliFrontend {
if (clusterId == null && !client.isDetached()) {
// terminate the cluster only if we have started it before and if it's not detached
try {
- clusterDescriptor.terminateCluster(client.getClusterId());
- } catch (FlinkException e) {
+ client.shutDownCluster();
+ } catch (final Exception e) {
LOG.info("Could not properly terminate the Flink cluster.", e);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterDescriptor.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterDescriptor.java
index f9f5d4b..e6b3922 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterDescriptor.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterDescriptor.java
@@ -72,5 +72,5 @@ public interface ClusterDescriptor<T> extends AutoCloseable {
* @param clusterId identifying the cluster to shut down
* @throws FlinkException if the cluster could not be terminated
*/
- void terminateCluster(T clusterId) throws FlinkException;
+ void killCluster(T clusterId) throws FlinkException;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-clients/src/main/java/org/apache/flink/client/deployment/LegacyStandaloneClusterDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/LegacyStandaloneClusterDescriptor.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/LegacyStandaloneClusterDescriptor.java
index 21e020c..b448970 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/LegacyStandaloneClusterDescriptor.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/LegacyStandaloneClusterDescriptor.java
@@ -65,7 +65,7 @@ public class LegacyStandaloneClusterDescriptor implements ClusterDescriptor<Stan
}
@Override
- public void terminateCluster(StandaloneClusterId clusterId) throws FlinkException {
+ public void killCluster(StandaloneClusterId clusterId) throws FlinkException {
throw new UnsupportedOperationException("Cannot terminate standalone clusters.");
}
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-clients/src/main/java/org/apache/flink/client/deployment/StandaloneClusterDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/StandaloneClusterDescriptor.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/StandaloneClusterDescriptor.java
index c4bcde6..bdf8faf 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/StandaloneClusterDescriptor.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/StandaloneClusterDescriptor.java
@@ -66,7 +66,7 @@ public class StandaloneClusterDescriptor implements ClusterDescriptor<Standalone
}
@Override
- public void terminateCluster(StandaloneClusterId clusterId) throws FlinkException {
+ public void killCluster(StandaloneClusterId clusterId) throws FlinkException {
throw new UnsupportedOperationException("Cannot terminate a standalone cluster.");
}
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
index 2ef0b2e..ac779a7 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
@@ -1036,4 +1036,7 @@ public abstract class ClusterClient<T> {
throw new UnsupportedOperationException("The " + getClass().getSimpleName() + " does not support rescaling.");
}
+ public void shutDownCluster() {
+ throw new UnsupportedOperationException("The " + getClass().getSimpleName() + " does not support shutDownCluster.");
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
index 3d50e93..5c04be7 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
@@ -67,6 +67,7 @@ import org.apache.flink.runtime.rest.messages.RequestBody;
import org.apache.flink.runtime.rest.messages.ResponseBody;
import org.apache.flink.runtime.rest.messages.TerminationModeQueryParameter;
import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.cluster.ShutdownHeaders;
import org.apache.flink.runtime.rest.messages.job.JobDetailsHeaders;
import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
import org.apache.flink.runtime.rest.messages.job.JobExecutionResultHeaders;
@@ -570,6 +571,21 @@ public class RestClusterClient<T> extends ClusterClient<T> implements NewCluster
});
}
+ @Override
+ public void shutDownCluster() {
+ try {
+ sendRetryableRequest(
+ ShutdownHeaders.getInstance(),
+ EmptyMessageParameters.getInstance(),
+ EmptyRequestBody.getInstance(),
+ isConnectionProblemException()).get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (ExecutionException e) {
+ log.error("Error while shutting down cluster", e);
+ }
+ }
+
/**
* Creates a {@code CompletableFuture} that polls a {@code AsynchronouslyCreatedResource} until
* its {@link AsynchronouslyCreatedResource#queueStatus() QueueStatus} becomes
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyClusterDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyClusterDescriptor.java b/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyClusterDescriptor.java
index df2f3f7..7620ae2 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyClusterDescriptor.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyClusterDescriptor.java
@@ -60,7 +60,7 @@ public class DummyClusterDescriptor<T> implements ClusterDescriptor<T> {
}
@Override
- public void terminateCluster(T clusterId) throws FlinkException {
+ public void killCluster(T clusterId) throws FlinkException {
throw new UnsupportedOperationException("Cannot terminate a dummy cluster.");
}
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
index 41efaab..c05255b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
@@ -518,6 +518,12 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
jobMasterGateway.triggerSavepoint(targetDirectory, cancelJob, timeout));
}
+ @Override
+ public CompletableFuture<Acknowledge> shutDownCluster() {
+ shutDown();
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ }
+
/**
* Cleans up the job related data from the dispatcher. If cleanupHA is true, then
* the data will also be removed from HA.
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/ShutdownHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/ShutdownHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/ShutdownHandler.java
new file mode 100644
index 0000000..73fca58
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/ShutdownHandler.java
@@ -0,0 +1,58 @@
+/*
+ * 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.flink.runtime.rest.handler.cluster;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.EmptyResponseBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * REST handler which allows to shut down the cluster.
+ */
+public class ShutdownHandler extends
+ AbstractRestHandler<RestfulGateway, EmptyRequestBody, EmptyResponseBody, EmptyMessageParameters> {
+
+ public ShutdownHandler(
+ final CompletableFuture<String> localRestAddress,
+ final GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ final Time timeout,
+ final Map<String, String> responseHeaders,
+ final MessageHeaders<EmptyRequestBody, EmptyResponseBody, EmptyMessageParameters> messageHeaders) {
+ super(localRestAddress, leaderRetriever, timeout, responseHeaders, messageHeaders);
+ }
+
+ @Override
+ protected CompletableFuture<EmptyResponseBody> handleRequest(
+ @Nonnull final HandlerRequest<EmptyRequestBody, EmptyMessageParameters> request,
+ @Nonnull final RestfulGateway gateway) throws RestHandlerException {
+ return gateway.shutDownCluster().thenApply(ignored -> EmptyResponseBody.getInstance());
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/ShutdownHeaders.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/ShutdownHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/ShutdownHeaders.java
new file mode 100644
index 0000000..75a1e99
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/ShutdownHeaders.java
@@ -0,0 +1,69 @@
+/*
+ * 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.flink.runtime.rest.messages.cluster;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.EmptyResponseBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+/**
+ * Message headers for {@link org.apache.flink.runtime.rest.handler.cluster.ShutdownHandler}.
+ */
+public class ShutdownHeaders implements MessageHeaders<EmptyRequestBody, EmptyResponseBody, EmptyMessageParameters> {
+
+ private static final ShutdownHeaders INSTANCE = new ShutdownHeaders();
+
+ @Override
+ public Class<EmptyResponseBody> getResponseClass() {
+ return EmptyResponseBody.class;
+ }
+
+ @Override
+ public HttpResponseStatus getResponseStatusCode() {
+ return HttpResponseStatus.OK;
+ }
+
+ @Override
+ public Class<EmptyRequestBody> getRequestClass() {
+ return EmptyRequestBody.class;
+ }
+
+ @Override
+ public EmptyMessageParameters getUnresolvedMessageParameters() {
+ return EmptyMessageParameters.getInstance();
+ }
+
+ @Override
+ public HttpMethodWrapper getHttpMethod() {
+ return HttpMethodWrapper.DELETE;
+ }
+
+ @Override
+ public String getTargetRestEndpointURL() {
+ return "/cluster";
+ }
+
+ public static ShutdownHeaders getInstance() {
+ return INSTANCE;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/RestfulGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/RestfulGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/RestfulGateway.java
index 6bb088c..6a6c34b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/RestfulGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/RestfulGateway.java
@@ -200,4 +200,8 @@ public interface RestfulGateway extends RpcGateway {
@RpcTimeout Time timeout) {
throw new UnsupportedOperationException();
}
+
+ default CompletableFuture<Acknowledge> shutDownCluster() {
+ throw new UnsupportedOperationException();
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
index 1a67d92..ef6721b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
@@ -34,6 +34,7 @@ import org.apache.flink.runtime.rest.handler.RestHandlerSpecification;
import org.apache.flink.runtime.rest.handler.cluster.ClusterConfigHandler;
import org.apache.flink.runtime.rest.handler.cluster.ClusterOverviewHandler;
import org.apache.flink.runtime.rest.handler.cluster.DashboardConfigHandler;
+import org.apache.flink.runtime.rest.handler.cluster.ShutdownHandler;
import org.apache.flink.runtime.rest.handler.job.JobAccumulatorsHandler;
import org.apache.flink.runtime.rest.handler.job.JobConfigHandler;
import org.apache.flink.runtime.rest.handler.job.JobDetailsHandler;
@@ -104,6 +105,7 @@ import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointConfigHeader
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointStatisticDetailsHeaders;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatisticsHeaders;
import org.apache.flink.runtime.rest.messages.checkpoints.TaskCheckpointStatisticsHeaders;
+import org.apache.flink.runtime.rest.messages.cluster.ShutdownHeaders;
import org.apache.flink.runtime.rest.messages.job.JobDetailsHeaders;
import org.apache.flink.runtime.rest.messages.job.JobExecutionResultHeaders;
import org.apache.flink.runtime.rest.messages.job.SubtaskCurrentAttemptDetailsHeaders;
@@ -558,6 +560,13 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp
timeout,
responseHeaders);
+ final ShutdownHandler shutdownHandler = new ShutdownHandler(
+ restAddressFuture,
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ ShutdownHeaders.getInstance());
+
final File webUiDir = restConfiguration.getWebUiDir();
Optional<StaticFileServerHandler<T>> optWebContent;
@@ -619,6 +628,8 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp
handlers.add(Tuple2.of(YarnCancelJobTerminationHeaders.getInstance(), jobCancelTerminationHandler));
handlers.add(Tuple2.of(YarnStopJobTerminationHeaders.getInstance(), jobStopTerminationHandler));
+ handlers.add(Tuple2.of(ShutdownHeaders.getInstance(), shutdownHandler));
+
optWebContent.ifPresent(
webContent -> {
handlers.add(Tuple2.of(WebContentHandlerSpecification.getInstance(), webContent));
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
index 3a2f957..a5f7c10 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
@@ -184,7 +184,7 @@ public class YarnConfigurationITCase extends YarnTestBase {
clusterClient.shutdown();
}
- clusterDescriptor.terminateCluster(clusterId);
+ clusterDescriptor.killCluster(clusterId);
} finally {
clusterDescriptor.close();
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-yarn-tests/src/test/java/org/apache/flink/yarn/util/NonDeployingYarnClusterDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/util/NonDeployingYarnClusterDescriptor.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/util/NonDeployingYarnClusterDescriptor.java
index 4916b73..de0e1d7 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/util/NonDeployingYarnClusterDescriptor.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/util/NonDeployingYarnClusterDescriptor.java
@@ -83,7 +83,7 @@ public class NonDeployingYarnClusterDescriptor extends AbstractYarnClusterDescri
}
@Override
- public void terminateCluster(ApplicationId clusterId) {
+ public void killCluster(ApplicationId clusterId) {
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
index aec5fdb..1948d0f 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
@@ -412,9 +412,12 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
}
@Override
- public void terminateCluster(ApplicationId applicationId) throws FlinkException {
+ public void killCluster(ApplicationId applicationId) throws FlinkException {
try {
yarnClient.killApplication(applicationId);
+ Utils.deleteApplicationFiles(Collections.singletonMap(
+ YarnConfigKeys.FLINK_YARN_FILES,
+ getYarnFilesDir(applicationId).toUri().toString()));
} catch (YarnException | IOException e) {
throw new FlinkException("Could not kill the Yarn Flink cluster with id " + applicationId + '.', e);
}
@@ -897,8 +900,7 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
}
}
- Path yarnFilesDir = new Path(homeDir, ".flink/" + appId + '/');
-
+ final Path yarnFilesDir = getYarnFilesDir(appId);
FsPermission permission = new FsPermission(FsAction.ALL, FsAction.NONE, FsAction.NONE);
fs.setPermission(yarnFilesDir, permission); // set permission for path.
@@ -1086,6 +1088,17 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
}
/**
+ * Returns the Path where the YARN application files should be uploaded to.
+ *
+ * @param appId YARN application id
+ */
+ private Path getYarnFilesDir(final ApplicationId appId) throws IOException {
+ final FileSystem fileSystem = FileSystem.get(yarnConfiguration);
+ final Path homeDir = fileSystem.getHomeDirectory();
+ return new Path(homeDir, ".flink/" + appId + '/');
+ }
+
+ /**
* Uploads and registers a single resource and adds it to <tt>localResources</tt>.
*
* @param key
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java b/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
index fb9a478..20e02e1 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
@@ -24,6 +24,7 @@ import org.apache.flink.runtime.clusterframework.BootstrapTools;
import org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters;
import org.apache.flink.runtime.util.HadoopUtils;
import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
@@ -173,6 +174,29 @@ public final class Utils {
}
/**
+ * Deletes the YARN application files, e.g., Flink binaries, libraries, etc., from the remote
+ * filesystem.
+ *
+ * @param env The environment variables.
+ */
+ public static void deleteApplicationFiles(final Map<String, String> env) {
+ final String applicationFilesDir = env.get(YarnConfigKeys.FLINK_YARN_FILES);
+ if (!StringUtils.isNullOrWhitespaceOnly(applicationFilesDir)) {
+ final org.apache.flink.core.fs.Path path = new org.apache.flink.core.fs.Path(applicationFilesDir);
+ try {
+ final org.apache.flink.core.fs.FileSystem fileSystem = path.getFileSystem();
+ if (!fileSystem.delete(path, true)) {
+ LOG.error("Deleting yarn application files under {} was unsuccessful.", applicationFilesDir);
+ }
+ } catch (final IOException e) {
+ LOG.error("Could not properly delete yarn application files directory {}.", applicationFilesDir, e);
+ }
+ } else {
+ LOG.debug("No yarn application files directory set. Therefore, cannot clean up the data.");
+ }
+ }
+
+ /**
* Creates a YARN resource for the remote object at the given location.
*
* @param remoteRsrcPath remote location of the resource
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
index 3edd6e4..0d7546e 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
@@ -24,6 +24,7 @@ import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.clusterframework.ApplicationStatus;
import org.apache.flink.runtime.clusterframework.messages.GetClusterStatus;
import org.apache.flink.runtime.clusterframework.messages.GetClusterStatusResponse;
import org.apache.flink.runtime.clusterframework.messages.InfoMessage;
@@ -263,6 +264,20 @@ public class YarnClusterClient extends ClusterClient<ApplicationId> {
}
}
+ @Override
+ public void shutDownCluster() {
+ LOG.info("Sending shutdown request to the Application Master");
+ try {
+ final Future<Object> response = Patterns.ask(applicationClient.get(),
+ new YarnMessages.LocalStopYarnSession(ApplicationStatus.SUCCEEDED,
+ "Flink YARN Client requested shutdown"),
+ new Timeout(akkaDuration));
+ Await.ready(response, akkaDuration);
+ } catch (final Exception e) {
+ LOG.warn("Error while stopping YARN cluster.", e);
+ }
+ }
+
public ApplicationId getApplicationId() {
return appId;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
index 220f4f2..22d0164 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
@@ -279,6 +279,8 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
} catch (Throwable t) {
log.error("Could not unregister the application master.", t);
}
+
+ Utils.deleteApplicationFiles(env);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
index 7596d68..698d48d 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
@@ -611,7 +611,7 @@ public class FlinkYarnSessionCli extends AbstractCustomCommandLine<ApplicationId
}
try {
- yarnClusterDescriptor.terminateCluster(yarnApplicationId);
+ yarnClusterDescriptor.killCluster(yarnApplicationId);
} catch (FlinkException fe) {
LOG.info("Could not properly terminate the Flink cluster.", fe);
}
@@ -644,18 +644,14 @@ public class FlinkYarnSessionCli extends AbstractCustomCommandLine<ApplicationId
LOG.info("Could not properly close the Yarn application status monitor.", e);
}
+ clusterClient.shutDownCluster();
+
try {
clusterClient.shutdown();
} catch (Exception e) {
LOG.info("Could not properly shutdown cluster client.", e);
}
- try {
- yarnClusterDescriptor.terminateCluster(yarnApplicationId);
- } catch (FlinkException e) {
- LOG.info("Could not properly terminate the Flink cluster.", e);
- }
-
// shut down the scheduled executor service
ExecutorUtils.gracefulShutdown(
1000L,
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala b/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala
index 6b439bd..7d17325 100644
--- a/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala
+++ b/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala
@@ -109,8 +109,6 @@ class YarnJobManager(
private def handleYarnShutdown: Receive = {
case msg: StopCluster =>
- super.handleMessage(msg)
-
// do global cleanup if the yarn files path has been set
yarnFilesPath match {
case Some(filePath) =>
@@ -135,5 +133,7 @@ class YarnJobManager(
log.debug("No yarn application files directory set. Therefore, cannot clean up " +
"the data.")
}
+
+ super.handleMessage(msg)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-yarn/src/test/java/org/apache/flink/yarn/UtilsTest.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/UtilsTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/UtilsTest.java
new file mode 100644
index 0000000..5fc3567
--- /dev/null
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/UtilsTest.java
@@ -0,0 +1,52 @@
+/*
+ * 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.flink.yarn;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for {@link Utils}.
+ */
+public class UtilsTest {
+
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Test
+ public void testDeleteApplicationFiles() throws Exception {
+ final Path applicationFilesDir = temporaryFolder.newFolder(".flink").toPath();
+ Files.createFile(applicationFilesDir.resolve("flink.jar"));
+ assertThat(Files.list(temporaryFolder.getRoot().toPath()).count(), equalTo(1L));
+ assertThat(Files.list(applicationFilesDir).count(), equalTo(1L));
+
+ Utils.deleteApplicationFiles(Collections.singletonMap(
+ YarnConfigKeys.FLINK_YARN_FILES,
+ applicationFilesDir.toString()));
+ assertThat(Files.list(temporaryFolder.getRoot().toPath()).count(), equalTo(0L));
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/37df3c87/flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java
index 0d37b8e..03f3ef2 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ResourceManagerOptions;
+import org.apache.flink.runtime.clusterframework.ApplicationStatus;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
@@ -76,6 +77,7 @@ import org.junit.rules.TemporaryFolder;
import javax.annotation.Nullable;
import java.io.File;
+import java.nio.file.Files;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -90,7 +92,9 @@ import static org.apache.flink.yarn.YarnConfigKeys.ENV_CLIENT_SHIP_FILES;
import static org.apache.flink.yarn.YarnConfigKeys.ENV_FLINK_CLASSPATH;
import static org.apache.flink.yarn.YarnConfigKeys.ENV_HADOOP_USER_NAME;
import static org.apache.flink.yarn.YarnConfigKeys.FLINK_JAR_PATH;
+import static org.apache.flink.yarn.YarnConfigKeys.FLINK_YARN_FILES;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.eq;
@@ -105,9 +109,9 @@ public class YarnResourceManagerTest extends TestLogger {
private static final Time TIMEOUT = Time.seconds(10L);
- private static Configuration flinkConfig = new Configuration();
+ private Configuration flinkConfig = new Configuration();
- private static Map<String, String> env = new HashMap<>();
+ private Map<String, String> env = new HashMap<>();
@Rule
public TemporaryFolder folder = new TemporaryFolder();
@@ -200,7 +204,7 @@ public class YarnResourceManagerTest extends TestLogger {
}
}
- static class Context {
+ class Context {
// services
final TestingRpcService rpcService;
@@ -388,4 +392,20 @@ public class YarnResourceManagerTest extends TestLogger {
assertTrue(resourceManager.getNumberOfRegisteredTaskManagers().get() == 0);
}};
}
+
+ /**
+ * Tests that application files are deleted when the YARN application master is de-registered.
+ */
+ @Test
+ public void testDeleteApplicationFiles() throws Exception {
+ new Context() {{
+ final File applicationDir = folder.newFolder(".flink");
+ env.put(FLINK_YARN_FILES, applicationDir.getCanonicalPath());
+
+ startResourceManager();
+
+ resourceManager.deregisterApplication(ApplicationStatus.SUCCEEDED, null);
+ assertFalse("YARN application directory was not removed", Files.exists(applicationDir.toPath()));
+ }};
+ }
}
[09/15] flink git commit: [hotfix] [tests] Rename UtilsTest to
YarnFlinkResourceManagerTest.
Posted by se...@apache.org.
[hotfix] [tests] Rename UtilsTest to YarnFlinkResourceManagerTest.
Test was misnamed.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9c49dfc7
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9c49dfc7
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9c49dfc7
Branch: refs/heads/master
Commit: 9c49dfc7f13a95425cefd994f0e74882888c4fb4
Parents: 02a9d3d
Author: gyao <ga...@data-artisans.com>
Authored: Thu Apr 19 11:18:32 2018 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Apr 30 21:21:04 2018 +0200
----------------------------------------------------------------------
.../java/org/apache/flink/yarn/UtilsTest.java | 244 -------------------
.../yarn/YarnFlinkResourceManagerTest.java | 244 +++++++++++++++++++
2 files changed, 244 insertions(+), 244 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/9c49dfc7/flink-yarn/src/test/java/org/apache/flink/yarn/UtilsTest.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/UtilsTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/UtilsTest.java
deleted file mode 100644
index b7a38b0..0000000
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/UtilsTest.java
+++ /dev/null
@@ -1,244 +0,0 @@
-/*
- * 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.flink.yarn;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.akka.AkkaUtils;
-import org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters;
-import org.apache.flink.runtime.clusterframework.messages.NotifyResourceStarted;
-import org.apache.flink.runtime.clusterframework.messages.RegisterResourceManager;
-import org.apache.flink.runtime.clusterframework.messages.RegisterResourceManagerSuccessful;
-import org.apache.flink.runtime.instance.AkkaActorGateway;
-import org.apache.flink.runtime.leaderretrieval.SettableLeaderRetrievalService;
-import org.apache.flink.runtime.messages.Acknowledge;
-import org.apache.flink.runtime.testingUtils.TestingUtils;
-import org.apache.flink.util.TestLogger;
-import org.apache.flink.yarn.messages.NotifyWhenResourcesRegistered;
-import org.apache.flink.yarn.messages.RequestNumberOfRegisteredResources;
-
-import akka.actor.ActorRef;
-import akka.actor.ActorSystem;
-import akka.actor.PoisonPill;
-import akka.actor.Props;
-import akka.testkit.JavaTestKit;
-import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
-import org.apache.hadoop.yarn.api.records.ApplicationId;
-import org.apache.hadoop.yarn.api.records.Container;
-import org.apache.hadoop.yarn.api.records.ContainerId;
-import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
-import org.apache.hadoop.yarn.api.records.NodeId;
-import org.apache.hadoop.yarn.client.api.AMRMClient;
-import org.apache.hadoop.yarn.client.api.NMClient;
-import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
-import org.apache.hadoop.yarn.conf.YarnConfiguration;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.mockito.Matchers;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
-
-import scala.Option;
-import scala.concurrent.Await;
-import scala.concurrent.Future;
-import scala.concurrent.duration.Deadline;
-import scala.concurrent.duration.FiniteDuration;
-
-import static org.junit.Assert.assertEquals;
-import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-/**
- * Tests for {@link Utils}.
- */
-public class UtilsTest extends TestLogger {
-
- private static ActorSystem system;
-
- @BeforeClass
- public static void setup() {
- system = AkkaUtils.createLocalActorSystem(new Configuration());
- }
-
- @AfterClass
- public static void teardown() {
- JavaTestKit.shutdownActorSystem(system);
- }
-
- @Test
- public void testYarnFlinkResourceManagerJobManagerLostLeadership() throws Exception {
- new JavaTestKit(system) {{
-
- final Deadline deadline = new FiniteDuration(3, TimeUnit.MINUTES).fromNow();
-
- Configuration flinkConfig = new Configuration();
- YarnConfiguration yarnConfig = new YarnConfiguration();
- SettableLeaderRetrievalService leaderRetrievalService = new SettableLeaderRetrievalService(
- null,
- null);
- String applicationMasterHostName = "localhost";
- String webInterfaceURL = "foobar";
- ContaineredTaskManagerParameters taskManagerParameters = new ContaineredTaskManagerParameters(
- 1L, 1L, 1L, 1, new HashMap<String, String>());
- ContainerLaunchContext taskManagerLaunchContext = mock(ContainerLaunchContext.class);
- int yarnHeartbeatIntervalMillis = 1000;
- int maxFailedContainers = 10;
- int numInitialTaskManagers = 5;
- final YarnResourceManagerCallbackHandler callbackHandler = new YarnResourceManagerCallbackHandler();
- AMRMClientAsync<AMRMClient.ContainerRequest> resourceManagerClient = mock(AMRMClientAsync.class);
- NMClient nodeManagerClient = mock(NMClient.class);
- UUID leaderSessionID = UUID.randomUUID();
-
- final List<Container> containerList = new ArrayList<>();
-
- for (int i = 0; i < numInitialTaskManagers; i++) {
- Container mockContainer = mock(Container.class);
- when(mockContainer.getId()).thenReturn(
- ContainerId.newInstance(
- ApplicationAttemptId.newInstance(
- ApplicationId.newInstance(System.currentTimeMillis(), 1),
- 1),
- i));
- when(mockContainer.getNodeId()).thenReturn(NodeId.newInstance("container", 1234));
- containerList.add(mockContainer);
- }
-
- doAnswer(new Answer() {
- int counter = 0;
- @Override
- public Object answer(InvocationOnMock invocation) throws Throwable {
- if (counter < containerList.size()) {
- callbackHandler.onContainersAllocated(
- Collections.singletonList(
- containerList.get(counter++)
- ));
- }
- return null;
- }
- }).when(resourceManagerClient).addContainerRequest(Matchers.any(AMRMClient.ContainerRequest.class));
-
- final CompletableFuture<AkkaActorGateway> resourceManagerFuture = new CompletableFuture<>();
- final CompletableFuture<AkkaActorGateway> leaderGatewayFuture = new CompletableFuture<>();
-
- doAnswer(
- (InvocationOnMock invocation) -> {
- Container container = (Container) invocation.getArguments()[0];
- resourceManagerFuture.thenCombine(leaderGatewayFuture,
- (resourceManagerGateway, leaderGateway) -> {
- resourceManagerGateway.tell(
- new NotifyResourceStarted(YarnFlinkResourceManager.extractResourceID(container)),
- leaderGateway);
- return null;
- });
- return null;
- })
- .when(nodeManagerClient)
- .startContainer(
- Matchers.any(Container.class),
- Matchers.any(ContainerLaunchContext.class));
-
- ActorRef resourceManager = null;
- ActorRef leader1;
-
- try {
- leader1 = system.actorOf(
- Props.create(
- TestingUtils.ForwardingActor.class,
- getRef(),
- Option.apply(leaderSessionID)
- ));
-
- resourceManager = system.actorOf(
- Props.create(
- TestingYarnFlinkResourceManager.class,
- flinkConfig,
- yarnConfig,
- leaderRetrievalService,
- applicationMasterHostName,
- webInterfaceURL,
- taskManagerParameters,
- taskManagerLaunchContext,
- yarnHeartbeatIntervalMillis,
- maxFailedContainers,
- numInitialTaskManagers,
- callbackHandler,
- resourceManagerClient,
- nodeManagerClient
- ));
-
- leaderRetrievalService.notifyListener(leader1.path().toString(), leaderSessionID);
-
- final AkkaActorGateway leader1Gateway = new AkkaActorGateway(leader1, leaderSessionID);
- final AkkaActorGateway resourceManagerGateway = new AkkaActorGateway(resourceManager, leaderSessionID);
-
- leaderGatewayFuture.complete(leader1Gateway);
- resourceManagerFuture.complete(resourceManagerGateway);
-
- expectMsgClass(deadline.timeLeft(), RegisterResourceManager.class);
-
- resourceManagerGateway.tell(new RegisterResourceManagerSuccessful(leader1, Collections.emptyList()));
-
- for (int i = 0; i < containerList.size(); i++) {
- expectMsgClass(deadline.timeLeft(), Acknowledge.class);
- }
-
- Future<Object> taskManagerRegisteredFuture = resourceManagerGateway.ask(new NotifyWhenResourcesRegistered(numInitialTaskManagers), deadline.timeLeft());
-
- Await.ready(taskManagerRegisteredFuture, deadline.timeLeft());
-
- leaderRetrievalService.notifyListener(null, null);
-
- leaderRetrievalService.notifyListener(leader1.path().toString(), leaderSessionID);
-
- expectMsgClass(deadline.timeLeft(), RegisterResourceManager.class);
-
- resourceManagerGateway.tell(new RegisterResourceManagerSuccessful(leader1, Collections.emptyList()));
-
- for (Container container: containerList) {
- resourceManagerGateway.tell(
- new NotifyResourceStarted(YarnFlinkResourceManager.extractResourceID(container)),
- leader1Gateway);
- }
-
- for (int i = 0; i < containerList.size(); i++) {
- expectMsgClass(deadline.timeLeft(), Acknowledge.class);
- }
-
- Future<Object> numberOfRegisteredResourcesFuture = resourceManagerGateway.ask(RequestNumberOfRegisteredResources.INSTANCE, deadline.timeLeft());
-
- int numberOfRegisteredResources = (Integer) Await.result(numberOfRegisteredResourcesFuture, deadline.timeLeft());
-
- assertEquals(numInitialTaskManagers, numberOfRegisteredResources);
- } finally {
- if (resourceManager != null) {
- resourceManager.tell(PoisonPill.getInstance(), ActorRef.noSender());
- }
- }
- }};
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/9c49dfc7/flink-yarn/src/test/java/org/apache/flink/yarn/YarnFlinkResourceManagerTest.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnFlinkResourceManagerTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnFlinkResourceManagerTest.java
new file mode 100644
index 0000000..10b2ce9
--- /dev/null
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnFlinkResourceManagerTest.java
@@ -0,0 +1,244 @@
+/*
+ * 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.flink.yarn;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters;
+import org.apache.flink.runtime.clusterframework.messages.NotifyResourceStarted;
+import org.apache.flink.runtime.clusterframework.messages.RegisterResourceManager;
+import org.apache.flink.runtime.clusterframework.messages.RegisterResourceManagerSuccessful;
+import org.apache.flink.runtime.instance.AkkaActorGateway;
+import org.apache.flink.runtime.leaderretrieval.SettableLeaderRetrievalService;
+import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.util.TestLogger;
+import org.apache.flink.yarn.messages.NotifyWhenResourcesRegistered;
+import org.apache.flink.yarn.messages.RequestNumberOfRegisteredResources;
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.PoisonPill;
+import akka.actor.Props;
+import akka.testkit.JavaTestKit;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
+import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.client.api.AMRMClient;
+import org.apache.hadoop.yarn.client.api.NMClient;
+import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Matchers;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import scala.Option;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Deadline;
+import scala.concurrent.duration.FiniteDuration;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link YarnFlinkResourceManager}.
+ */
+public class YarnFlinkResourceManagerTest extends TestLogger {
+
+ private static ActorSystem system;
+
+ @BeforeClass
+ public static void setup() {
+ system = AkkaUtils.createLocalActorSystem(new Configuration());
+ }
+
+ @AfterClass
+ public static void teardown() {
+ JavaTestKit.shutdownActorSystem(system);
+ }
+
+ @Test
+ public void testYarnFlinkResourceManagerJobManagerLostLeadership() throws Exception {
+ new JavaTestKit(system) {{
+
+ final Deadline deadline = new FiniteDuration(3, TimeUnit.MINUTES).fromNow();
+
+ Configuration flinkConfig = new Configuration();
+ YarnConfiguration yarnConfig = new YarnConfiguration();
+ SettableLeaderRetrievalService leaderRetrievalService = new SettableLeaderRetrievalService(
+ null,
+ null);
+ String applicationMasterHostName = "localhost";
+ String webInterfaceURL = "foobar";
+ ContaineredTaskManagerParameters taskManagerParameters = new ContaineredTaskManagerParameters(
+ 1L, 1L, 1L, 1, new HashMap<String, String>());
+ ContainerLaunchContext taskManagerLaunchContext = mock(ContainerLaunchContext.class);
+ int yarnHeartbeatIntervalMillis = 1000;
+ int maxFailedContainers = 10;
+ int numInitialTaskManagers = 5;
+ final YarnResourceManagerCallbackHandler callbackHandler = new YarnResourceManagerCallbackHandler();
+ AMRMClientAsync<AMRMClient.ContainerRequest> resourceManagerClient = mock(AMRMClientAsync.class);
+ NMClient nodeManagerClient = mock(NMClient.class);
+ UUID leaderSessionID = UUID.randomUUID();
+
+ final List<Container> containerList = new ArrayList<>();
+
+ for (int i = 0; i < numInitialTaskManagers; i++) {
+ Container mockContainer = mock(Container.class);
+ when(mockContainer.getId()).thenReturn(
+ ContainerId.newInstance(
+ ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(System.currentTimeMillis(), 1),
+ 1),
+ i));
+ when(mockContainer.getNodeId()).thenReturn(NodeId.newInstance("container", 1234));
+ containerList.add(mockContainer);
+ }
+
+ doAnswer(new Answer() {
+ int counter = 0;
+ @Override
+ public Object answer(InvocationOnMock invocation) throws Throwable {
+ if (counter < containerList.size()) {
+ callbackHandler.onContainersAllocated(
+ Collections.singletonList(
+ containerList.get(counter++)
+ ));
+ }
+ return null;
+ }
+ }).when(resourceManagerClient).addContainerRequest(Matchers.any(AMRMClient.ContainerRequest.class));
+
+ final CompletableFuture<AkkaActorGateway> resourceManagerFuture = new CompletableFuture<>();
+ final CompletableFuture<AkkaActorGateway> leaderGatewayFuture = new CompletableFuture<>();
+
+ doAnswer(
+ (InvocationOnMock invocation) -> {
+ Container container = (Container) invocation.getArguments()[0];
+ resourceManagerFuture.thenCombine(leaderGatewayFuture,
+ (resourceManagerGateway, leaderGateway) -> {
+ resourceManagerGateway.tell(
+ new NotifyResourceStarted(YarnFlinkResourceManager.extractResourceID(container)),
+ leaderGateway);
+ return null;
+ });
+ return null;
+ })
+ .when(nodeManagerClient)
+ .startContainer(
+ Matchers.any(Container.class),
+ Matchers.any(ContainerLaunchContext.class));
+
+ ActorRef resourceManager = null;
+ ActorRef leader1;
+
+ try {
+ leader1 = system.actorOf(
+ Props.create(
+ TestingUtils.ForwardingActor.class,
+ getRef(),
+ Option.apply(leaderSessionID)
+ ));
+
+ resourceManager = system.actorOf(
+ Props.create(
+ TestingYarnFlinkResourceManager.class,
+ flinkConfig,
+ yarnConfig,
+ leaderRetrievalService,
+ applicationMasterHostName,
+ webInterfaceURL,
+ taskManagerParameters,
+ taskManagerLaunchContext,
+ yarnHeartbeatIntervalMillis,
+ maxFailedContainers,
+ numInitialTaskManagers,
+ callbackHandler,
+ resourceManagerClient,
+ nodeManagerClient
+ ));
+
+ leaderRetrievalService.notifyListener(leader1.path().toString(), leaderSessionID);
+
+ final AkkaActorGateway leader1Gateway = new AkkaActorGateway(leader1, leaderSessionID);
+ final AkkaActorGateway resourceManagerGateway = new AkkaActorGateway(resourceManager, leaderSessionID);
+
+ leaderGatewayFuture.complete(leader1Gateway);
+ resourceManagerFuture.complete(resourceManagerGateway);
+
+ expectMsgClass(deadline.timeLeft(), RegisterResourceManager.class);
+
+ resourceManagerGateway.tell(new RegisterResourceManagerSuccessful(leader1, Collections.emptyList()));
+
+ for (int i = 0; i < containerList.size(); i++) {
+ expectMsgClass(deadline.timeLeft(), Acknowledge.class);
+ }
+
+ Future<Object> taskManagerRegisteredFuture = resourceManagerGateway.ask(new NotifyWhenResourcesRegistered(numInitialTaskManagers), deadline.timeLeft());
+
+ Await.ready(taskManagerRegisteredFuture, deadline.timeLeft());
+
+ leaderRetrievalService.notifyListener(null, null);
+
+ leaderRetrievalService.notifyListener(leader1.path().toString(), leaderSessionID);
+
+ expectMsgClass(deadline.timeLeft(), RegisterResourceManager.class);
+
+ resourceManagerGateway.tell(new RegisterResourceManagerSuccessful(leader1, Collections.emptyList()));
+
+ for (Container container: containerList) {
+ resourceManagerGateway.tell(
+ new NotifyResourceStarted(YarnFlinkResourceManager.extractResourceID(container)),
+ leader1Gateway);
+ }
+
+ for (int i = 0; i < containerList.size(); i++) {
+ expectMsgClass(deadline.timeLeft(), Acknowledge.class);
+ }
+
+ Future<Object> numberOfRegisteredResourcesFuture = resourceManagerGateway.ask(RequestNumberOfRegisteredResources.INSTANCE, deadline.timeLeft());
+
+ int numberOfRegisteredResources = (Integer) Await.result(numberOfRegisteredResourcesFuture, deadline.timeLeft());
+
+ assertEquals(numInitialTaskManagers, numberOfRegisteredResources);
+ } finally {
+ if (resourceManager != null) {
+ resourceManager.tell(PoisonPill.getInstance(), ActorRef.noSender());
+ }
+ }
+ }};
+ }
+}
[05/15] flink git commit: [hotfix] Indent method parameters.
Posted by se...@apache.org.
[hotfix] Indent method parameters.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b209e6ef
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b209e6ef
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b209e6ef
Branch: refs/heads/master
Commit: b209e6ef594f1147a0a2e001aca5b13c010c927f
Parents: a3e3bed
Author: gyao <ga...@data-artisans.com>
Authored: Thu Apr 19 11:03:20 2018 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Apr 30 21:20:56 2018 +0200
----------------------------------------------------------------------
.../src/main/java/org/apache/flink/yarn/YarnResourceManager.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b209e6ef/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
index 4eb4fc9..0ea66b8 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
@@ -267,8 +267,8 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
@Override
protected void internalDeregisterApplication(
- ApplicationStatus finalStatus,
- @Nullable String diagnostics) {
+ ApplicationStatus finalStatus,
+ @Nullable String diagnostics) {
// first, de-register from YARN
FinalApplicationStatus yarnStatus = getYarnStatus(finalStatus);
[02/15] flink git commit: [hotfix] Update README.md building
prerequisites
Posted by se...@apache.org.
[hotfix] Update README.md building prerequisites
This closes #5924
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ad7a4ac9
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ad7a4ac9
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ad7a4ac9
Branch: refs/heads/master
Commit: ad7a4ac9d82f6d3a96191936a3ac1557a78458d6
Parents: bd8cb86
Author: 陈梓立 <wa...@gmail.com>
Authored: Fri Apr 27 09:07:33 2018 +0800
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Apr 30 21:07:55 2018 +0200
----------------------------------------------------------------------
README.md | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ad7a4ac9/README.md
----------------------------------------------------------------------
diff --git a/README.md b/README.md
index 095f652a..7966ecd 100644
--- a/README.md
+++ b/README.md
@@ -67,10 +67,10 @@ counts.writeAsCsv(outputPath)
Prerequisites for building Flink:
-* Unix-like environment (We use Linux, Mac OS X, Cygwin)
+* Unix-like environment (we use Linux, Mac OS X, Cygwin)
* git
* Maven (we recommend version 3.2.5)
-* Java 8
+* Java 8 (Java 9 and 10 are not yet supported)
```
git clone https://github.com/apache/flink.git
[12/15] flink git commit: [FLINK-9275] [streaming] Add taskName to
the output flusher thread's name
Posted by se...@apache.org.
[FLINK-9275] [streaming] Add taskName to the output flusher thread's name
This closes #5943
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a8f4515d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a8f4515d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a8f4515d
Branch: refs/heads/master
Commit: a8f4515d9b62a95cf6b10d59c95ce36f6198ee28
Parents: ff0622b
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Mon Apr 30 15:02:43 2018 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Apr 30 21:24:32 2018 +0200
----------------------------------------------------------------------
.../org/apache/flink/streaming/runtime/io/StreamRecordWriter.java | 3 ++-
.../java/org/apache/flink/streaming/runtime/tasks/StreamTask.java | 2 +-
2 files changed, 3 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a8f4515d/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamRecordWriter.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamRecordWriter.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamRecordWriter.java
index dad680c..9fedf70 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamRecordWriter.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamRecordWriter.java
@@ -67,7 +67,8 @@ public class StreamRecordWriter<T extends IOReadableWritable> extends RecordWrit
}
else {
String threadName = taskName == null ?
- DEFAULT_OUTPUT_FLUSH_THREAD_NAME : "Output Timeout Flusher - " + taskName;
+ DEFAULT_OUTPUT_FLUSH_THREAD_NAME :
+ DEFAULT_OUTPUT_FLUSH_THREAD_NAME + " for " + taskName;
outputFlusher = new OutputFlusher(threadName, timeout);
outputFlusher.start();
http://git-wip-us.apache.org/repos/asf/flink/blob/a8f4515d/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index 6812871..6790949 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -1204,7 +1204,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
}
StreamRecordWriter<SerializationDelegate<StreamRecord<OUT>>> output =
- new StreamRecordWriter<>(bufferWriter, outputPartitioner, bufferTimeout);
+ new StreamRecordWriter<>(bufferWriter, outputPartitioner, bufferTimeout, taskName);
output.setMetricGroup(environment.getMetricGroup().getIOMetricGroup());
return output;
}
[03/15] flink git commit: [hotfix] Replace String concatenation with
Slf4j placeholders.
Posted by se...@apache.org.
[hotfix] Replace String concatenation with Slf4j placeholders.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a3e3bedf
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a3e3bedf
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a3e3bedf
Branch: refs/heads/master
Commit: a3e3bedfef879e5e8659e78ae06cbca22d53beba
Parents: ad7a4ac
Author: gyao <ga...@data-artisans.com>
Authored: Thu Apr 19 10:29:43 2018 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Apr 30 21:20:55 2018 +0200
----------------------------------------------------------------------
flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a3e3bedf/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java b/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
index 2ca4cb6..fb9a478 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
@@ -156,7 +156,7 @@ public final class Utils {
Path dst = new Path(homedir, suffix);
- LOG.debug("Copying from " + localSrcPath + " to " + dst);
+ LOG.debug("Copying from {} to {}", localSrcPath, dst);
fs.copyFromLocalFile(false, true, localSrcPath, dst);
[06/15] flink git commit: [hotfix] Fix raw types warning.
Posted by se...@apache.org.
[hotfix] Fix raw types warning.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4fffd607
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4fffd607
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4fffd607
Branch: refs/heads/master
Commit: 4fffd60769bf1fa8501b4f371ba0117bddaec11f
Parents: 290b11b
Author: gyao <ga...@data-artisans.com>
Authored: Thu Apr 19 11:05:05 2018 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Apr 30 21:20:56 2018 +0200
----------------------------------------------------------------------
.../src/main/java/org/apache/flink/yarn/YarnResourceManager.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4fffd607/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
index a25e4c4..220f4f2 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
@@ -435,7 +435,7 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
String[] hostPort = address.split("@")[1].split(":");
String host = hostPort[0];
String port = hostPort[1].split("/")[0];
- return new Tuple2(host, Integer.valueOf(port));
+ return new Tuple2<>(host, Integer.valueOf(port));
}
private void requestYarnContainer(Resource resource, Priority priority) {
[15/15] flink git commit: [FLINK-9256] [network] Fix NPE in
SingleInputGate#updateInputChannel() for non-credit based flow control
Posted by se...@apache.org.
[FLINK-9256] [network] Fix NPE in SingleInputGate#updateInputChannel() for non-credit based flow control
This closes #5914
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ecde6c32
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ecde6c32
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ecde6c32
Branch: refs/heads/master
Commit: ecde6c328b38d6f6efea4b0d62f4ec8fe0040240
Parents: b957bf2
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Wed Apr 25 18:28:48 2018 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Apr 30 21:35:06 2018 +0200
----------------------------------------------------------------------
.../runtime/io/network/NetworkEnvironment.java | 4 +
.../partition/consumer/SingleInputGate.java | 20 +-
.../io/network/NetworkEnvironmentTest.java | 5 +-
.../PartitionRequestClientHandlerTest.java | 3 +-
.../partition/InputGateConcurrentTest.java | 9 +-
.../partition/InputGateFairnessTest.java | 17 +-
.../consumer/LocalInputChannelTest.java | 6 +-
.../consumer/RemoteInputChannelTest.java | 3 +-
.../partition/consumer/SingleInputGateTest.java | 318 +++++++++++++++----
.../partition/consumer/TestSingleInputGate.java | 3 +-
.../partition/consumer/UnionInputGateTest.java | 6 +-
11 files changed, 301 insertions(+), 93 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ecde6c32/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
index 0a9dc0f..f254756 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
@@ -157,6 +157,10 @@ public class NetworkEnvironment {
return partitionRequestMaxBackoff;
}
+ public boolean isCreditBased() {
+ return enableCreditBased;
+ }
+
public KvStateRegistry getKvStateRegistry() {
return kvStateRegistry;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ecde6c32/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
index b9091b2..06e80ff 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
@@ -157,9 +157,11 @@ public class SingleInputGate implements InputGate {
*/
private BufferPool bufferPool;
- /** Global network buffer pool to request and recycle exclusive buffers. */
+ /** Global network buffer pool to request and recycle exclusive buffers (only for credit-based). */
private NetworkBufferPool networkBufferPool;
+ private final boolean isCreditBased;
+
private boolean hasReceivedAllEndOfPartitionEvents;
/** Flag indicating whether partitions have been requested. */
@@ -189,7 +191,8 @@ public class SingleInputGate implements InputGate {
int consumedSubpartitionIndex,
int numberOfInputChannels,
TaskActions taskActions,
- TaskIOMetricGroup metrics) {
+ TaskIOMetricGroup metrics,
+ boolean isCreditBased) {
this.owningTaskName = checkNotNull(owningTaskName);
this.jobId = checkNotNull(jobId);
@@ -208,6 +211,7 @@ public class SingleInputGate implements InputGate {
this.enqueuedInputChannelsWithData = new BitSet(numberOfInputChannels);
this.taskActions = checkNotNull(taskActions);
+ this.isCreditBased = isCreditBased;
}
// ------------------------------------------------------------------------
@@ -288,6 +292,7 @@ public class SingleInputGate implements InputGate {
* @param networkBuffersPerChannel The number of exclusive buffers for each channel
*/
public void assignExclusiveSegments(NetworkBufferPool networkBufferPool, int networkBuffersPerChannel) throws IOException {
+ checkState(this.isCreditBased, "Bug in input gate setup logic: exclusive buffers only exist with credit-based flow control.");
checkState(this.networkBufferPool == null, "Bug in input gate setup logic: global buffer pool has" +
"already been set for this input gate.");
@@ -347,8 +352,13 @@ public class SingleInputGate implements InputGate {
}
else if (partitionLocation.isRemote()) {
newChannel = unknownChannel.toRemoteInputChannel(partitionLocation.getConnectionId());
- ((RemoteInputChannel)newChannel).assignExclusiveSegments(
- networkBufferPool.requestMemorySegments(networkBuffersPerChannel));
+
+ if (this.isCreditBased) {
+ checkState(this.networkBufferPool != null, "Bug in input gate setup logic: " +
+ "global buffer pool has not been set for this input gate.");
+ ((RemoteInputChannel) newChannel).assignExclusiveSegments(
+ networkBufferPool.requestMemorySegments(networkBuffersPerChannel));
+ }
}
else {
throw new IllegalStateException("Tried to update unknown channel with unknown channel.");
@@ -661,7 +671,7 @@ public class SingleInputGate implements InputGate {
final SingleInputGate inputGate = new SingleInputGate(
owningTaskName, jobId, consumedResultId, consumedPartitionType, consumedSubpartitionIndex,
- icdd.length, taskActions, metrics);
+ icdd.length, taskActions, metrics, networkEnvironment.isCreditBased());
// Create the input channels. There is one input channel for each consumed partition.
final InputChannel[] inputChannels = new InputChannel[icdd.length];
http://git-wip-us.apache.org/repos/asf/flink/blob/ecde6c32/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
index 317a214..f790b5f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
@@ -329,7 +329,7 @@ public class NetworkEnvironmentTest {
*
* @return input gate with some fake settings
*/
- private static SingleInputGate createSingleInputGate(
+ private SingleInputGate createSingleInputGate(
final ResultPartitionType partitionType, final int channels) {
return spy(new SingleInputGate(
"Test Task Name",
@@ -339,7 +339,8 @@ public class NetworkEnvironmentTest {
0,
channels,
mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup()));
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup(),
+ enableCreditBasedFlowControl));
}
private static void createRemoteInputChannel(
http://git-wip-us.apache.org/repos/asf/flink/blob/ecde6c32/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
index 13f7510..842aed8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
@@ -221,7 +221,8 @@ public class PartitionRequestClientHandlerTest {
0,
1,
mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup(),
+ true);
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/ecde6c32/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateConcurrentTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateConcurrentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateConcurrentTest.java
index 289a398..73f3cfb 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateConcurrentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateConcurrentTest.java
@@ -66,7 +66,8 @@ public class InputGateConcurrentTest {
new IntermediateDataSetID(), ResultPartitionType.PIPELINED,
0, numChannels,
mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup(),
+ true);
for (int i = 0; i < numChannels; i++) {
LocalInputChannel channel = new LocalInputChannel(gate, i, new ResultPartitionID(),
@@ -102,7 +103,8 @@ public class InputGateConcurrentTest {
0,
numChannels,
mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup(),
+ true);
for (int i = 0; i < numChannels; i++) {
RemoteInputChannel channel = new RemoteInputChannel(
@@ -151,7 +153,8 @@ public class InputGateConcurrentTest {
0,
numChannels,
mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup(),
+ true);
for (int i = 0, local = 0; i < numChannels; i++) {
if (localOrRemote.get(i)) {
http://git-wip-us.apache.org/repos/asf/flink/blob/ecde6c32/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateFairnessTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateFairnessTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateFairnessTest.java
index 45df56f..82a27cc 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateFairnessTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateFairnessTest.java
@@ -93,7 +93,8 @@ public class InputGateFairnessTest {
new IntermediateDataSetID(),
0, numChannels,
mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup(),
+ true);
for (int i = 0; i < numChannels; i++) {
LocalInputChannel channel = new LocalInputChannel(gate, i, new ResultPartitionID(),
@@ -146,7 +147,8 @@ public class InputGateFairnessTest {
new IntermediateDataSetID(),
0, numChannels,
mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup(),
+ true);
for (int i = 0; i < numChannels; i++) {
LocalInputChannel channel = new LocalInputChannel(gate, i, new ResultPartitionID(),
@@ -196,7 +198,8 @@ public class InputGateFairnessTest {
new IntermediateDataSetID(),
0, numChannels,
mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup(),
+ true);
final ConnectionManager connManager = createDummyConnectionManager();
@@ -251,7 +254,8 @@ public class InputGateFairnessTest {
new IntermediateDataSetID(),
0, numChannels,
mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup(),
+ true);
final ConnectionManager connManager = createDummyConnectionManager();
@@ -349,11 +353,12 @@ public class InputGateFairnessTest {
int consumedSubpartitionIndex,
int numberOfInputChannels,
TaskActions taskActions,
- TaskIOMetricGroup metrics) {
+ TaskIOMetricGroup metrics,
+ boolean isCreditBased) {
super(owningTaskName, jobId, consumedResultId, ResultPartitionType.PIPELINED,
consumedSubpartitionIndex,
- numberOfInputChannels, taskActions, metrics);
+ numberOfInputChannels, taskActions, metrics, isCreditBased);
try {
Field f = SingleInputGate.class.getDeclaredField("inputChannelsWithData");
http://git-wip-us.apache.org/repos/asf/flink/blob/ecde6c32/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java
index c78b7b9..1ecb67f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java
@@ -293,7 +293,8 @@ public class LocalInputChannelTest {
0,
1,
mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup()
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup(),
+ true
);
ResultPartitionManager partitionManager = mock(ResultPartitionManager.class);
@@ -490,7 +491,8 @@ public class LocalInputChannelTest {
subpartitionIndex,
numberOfInputChannels,
mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup(),
+ true);
// Set buffer pool
inputGate.setBufferPool(bufferPool);
http://git-wip-us.apache.org/repos/asf/flink/blob/ecde6c32/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
index 97a5688..802cb93 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
@@ -889,7 +889,8 @@ public class RemoteInputChannelTest {
0,
1,
mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup(),
+ true);
}
private RemoteInputChannel createRemoteInputChannel(SingleInputGate inputGate)
http://git-wip-us.apache.org/repos/asf/flink/blob/ecde6c32/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
index 8c54c1f..c244668 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
@@ -19,13 +19,13 @@
package org.apache.flink.runtime.io.network.partition.consumer;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.runtime.deployment.InputChannelDeploymentDescriptor;
import org.apache.flink.runtime.deployment.InputGateDeploymentDescriptor;
import org.apache.flink.runtime.deployment.ResultPartitionLocation;
import org.apache.flink.runtime.event.TaskEvent;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.network.ConnectionID;
import org.apache.flink.runtime.io.network.ConnectionManager;
import org.apache.flink.runtime.io.network.LocalConnectionManager;
@@ -45,31 +45,51 @@ import org.apache.flink.runtime.io.network.util.TestTaskEvent;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.query.KvStateRegistry;
import org.apache.flink.runtime.taskmanager.TaskActions;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.util.Arrays;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyInt;
-import static org.mockito.Matchers.anyListOf;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+/**
+ * Tests for {@link SingleInputGate}.
+ */
+@RunWith(Parameterized.class)
public class SingleInputGateTest {
+ @Parameterized.Parameter
+ public boolean enableCreditBasedFlowControl;
+
+ @Parameterized.Parameters(name = "Credit-based = {0}")
+ public static List<Boolean> parameters() {
+ return Arrays.asList(Boolean.TRUE, Boolean.FALSE);
+ }
+
/**
* Tests basic correctness of buffer-or-event interleaving and correct <code>null</code> return
* value after receiving all end-of-partition events.
@@ -324,12 +344,7 @@ public class SingleInputGateTest {
int initialBackoff = 137;
int maxBackoff = 1001;
- NetworkEnvironment netEnv = mock(NetworkEnvironment.class);
- when(netEnv.getResultPartitionManager()).thenReturn(new ResultPartitionManager());
- when(netEnv.getTaskEventDispatcher()).thenReturn(new TaskEventDispatcher());
- when(netEnv.getPartitionRequestInitialBackoff()).thenReturn(initialBackoff);
- when(netEnv.getPartitionRequestMaxBackoff()).thenReturn(maxBackoff);
- when(netEnv.getConnectionManager()).thenReturn(new LocalConnectionManager());
+ final NetworkEnvironment netEnv = createNetworkEnvironment(2, 8, initialBackoff, maxBackoff);
SingleInputGate gate = SingleInputGate.create(
"TestTask",
@@ -340,37 +355,43 @@ public class SingleInputGateTest {
mock(TaskActions.class),
UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
- assertEquals(gateDesc.getConsumedPartitionType(), gate.getConsumedPartitionType());
+ try {
+ assertEquals(gateDesc.getConsumedPartitionType(), gate.getConsumedPartitionType());
- Map<IntermediateResultPartitionID, InputChannel> channelMap = gate.getInputChannels();
+ Map<IntermediateResultPartitionID, InputChannel> channelMap = gate.getInputChannels();
- assertEquals(3, channelMap.size());
- InputChannel localChannel = channelMap.get(partitionIds[0].getPartitionId());
- assertEquals(LocalInputChannel.class, localChannel.getClass());
+ assertEquals(3, channelMap.size());
+ InputChannel localChannel = channelMap.get(partitionIds[0].getPartitionId());
+ assertEquals(LocalInputChannel.class, localChannel.getClass());
- InputChannel remoteChannel = channelMap.get(partitionIds[1].getPartitionId());
- assertEquals(RemoteInputChannel.class, remoteChannel.getClass());
+ InputChannel remoteChannel = channelMap.get(partitionIds[1].getPartitionId());
+ assertEquals(RemoteInputChannel.class, remoteChannel.getClass());
- InputChannel unknownChannel = channelMap.get(partitionIds[2].getPartitionId());
- assertEquals(UnknownInputChannel.class, unknownChannel.getClass());
+ InputChannel unknownChannel = channelMap.get(partitionIds[2].getPartitionId());
+ assertEquals(UnknownInputChannel.class, unknownChannel.getClass());
- InputChannel[] channels = new InputChannel[]{localChannel, remoteChannel, unknownChannel};
- for (InputChannel ch : channels) {
- assertEquals(0, ch.getCurrentBackoff());
+ InputChannel[] channels =
+ new InputChannel[] {localChannel, remoteChannel, unknownChannel};
+ for (InputChannel ch : channels) {
+ assertEquals(0, ch.getCurrentBackoff());
- assertTrue(ch.increaseBackoff());
- assertEquals(initialBackoff, ch.getCurrentBackoff());
+ assertTrue(ch.increaseBackoff());
+ assertEquals(initialBackoff, ch.getCurrentBackoff());
- assertTrue(ch.increaseBackoff());
- assertEquals(initialBackoff * 2, ch.getCurrentBackoff());
+ assertTrue(ch.increaseBackoff());
+ assertEquals(initialBackoff * 2, ch.getCurrentBackoff());
- assertTrue(ch.increaseBackoff());
- assertEquals(initialBackoff * 2 * 2, ch.getCurrentBackoff());
+ assertTrue(ch.increaseBackoff());
+ assertEquals(initialBackoff * 2 * 2, ch.getCurrentBackoff());
- assertTrue(ch.increaseBackoff());
- assertEquals(maxBackoff, ch.getCurrentBackoff());
+ assertTrue(ch.increaseBackoff());
+ assertEquals(maxBackoff, ch.getCurrentBackoff());
- assertFalse(ch.increaseBackoff());
+ assertFalse(ch.increaseBackoff());
+ }
+ } finally {
+ gate.releaseAllResources();
+ netEnv.shutdown();
}
}
@@ -379,26 +400,39 @@ public class SingleInputGateTest {
*/
@Test
public void testRequestBuffersWithRemoteInputChannel() throws Exception {
- final SingleInputGate inputGate = new SingleInputGate(
- "t1",
- new JobID(),
- new IntermediateDataSetID(),
- ResultPartitionType.PIPELINED_BOUNDED,
- 0,
- 1,
- mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
-
- RemoteInputChannel remote = mock(RemoteInputChannel.class);
- inputGate.setInputChannel(new IntermediateResultPartitionID(), remote);
-
- final int buffersPerChannel = 2;
- NetworkBufferPool network = mock(NetworkBufferPool.class);
- // Trigger requests of segments from global pool and assign buffers to remote input channel
- inputGate.assignExclusiveSegments(network, buffersPerChannel);
-
- verify(network, times(1)).requestMemorySegments(buffersPerChannel);
- verify(remote, times(1)).assignExclusiveSegments(anyListOf(MemorySegment.class));
+ final SingleInputGate inputGate = createInputGate(1, ResultPartitionType.PIPELINED_BOUNDED);
+ int buffersPerChannel = 2;
+ int extraNetworkBuffersPerGate = 8;
+ final NetworkEnvironment network = createNetworkEnvironment(buffersPerChannel,
+ extraNetworkBuffersPerGate, 0, 0);
+
+ try {
+ final ResultPartitionID resultPartitionId = new ResultPartitionID();
+ final ConnectionID connectionId = new ConnectionID(new InetSocketAddress("localhost", 5000), 0);
+ addRemoteInputChannel(network, inputGate, connectionId, resultPartitionId, 0);
+
+ network.setupInputGate(inputGate);
+
+ NetworkBufferPool bufferPool = network.getNetworkBufferPool();
+ if (enableCreditBasedFlowControl) {
+ verify(bufferPool,
+ times(1)).requestMemorySegments(buffersPerChannel);
+ RemoteInputChannel remote = (RemoteInputChannel) inputGate.getInputChannels()
+ .get(resultPartitionId.getPartitionId());
+ // only the exclusive buffers should be assigned/available now
+ assertEquals(buffersPerChannel, remote.getNumberOfAvailableBuffers());
+
+ assertEquals(bufferPool.getTotalNumberOfMemorySegments() - buffersPerChannel,
+ bufferPool.getNumberOfAvailableMemorySegments());
+ // note: exclusive buffers are not handed out into LocalBufferPool and are thus not counted
+ assertEquals(extraNetworkBuffersPerGate, bufferPool.countBuffers());
+ } else {
+ assertEquals(buffersPerChannel + extraNetworkBuffersPerGate, bufferPool.countBuffers());
+ }
+ } finally {
+ inputGate.releaseAllResources();
+ network.shutdown();
+ }
}
/**
@@ -407,51 +441,195 @@ public class SingleInputGateTest {
*/
@Test
public void testRequestBuffersWithUnknownInputChannel() throws Exception {
- final SingleInputGate inputGate = createInputGate(1);
+ final SingleInputGate inputGate = createInputGate(1, ResultPartitionType.PIPELINED_BOUNDED);
+ int buffersPerChannel = 2;
+ int extraNetworkBuffersPerGate = 8;
+ final NetworkEnvironment network = createNetworkEnvironment(buffersPerChannel, extraNetworkBuffersPerGate, 0, 0);
- UnknownInputChannel unknown = mock(UnknownInputChannel.class);
- final ResultPartitionID resultPartitionId = new ResultPartitionID();
- inputGate.setInputChannel(resultPartitionId.getPartitionId(), unknown);
+ try {
+ final ResultPartitionID resultPartitionId = new ResultPartitionID();
+ addUnknownInputChannel(network, inputGate, resultPartitionId, 0);
- RemoteInputChannel remote = mock(RemoteInputChannel.class);
- final ConnectionID connectionId = new ConnectionID(new InetSocketAddress("localhost", 5000), 0);
- when(unknown.toRemoteInputChannel(connectionId)).thenReturn(remote);
+ network.setupInputGate(inputGate);
+ NetworkBufferPool bufferPool = network.getNetworkBufferPool();
- final int buffersPerChannel = 2;
- NetworkBufferPool network = mock(NetworkBufferPool.class);
- inputGate.assignExclusiveSegments(network, buffersPerChannel);
+ if (enableCreditBasedFlowControl) {
+ verify(bufferPool, times(0)).requestMemorySegments(buffersPerChannel);
- // Trigger updates to remote input channel from unknown input channel
- inputGate.updateInputChannel(new InputChannelDeploymentDescriptor(
- resultPartitionId,
- ResultPartitionLocation.createRemote(connectionId)));
+ assertEquals(bufferPool.getTotalNumberOfMemorySegments(),
+ bufferPool.getNumberOfAvailableMemorySegments());
+ // note: exclusive buffers are not handed out into LocalBufferPool and are thus not counted
+ assertEquals(extraNetworkBuffersPerGate, bufferPool.countBuffers());
+ } else {
+ assertEquals(buffersPerChannel + extraNetworkBuffersPerGate, bufferPool.countBuffers());
+ }
+
+ // Trigger updates to remote input channel from unknown input channel
+ final ConnectionID connectionId = new ConnectionID(new InetSocketAddress("localhost", 5000), 0);
+ inputGate.updateInputChannel(new InputChannelDeploymentDescriptor(
+ resultPartitionId,
+ ResultPartitionLocation.createRemote(connectionId)));
+
+ if (enableCreditBasedFlowControl) {
+ verify(bufferPool,
+ times(1)).requestMemorySegments(buffersPerChannel);
+ RemoteInputChannel remote = (RemoteInputChannel) inputGate.getInputChannels()
+ .get(resultPartitionId.getPartitionId());
+ // only the exclusive buffers should be assigned/available now
+ assertEquals(buffersPerChannel, remote.getNumberOfAvailableBuffers());
+
+ assertEquals(bufferPool.getTotalNumberOfMemorySegments() - buffersPerChannel,
+ bufferPool.getNumberOfAvailableMemorySegments());
+ // note: exclusive buffers are not handed out into LocalBufferPool and are thus not counted
+ assertEquals(extraNetworkBuffersPerGate, bufferPool.countBuffers());
+ } else {
+ assertEquals(buffersPerChannel + extraNetworkBuffersPerGate, bufferPool.countBuffers());
+ }
+ } finally {
+ inputGate.releaseAllResources();
+ network.shutdown();
+ }
+ }
- verify(network, times(1)).requestMemorySegments(buffersPerChannel);
- verify(remote, times(1)).assignExclusiveSegments(anyListOf(MemorySegment.class));
+ /**
+ * Tests that input gate can successfully convert unknown input channels into local and remote
+ * channels.
+ */
+ @Test
+ public void testUpdateUnknownInputChannel() throws Exception {
+ final SingleInputGate inputGate = createInputGate(2);
+ int buffersPerChannel = 2;
+ final NetworkEnvironment network = createNetworkEnvironment(buffersPerChannel, 8, 0, 0);
+
+ try {
+ final ResultPartitionID localResultPartitionId = new ResultPartitionID();
+ addUnknownInputChannel(network, inputGate, localResultPartitionId, 0);
+
+ final ResultPartitionID remoteResultPartitionId = new ResultPartitionID();
+ addUnknownInputChannel(network, inputGate, remoteResultPartitionId, 1);
+
+ network.setupInputGate(inputGate);
+
+ assertThat(inputGate.getInputChannels().get(remoteResultPartitionId.getPartitionId()),
+ is(instanceOf((UnknownInputChannel.class))));
+ assertThat(inputGate.getInputChannels().get(localResultPartitionId.getPartitionId()),
+ is(instanceOf((UnknownInputChannel.class))));
+
+ // Trigger updates to remote input channel from unknown input channel
+ final ConnectionID remoteConnectionId = new ConnectionID(new InetSocketAddress("localhost", 5000), 0);
+ inputGate.updateInputChannel(new InputChannelDeploymentDescriptor(
+ remoteResultPartitionId,
+ ResultPartitionLocation.createRemote(remoteConnectionId)));
+
+ assertThat(inputGate.getInputChannels().get(remoteResultPartitionId.getPartitionId()),
+ is(instanceOf((RemoteInputChannel.class))));
+ assertThat(inputGate.getInputChannels().get(localResultPartitionId.getPartitionId()),
+ is(instanceOf((UnknownInputChannel.class))));
+
+ // Trigger updates to local input channel from unknown input channel
+ inputGate.updateInputChannel(new InputChannelDeploymentDescriptor(
+ localResultPartitionId,
+ ResultPartitionLocation.createLocal()));
+
+ assertThat(inputGate.getInputChannels().get(remoteResultPartitionId.getPartitionId()),
+ is(instanceOf((RemoteInputChannel.class))));
+ assertThat(inputGate.getInputChannels().get(localResultPartitionId.getPartitionId()),
+ is(instanceOf((LocalInputChannel.class))));
+ } finally {
+ inputGate.releaseAllResources();
+ network.shutdown();
+ }
}
// ---------------------------------------------------------------------------------------------
- private static SingleInputGate createInputGate() {
+ private NetworkEnvironment createNetworkEnvironment(
+ int buffersPerChannel,
+ int extraNetworkBuffersPerGate,
+ int initialBackoff,
+ int maxBackoff) {
+ return new NetworkEnvironment(
+ spy(new NetworkBufferPool(100, 32)),
+ new LocalConnectionManager(),
+ new ResultPartitionManager(),
+ new TaskEventDispatcher(),
+ new KvStateRegistry(),
+ null,
+ null,
+ IOManager.IOMode.SYNC,
+ initialBackoff,
+ maxBackoff,
+ buffersPerChannel,
+ extraNetworkBuffersPerGate,
+ enableCreditBasedFlowControl);
+ }
+
+ private SingleInputGate createInputGate() {
return createInputGate(2);
}
- private static SingleInputGate createInputGate(int numberOfInputChannels) {
+ private SingleInputGate createInputGate(int numberOfInputChannels) {
+ return createInputGate(numberOfInputChannels, ResultPartitionType.PIPELINED);
+ }
+
+ private SingleInputGate createInputGate(
+ int numberOfInputChannels, ResultPartitionType partitionType) {
SingleInputGate inputGate = new SingleInputGate(
"Test Task Name",
new JobID(),
new IntermediateDataSetID(),
- ResultPartitionType.PIPELINED,
+ partitionType,
0,
numberOfInputChannels,
mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup(),
+ enableCreditBasedFlowControl);
- assertEquals(ResultPartitionType.PIPELINED, inputGate.getConsumedPartitionType());
+ assertEquals(partitionType, inputGate.getConsumedPartitionType());
return inputGate;
}
+ private void addUnknownInputChannel(
+ NetworkEnvironment network,
+ SingleInputGate inputGate,
+ ResultPartitionID partitionId,
+ int channelIndex) {
+ UnknownInputChannel unknown =
+ createUnknownInputChannel(network, inputGate, partitionId, channelIndex);
+ inputGate.setInputChannel(partitionId.getPartitionId(), unknown);
+ }
+
+ private UnknownInputChannel createUnknownInputChannel(
+ NetworkEnvironment network,
+ SingleInputGate inputGate,
+ ResultPartitionID partitionId,
+ int channelIndex) {
+ return new UnknownInputChannel(
+ inputGate,
+ channelIndex,
+ partitionId,
+ network.getResultPartitionManager(),
+ network.getTaskEventDispatcher(),
+ network.getConnectionManager(),
+ network.getPartitionRequestInitialBackoff(),
+ network.getPartitionRequestMaxBackoff(),
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup()
+ );
+ }
+
+ private void addRemoteInputChannel(
+ NetworkEnvironment network,
+ SingleInputGate inputGate,
+ ConnectionID connectionId,
+ ResultPartitionID partitionId,
+ int channelIndex) {
+ RemoteInputChannel remote =
+ createUnknownInputChannel(network, inputGate, partitionId, channelIndex)
+ .toRemoteInputChannel(connectionId);
+ inputGate.setInputChannel(partitionId.getPartitionId(), remote);
+ }
+
static void verifyBufferOrEvent(
InputGate inputGate,
boolean expectedIsBuffer,
http://git-wip-us.apache.org/repos/asf/flink/blob/ecde6c32/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestSingleInputGate.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestSingleInputGate.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestSingleInputGate.java
index 0ae6e74..33dc1ca 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestSingleInputGate.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestSingleInputGate.java
@@ -60,7 +60,8 @@ public class TestSingleInputGate {
0,
numberOfInputChannels,
mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup(),
+ true);
this.inputGate = spy(realGate);
http://git-wip-us.apache.org/repos/asf/flink/blob/ecde6c32/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java
index 912cd5b..081d97d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java
@@ -50,13 +50,15 @@ public class UnionInputGateTest {
new IntermediateDataSetID(), ResultPartitionType.PIPELINED,
0, 3,
mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup(),
+ true);
final SingleInputGate ig2 = new SingleInputGate(
testTaskName, new JobID(),
new IntermediateDataSetID(), ResultPartitionType.PIPELINED,
0, 5,
mock(TaskActions.class),
- UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup());
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup(),
+ true);
final UnionInputGate union = new UnionInputGate(new SingleInputGate[]{ig1, ig2});
[11/15] flink git commit: [hotfix] [runtime] Fix a debug statement in
StreamTask
Posted by se...@apache.org.
[hotfix] [runtime] Fix a debug statement in StreamTask
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ff0622b8
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ff0622b8
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ff0622b8
Branch: refs/heads/master
Commit: ff0622b8dbd29d445dc7effa6db60f3e7156feee
Parents: 37df3c8
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Mon Apr 30 15:02:26 2018 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Apr 30 21:24:17 2018 +0200
----------------------------------------------------------------------
.../java/org/apache/flink/streaming/runtime/tasks/StreamTask.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ff0622b8/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index 182543d..6812871 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -1191,7 +1191,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
@SuppressWarnings("unchecked")
StreamPartitioner<OUT> outputPartitioner = (StreamPartitioner<OUT>) edge.getPartitioner();
- LOG.debug("Using partitioner {} for output {} of task ", outputPartitioner, outputIndex, taskName);
+ LOG.debug("Using partitioner {} for output {} of task {}", outputPartitioner, outputIndex, taskName);
ResultPartitionWriter bufferWriter = environment.getWriter(outputIndex);
[14/15] flink git commit: [hotfix] [tests] Remove redundant rebalance
in SuccessAfterNetworkBuffersFailureITCase
Posted by se...@apache.org.
[hotfix] [tests] Remove redundant rebalance in SuccessAfterNetworkBuffersFailureITCase
This closes #5916
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b957bf26
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b957bf26
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b957bf26
Branch: refs/heads/master
Commit: b957bf26fb588cab072e51240e9026456b862ce7
Parents: fa63c33
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Thu Apr 26 10:35:55 2018 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Apr 30 21:30:54 2018 +0200
----------------------------------------------------------------------
.../flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java | 1 -
1 file changed, 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b957bf26/flink-tests/src/test/java/org/apache/flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java b/flink-tests/src/test/java/org/apache/flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java
index c5c1882..e149f43 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java
@@ -136,7 +136,6 @@ public class SuccessAfterNetworkBuffersFailureITCase extends TestLogger {
// add some re-partitions to increase network buffer use
DataSet<KMeans.Centroid> newCentroids = points
- .rebalance()
// compute closest centroid for each point
.map(new KMeans.SelectNearestCenter()).withBroadcastSet(loop, "centroids")
.rebalance()
[08/15] flink git commit: [hotfix] [yarn] Remove unused field
appReport in YarnClusterClient.
Posted by se...@apache.org.
[hotfix] [yarn] Remove unused field appReport in YarnClusterClient.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/02a9d3d8
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/02a9d3d8
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/02a9d3d8
Branch: refs/heads/master
Commit: 02a9d3d878098ab25c4989680b07aa11ce96c859
Parents: 17c68e5
Author: gyao <ga...@data-artisans.com>
Authored: Thu Apr 26 17:44:56 2018 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Apr 30 21:21:00 2018 +0200
----------------------------------------------------------------------
.../src/main/java/org/apache/flink/yarn/YarnClusterClient.java | 2 --
1 file changed, 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/02a9d3d8/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
index 2ac9664..3edd6e4 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
@@ -68,7 +68,6 @@ public class YarnClusterClient extends ClusterClient<ApplicationId> {
private final int slotsPerTaskManager;
private final LazApplicationClientLoader applicationClient;
private final FiniteDuration akkaDuration;
- private final ApplicationReport appReport;
private final ApplicationId appId;
private final String trackingURL;
@@ -101,7 +100,6 @@ public class YarnClusterClient extends ClusterClient<ApplicationId> {
this.clusterDescriptor = clusterDescriptor;
this.numberTaskManagers = numberTaskManagers;
this.slotsPerTaskManager = slotsPerTaskManager;
- this.appReport = appReport;
this.appId = appReport.getApplicationId();
this.trackingURL = appReport.getTrackingUrl();
this.newlyCreatedCluster = newlyCreatedCluster;
[13/15] flink git commit: [FLINK-9274] [kafka] Add thread name for
partition discovery
Posted by se...@apache.org.
[FLINK-9274] [kafka] Add thread name for partition discovery
This closes #5942
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/fa63c335
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/fa63c335
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/fa63c335
Branch: refs/heads/master
Commit: fa63c3356b3fa89e549c058cbb6c6ecc19e61d8a
Parents: a8f4515
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Mon Apr 30 14:22:40 2018 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Apr 30 21:28:54 2018 +0200
----------------------------------------------------------------------
.../flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/fa63c335/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
index 82ac2c3..cfb5b6d 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
@@ -715,7 +715,7 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
}
}
}
- });
+ }, "Kafka Partition Discovery for " + getRuntimeContext().getTaskNameWithSubtasks());
discoveryLoopThread.start();
kafkaFetcher.runFetchLoop();
[07/15] flink git commit: [hotfix] Remove unnecessary int cast.
Posted by se...@apache.org.
[hotfix] Remove unnecessary int cast.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/290b11b6
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/290b11b6
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/290b11b6
Branch: refs/heads/master
Commit: 290b11b6a562f45c2463d3d13692be1ab86245c4
Parents: b209e6e
Author: gyao <ga...@data-artisans.com>
Authored: Thu Apr 19 11:04:27 2018 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Apr 30 21:20:56 2018 +0200
----------------------------------------------------------------------
.../src/main/java/org/apache/flink/yarn/YarnResourceManager.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/290b11b6/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
index 0ea66b8..a25e4c4 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
@@ -286,7 +286,7 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
// Priority for worker containers - priorities are intra-application
//TODO: set priority according to the resource allocated
Priority priority = Priority.newInstance(generatePriority(resourceProfile));
- int mem = resourceProfile.getMemoryInMB() < 0 ? defaultTaskManagerMemoryMB : (int) resourceProfile.getMemoryInMB();
+ int mem = resourceProfile.getMemoryInMB() < 0 ? defaultTaskManagerMemoryMB : resourceProfile.getMemoryInMB();
int vcore = resourceProfile.getCpuCores() < 1 ? defaultCpus : (int) resourceProfile.getCpuCores();
Resource capability = Resource.newInstance(mem, vcore);
requestYarnContainer(capability, priority);