You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2023/01/05 10:17:07 UTC
[GitHub] [flink] gaborgsomogyi opened a new pull request, #21604: [FLINK-30425][runtime][security] Generalize token receive side
gaborgsomogyi opened a new pull request, #21604:
URL: https://github.com/apache/flink/pull/21604
## What is the purpose of the change
Flink delegation token obtain side is generic already but on the receiver side there is still an assumption that only Hadoop tokens can come. In this PR I've introduced `DelegationTokenReceiver` interface which can be implemented by every.
## Brief change log
* Added `DelegationTokenReceiver` interface
* Added `HadoopFSDelegationTokenReceiver` for Hadoop FS
* Added `HBaseDelegationTokenReceiver` for HBase
* Added `DelegationTokenReceiverRepository` which loads all receivers, and propagates new tokens to them
* New/updated automated tests
## Verifying this change
Existing/new unit/integration tests + manually on [minikube](https://gist.github.com/gaborgsomogyi/ac4f71ead8494da2f5c35265bcb1e885).
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1063420746
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/hadoop/HadoopDelegationTokenReceiver.java:
##########
@@ -19,46 +19,43 @@
package org.apache.flink.runtime.security.token.hadoop;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.runtime.security.token.DelegationTokenContainer;
-import org.apache.flink.util.InstantiationUtil;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.security.token.DelegationTokenReceiver;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/** Delegation token updater functionality. */
+/** Hadoop delegation token receiver base class. */
@Internal
Review Comment:
You could but since it's `@Internal` it's subject to change.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on PR #21604:
URL: https://github.com/apache/flink/pull/21604#issuecomment-1372029173
FYI, the manual test gist updated to use the operator. This way we're not bound to local machine at all.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on PR #21604:
URL: https://github.com/apache/flink/pull/21604#issuecomment-1372436519
Waiting w/ the next push to see a full run whether there are test failures...
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
XComp commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1062636835
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -118,10 +133,13 @@ private Map<String, DelegationTokenProvider> loadProviders() {
provider.serviceName());
}
} catch (Exception | NoClassDefFoundError e) {
- LOG.warn(
+ // The intentional general rule is that if a provider's init method throws exception
+ // then stop the workload
+ LOG.error(
"Failed to initialize delegation token provider {}",
provider.serviceName(),
e);
+ throw new RuntimeException(e);
Review Comment:
Shouldn't we do either logging or throwing the exception and if we decide to throw the exception, adding the log message as the error message might be useful. :thinking: `FlinkRuntimeException` might be useful since it indicates that the exception was actually initiated by Flink code. ...just as an idea.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -58,6 +59,13 @@
@Internal
public class DefaultDelegationTokenManager implements DelegationTokenManager {
+ private static final String PROVIDER_RECEIVER_INCONSISTENCY_ERROR =
+ " There is an inconsistency between loaded delegation token providers and receivers. "
Review Comment:
```suggestion
"There is an inconsistency between loaded delegation token providers and receivers. "
```
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1062465034
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -120,8 +135,8 @@ private Map<String, DelegationTokenProvider> loadProviders() {
} catch (Exception | NoClassDefFoundError e) {
LOG.warn(
Review Comment:
Well, this part has been reworked but the main statement is the same. We're not writing out stacktrace unless we stop the workload.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #21604:
URL: https://github.com/apache/flink/pull/21604#issuecomment-1372031032
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "de992363fb09b0e2ab98bdf7c55cc7e51c55d2a9",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "de992363fb09b0e2ab98bdf7c55cc7e51c55d2a9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* de992363fb09b0e2ab98bdf7c55cc7e51c55d2a9 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1062455873
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/hadoop/HadoopFSDelegationTokenProvider.java:
##########
@@ -69,12 +69,35 @@ public String serviceName() {
@Override
public void init(Configuration configuration) throws Exception {
flinkConfiguration = configuration;
- hadoopConfiguration = HadoopUtils.getHadoopConfiguration(configuration);
- kerberosLoginProvider = new KerberosLoginProvider(configuration);
+ try {
+ hadoopConfiguration = HadoopUtils.getHadoopConfiguration(configuration);
+ kerberosLoginProvider = new KerberosLoginProvider(configuration);
+ } catch (NoClassDefFoundError e) {
+ LOG.info(
+ "Hadoop FS is not available (not packaged with this application): {} : \"{}\".",
+ e.getClass().getSimpleName(),
+ e.getMessage());
+ }
}
@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.
+ *
+ * <p>This null check is required because the Hadoop FS provider is always on classpath but
+ * Hadoop FS jars are optional. Such case configuration is not able to be loaded. This
+ * construct is intended to be removed when HBase provider/receiver pair can be externalized
+ * (namely if a provider/receiver throws an exception then workload must be stopped).
+ */
+ if (hadoopConfiguration == null) {
+ LOG.debug(
+ "Hadoop FS is not available (not packaged with this application), hence no "
+ + "tokens will be acquired.");
+ return false;
+ }
Review Comment:
This is temporary solution as mentioned before.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on PR #21604:
URL: https://github.com/apache/flink/pull/21604#issuecomment-1372031310
I've tested w/ keytab and TGT cache scenario and both worked. I've added expected exception throwing on the receiver side as well (just like it exists on the provider side) to simulate temporary failures.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1062460423
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DelegationTokenReceiverRepository.java:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.InstantiationUtil;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.ServiceLoader;
+
+import static org.apache.flink.runtime.security.token.DefaultDelegationTokenManager.isProviderEnabled;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/** Repository for delegation token receivers. */
+@Internal
+public class DelegationTokenReceiverRepository {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(DelegationTokenReceiverRepository.class);
+
+ private final Configuration configuration;
+
+ @VisibleForTesting final Map<String, DelegationTokenReceiver> delegationTokenReceivers;
+
+ public DelegationTokenReceiverRepository(Configuration configuration) {
+ this.configuration = checkNotNull(configuration, "Flink configuration must not be null");
+ this.delegationTokenReceivers = loadReceivers();
+ }
+
+ private Map<String, DelegationTokenReceiver> loadReceivers() {
+ LOG.info("Loading delegation token receivers");
+
+ ServiceLoader<DelegationTokenReceiver> serviceLoader =
+ ServiceLoader.load(DelegationTokenReceiver.class);
+
+ Map<String, DelegationTokenReceiver> receivers = new HashMap<>();
+ for (DelegationTokenReceiver receiver : serviceLoader) {
+ try {
+ if (isProviderEnabled(configuration, receiver.serviceName())) {
+ receiver.init(configuration);
+ LOG.info(
+ "Delegation token receiver {} loaded and initialized",
+ receiver.serviceName());
+ checkState(
+ !receivers.containsKey(receiver.serviceName()),
+ "Delegation token receiver with service name {} has multiple implementations",
+ receiver.serviceName());
+ receivers.put(receiver.serviceName(), receiver);
+ } else {
+ LOG.info(
+ "Delegation token receiver {} is disabled so not loaded",
+ receiver.serviceName());
+ }
+ } catch (Exception | NoClassDefFoundError e) {
+ // The intentional general rule is that if a receiver's init method throws exception
+ // then stop the workload
+ LOG.error(
+ "Failed to initialize delegation token receiver {}",
+ receiver.serviceName(),
+ e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ LOG.info("Delegation token receivers loaded successfully");
+
+ return receivers;
+ }
+
+ @VisibleForTesting
+ boolean isReceiverLoaded(String serviceName) {
+ return delegationTokenReceivers.containsKey(serviceName);
+ }
+
+ /**
+ * Callback function when new delegation tokens obtained.
+ *
+ * @param containerBytes Serialized form of a DelegationTokenContainer. All the available tokens
+ * will be forwarded to the appropriate {@link DelegationTokenReceiver} based on service
+ * name.
+ */
+ public void onNewTokensObtained(byte[] containerBytes) throws Exception {
+ if (containerBytes == null || containerBytes.length == 0) {
+ throw new IllegalArgumentException("Illegal container tried to be processed");
+ }
+ DelegationTokenContainer container =
+ InstantiationUtil.deserializeObject(
+ containerBytes, DelegationTokenContainer.class.getClassLoader());
+ onNewTokensObtained(container);
+ }
+
+ /**
+ * Callback function when new delegation tokens obtained.
+ *
+ * @param container Serialized form of delegation tokens stored in DelegationTokenContainer. All
+ * the available tokens will be forwarded to the appropriate {@link DelegationTokenReceiver}
+ * based on service name.
+ */
+ public void onNewTokensObtained(DelegationTokenContainer container) throws Exception {
+ LOG.info("New delegation tokens arrived, sending them to receivers");
+ for (Map.Entry<String, byte[]> entry : container.getTokens().entrySet()) {
+ String serviceName = entry.getKey();
+ byte[] tokens = entry.getValue();
+ if (!delegationTokenReceivers.containsKey(serviceName)) {
+ throw new IllegalStateException(
+ "Tokens arrived for service but no receiver found for it: " + serviceName);
+ }
+ try {
+ delegationTokenReceivers.get(serviceName).onNewTokensObtained(tokens);
+ } catch (Exception e) {
+ LOG.warn("Failed to send tokens to delegation token receiver {}", serviceName, e);
Review Comment:
We handle receiver failures as temporary issue and we're going forward to increase resiliency. This way maybe not all of the tokens stored but still we're increasing the availability of the overall system.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1062388936
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -58,6 +59,13 @@
@Internal
public class DefaultDelegationTokenManager implements DelegationTokenManager {
+ private static final String PROVIDER_RECEIVER_INCONSISTENCY_ERROR =
+ " There is an inconsistency between loaded delegation token providers and receivers. "
+ + "One must implement a DelegationTokenProvider and a DelegationTokenReceiver "
+ + "with the same service name and put them together to the classpath to make "
Review Comment:
Fixed.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on PR #21604:
URL: https://github.com/apache/flink/pull/21604#issuecomment-1381090803
Thanks, seems like nowadays we have less and less flaky tests and passed🙂
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1062399853
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -120,8 +135,8 @@ private Map<String, DelegationTokenProvider> loadProviders() {
} catch (Exception | NoClassDefFoundError e) {
LOG.warn(
Review Comment:
@XComp This is the part what I've meant. No more stack trace going to be printed in WARN level. Though we print out the failure for debugging purposes.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pvary commented on pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
pvary commented on PR #21604:
URL: https://github.com/apache/flink/pull/21604#issuecomment-1373553600
Do we have test cases for the specific providers/receivers, like HBase/HadoopFS?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pvary commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
pvary commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1063410971
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/hadoop/HadoopDelegationTokenReceiver.java:
##########
@@ -19,46 +19,43 @@
package org.apache.flink.runtime.security.token.hadoop;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.runtime.security.token.DelegationTokenContainer;
-import org.apache.flink.util.InstantiationUtil;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.security.token.DelegationTokenReceiver;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/** Delegation token updater functionality. */
+/** Hadoop delegation token receiver base class. */
@Internal
Review Comment:
Could I use this class for the HMSDelegationTokenReceiver implementation?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1062653461
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -118,10 +133,13 @@ private Map<String, DelegationTokenProvider> loadProviders() {
provider.serviceName());
}
} catch (Exception | NoClassDefFoundError e) {
- LOG.warn(
+ // The intentional general rule is that if a provider's init method throws exception
+ // then stop the workload
+ LOG.error(
"Failed to initialize delegation token provider {}",
provider.serviceName(),
e);
+ throw new RuntimeException(e);
Review Comment:
Makes sense to use `FlinkRuntimeException`, changed on both places.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on PR #21604:
URL: https://github.com/apache/flink/pull/21604#issuecomment-1373577228
There is an end-to-end test for HDFS but for HBase we don't have anything. Honestly it's horror complex to setup everything within a single process, like KDC, HDFS, HBase, Flink app. It's definitely doable but I personally think that it must be included within the connector when externalized.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1062654455
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -58,6 +59,13 @@
@Internal
public class DefaultDelegationTokenManager implements DelegationTokenManager {
+ private static final String PROVIDER_RECEIVER_INCONSISTENCY_ERROR =
+ " There is an inconsistency between loaded delegation token providers and receivers. "
Review Comment:
Fixed.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1062389299
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -58,6 +59,13 @@
@Internal
public class DefaultDelegationTokenManager implements DelegationTokenManager {
+ private static final String PROVIDER_RECEIVER_INCONSISTENCY_ERROR =
+ " There is an inconsistency between loaded delegation token providers and receivers. "
+ + "One must implement a DelegationTokenProvider and a DelegationTokenReceiver "
+ + "with the same service name and put them together to the classpath to make "
+ + "the system consistent. The mentioned classes are loaded with Java's service "
+ + "loader so the appropriate META-INF registration must not be forgotten.";
Review Comment:
Fixed.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1062456905
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -118,10 +133,13 @@ private Map<String, DelegationTokenProvider> loadProviders() {
provider.serviceName());
}
} catch (Exception | NoClassDefFoundError e) {
- LOG.warn(
+ // The intentional general rule is that if a provider's init method throws exception
+ // then stop the workload
+ LOG.error(
"Failed to initialize delegation token provider {}",
provider.serviceName(),
e);
+ throw new RuntimeException(e);
Review Comment:
I think this will be the most clean statement for later development. The actually existing HBase + HadoopFS providers are exceptional because the're not externalized. The problem w/ them is that the provider/receiver is always on classpath but HadoopFS/HBase dependency is optional.
* The temporary solution is to catch no class found exception within the provider
* The long term solution is to externalize both of them
I've added comments in the respective files [here](https://github.com/apache/flink/pull/21604/files#r1062455662) and [here](https://github.com/apache/flink/pull/21604/files#r1062455873).
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1062462020
##########
flink-runtime/src/test/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManagerTest.java:
##########
@@ -43,37 +46,45 @@ public class DefaultDelegationTokenManagerTest {
@BeforeEach
public void beforeEach() {
ExceptionThrowingDelegationTokenProvider.reset();
+ ExceptionThrowingDelegationTokenReceiver.reset();
}
- @AfterAll
- public static void afterAll() {
+ @AfterEach
+ public void afterEach() {
Review Comment:
This is just a minor improvement which is not related. We need to do reset after each test and not after class.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on PR #21604:
URL: https://github.com/apache/flink/pull/21604#issuecomment-1372028355
cc @gyfora @mbalassi
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1062456905
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -118,10 +133,13 @@ private Map<String, DelegationTokenProvider> loadProviders() {
provider.serviceName());
}
} catch (Exception | NoClassDefFoundError e) {
- LOG.warn(
+ // The intentional general rule is that if a provider's init method throws exception
+ // then stop the workload
+ LOG.error(
"Failed to initialize delegation token provider {}",
provider.serviceName(),
e);
+ throw new RuntimeException(e);
Review Comment:
I think this will be the most clean statement for later development. The actually existing HBase + HadoopFS providers are exceptional because the're not externalized. The problem w/ them is that the provider/receiver is always on classpath but HadoopFS/HBase dependency is optional.
* The temporary solution is to catch no class found exception within the provider and ignore the provider in `delegationTokensRequired` if conf is null
* The long term solution is to externalize both of them and expect provider/receiver + all dependencies together on the classpath
I've added comments in the respective files [here](https://github.com/apache/flink/pull/21604/files#r1062455662) and [here](https://github.com/apache/flink/pull/21604/files#r1062455873).
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1062455662
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/hadoop/HBaseDelegationTokenProvider.java:
##########
@@ -91,6 +90,22 @@ private org.apache.hadoop.conf.Configuration getHBaseConfiguration(Configuration
@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.
+ *
+ * <p>This null check is required because the HBase provider is always on classpath but
+ * HBase jars are optional. Such case configuration is not able to be loaded. This construct
+ * is intended to be removed when HBase provider/receiver pair can be externalized (namely
+ * if a provider/receiver throws an exception then workload must be stopped).
+ */
+ if (hbaseConf == null) {
+ LOG.debug(
+ "HBase is not available (not packaged with this application), hence no "
+ + "tokens will be acquired.");
+ return false;
+ }
Review Comment:
This is temporary solution as mentioned before.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1062464051
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -120,8 +135,8 @@ private Map<String, DelegationTokenProvider> loadProviders() {
} catch (Exception | NoClassDefFoundError e) {
LOG.warn(
"Failed to initialize delegation token provider {}",
- provider.serviceName(),
- e);
+ provider.serviceName());
+ LOG.debug(e.toString());
Review Comment:
This part has beeen changed significantly so I think worth a second look.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -120,8 +135,8 @@ private Map<String, DelegationTokenProvider> loadProviders() {
} catch (Exception | NoClassDefFoundError e) {
LOG.warn(
"Failed to initialize delegation token provider {}",
- provider.serviceName(),
- e);
+ provider.serviceName());
+ LOG.debug(e.toString());
Review Comment:
This part has been changed significantly so I think worth a second look.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mbalassi commented on a diff in pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
mbalassi commented on code in PR #21604:
URL: https://github.com/apache/flink/pull/21604#discussion_r1062378909
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -58,6 +59,13 @@
@Internal
public class DefaultDelegationTokenManager implements DelegationTokenManager {
+ private static final String PROVIDER_RECEIVER_INCONSISTENCY_ERROR =
+ " There is an inconsistency between loaded delegation token providers and receivers. "
+ + "One must implement a DelegationTokenProvider and a DelegationTokenReceiver "
+ + "with the same service name and put them together to the classpath to make "
Review Comment:
nit: put them together -> _add_ them together
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -58,6 +59,13 @@
@Internal
public class DefaultDelegationTokenManager implements DelegationTokenManager {
+ private static final String PROVIDER_RECEIVER_INCONSISTENCY_ERROR =
+ " There is an inconsistency between loaded delegation token providers and receivers. "
+ + "One must implement a DelegationTokenProvider and a DelegationTokenReceiver "
+ + "with the same service name and put them together to the classpath to make "
+ + "the system consistent. The mentioned classes are loaded with Java's service "
+ + "loader so the appropriate META-INF registration must not be forgotten.";
Review Comment:
nit: must not be forgotten -> also needs to be created.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -120,8 +135,8 @@ private Map<String, DelegationTokenProvider> loadProviders() {
} catch (Exception | NoClassDefFoundError e) {
LOG.warn(
"Failed to initialize delegation token provider {}",
- provider.serviceName(),
- e);
+ provider.serviceName());
+ LOG.debug(e.toString());
Review Comment:
Add some explanation to this log message since given the multi-threaded nature of Flink it might not directly follow the previous line.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mbalassi merged pull request #21604: [FLINK-30425][runtime][security] Generalize token receive side
Posted by GitBox <gi...@apache.org>.
mbalassi merged PR #21604:
URL: https://github.com/apache/flink/pull/21604
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org