[
https://issues.apache.org/jira/browse/FLINK-6630?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16133055#comment-16133055
]
ASF GitHub Bot commented on FLINK-6630:
---------------------------------------
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/4555#discussion_r133974223
--- Diff:
flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosEntrypointUtils.java
---
@@ -0,0 +1,175 @@
+/*
+ * 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.entrypoint;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.configuration.IllegalConfigurationException;
+import org.apache.flink.mesos.configuration.MesosOptions;
+import
org.apache.flink.mesos.runtime.clusterframework.MesosTaskManagerParameters;
+import org.apache.flink.mesos.util.MesosConfiguration;
+import org.apache.flink.runtime.clusterframework.BootstrapTools;
+import org.apache.flink.runtime.clusterframework.ContainerSpecification;
+import
org.apache.flink.runtime.clusterframework.overlays.CompositeContainerOverlay;
+import
org.apache.flink.runtime.clusterframework.overlays.FlinkDistributionOverlay;
+import
org.apache.flink.runtime.clusterframework.overlays.HadoopConfOverlay;
+import
org.apache.flink.runtime.clusterframework.overlays.HadoopUserOverlay;
+import org.apache.flink.runtime.clusterframework.overlays.KeytabOverlay;
+import org.apache.flink.runtime.clusterframework.overlays.Krb5ConfOverlay;
+import org.apache.flink.runtime.clusterframework.overlays.SSLStoreOverlay;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.mesos.Protos;
+import org.slf4j.Logger;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+import scala.concurrent.duration.Duration;
+import scala.concurrent.duration.FiniteDuration;
+
+/**
+ * Utils for Mesos entrpoints.
+ */
+public class MesosEntrypointUtils {
+
+ @Deprecated
+ public static Configuration loadConfiguration(CommandLine cmd) {
+
+ // merge the dynamic properties from the command-line
+ Configuration dynamicProperties =
BootstrapTools.parseDynamicProperties(cmd);
+ GlobalConfiguration.setDynamicProperties(dynamicProperties);
+ Configuration config = GlobalConfiguration.loadConfiguration();
+
+ return config;
+ }
+
+ /**
+ * Loads and validates the Mesos scheduler configuration.
+ * @param flinkConfig the global configuration.
+ * @param hostname the hostname to advertise to the Mesos master.
+ */
+ public static MesosConfiguration
createMesosSchedulerConfiguration(Configuration flinkConfig, String hostname) {
+
+ Protos.FrameworkInfo.Builder frameworkInfo =
Protos.FrameworkInfo.newBuilder()
+ .setHostname(hostname);
+ Protos.Credential.Builder credential = null;
+
+ if (!flinkConfig.contains(MesosOptions.MASTER_URL)) {
+ throw new
IllegalConfigurationException(MesosOptions.MASTER_URL.key() + " must be
configured.");
+ }
+ String masterUrl =
flinkConfig.getString(MesosOptions.MASTER_URL);
+
+ Duration failoverTimeout = FiniteDuration.apply(
+ flinkConfig.getInteger(
+ MesosOptions.FAILOVER_TIMEOUT_SECONDS),
+ TimeUnit.SECONDS);
+ frameworkInfo.setFailoverTimeout(failoverTimeout.toSeconds());
+
+ frameworkInfo.setName(flinkConfig.getString(
+ MesosOptions.RESOURCEMANAGER_FRAMEWORK_NAME));
+
+ frameworkInfo.setRole(flinkConfig.getString(
+ MesosOptions.RESOURCEMANAGER_FRAMEWORK_ROLE));
+
+ frameworkInfo.setUser(flinkConfig.getString(
+ MesosOptions.RESOURCEMANAGER_FRAMEWORK_USER));
+
+ if
(flinkConfig.contains(MesosOptions.RESOURCEMANAGER_FRAMEWORK_PRINCIPAL)) {
+ frameworkInfo.setPrincipal(flinkConfig.getString(
+
MesosOptions.RESOURCEMANAGER_FRAMEWORK_PRINCIPAL));
+
+ credential = Protos.Credential.newBuilder();
+ credential.setPrincipal(frameworkInfo.getPrincipal());
+
+ // some environments use a side-channel to communicate
the secret to Mesos,
+ // and thus don't set the 'secret' configuration setting
+ if
(flinkConfig.contains(MesosOptions.RESOURCEMANAGER_FRAMEWORK_SECRET)) {
+ credential.setSecret(flinkConfig.getString(
+
MesosOptions.RESOURCEMANAGER_FRAMEWORK_SECRET));
+ }
+ }
+
+ MesosConfiguration mesos =
+ new MesosConfiguration(masterUrl, frameworkInfo,
scala.Option.apply(credential));
+
+ return mesos;
+ }
+
+ public static MesosTaskManagerParameters
createTmParameters(Configuration configuration, Logger log) {
+ // TM configuration
+ final MesosTaskManagerParameters taskManagerParameters =
MesosTaskManagerParameters.create(configuration);
+
+ log.info("TaskManagers will be created with {} task slots",
+
taskManagerParameters.containeredParameters().numSlots());
+ log.info("TaskManagers will be started with container size {}
MB, JVM heap size {} MB, " +
+ "JVM direct memory limit {} MB, {} cpus",
+
taskManagerParameters.containeredParameters().taskManagerTotalMemoryMB(),
+
taskManagerParameters.containeredParameters().taskManagerHeapSizeMB(),
+
taskManagerParameters.containeredParameters().taskManagerDirectMemoryLimitMB(),
+ taskManagerParameters.cpus());
+
+ return taskManagerParameters;
+ }
+
+ public static ContainerSpecification createContainerSpec(Configuration
configuration, Configuration dynamicProperties)
+ throws Exception {
+ // generate a container spec which conveys the artifacts/vars
needed to launch a TM
+ ContainerSpecification spec = new ContainerSpecification();
+
+ // propagate the AM dynamic configuration to the TM
+ spec.getDynamicConfiguration().addAll(dynamicProperties);
+
+// // propagate newly-generated configuration elements
+// final Configuration taskManagerConfig =
BootstrapTools.generateTaskManagerConfiguration(
+// new Configuration(), null, 0,
taskManagerParameters.containeredParameters().numSlots(),
+// TASKMANAGER_REGISTRATION_TIMEOUT);
+// spec.getDynamicConfiguration().addAll(taskManagerConfig);
--- End diff --
Can we remove this code?
> Implement FLIP-6 MesosAppMasterRunner
> -------------------------------------
>
> Key: FLINK-6630
> URL: https://issues.apache.org/jira/browse/FLINK-6630
> Project: Flink
> Issue Type: Sub-task
> Components: Mesos
> Reporter: Eron Wright
> Assignee: Eron Wright
>
> A new runner must be developed for the FLIP-6 RM. Target the "single job"
> scenario.
> Take some time to consider a general solution or a base implementation that
> is shared with the old implementation.
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
