[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1269772728
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveServer2DelegationTokenProvider.java:
##
@@ -0,0 +1,232 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.time.Clock;
+import java.util.Optional;
+
+import static org.apache.flink.runtime.hadoop.HadoopUserUtils.getIssueDate;
+
+/** Delegation token provider for HiveServer2. */
+@Internal
+public class HiveServer2DelegationTokenProvider implements
DelegationTokenProvider {
+
+private static final Logger LOG =
+LoggerFactory.getLogger(HiveServer2DelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private Long tokenRenewalInterval;
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
+LOG.warn("Fail to create HiveServer2 Configuration", e);
+}
+return hiveConf;
+}
+
+@Override
+public boolean delegationTokensRequired() throws Exception {
+/**
+ * The general rule how a provider/receiver must behave is the
following: The provider and
+ * the receiver must be added to the classpath together with all the
additionally required
+ * dependencies.
+ *
+ * This null check is required because the HiveServer2 provider is
always on classpath
+ * but Hive jars are optional. Such case configuration is not able to
be loaded. This
+ * construct is intended to be removed when HiveServer2
provider/receiver pair can be
+ * externalized (namely if a provider/receiver throws an exception
then workload must be
+ * stopped).
+ */
+if (hiveConf == null) {
+LOG.debug(
+"HiveServer2 is not available (not packaged with this
application), hence no "
++ "hiveServer2 tokens will be acquired.");
+return false;
+}
+try {
+if
(!HadoopUtils.isKerberosSecurityEnabled(UserGroupInformation.getCurrentUser()))
{
+LOG.debug(
+"Hadoop Kerberos is not enabled,hence no hiveServer2
tokens will be acquired.");
+return false;
+}
+} catch (IOException e) {
+LOG.debug(
+"Hadoop Kerberos is
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1267947674
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveServer2DelegationTokenProvider.java:
##
@@ -0,0 +1,232 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.time.Clock;
+import java.util.Optional;
+
+import static org.apache.flink.runtime.hadoop.HadoopUserUtils.getIssueDate;
+
+/** Delegation token provider for HiveServer2. */
+@Internal
+public class HiveServer2DelegationTokenProvider implements
DelegationTokenProvider {
+
+private static final Logger LOG =
+LoggerFactory.getLogger(HiveServer2DelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private Long tokenRenewalInterval;
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
+LOG.warn("Fail to create HiveServer2 Configuration", e);
+}
+return hiveConf;
+}
+
+@Override
+public boolean delegationTokensRequired() throws Exception {
+/**
+ * The general rule how a provider/receiver must behave is the
following: The provider and
+ * the receiver must be added to the classpath together with all the
additionally required
+ * dependencies.
+ *
+ * This null check is required because the HiveServer2 provider is
always on classpath
+ * but Hive jars are optional. Such case configuration is not able to
be loaded. This
+ * construct is intended to be removed when HiveServer2
provider/receiver pair can be
+ * externalized (namely if a provider/receiver throws an exception
then workload must be
+ * stopped).
+ */
+if (hiveConf == null) {
+LOG.debug(
+"HiveServer2 is not available (not packaged with this
application), hence no "
++ "hiveServer2 tokens will be acquired.");
+return false;
+}
+try {
+if
(!HadoopUtils.isKerberosSecurityEnabled(UserGroupInformation.getCurrentUser()))
{
+LOG.debug(
+"Hadoop Kerberos is not enabled,hence no hiveServer2
tokens will be acquired.");
+return false;
+}
+} catch (IOException e) {
+LOG.debug(
+"Hadoop Kerberos is
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1263982565
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveServer2DelegationTokenProvider.java:
##
@@ -0,0 +1,235 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.time.Clock;
+import java.util.Optional;
+
+import static org.apache.flink.runtime.hadoop.HadoopUserUtils.getIssueDate;
+
+/** Delegation token provider for HiveServer2. */
+@Internal
+public class HiveServer2DelegationTokenProvider implements
DelegationTokenProvider {
+
+private static final Logger LOG =
+LoggerFactory.getLogger(HiveServer2DelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private Optional tokenRenewalInterval;
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
+LOG.warn("Fail to create HiveServer2 Configuration", e);
+}
+return hiveConf;
+}
+
+@Override
+public boolean delegationTokensRequired() throws Exception {
+/**
+ * The general rule how a provider/receiver must behave is the
following: The provider and
+ * the receiver must be added to the classpath together with all the
additionally required
+ * dependencies.
+ *
+ * This null check is required because the HiveServer2 provider is
always on classpath
+ * but Hive jars are optional. Such case configuration is not able to
be loaded. This
+ * construct is intended to be removed when HiveServer2
provider/receiver pair can be
+ * externalized (namely if a provider/receiver throws an exception
then workload must be
+ * stopped).
+ */
+if (hiveConf == null) {
+LOG.debug(
+"HiveServer2 is not available (not packaged with this
application), hence no "
++ "hiveServer2 tokens will be acquired.");
+return false;
+}
+try {
+if
(!HadoopUtils.isKerberosSecurityEnabled(UserGroupInformation.getCurrentUser()))
{
+LOG.debug(
+"Hadoop Kerberos is not enabled,hence no hiveServer2
tokens will be acquired.");
+return false;
+}
+} catch (IOException e) {
+LOG.debug(
+"Hadoop Kerberos is not enabled,hence no hiveServer2
tokens will be acquired.",
+e);
+return false;
+
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1263982032
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveServer2DelegationTokenProvider.java:
##
@@ -0,0 +1,235 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.time.Clock;
+import java.util.Optional;
+
+import static org.apache.flink.runtime.hadoop.HadoopUserUtils.getIssueDate;
+
+/** Delegation token provider for HiveServer2. */
+@Internal
+public class HiveServer2DelegationTokenProvider implements
DelegationTokenProvider {
+
+private static final Logger LOG =
+LoggerFactory.getLogger(HiveServer2DelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private Optional tokenRenewalInterval;
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
+LOG.warn("Fail to create HiveServer2 Configuration", e);
+}
+return hiveConf;
+}
+
+@Override
+public boolean delegationTokensRequired() throws Exception {
+/**
+ * The general rule how a provider/receiver must behave is the
following: The provider and
+ * the receiver must be added to the classpath together with all the
additionally required
+ * dependencies.
+ *
+ * This null check is required because the HiveServer2 provider is
always on classpath
+ * but Hive jars are optional. Such case configuration is not able to
be loaded. This
+ * construct is intended to be removed when HiveServer2
provider/receiver pair can be
+ * externalized (namely if a provider/receiver throws an exception
then workload must be
+ * stopped).
+ */
+if (hiveConf == null) {
+LOG.debug(
+"HiveServer2 is not available (not packaged with this
application), hence no "
++ "hiveServer2 tokens will be acquired.");
+return false;
+}
+try {
+if
(!HadoopUtils.isKerberosSecurityEnabled(UserGroupInformation.getCurrentUser()))
{
+LOG.debug(
+"Hadoop Kerberos is not enabled,hence no hiveServer2
tokens will be acquired.");
+return false;
+}
+} catch (IOException e) {
+LOG.debug(
+"Hadoop Kerberos is not enabled,hence no hiveServer2
tokens will be acquired.",
+e);
+return false;
+
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1263980114
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveServer2DelegationTokenProvider.java:
##
@@ -0,0 +1,235 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.time.Clock;
+import java.util.Optional;
+
+import static org.apache.flink.runtime.hadoop.HadoopUserUtils.getIssueDate;
+
+/** Delegation token provider for HiveServer2. */
+@Internal
+public class HiveServer2DelegationTokenProvider implements
DelegationTokenProvider {
+
+private static final Logger LOG =
+LoggerFactory.getLogger(HiveServer2DelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private Optional tokenRenewalInterval;
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
+LOG.warn("Fail to create HiveServer2 Configuration", e);
+}
+return hiveConf;
+}
+
+@Override
+public boolean delegationTokensRequired() throws Exception {
+/**
+ * The general rule how a provider/receiver must behave is the
following: The provider and
+ * the receiver must be added to the classpath together with all the
additionally required
+ * dependencies.
+ *
+ * This null check is required because the HiveServer2 provider is
always on classpath
+ * but Hive jars are optional. Such case configuration is not able to
be loaded. This
+ * construct is intended to be removed when HiveServer2
provider/receiver pair can be
+ * externalized (namely if a provider/receiver throws an exception
then workload must be
+ * stopped).
+ */
+if (hiveConf == null) {
+LOG.debug(
+"HiveServer2 is not available (not packaged with this
application), hence no "
++ "hiveServer2 tokens will be acquired.");
+return false;
+}
+try {
+if
(!HadoopUtils.isKerberosSecurityEnabled(UserGroupInformation.getCurrentUser()))
{
+LOG.debug(
+"Hadoop Kerberos is not enabled,hence no hiveServer2
tokens will be acquired.");
+return false;
+}
+} catch (IOException e) {
+LOG.debug(
+"Hadoop Kerberos is not enabled,hence no hiveServer2
tokens will be acquired.",
+e);
+return false;
+
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1263979044
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveServer2DelegationTokenProvider.java:
##
@@ -0,0 +1,235 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.time.Clock;
+import java.util.Optional;
+
+import static org.apache.flink.runtime.hadoop.HadoopUserUtils.getIssueDate;
+
+/** Delegation token provider for HiveServer2. */
+@Internal
+public class HiveServer2DelegationTokenProvider implements
DelegationTokenProvider {
+
+private static final Logger LOG =
+LoggerFactory.getLogger(HiveServer2DelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private Optional tokenRenewalInterval;
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
+LOG.warn("Fail to create HiveServer2 Configuration", e);
+}
+return hiveConf;
+}
+
+@Override
+public boolean delegationTokensRequired() throws Exception {
+/**
+ * The general rule how a provider/receiver must behave is the
following: The provider and
+ * the receiver must be added to the classpath together with all the
additionally required
+ * dependencies.
+ *
+ * This null check is required because the HiveServer2 provider is
always on classpath
+ * but Hive jars are optional. Such case configuration is not able to
be loaded. This
+ * construct is intended to be removed when HiveServer2
provider/receiver pair can be
+ * externalized (namely if a provider/receiver throws an exception
then workload must be
+ * stopped).
+ */
+if (hiveConf == null) {
+LOG.debug(
+"HiveServer2 is not available (not packaged with this
application), hence no "
++ "hiveServer2 tokens will be acquired.");
+return false;
+}
+try {
+if
(!HadoopUtils.isKerberosSecurityEnabled(UserGroupInformation.getCurrentUser()))
{
+LOG.debug(
+"Hadoop Kerberos is not enabled,hence no hiveServer2
tokens will be acquired.");
+return false;
+}
+} catch (IOException e) {
+LOG.debug(
+"Hadoop Kerberos is not enabled,hence no hiveServer2
tokens will be acquired.",
+e);
+return false;
+
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1263974727
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveServer2DelegationTokenProvider.java:
##
@@ -0,0 +1,235 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.time.Clock;
+import java.util.Optional;
+
+import static org.apache.flink.runtime.hadoop.HadoopUserUtils.getIssueDate;
+
+/** Delegation token provider for HiveServer2. */
+@Internal
+public class HiveServer2DelegationTokenProvider implements
DelegationTokenProvider {
+
+private static final Logger LOG =
+LoggerFactory.getLogger(HiveServer2DelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private Optional tokenRenewalInterval;
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
+LOG.warn("Fail to create HiveServer2 Configuration", e);
+}
+return hiveConf;
+}
+
+@Override
+public boolean delegationTokensRequired() throws Exception {
+/**
+ * The general rule how a provider/receiver must behave is the
following: The provider and
+ * the receiver must be added to the classpath together with all the
additionally required
+ * dependencies.
+ *
+ * This null check is required because the HiveServer2 provider is
always on classpath
+ * but Hive jars are optional. Such case configuration is not able to
be loaded. This
+ * construct is intended to be removed when HiveServer2
provider/receiver pair can be
+ * externalized (namely if a provider/receiver throws an exception
then workload must be
+ * stopped).
+ */
+if (hiveConf == null) {
+LOG.debug(
+"HiveServer2 is not available (not packaged with this
application), hence no "
++ "hiveServer2 tokens will be acquired.");
+return false;
+}
+try {
+if
(!HadoopUtils.isKerberosSecurityEnabled(UserGroupInformation.getCurrentUser()))
{
+LOG.debug(
+"Hadoop Kerberos is not enabled,hence no hiveServer2
tokens will be acquired.");
+return false;
+}
+} catch (IOException e) {
+LOG.debug(
+"Hadoop Kerberos is not enabled,hence no hiveServer2
tokens will be acquired.",
+e);
+return false;
+
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1227217764
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveDelegationTokenProvider.java:
##
@@ -0,0 +1,148 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier;
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.Optional;
+
+/** Delegation token provider for Hive. */
+@Internal
+public class HiveDelegationTokenProvider implements DelegationTokenProvider {
+
+private static final Logger LOG =
LoggerFactory.getLogger(HiveDelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private static final Text TOKEN_ALIAS = new
Text("hive.server2.delegation.token");
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
+LOG.warn("Fail to create Hive Configuration", e);
+}
+return hiveConf;
+}
+
+@Override
+public boolean delegationTokensRequired() throws Exception {
+/**
+ * The general rule how a provider/receiver must behave is the
following: The provider and
+ * the receiver must be added to the classpath together with all the
additionally required
+ * dependencies.
+ *
+ * This null check is required because the Hive provider is always
on classpath but Hive
+ * jars are optional. Such case configuration is not able to be
loaded. This construct is
+ * intended to be removed when Hive provider/receiver pair can be
externalized (namely if a
+ * provider/receiver throws an exception then workload must be
stopped).
+ */
+if (hiveConf == null) {
+LOG.debug(
+"Hive is not available (not packaged with this
application), hence no "
++ "tokens will be acquired.");
+return false;
+}
+try {
+if
(!HadoopUtils.isKerberosSecurityEnabled(UserGroupInformation.getCurrentUser()))
{
+return false;
+}
+} catch (IOException e) {
+LOG.debug("Hadoop Kerberos is not enabled.");
+return false;
+}
+return !hiveConf.getTrimmed("hive.metastore.uris", "").isEmpty()
+&& kerberosLoginProvider.isLoginPossible(false);
+}
+
+@Override
+public ObtainedDelegationTokens obtainDelegationTokens() throws Exception {
+UserGroupInformation freshUGI =
kerberosLoginProvider.doLoginAndReturnUGI();
+
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1223958279
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveDelegationTokenProvider.java:
##
@@ -0,0 +1,148 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier;
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.Optional;
+
+/** Delegation token provider for Hive. */
+@Internal
+public class HiveDelegationTokenProvider implements DelegationTokenProvider {
+
+private static final Logger LOG =
LoggerFactory.getLogger(HiveDelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private static final Text TOKEN_ALIAS = new
Text("hive.server2.delegation.token");
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
+LOG.warn("Fail to create Hive Configuration", e);
+}
+return hiveConf;
+}
+
+@Override
+public boolean delegationTokensRequired() throws Exception {
+/**
+ * The general rule how a provider/receiver must behave is the
following: The provider and
+ * the receiver must be added to the classpath together with all the
additionally required
+ * dependencies.
+ *
+ * This null check is required because the Hive provider is always
on classpath but Hive
+ * jars are optional. Such case configuration is not able to be
loaded. This construct is
+ * intended to be removed when Hive provider/receiver pair can be
externalized (namely if a
+ * provider/receiver throws an exception then workload must be
stopped).
+ */
+if (hiveConf == null) {
+LOG.debug(
+"Hive is not available (not packaged with this
application), hence no "
++ "tokens will be acquired.");
+return false;
+}
+try {
+if
(!HadoopUtils.isKerberosSecurityEnabled(UserGroupInformation.getCurrentUser()))
{
+return false;
+}
+} catch (IOException e) {
+LOG.debug("Hadoop Kerberos is not enabled.");
+return false;
+}
+return !hiveConf.getTrimmed("hive.metastore.uris", "").isEmpty()
+&& kerberosLoginProvider.isLoginPossible(false);
+}
+
+@Override
+public ObtainedDelegationTokens obtainDelegationTokens() throws Exception {
+UserGroupInformation freshUGI =
kerberosLoginProvider.doLoginAndReturnUGI();
+
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1223955128
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveDelegationTokenProvider.java:
##
@@ -0,0 +1,148 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier;
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.Optional;
+
+/** Delegation token provider for Hive. */
+@Internal
+public class HiveDelegationTokenProvider implements DelegationTokenProvider {
+
+private static final Logger LOG =
LoggerFactory.getLogger(HiveDelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private static final Text TOKEN_ALIAS = new
Text("hive.server2.delegation.token");
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
+LOG.warn("Fail to create Hive Configuration", e);
+}
+return hiveConf;
+}
+
+@Override
+public boolean delegationTokensRequired() throws Exception {
+/**
+ * The general rule how a provider/receiver must behave is the
following: The provider and
+ * the receiver must be added to the classpath together with all the
additionally required
+ * dependencies.
+ *
+ * This null check is required because the Hive provider is always
on classpath but Hive
+ * jars are optional. Such case configuration is not able to be
loaded. This construct is
+ * intended to be removed when Hive provider/receiver pair can be
externalized (namely if a
+ * provider/receiver throws an exception then workload must be
stopped).
+ */
+if (hiveConf == null) {
+LOG.debug(
+"Hive is not available (not packaged with this
application), hence no "
++ "tokens will be acquired.");
+return false;
+}
+try {
+if
(!HadoopUtils.isKerberosSecurityEnabled(UserGroupInformation.getCurrentUser()))
{
+return false;
+}
+} catch (IOException e) {
+LOG.debug("Hadoop Kerberos is not enabled.");
+return false;
+}
+return !hiveConf.getTrimmed("hive.metastore.uris", "").isEmpty()
+&& kerberosLoginProvider.isLoginPossible(false);
+}
+
+@Override
+public ObtainedDelegationTokens obtainDelegationTokens() throws Exception {
+UserGroupInformation freshUGI =
kerberosLoginProvider.doLoginAndReturnUGI();
+
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1223879529
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveDelegationTokenProvider.java:
##
@@ -0,0 +1,148 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier;
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.Optional;
+
+/** Delegation token provider for Hive. */
+@Internal
+public class HiveDelegationTokenProvider implements DelegationTokenProvider {
Review Comment:
Nit: maybe `HiveServer2DelegationTokenProvider`?
--
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
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1223878792
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveDelegationTokenProvider.java:
##
@@ -0,0 +1,148 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier;
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.Optional;
+
+/** Delegation token provider for Hive. */
+@Internal
+public class HiveDelegationTokenProvider implements DelegationTokenProvider {
+
+private static final Logger LOG =
LoggerFactory.getLogger(HiveDelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private static final Text TOKEN_ALIAS = new
Text("hive.server2.delegation.token");
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
+LOG.warn("Fail to create Hive Configuration", e);
+}
+return hiveConf;
+}
+
+@Override
+public boolean delegationTokensRequired() throws Exception {
+/**
+ * The general rule how a provider/receiver must behave is the
following: The provider and
+ * the receiver must be added to the classpath together with all the
additionally required
+ * dependencies.
+ *
+ * This null check is required because the Hive provider is always
on classpath but Hive
+ * jars are optional. Such case configuration is not able to be
loaded. This construct is
+ * intended to be removed when Hive provider/receiver pair can be
externalized (namely if a
+ * provider/receiver throws an exception then workload must be
stopped).
+ */
+if (hiveConf == null) {
+LOG.debug(
+"Hive is not available (not packaged with this
application), hence no "
++ "tokens will be acquired.");
+return false;
+}
+try {
+if
(!HadoopUtils.isKerberosSecurityEnabled(UserGroupInformation.getCurrentUser()))
{
+return false;
+}
+} catch (IOException e) {
+LOG.debug("Hadoop Kerberos is not enabled.");
+return false;
+}
+return !hiveConf.getTrimmed("hive.metastore.uris", "").isEmpty()
+&& kerberosLoginProvider.isLoginPossible(false);
+}
+
+@Override
+public ObtainedDelegationTokens obtainDelegationTokens() throws Exception {
+UserGroupInformation freshUGI =
kerberosLoginProvider.doLoginAndReturnUGI();
+
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1223877338
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveDelegationTokenProvider.java:
##
@@ -0,0 +1,148 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier;
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.Optional;
+
+/** Delegation token provider for Hive. */
+@Internal
+public class HiveDelegationTokenProvider implements DelegationTokenProvider {
+
+private static final Logger LOG =
LoggerFactory.getLogger(HiveDelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private static final Text TOKEN_ALIAS = new
Text("hive.server2.delegation.token");
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
+LOG.warn("Fail to create Hive Configuration", e);
+}
+return hiveConf;
+}
+
+@Override
+public boolean delegationTokensRequired() throws Exception {
+/**
+ * The general rule how a provider/receiver must behave is the
following: The provider and
+ * the receiver must be added to the classpath together with all the
additionally required
+ * dependencies.
+ *
+ * This null check is required because the Hive provider is always
on classpath but Hive
+ * jars are optional. Such case configuration is not able to be
loaded. This construct is
+ * intended to be removed when Hive provider/receiver pair can be
externalized (namely if a
+ * provider/receiver throws an exception then workload must be
stopped).
+ */
+if (hiveConf == null) {
+LOG.debug(
+"Hive is not available (not packaged with this
application), hence no "
++ "tokens will be acquired.");
+return false;
+}
+try {
+if
(!HadoopUtils.isKerberosSecurityEnabled(UserGroupInformation.getCurrentUser()))
{
+return false;
+}
+} catch (IOException e) {
+LOG.debug("Hadoop Kerberos is not enabled.");
+return false;
+}
+return !hiveConf.getTrimmed("hive.metastore.uris", "").isEmpty()
+&& kerberosLoginProvider.isLoginPossible(false);
+}
+
+@Override
+public ObtainedDelegationTokens obtainDelegationTokens() throws Exception {
+UserGroupInformation freshUGI =
kerberosLoginProvider.doLoginAndReturnUGI();
+
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1223876233
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveDelegationTokenProvider.java:
##
@@ -0,0 +1,148 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier;
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.Optional;
+
+/** Delegation token provider for Hive. */
+@Internal
+public class HiveDelegationTokenProvider implements DelegationTokenProvider {
+
+private static final Logger LOG =
LoggerFactory.getLogger(HiveDelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private static final Text TOKEN_ALIAS = new
Text("hive.server2.delegation.token");
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
+LOG.warn("Fail to create Hive Configuration", e);
+}
+return hiveConf;
+}
+
+@Override
+public boolean delegationTokensRequired() throws Exception {
+/**
+ * The general rule how a provider/receiver must behave is the
following: The provider and
+ * the receiver must be added to the classpath together with all the
additionally required
+ * dependencies.
+ *
+ * This null check is required because the Hive provider is always
on classpath but Hive
+ * jars are optional. Such case configuration is not able to be
loaded. This construct is
+ * intended to be removed when Hive provider/receiver pair can be
externalized (namely if a
+ * provider/receiver throws an exception then workload must be
stopped).
+ */
+if (hiveConf == null) {
+LOG.debug(
+"Hive is not available (not packaged with this
application), hence no "
++ "tokens will be acquired.");
+return false;
+}
+try {
+if
(!HadoopUtils.isKerberosSecurityEnabled(UserGroupInformation.getCurrentUser()))
{
+return false;
+}
+} catch (IOException e) {
+LOG.debug("Hadoop Kerberos is not enabled.");
+return false;
+}
+return !hiveConf.getTrimmed("hive.metastore.uris", "").isEmpty()
Review Comment:
nit: maybe a minimally a debug log here to show why it is not working if the
result is false
--
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
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1223876105
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveDelegationTokenProvider.java:
##
@@ -0,0 +1,148 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier;
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.Optional;
+
+/** Delegation token provider for Hive. */
+@Internal
+public class HiveDelegationTokenProvider implements DelegationTokenProvider {
+
+private static final Logger LOG =
LoggerFactory.getLogger(HiveDelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private static final Text TOKEN_ALIAS = new
Text("hive.server2.delegation.token");
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
+LOG.warn("Fail to create Hive Configuration", e);
+}
+return hiveConf;
+}
+
+@Override
+public boolean delegationTokensRequired() throws Exception {
+/**
+ * The general rule how a provider/receiver must behave is the
following: The provider and
+ * the receiver must be added to the classpath together with all the
additionally required
+ * dependencies.
+ *
+ * This null check is required because the Hive provider is always
on classpath but Hive
+ * jars are optional. Such case configuration is not able to be
loaded. This construct is
+ * intended to be removed when Hive provider/receiver pair can be
externalized (namely if a
+ * provider/receiver throws an exception then workload must be
stopped).
+ */
+if (hiveConf == null) {
+LOG.debug(
+"Hive is not available (not packaged with this
application), hence no "
++ "tokens will be acquired.");
+return false;
+}
+try {
+if
(!HadoopUtils.isKerberosSecurityEnabled(UserGroupInformation.getCurrentUser()))
{
+return false;
Review Comment:
nit: maybe a minimally a debug log here to show why it is not working
--
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
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1223870641
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveDelegationTokenProvider.java:
##
@@ -0,0 +1,148 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier;
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.Optional;
+
+/** Delegation token provider for Hive. */
+@Internal
+public class HiveDelegationTokenProvider implements DelegationTokenProvider {
+
+private static final Logger LOG =
LoggerFactory.getLogger(HiveDelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private static final Text TOKEN_ALIAS = new
Text("hive.server2.delegation.token");
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
Review Comment:
Ohh... I see. The answer is yes based on the comment below
--
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
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1223870641
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveDelegationTokenProvider.java:
##
@@ -0,0 +1,148 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier;
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.Optional;
+
+/** Delegation token provider for Hive. */
+@Internal
+public class HiveDelegationTokenProvider implements DelegationTokenProvider {
+
+private static final Logger LOG =
LoggerFactory.getLogger(HiveDelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private static final Text TOKEN_ALIAS = new
Text("hive.server2.delegation.token");
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
Review Comment:
Ohh... I see
--
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
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1223870388
##
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveDelegationTokenProvider.java:
##
@@ -0,0 +1,148 @@
+/*
+ * 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.table.security.token;
+
+import org.apache.flink.annotation.Internal;
+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.flink.runtime.security.token.hadoop.KerberosLoginProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier;
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.Optional;
+
+/** Delegation token provider for Hive. */
+@Internal
+public class HiveDelegationTokenProvider implements DelegationTokenProvider {
+
+private static final Logger LOG =
LoggerFactory.getLogger(HiveDelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private static final Text TOKEN_ALIAS = new
Text("hive.server2.delegation.token");
+
+@Override
+public String serviceName() {
+return "HiveServer2";
+}
+
+@Override
+public void init(Configuration configuration) throws Exception {
+hiveConf = getHiveConfiguration(configuration);
+kerberosLoginProvider = new KerberosLoginProvider(configuration);
+}
+
+private org.apache.hadoop.conf.Configuration
getHiveConfiguration(Configuration conf) {
+try {
+org.apache.hadoop.conf.Configuration hadoopConf =
+HadoopUtils.getHadoopConfiguration(conf);
+hiveConf = new HiveConf(hadoopConf, HiveConf.class);
+} catch (Exception | NoClassDefFoundError e) {
Review Comment:
Question: If we put the provider to the Hive module, is it still possible to
miss Hive from the classpath?
--
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
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694: URL: https://github.com/apache/flink/pull/22694#discussion_r1223869035 ## flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveDelegationTokenProvider.java: ## @@ -0,0 +1,148 @@ +/* + * 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.table.security.token; + +import org.apache.flink.annotation.Internal; +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.flink.runtime.security.token.hadoop.KerberosLoginProvider; +import org.apache.flink.runtime.util.HadoopUtils; +import org.apache.flink.util.FlinkRuntimeException; +import org.apache.flink.util.Preconditions; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.ql.metadata.Hive; +import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier; +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.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.security.PrivilegedExceptionAction; +import java.util.Optional; + +/** Delegation token provider for Hive. */ Review Comment: Nit: maybe we could add hiveserver2 to the comment as well -- 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
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694: URL: https://github.com/apache/flink/pull/22694#discussion_r1213177848 ## flink-runtime/src/main/java/org/apache/flink/runtime/security/token/hadoop/HiveDelegationTokenProvider.java: ## @@ -0,0 +1,177 @@ +/* + * 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.security.token.hadoop; Review Comment: Should we put this class to the hive connector module instead? In that case we might be able to get rid of the `Class.forname` stuff which is ugly and error prone -- 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
[GitHub] [flink] pvary commented on a diff in pull request #22694: [FLINK-32223][runtime][security] Add Hive delegation token support
pvary commented on code in PR #22694:
URL: https://github.com/apache/flink/pull/22694#discussion_r1213175818
##
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/hadoop/HiveDelegationTokenProvider.java:
##
@@ -0,0 +1,177 @@
+/*
+ * 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.security.token.hadoop;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.security.token.DelegationTokenProvider;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import
org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.security.PrivilegedExceptionAction;
+import java.util.Optional;
+
+/** Delegation token provider for Hive. */
+@Internal
+public class HiveDelegationTokenProvider implements DelegationTokenProvider {
+
+private static final Logger LOG =
LoggerFactory.getLogger(HiveDelegationTokenProvider.class);
+
+org.apache.hadoop.conf.Configuration hiveConf;
+
+private KerberosLoginProvider kerberosLoginProvider;
+
+private static final Text TOKEN_ALIAS = new
Text("hive.server2.delegation.token");
+
+@Override
+public String serviceName() {
+return "hive";
Review Comment:
Maybe using `HiveServer2` instead of `hive` would be better.
In some uses-cases the client wants to connect to the metastore and does not
run any HS2 instances. For these cases we need `HiveMetastore` or `HMS`. And
having `hive` could be confusing in this case.
--
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
