[ZEPPELIN-4851]. Add property to flink interpreter to allow replace the yarn address

docs/interpreter/flink.md

@@ -139,7 +139,12 @@ You can also set other flink properties which are not listed in the table. For a
   <tr>
     <td>flink.webui.yarn.useProxy</td>
     <td>false</td>
-    <td>whether use yarn proxy url as flink weburl, e.g. http://localhost:8088/proxy/application_1583396598068_0004</td>
+    <td>whether use yarn proxy url as flink weburl, e.g. http://resource-manager:8088/proxy/application_1583396598068_0004</td>
+  </tr>
+  <tr>
+    <td>flink.webui.yarn.yarnAddress</td>
+    <td></td>
+    <td>Set this value only when your yarn address is mapped to some other address, e.g. some cloud vender will map `http://resource-manager:8088` to `https://xxx-yarn.yy.cn/gateway/kkk/yarn`</td>
   </tr>
   <tr>
     <td>flink.udf.jars</td>

flink/interpreter/pom.xml

@@ -626,6 +626,13 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>org.scalatest</groupId>
+      <artifactId>scalatest_2.11</artifactId>
+      <version>3.0.8</version>
+      <scope>test</scope>
+    </dependency>
+
   </dependencies>
 
   <build>
@@ -724,6 +731,12 @@
         </configuration>
       </plugin>
 
+      <!-- Scalatest runs all Scala tests -->
+      <plugin>
+        <groupId>org.scalatest</groupId>
+        <artifactId>scalatest-maven-plugin</artifactId>
+      </plugin>
+
       <!-- Eclipse Integration -->
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>

flink/interpreter/src/main/resources/interpreter-setting.json

@@ -96,6 +96,13 @@
         "description": "Whether use yarn proxy url as flink weburl, e.g. http://localhost:8088/proxy/application_1583396598068_0004",
         "type": "checkbox"
       },
+      "flink.webui.yarn.yarnAddress": {
+        "envName": null,
+        "propertyName": null,
+        "defaultValue": "",
+        "description": "Set this value only when your yarn address is mapped to some other address, e.g. some cloud vender will map `http://resource-manager:8088` to `https://xxx-yarn.yy.cn/gateway/kkk/yarn`",
+        "type": "checkbox"
+      },
       "flink.udf.jars": {
         "envName": null,
         "propertyName": null,

flink/interpreter/src/main/scala/org/apache/zeppelin/flink/FlinkScalaInterpreter.scala

@@ -24,6 +24,7 @@ import java.nio.file.Files
 import java.util.Properties
 import java.util.concurrent.TimeUnit
 import java.util.jar.JarFile
+import java.util.regex.Pattern
 
 import org.apache.commons.lang3.StringUtils
 import org.apache.commons.lang3.exception.ExceptionUtils
@@ -259,6 +260,11 @@ class FlinkScalaInterpreter(val properties: Properties) {
             LOGGER.info("Starting FlinkCluster in yarn mode")
             if (properties.getProperty("flink.webui.yarn.useProxy", "false").toBoolean) {
               this.jmWebUrl = HadoopUtils.getYarnAppTrackingUrl(clusterClient)
+              // for some cloud vender, the yarn address may be mapped to some other address.
+              val yarnAddress = properties.getProperty("flink.webui.yarn.address")
+              if (!StringUtils.isBlank(yarnAddress)) {
+                this.jmWebUrl = replaceYarnAddress(this.jmWebUrl, yarnAddress)
+              }
             } else {
               this.jmWebUrl = clusterClient.getWebInterfaceURL
             }
@@ -829,6 +835,12 @@ class FlinkScalaInterpreter(val properties: Properties) {
       }
     }
   }
+
+  def replaceYarnAddress(webURL: String, yarnAddress: String): String = {
+    val pattern = "(https?://.*:\\d+)(.*)".r
+    val pattern(prefix, remaining) = webURL
+    yarnAddress + remaining
+  }
 }
 
 

flink/interpreter/src/test/scala/org/apache/zeppelin/flink/FlinkScalaInterpreterTest.scala

@@ -0,0 +1,42 @@
+/*
+ * 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.zeppelin.flink
+
+
+import java.util.Properties
+
+import org.junit.Assert.assertEquals
+import org.scalatest.FunSuite
+
+class FlinkScalaInterpreterTest extends FunSuite {
+
+  test("testReplaceYarnAddress") {
+    val flinkScalaInterpreter = new FlinkScalaInterpreter(new Properties())
+    var targetURL = flinkScalaInterpreter.replaceYarnAddress("http://localhost:8081",
+      "http://my-server:9090/gateway")
+    assertEquals("http://my-server:9090/gateway", targetURL)
+
+    targetURL = flinkScalaInterpreter.replaceYarnAddress("https://localhost:8081/",
+      "https://my-server:9090/gateway")
+    assertEquals("https://my-server:9090/gateway/", targetURL)
+
+    targetURL = flinkScalaInterpreter.replaceYarnAddress("https://localhost:8081/proxy/app_1",
+      "https://my-server:9090/gateway")
+    assertEquals("https://my-server:9090/gateway/proxy/app_1", targetURL)
+  }
+}