KarmaGYZ commented on a change in pull request #18087:
URL: https://github.com/apache/flink/pull/18087#discussion_r770313868
##########
File path:
flink-yarn/src/main/java/org/apache/flink/yarn/configuration/YarnConfigOptions.java
##########
@@ -296,6 +296,12 @@
.noDefaultValue()
.withDescription("Specify YARN node label for the YARN
application.");
+ public static final ConfigOption<String> TASK_MANAGER_NODE_LABEL =
+ key("yarn.taskmanager.node-label")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Specify YARN node label for the Flink
TaskManagers.");
Review comment:
Better to describe whether it will override the
`yarn.application.node-label` if both are set.
##########
File path:
flink-yarn/src/main/java/org/apache/flink/yarn/ContainerRequestReflector.java
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.client.api.AMRMClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+
+/**
+ * Use reflection to determine whether the Hadoop supports node-label,
depending on the Hadoop
+ * version, may or may not be supported. If not, nothing happened.
+ *
+ * <p>The node label mechanism is supported by Hadoop version greater than
2.6.0
+ */
+class ContainerRequestReflector {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ContainerRequestReflector.class);
+
+ static final ContainerRequestReflector INSTANCE = new
ContainerRequestReflector();
+
+ private Constructor<AMRMClient.ContainerRequest> defaultConstructor;
+
+ private ContainerRequestReflector() {
+ Class<AMRMClient.ContainerRequest> requestCls =
AMRMClient.ContainerRequest.class;
+ try {
+ defaultConstructor =
+ requestCls.getDeclaredConstructor(
+ Resource.class,
+ String[].class,
+ String[].class,
+ Priority.class,
+ boolean.class,
+ String.class);
+ } catch (NoSuchMethodException exception) {
+ LOG.info("The node-label mechanism in Yarn don't be supported in
this Hadoop version.");
+ }
+ }
+
+ public AMRMClient.ContainerRequest getContainerRequest(
+ Resource containerResource, Priority priority, String nodeLabel) {
+ if (nodeLabel == null || defaultConstructor == null) {
+ return new AMRMClient.ContainerRequest(containerResource, null,
null, priority);
+ }
+
+ try {
+ return defaultConstructor.newInstance(
+ containerResource, null, null, priority, true, nodeLabel);
Review comment:
Why do we arbitrarily set it to `true` here?
##########
File path:
flink-yarn/src/main/java/org/apache/flink/yarn/ContainerRequestReflector.java
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.client.api.AMRMClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+
+/**
+ * Use reflection to determine whether the Hadoop supports node-label,
depending on the Hadoop
+ * version, may or may not be supported. If not, nothing happened.
+ *
+ * <p>The node label mechanism is supported by Hadoop version greater than
2.6.0
+ */
+class ContainerRequestReflector {
Review comment:
Could we have a test for it, you can refer to
`ResourceInformationReflectorTest`.
##########
File path:
flink-yarn/src/main/java/org/apache/flink/yarn/ContainerRequestReflector.java
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.client.api.AMRMClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+
+/**
+ * Use reflection to determine whether the Hadoop supports node-label,
depending on the Hadoop
+ * version, may or may not be supported. If not, nothing happened.
+ *
+ * <p>The node label mechanism is supported by Hadoop version greater than
2.6.0
+ */
+class ContainerRequestReflector {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ContainerRequestReflector.class);
+
+ static final ContainerRequestReflector INSTANCE = new
ContainerRequestReflector();
+
+ private Constructor<AMRMClient.ContainerRequest> defaultConstructor;
Review comment:
```suggestion
@Nullable private Constructor<AMRMClient.ContainerRequest>
defaultConstructor;
```
##########
File path:
flink-yarn/src/main/java/org/apache/flink/yarn/ContainerRequestReflector.java
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.client.api.AMRMClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+
+/**
+ * Use reflection to determine whether the Hadoop supports node-label,
depending on the Hadoop
+ * version, may or may not be supported. If not, nothing happened.
+ *
+ * <p>The node label mechanism is supported by Hadoop version greater than
2.6.0
+ */
+class ContainerRequestReflector {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ContainerRequestReflector.class);
+
+ static final ContainerRequestReflector INSTANCE = new
ContainerRequestReflector();
+
+ private Constructor<AMRMClient.ContainerRequest> defaultConstructor;
+
+ private ContainerRequestReflector() {
+ Class<AMRMClient.ContainerRequest> requestCls =
AMRMClient.ContainerRequest.class;
+ try {
+ defaultConstructor =
+ requestCls.getDeclaredConstructor(
+ Resource.class,
+ String[].class,
+ String[].class,
+ Priority.class,
+ boolean.class,
+ String.class);
+ } catch (NoSuchMethodException exception) {
+ LOG.info("The node-label mechanism in Yarn don't be supported in
this Hadoop version.");
+ }
+ }
+
+ public AMRMClient.ContainerRequest getContainerRequest(
+ Resource containerResource, Priority priority, String nodeLabel) {
+ if (nodeLabel == null || defaultConstructor == null) {
+ return new AMRMClient.ContainerRequest(containerResource, null,
null, priority);
+ }
+
+ try {
+ return defaultConstructor.newInstance(
+ containerResource, null, null, priority, true, nodeLabel);
Review comment:
What if `nodeLabel` is empty?
##########
File path:
flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManagerDriver.java
##########
@@ -550,7 +559,33 @@ private FinalApplicationStatus
getYarnStatus(ApplicationStatus status) {
@Nonnull
@VisibleForTesting
static AMRMClient.ContainerRequest getContainerRequest(
- Resource containerResource, Priority priority) {
+ Resource containerResource, Priority priority, String nodeLabel) {
Review comment:
We don't need it anymore.
##########
File path:
flink-yarn/src/main/java/org/apache/flink/yarn/ContainerRequestReflector.java
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.client.api.AMRMClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+
+/**
+ * Use reflection to determine whether the Hadoop supports node-label,
depending on the Hadoop
+ * version, may or may not be supported. If not, nothing happened.
+ *
+ * <p>The node label mechanism is supported by Hadoop version greater than
2.6.0
+ */
+class ContainerRequestReflector {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ContainerRequestReflector.class);
+
+ static final ContainerRequestReflector INSTANCE = new
ContainerRequestReflector();
+
+ private Constructor<AMRMClient.ContainerRequest> defaultConstructor;
+
+ private ContainerRequestReflector() {
+ Class<AMRMClient.ContainerRequest> requestCls =
AMRMClient.ContainerRequest.class;
+ try {
+ defaultConstructor =
+ requestCls.getDeclaredConstructor(
Review comment:
Could you comment on which constructor you use and what each param means?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
