apache
diff --git a/‎docs/configuration/settings.md‎
Lines changed: 1 addition & 0 deletions b/‎docs/configuration/settings.md‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎extensions/flink/kyuubi-flink-token-provider/pom.xml‎
Lines changed: 78 additions & 0 deletions b/‎extensions/flink/kyuubi-flink-token-provider/pom.xml‎
Lines changed: 78 additions & 0 deletions
diff --git a/‎extensions/flink/kyuubi-flink-token-provider/src/main/java/org/apache/kyuubi/flink/security/token/KyuubiDelegationTokenProvider.java‎
Lines changed: 97 additions & 0 deletions b/‎extensions/flink/kyuubi-flink-token-provider/src/main/java/org/apache/kyuubi/flink/security/token/KyuubiDelegationTokenProvider.java‎
Lines changed: 97 additions & 0 deletions
diff --git a/‎extensions/flink/kyuubi-flink-token-provider/src/main/java/org/apache/kyuubi/flink/security/token/KyuubiDelegationTokenReceiver.java‎
Lines changed: 28 additions & 0 deletions b/‎extensions/flink/kyuubi-flink-token-provider/src/main/java/org/apache/kyuubi/flink/security/token/KyuubiDelegationTokenReceiver.java‎
Lines changed: 28 additions & 0 deletions
@@ -149,6 +149,7 @@ You can configure the Kyuubi properties in `$KYUUBI_HOME/conf/kyuubi-defaults.co
 | kyuubi.engine.event.json.log.path                        | file:///tmp/kyuubi/events | The location where all the engine events go for the built-in JSON logger.<ul><li>Local Path: start with 'file://'</li><li>HDFS Path: start with 'hdfs://'</li></ul>                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             | string   | 1.3.0  |
 | kyuubi.engine.event.loggers                              | SPARK                     | A comma-separated list of engine history loggers, where engine/session/operation etc events go.<ul> <li>SPARK: the events will be written to the Spark listener bus.</li> <li>JSON: the events will be written to the location of kyuubi.engine.event.json.log.path</li> <li>JDBC: to be done</li> <li>CUSTOM: User-defined event handlers.</li></ul> Note that: Kyuubi supports custom event handlers with the Java SPI. To register a custom event handler, the user needs to implement a subclass of `org.apache.kyuubi.events.handler.CustomEventHandlerProvider` which has a zero-arg constructor.                                                                                                                                                                                                                                                                                                                                                                                                                                                         | seq      | 1.3.0  |
 | kyuubi.engine.flink.application.jars                     | &lt;undefined&gt;         | A comma-separated list of the local jars to be shipped with the job to the cluster. For example, SQL UDF jars. Only effective in yarn application mode.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         | string   | 1.8.0  |
+| kyuubi.engine.flink.doAs.enabled                         | false                     | When enabled, the session user is used as the proxy user to launch the Flink engine, otherwise, the server user. Note, due to the limitation of Apache Flink, it can only be enabled on Kerberized environment.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                 | boolean  | 1.10.0 |
 | kyuubi.engine.flink.extra.classpath                      | &lt;undefined&gt;         | The extra classpath for the Flink SQL engine, for configuring the location of hadoop client jars, etc. Only effective in yarn session mode.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     | string   | 1.6.0  |
 | kyuubi.engine.flink.initialize.sql                       | SHOW DATABASES            | The initialize sql for Flink engine. It fallback to `kyuubi.engine.initialize.sql`.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             | seq      | 1.8.1  |
 | kyuubi.engine.flink.java.options                         | &lt;undefined&gt;         | The extra Java options for the Flink SQL engine. Only effective in yarn session mode.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                           | string   | 1.6.0  |
 
@@ -0,0 +1,78 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.kyuubi</groupId>
+        <artifactId>kyuubi-parent</artifactId>
+        <version>1.10.0-SNAPSHOT</version>
+        <relativePath>../../../pom.xml</relativePath>
+    </parent>
+
+    <artifactId>kyuubi-flink-token-provider</artifactId>
+    <packaging>jar</packaging>
+    <name>Kyuubi Flink Token Provider</name>
+    <url>https://kyuubi.apache.org/</url>
+
+    <dependencies>
+        <!-- flink -->
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-core</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-runtime</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-clients</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- hadoop client -->
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-client-api</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-client-runtime</artifactId>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>net.alchim31.maven</groupId>
+                <artifactId>scala-maven-plugin</artifactId>
+                <configuration>
+                    <skip>true</skip>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+</project>
@@ -0,0 +1,97 @@
+/*
+ * 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.kyuubi.flink.security.token;
+
+import static org.apache.flink.client.deployment.application.ApplicationConfiguration.APPLICATION_ARGS;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.configuration.ConfigUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.security.token.DelegationTokenProvider;
+import org.apache.flink.runtime.security.token.hadoop.HadoopDelegationTokenConverter;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.TokenIdentifier;
+import org.apache.kyuubi.flink.security.token.utils.KyuubiUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class KyuubiDelegationTokenProvider implements DelegationTokenProvider {
+
+  private static final Logger LOG = LoggerFactory.getLogger(KyuubiDelegationTokenProvider.class);
+
+  public static volatile Map<Text, Token<? extends TokenIdentifier>> previousTokens;
+
+  private long renewalInterval;
+
+  @Override
+  public void init(Configuration configuration) throws Exception {
+    final List<String> programArgsList =
+        ConfigUtils.decodeListFromConfig(configuration, APPLICATION_ARGS, String::new);
+    Map<String, String> kyuubiConf = KyuubiUtils.fromCommandLineArgs(programArgsList);
+    String engineCredentials =
+        kyuubiConf.getOrDefault(KyuubiUtils.KYUUBI_ENGINE_CREDENTIALS_KEY, "");
+    if (StringUtils.isNotBlank(engineCredentials)) {
+      LOG.info("Renew delegation token with engine credentials: {}", engineCredentials);
+      KyuubiUtils.renewDelegationToken(engineCredentials);
+    }
+    Credentials credentials = UserGroupInformation.getCurrentUser().getCredentials();
+    previousTokens = new HashMap<>(credentials.getTokenMap());
+    String interval = kyuubiConf.get(KyuubiUtils.KYUUBI_CREDENTIALS_RENEWAL_INTERVAL_KEY);
+    if (StringUtils.isNotBlank(interval)) {
+      renewalInterval = Long.parseLong(interval);
+    } else {
+      renewalInterval = KyuubiUtils.KYUUBI_CREDENTIALS_RENEWAL_INTERVAL_DEFAULT;
+    }
+  }
+
+  @Override
+  public ObtainedDelegationTokens obtainDelegationTokens() throws Exception {
+    Credentials credentials = UserGroupInformation.getCurrentUser().getCredentials();
+    Credentials newCredentials = new Credentials();
+    for (Map.Entry<Text, Token<? extends TokenIdentifier>> tokenEntry :
+        credentials.getTokenMap().entrySet()) {
+      Text alias = tokenEntry.getKey();
+      Token<? extends TokenIdentifier> token = tokenEntry.getValue();
+      Token<? extends TokenIdentifier> previousToken = previousTokens.get(alias);
+      if (previousToken == null || KyuubiUtils.compareIssueDate(token, previousToken) > 0) {
+        newCredentials.addToken(alias, token);
+      }
+    }
+    previousTokens = new HashMap<>(credentials.getTokenMap());
+    Optional<Long> validUntil = Optional.of(System.currentTimeMillis() + renewalInterval);
+    return new ObtainedDelegationTokens(
+        HadoopDelegationTokenConverter.serialize(credentials), validUntil);
+  }
+
+  @Override
+  public boolean delegationTokensRequired() throws Exception {
+    return true;
+  }
+
+  @Override
+  public String serviceName() {
+    return "kyuubi";
+  }
+}
@@ -0,0 +1,28 @@
+/*
+ * 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.kyuubi.flink.security.token;
+
+import org.apache.flink.runtime.security.token.hadoop.HadoopDelegationTokenReceiver;
+
+public class KyuubiDelegationTokenReceiver extends HadoopDelegationTokenReceiver {
+
+  @Override
+  public String serviceName() {
+    return "kyuubi";
+  }
+}