stevenwarejones commented on code in PR #3785:
URL: https://github.com/apache/parquet-java/pull/3785#discussion_r3989330762
##########
parquet-hadoop/src/main/java/org/apache/parquet/crypto/keytools/KeyToolkit.java:
##########
@@ -281,15 +343,64 @@ public static void rotateMasterKeys(String folderPath,
Configuration hadoopConfi
* @param accessToken access token
*/
public static void removeCacheEntriesForToken(String accessToken) {
- KMS_CLIENT_CACHE_PER_TOKEN.removeCacheEntriesForToken(accessToken);
- KEK_WRITE_CACHE_PER_TOKEN.removeCacheEntriesForToken(accessToken);
- KEK_READ_CACHE_PER_TOKEN.removeCacheEntriesForToken(accessToken);
+ DEFAULT_KMS_CLIENT_CACHE_CONTEXT.removeCacheEntriesForToken(accessToken);
+ synchronized (KMS_CLIENT_FACTORY_REGISTRATIONS) {
+ for (KmsClientCacheContext cacheContext :
KMS_CLIENT_FACTORY_REGISTRATIONS.values()) {
+ cacheContext.removeCacheEntriesForToken(accessToken);
+ }
+ }
}
public static void removeCacheEntriesForAllTokens() {
- KMS_CLIENT_CACHE_PER_TOKEN.clear();
- KEK_WRITE_CACHE_PER_TOKEN.clear();
- KEK_READ_CACHE_PER_TOKEN.clear();
+ DEFAULT_KMS_CLIENT_CACHE_CONTEXT.clear();
+ synchronized (KMS_CLIENT_FACTORY_REGISTRATIONS) {
+ for (KmsClientCacheContext cacheContext :
KMS_CLIENT_FACTORY_REGISTRATIONS.values()) {
+ cacheContext.clear();
+ }
+ }
+ }
+
+ /**
+ * Sets the factory used to create KMS clients for the supplied
configuration.
+ *
+ * <p>The factory is local to this JVM and must be set before constructing a
reader or writer.
+ * Reflection through {@link #KMS_CLIENT_CLASS_PROPERTY_NAME} remains the
default for other
+ * configurations. Clients returned by the factory are initialized and
cached by {@link
+ * KeyToolkit} in the same way as reflectively constructed clients. The KMS
client and key
+ * encryption key caches are isolated from registrations for other
configurations.
+ *
+ * <p>The association is not serialized, and configuration copies must
register their own
+ * factory. The caller must invoke {@link
#removeKmsClientFactory(Configuration)} after all
+ * readers and writers using the configuration have closed. Replacing a
factory clears the
+ * previous registration and its caches.
+ *
+ * @param configuration Hadoop configuration associated with the factory
+ * @param kmsClientFactory factory used to create KMS clients
+ */
+ public static void setKmsClientFactory(Configuration configuration,
KmsClientFactory kmsClientFactory) {
+ Objects.requireNonNull(configuration, "configuration");
+ Objects.requireNonNull(kmsClientFactory, "kmsClientFactory");
+ KmsClientCacheContext previous =
+ KMS_CLIENT_FACTORY_REGISTRATIONS.put(configuration, new
KmsClientCacheContext(kmsClientFactory));
Review Comment:
Good point. That would be a copied `Configuration`, so the identity-based
registration would not be found.
One option is to store an opaque registration ID in the configuration. That
would support copies within the same JVM, but it could not carry the live
factory into another JVM. Transparent cross-JVM support would require
Spark/Flink-side integration to register or create the factory on each worker.
Would same-JVM copy support, with an explicit error when no local factory is
registered, be the right scope here? Or would you suggest a different approach?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
