This is an automated email from the ASF dual-hosted git repository.
difin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new abf94a59937 HIVE-29359: Support Credential Vending in Hive Iceberg
REST Catalog Client (#6474)
abf94a59937 is described below
commit abf94a5993763523f5249df69dc7ef5ca35bff89
Author: Dmitriy Fingerman <[email protected]>
AuthorDate: Wed Jul 15 18:48:05 2026 -0400
HIVE-29359: Support Credential Vending in Hive Iceberg REST Catalog Client
(#6474)
* HIVE-29359: Support S3 credential vending in Hive Iceberg REST catalog
client.
---------
Co-authored-by: Dmitriy Fingerman <[email protected]>
Co-authored-by: Dmitriy Fingerman <[email protected]>
---
.../java/org/apache/hadoop/hive/conf/HiveConf.java | 5 +
.../iceberg/hive/IcebergCatalogProperties.java | 15 +
.../hive/client/RestAccessDelegationMode.java | 80 ++
.../hive/client/TestRestAccessDelegationMode.java | 75 ++
iceberg/iceberg-handler/pom.xml | 8 +
.../main/java/org/apache/iceberg/mr/Catalogs.java | 15 +-
.../org/apache/iceberg/mr/InputFormatConfig.java | 12 +
.../mr/hive/HiveIcebergOutputCommitter.java | 8 +-
.../iceberg/mr/hive/HiveIcebergStorageHandler.java | 100 ++-
.../org/apache/iceberg/mr/hive/HiveTableUtil.java | 7 +-
.../mr/hive/IcebergVendedCredentialUtil.java | 499 ++++++++++++
.../iceberg/mr/hive/writer/HiveIcebergWriter.java | 3 +
.../mr/hive/writer/HiveIcebergWriterBase.java | 5 +
.../iceberg/mr/mapreduce/IcebergInputFormat.java | 2 +
.../mr/hive/TestIcebergVendedCredentialUtil.java | 885 +++++++++++++++++++++
.../positive/iceberg_rest_catalog_gravitino.q | 14 +-
.../llap/iceberg_rest_catalog_gravitino.q.out | 14 +-
itests/qtest-iceberg/pom.xml | 26 +
...bergRESTCatalogGravitinoLlapLocalCliDriver.java | 257 +++---
.../gravitino-s3-vended-oauth-template.conf | 37 +
itests/util/pom.xml | 4 +
.../hadoop/hive/cli/OzoneS3GatewayContainers.java | 203 +++++
.../apache/hadoop/hive/ql/exec/FetchOperator.java | 7 +
.../hadoop/hive/ql/exec/FileSinkOperator.java | 10 +
.../org/apache/hadoop/hive/ql/exec/Utilities.java | 8 +-
25 files changed, 2126 insertions(+), 173 deletions(-)
diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
index b10509b85f1..65ad8a82bbe 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -5600,6 +5600,11 @@ public static enum ConfVars {
+ ",fs.s3a.access.key"
+ ",fs.s3a.secret.key"
+ ",fs.s3a.proxy.password"
+ + ",iceberg.vended.storage.credentials"
+ // Iceberg FileIO vended credential keys (S3FileIOProperties)
+ + ",s3.access-key-id"
+ + ",s3.secret-access-key"
+ + ",s3.session-token"
+ ",dfs.adls.oauth2.credential"
+ ",fs.adl.oauth2.credential"
+ ",fs.azure.account.oauth2.client.secret"
diff --git
a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/IcebergCatalogProperties.java
b/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/IcebergCatalogProperties.java
index 424f8e10c35..6d7f047c05d 100644
---
a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/IcebergCatalogProperties.java
+++
b/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/IcebergCatalogProperties.java
@@ -27,6 +27,7 @@
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.hive.client.RestAccessDelegationMode;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
public class IcebergCatalogProperties {
@@ -38,6 +39,7 @@ public class IcebergCatalogProperties {
public static final String ICEBERG_HADOOP_TABLE_NAME =
"location_based_table";
public static final String ICEBERG_DEFAULT_CATALOG_NAME = "default_iceberg";
public static final String NO_CATALOG_TYPE = "no catalog";
+ public static final String REST_ACCESS_DELEGATION_HEADER_PROPERTY =
"header.X-Iceberg-Access-Delegation";
private IcebergCatalogProperties() {
@@ -104,6 +106,19 @@ public static String catalogPropertyConfigKey(String
catalogName, String catalog
return String.format("%s%s.%s", CATALOG_CONFIG_PREFIX, catalogName,
catalogProperty);
}
+ /**
+ * Returns true when the catalog is configured to request REST vended
storage credentials via
+ * {@link #REST_ACCESS_DELEGATION_HEADER_PROPERTY}.
+ */
+ public static boolean requestsVendedCredentials(String catalogName,
Configuration conf) {
+ if (conf == null || StringUtils.isEmpty(catalogName)) {
+ return false;
+ }
+ String headerValue =
+ conf.get(catalogPropertyConfigKey(catalogName,
REST_ACCESS_DELEGATION_HEADER_PROPERTY));
+ return
RestAccessDelegationMode.headerRequestsVendedCredentials(headerValue);
+ }
+
/**
* Return the catalog type based on the catalog name.
* <p>
diff --git
a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/client/RestAccessDelegationMode.java
b/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/client/RestAccessDelegationMode.java
new file mode 100644
index 00000000000..dd2d435a9b0
--- /dev/null
+++
b/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/client/RestAccessDelegationMode.java
@@ -0,0 +1,80 @@
+/*
+ * 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.iceberg.hive.client;
+
+import java.util.Arrays;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.iceberg.hive.IcebergCatalogProperties;
+
+/**
+ * Values for the Iceberg REST catalog {@code X-Iceberg-Access-Delegation}
request header. The header
+ * accepts a comma-separated list of these modes; configure via
+ * {@link IcebergCatalogProperties#REST_ACCESS_DELEGATION_HEADER_PROPERTY}.
+ *
+ * @see <a
href="https://github.com/apache/iceberg/blob/main/open-api/rest-catalog-open-api.yaml";>REST
catalog spec</a>
+ */
+public enum RestAccessDelegationMode {
+ VENDED_CREDENTIALS("vended-credentials"),
+ REMOTE_SIGNING("remote-signing");
+
+ private final String modeName;
+
+ RestAccessDelegationMode(String modeName) {
+ this.modeName = modeName;
+ }
+
+ /** Spec-defined header token for this delegation mode. */
+ public String modeName() {
+ return modeName;
+ }
+
+ /** Comma-separated list suitable for {@link
IcebergCatalogProperties#REST_ACCESS_DELEGATION_HEADER_PROPERTY}. */
+ public static String toHeaderValue(RestAccessDelegationMode... modes) {
+ return
Arrays.stream(modes).map(RestAccessDelegationMode::modeName).collect(Collectors.joining(","));
+ }
+
+ /** Parses a single mode name (case-insensitive); throws if unknown. */
+ public static RestAccessDelegationMode fromModeName(String modeName) {
+ for (RestAccessDelegationMode mode : values()) {
+ if (mode.modeName.equalsIgnoreCase(modeName.trim())) {
+ return mode;
+ }
+ }
+ throw new IllegalArgumentException(
+ String.format(
+ "Unknown REST access delegation mode: %s. Valid values are: %s",
+ modeName,
+
Arrays.stream(values()).map(RestAccessDelegationMode::modeName).collect(Collectors.joining(",
"))));
+ }
+
+ /** Returns true if the {@code X-Iceberg-Access-Delegation} header value
includes vended credentials. */
+ public static boolean headerRequestsVendedCredentials(String headerValue) {
+ if (StringUtils.isBlank(headerValue)) {
+ return false;
+ }
+ for (String token : headerValue.split(",")) {
+ if (VENDED_CREDENTIALS.modeName.equalsIgnoreCase(token.trim())) {
+ return true;
+ }
+ }
+ return false;
+ }
+}
diff --git
a/iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/client/TestRestAccessDelegationMode.java
b/iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/client/TestRestAccessDelegationMode.java
new file mode 100644
index 00000000000..dce5106b5cf
--- /dev/null
+++
b/iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/client/TestRestAccessDelegationMode.java
@@ -0,0 +1,75 @@
+/*
+ * 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.iceberg.hive.client;
+
+import org.apache.iceberg.hive.IcebergCatalogProperties;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class TestRestAccessDelegationMode {
+
+ @Test
+ void vendedCredentialsModeName() {
+
assertThat(RestAccessDelegationMode.VENDED_CREDENTIALS.modeName()).isEqualTo("vended-credentials");
+ }
+
+ @Test
+ void toHeaderValueJoinsModes() {
+ assertThat(
+ RestAccessDelegationMode.toHeaderValue(
+ RestAccessDelegationMode.VENDED_CREDENTIALS,
RestAccessDelegationMode.REMOTE_SIGNING))
+ .isEqualTo("vended-credentials,remote-signing");
+ }
+
+ @Test
+ void fromModeNameParsesCaseInsensitive() {
+ assertThat(RestAccessDelegationMode.fromModeName("VENDED-CREDENTIALS"))
+ .isEqualTo(RestAccessDelegationMode.VENDED_CREDENTIALS);
+ }
+
+ @Test
+ void fromModeNameRejectsUnknown() {
+ assertThatThrownBy(() -> RestAccessDelegationMode.fromModeName("unknown"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("unknown");
+ }
+
+ @Test
+ void headerRequestsVendedCredentials() {
+
assertThat(RestAccessDelegationMode.headerRequestsVendedCredentials(null)).isFalse();
+
assertThat(RestAccessDelegationMode.headerRequestsVendedCredentials("remote-signing")).isFalse();
+
assertThat(RestAccessDelegationMode.headerRequestsVendedCredentials("vended-credentials")).isTrue();
+
assertThat(RestAccessDelegationMode.headerRequestsVendedCredentials("VENDED-CREDENTIALS,remote-signing"))
+ .isTrue();
+ }
+
+ @Test
+ void requestsVendedCredentialsFromConfiguration() {
+ org.apache.hadoop.conf.Configuration conf = new
org.apache.hadoop.conf.Configuration();
+ assertThat(IcebergCatalogProperties.requestsVendedCredentials("ice01",
conf)).isFalse();
+
+ conf.set(
+ "iceberg.catalog.ice01.header.X-Iceberg-Access-Delegation",
+ RestAccessDelegationMode.VENDED_CREDENTIALS.modeName());
+ assertThat(IcebergCatalogProperties.requestsVendedCredentials("ice01",
conf)).isTrue();
+ }
+}
diff --git a/iceberg/iceberg-handler/pom.xml b/iceberg/iceberg-handler/pom.xml
index ef20ab1aab4..f00415ecbe2 100644
--- a/iceberg/iceberg-handler/pom.xml
+++ b/iceberg/iceberg-handler/pom.xml
@@ -34,6 +34,14 @@
<artifactId>hive-iceberg-catalog</artifactId>
<optional>true</optional>
</dependency>
+ <dependency>
+ <!-- compile-time only, for S3FileIOProperties/AwsClientProperties
string constants
+ (inlined by javac); the classes are not referenced at runtime -->
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-aws</artifactId>
+ <version>${iceberg.version}</version>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/Catalogs.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/Catalogs.java
index 9a416f8277a..7b6f0810461 100644
--- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/Catalogs.java
+++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/Catalogs.java
@@ -229,16 +229,27 @@ public static void renameTable(Configuration conf,
Properties props, TableIdenti
}
static Optional<Catalog> loadCatalog(Configuration conf, String catalogName)
{
- String catalogType = IcebergCatalogProperties.getCatalogType(conf,
catalogName);
+ String resolvedName = catalogName != null ? catalogName :
configuredDefaultCatalogName(conf);
+ String catalogType = IcebergCatalogProperties.getCatalogType(conf,
resolvedName);
if (NO_CATALOG_TYPE.equalsIgnoreCase(catalogType)) {
return Optional.empty();
} else {
- String name = catalogName == null ? ICEBERG_DEFAULT_CATALOG_NAME :
catalogName;
+ String name = resolvedName == null ? ICEBERG_DEFAULT_CATALOG_NAME :
resolvedName;
return Optional.of(CatalogUtil.buildIcebergCatalog(name,
IcebergCatalogProperties.getCatalogProperties(conf, name), conf));
}
}
+ /**
+ * Tables created without an explicit catalog in their properties belong to
the session default
+ * catalog when the configuration defines one (e.g. a REST catalog); returns
{@code null}
+ * otherwise, preserving the legacy default-catalog behavior.
+ */
+ private static String configuredDefaultCatalogName(Configuration conf) {
+ String defaultCatalogName = IcebergCatalogProperties.getCatalogName(conf);
+ return IcebergCatalogProperties.getCatalogType(conf, defaultCatalogName)
!= null ? defaultCatalogName : null;
+ }
+
/**
* Parse the table schema from the properties
* @param props the controlling properties
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/InputFormatConfig.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/InputFormatConfig.java
index 34ba9f2404c..46c281ce0f0 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/InputFormatConfig.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/InputFormatConfig.java
@@ -80,6 +80,18 @@ private InputFormatConfig() {
public static final String CATALOG_CONFIG_PREFIX = "iceberg.catalog.";
+ /**
+ * Base64-serialized list of {@link org.apache.iceberg.io.StorageCredential}
for Tez/LLAP executors.
+ * Stored in {@code TableDesc#jobSecrets} (HIVE-20651), not in job
properties. REST catalogs mint
+ * credentials per table, so entries are keyed via {@link
#vendedCredentialsKey(String)}; this
+ * constant is the key prefix (and the {@code hive.conf.hidden.list} entry,
matched by prefix).
+ */
+ public static final String VENDED_STORAGE_CREDENTIALS =
"iceberg.vended.storage.credentials";
+
+ public static String vendedCredentialsKey(String tableName) {
+ return VENDED_STORAGE_CREDENTIALS + "." + tableName;
+ }
+
public static final String SORT_ORDER = "sort.order";
public static final String SORT_COLUMNS = "sort.columns";
public static final String ZORDER = "ZORDER";
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
index 9fabe10488b..2a870137908 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
@@ -175,9 +175,12 @@ public void commitTask(TaskAttemptContext originalContext)
throws IOException {
referencedDataFiles.addAll(files.referencedDataFiles());
rewrittenDeleteFiles.addAll(files.rewrittenDeleteFiles());
}
+ // the writers' FileIO carries the table storage credentials;
the table deserialized
+ // from the committer conf does not, since that conf is built
at DAG time (secret-free)
+ FileIO io = writers.get(output).get(0).io();
createFileForCommit(
new FilesForCommit(dataFiles, deleteFiles,
replacedDataFiles, referencedDataFiles,
- rewrittenDeleteFiles, mergedPaths),
fileForCommitLocation, table.io());
+ rewrittenDeleteFiles, mergedPaths),
fileForCommitLocation, io);
} else {
LOG.info("CommitTask found no writer for specific table: {},
attemptID: {}", output, attemptID);
createFileForCommit(FilesForCommit.empty(),
fileForCommitLocation, table.io());
@@ -313,6 +316,9 @@ private static Multimap<OutputTable, JobContext>
collectOutputs(List<JobContext>
.orElseGet(() ->
HiveTableUtil.deserializeTable(jobContext.getJobConf(), output));
if (table != null) {
String catalogName = catalogName(jobContext.getJobConf(), output);
+ // pass the per-table catalog name: the job-level CATALOG_NAME is
absent in the DAG conf,
+ // and without it the session endpoint override is not applied to
the commit-side FileIO
+ IcebergVendedCredentialUtil.applyFromJobConf(table, catalogName,
jobContext.getJobConf());
outputs.put(new OutputTable(catalogName, output, table), jobContext);
} else {
LOG.info("Found no table object in QueryState or conf for: {}.
Skipping job commit.", output);
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
index 0fa610e6f3d..2875890efdc 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
@@ -284,6 +284,21 @@ public HiveAuthorizationProvider
getAuthorizationProvider() {
return null;
}
+ @Override
+ public void configureInputJobCredentials(TableDesc tableDesc, Map<String,
String> secrets) {
+ if
(!IcebergVendedCredentialUtil.requestsVendedCredentials(tableDesc.getProperties(),
conf)) {
+ return;
+ }
+ try {
+ Table table =
+
IcebergVendedCredentialUtil.getTableWithVendedCredentials(tableDesc.getProperties(),
conf);
+ String catalogName =
tableDesc.getProperties().getProperty(InputFormatConfig.CATALOG_NAME);
+ IcebergVendedCredentialUtil.propagateToJob(table, catalogName, null,
secrets, conf);
+ } catch (NoSuchTableException ex) {
+ // Table may not exist yet for CTAS; credentials will not be available.
+ }
+ }
+
@Override
public void configureInputJobProperties(TableDesc tableDesc, Map<String,
String> map) {
overlayTableProperties(conf, tableDesc, map);
@@ -335,32 +350,9 @@ public void commitJob(JobContext originalContext) {
@Override
public void configureJobConf(TableDesc tableDesc, JobConf jobConf) {
setCommonJobConf(jobConf);
- if (tableDesc != null && tableDesc.getProperties() != null &&
- tableDesc.getProperties().get(InputFormatConfig.OPERATION_TYPE_PREFIX
+ tableDesc.getTableName()) != null) {
- String tableName = tableDesc.getTableName();
- String opKey = InputFormatConfig.OPERATION_TYPE_PREFIX + tableName;
- // set operation type into job conf too
- jobConf.set(opKey, tableDesc.getProperties().getProperty(opKey));
- Preconditions.checkArgument(!tableName.contains(TABLE_NAME_SEPARATOR),
- "Can not handle table " + tableName + ". Its name contains '" +
TABLE_NAME_SEPARATOR + "'");
- if
(HiveCustomStorageHandlerUtils.getWriteOperation(tableDesc.getProperties()::getProperty,
tableName) != null) {
- HiveCustomStorageHandlerUtils.setWriteOperation(jobConf, tableName,
- Operation.valueOf(tableDesc.getProperties().getProperty(
- HiveCustomStorageHandlerUtils.WRITE_OPERATION_CONFIG_PREFIX +
tableName)));
- }
- boolean isMergeTaskEnabled = Boolean.parseBoolean(tableDesc.getProperty(
- HiveCustomStorageHandlerUtils.MERGE_TASK_ENABLED + tableName));
- if (isMergeTaskEnabled) {
- HiveCustomStorageHandlerUtils.setMergeTaskEnabled(jobConf, tableName,
true);
- }
- String tables = jobConf.get(InputFormatConfig.OUTPUT_TABLES);
- tables = (tables == null) ? tableName : tables + TABLE_NAME_SEPARATOR +
tableName;
- jobConf.set(InputFormatConfig.OUTPUT_TABLES, tables);
-
- String catalogName =
tableDesc.getProperties().getProperty(InputFormatConfig.CATALOG_NAME);
- if (catalogName != null) {
- jobConf.set(InputFormatConfig.TABLE_CATALOG_PREFIX + tableName,
catalogName);
- }
+ configureOutputTableJobConf(tableDesc, jobConf);
+ if
(IcebergVendedCredentialUtil.requestsVendedCredentials(tableDesc.getProperties(),
conf)) {
+ IcebergVendedCredentialUtil.refreshVendedCredentialsIfMissing(tableDesc,
jobConf, conf);
}
try {
if (!jobConf.getBoolean(ConfVars.HIVE_IN_TEST_IDE.varname, false)) {
@@ -373,6 +365,37 @@ public void configureJobConf(TableDesc tableDesc, JobConf
jobConf) {
}
}
+ private static void configureOutputTableJobConf(TableDesc tableDesc, JobConf
jobConf) {
+ if (tableDesc == null || tableDesc.getProperties() == null ||
+ tableDesc.getProperties().get(InputFormatConfig.OPERATION_TYPE_PREFIX
+ tableDesc.getTableName()) == null) {
+ return;
+ }
+ String tableName = tableDesc.getTableName();
+ String opKey = InputFormatConfig.OPERATION_TYPE_PREFIX + tableName;
+ // set operation type into job conf too
+ jobConf.set(opKey, tableDesc.getProperties().getProperty(opKey));
+ Preconditions.checkArgument(!tableName.contains(TABLE_NAME_SEPARATOR),
+ "Can not handle table " + tableName + ". Its name contains '" +
TABLE_NAME_SEPARATOR + "'");
+ if
(HiveCustomStorageHandlerUtils.getWriteOperation(tableDesc.getProperties()::getProperty,
tableName) != null) {
+ HiveCustomStorageHandlerUtils.setWriteOperation(jobConf, tableName,
+ Operation.valueOf(tableDesc.getProperties().getProperty(
+ HiveCustomStorageHandlerUtils.WRITE_OPERATION_CONFIG_PREFIX +
tableName)));
+ }
+ boolean isMergeTaskEnabled = Boolean.parseBoolean(tableDesc.getProperty(
+ HiveCustomStorageHandlerUtils.MERGE_TASK_ENABLED + tableName));
+ if (isMergeTaskEnabled) {
+ HiveCustomStorageHandlerUtils.setMergeTaskEnabled(jobConf, tableName,
true);
+ }
+ String tables = jobConf.get(InputFormatConfig.OUTPUT_TABLES);
+ tables = (tables == null) ? tableName : tables + TABLE_NAME_SEPARATOR +
tableName;
+ jobConf.set(InputFormatConfig.OUTPUT_TABLES, tables);
+
+ String catalogName =
tableDesc.getProperties().getProperty(InputFormatConfig.CATALOG_NAME);
+ if (catalogName != null) {
+ jobConf.set(InputFormatConfig.TABLE_CATALOG_PREFIX + tableName,
catalogName);
+ }
+ }
+
@Override
public boolean directInsert() {
return true;
@@ -1681,13 +1704,12 @@ static void overlayTableProperties(Configuration
configuration, TableDesc tableD
PartitionSpec spec;
String bytes;
try {
- Table table = IcebergTableUtil.getTable(configuration, props);
+ boolean isVendedCredentials =
+ IcebergVendedCredentialUtil.requestsVendedCredentials(props,
configuration);
+ Table table = isVendedCredentials ?
+ IcebergVendedCredentialUtil.getTableWithVendedCredentials(props,
configuration) :
+ IcebergTableUtil.getTable(configuration, props);
location = table.location();
- // set table format-version and write-mode information from tableDesc
- bytes = HiveTableUtil.serializeTable(table, configuration, props,
- ImmutableList.of(
- TableProperties.FORMAT_VERSION,
- TableProperties.DELETE_MODE, TableProperties.UPDATE_MODE,
TableProperties.MERGE_MODE));
schema = table.schema();
spec = table.spec();
@@ -1704,6 +1726,20 @@ static void overlayTableProperties(Configuration
configuration, TableDesc tableD
}
}
+ if (isVendedCredentials) {
+ String catalogName = props.getProperty(InputFormatConfig.CATALOG_NAME);
+ IcebergVendedCredentialUtil.propagateToJob(table, catalogName, map,
null, configuration);
+ // Serializing the table as-is would embed its FileIO, and with it the
vended credentials,
+ // into plain job properties: ship a copy over a secret-free FileIO
instead; executors
+ // restore the credentials from the Credentials channel
(setCredentials).
+ table = IcebergVendedCredentialUtil.secretFreeCopy(table,
configuration);
+ }
+ // set table format-version and write-mode information from tableDesc
+ bytes = HiveTableUtil.serializeTable(table, configuration, props,
+ ImmutableList.of(
+ TableProperties.FORMAT_VERSION,
+ TableProperties.DELETE_MODE, TableProperties.UPDATE_MODE,
TableProperties.MERGE_MODE));
+
} catch (NoSuchTableException ex) {
if (!HiveTableUtil.isCtas(props)) {
throw ex;
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveTableUtil.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveTableUtil.java
index a129266f6db..94e7217cefc 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveTableUtil.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveTableUtil.java
@@ -246,6 +246,7 @@ public static Table deserializeTable(Configuration config,
String name) {
table = readTableObjectFromFile(location, config);
}
checkAndSetIoConfig(config, table);
+ IcebergVendedCredentialUtil.applyFromJobConf(table, config);
// For intra-txn read-after-write: if a metadata file was written for
uncommitted in-txn state,
// reconstruct a BaseTable from it so the Tez side sees changes from prior
statements.
@@ -362,8 +363,10 @@ static void cleanupTableObjectFile(String location,
Configuration configuration)
try {
FileSystem fs = Util.getFs(toDelete, configuration);
fs.delete(toDelete, true);
- } catch (IOException ex) {
- throw new UncheckedIOException(ex);
+ } catch (IOException e) {
+ // best effort: never fail the commit over the temp table-object file;
with catalog-vended
+ // credentials the Hadoop filesystem has no keys and the query-scoped
file is left behind
+ LOG.warn("Could not remove temp table object file: {}", filePath, e);
}
}
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergVendedCredentialUtil.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergVendedCredentialUtil.java
new file mode 100644
index 00000000000..8d110f40f70
--- /dev/null
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergVendedCredentialUtil.java
@@ -0,0 +1,499 @@
+/*
+ * 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.iceberg.mr.hive;
+
+import java.util.Base64;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.conf.HiveConfUtil;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.iceberg.BaseMetadataTable;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.StaticTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.aws.AwsClientProperties;
+import org.apache.iceberg.aws.s3.S3FileIOProperties;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hive.IcebergCatalogProperties;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.StorageCredential;
+import org.apache.iceberg.io.SupportsStorageCredentials;
+import org.apache.iceberg.mr.InputFormatConfig;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.SerializationUtil;
+
+/**
+ * Propagates vended storage credentials from an Iceberg {@link Table}'s
{@link FileIO} to Hive job
+ * configuration so Tez/LLAP executors can access object storage without
static catalog keys.
+ *
+ * <p>Limitation: credentials are minted at compile/job launch and are not
refreshed — work that
+ * outlives the vended token lifetime fails authentication.
+ */
+public final class IcebergVendedCredentialUtil {
+
+ private IcebergVendedCredentialUtil() {
+ }
+
+ /**
+ * Copies vended credentials from the table FileIO into Hive job
configuration.
+ *
+ * <p>Follows the HIVE-20651 split used by {@code JdbcStorageHandler}:
sensitive values (including
+ * the serialized {@link StorageCredential} list) go to {@code jobSecrets};
non-secret config such
+ * as endpoint and path-style access go to {@code jobProperties}.
+ *
+ * @param table loaded Iceberg table; its catalog-qualified name keys the
credentials blob,
+ * since REST catalogs mint credentials per table (the table pointer
carries the same
+ * name, so executors resolve the matching key)
+ * @param catalogName Hive catalog name ({@link
InputFormatConfig#CATALOG_NAME})
+ * @param jobProperties Tez/MR non-secret job properties; may be {@code null}
+ * @param jobSecrets sensitive keys and serialized credentials; may be
{@code null}
+ * @param conf session conf used to preserve host-side endpoint overrides
+ */
+ public static void propagateToJob(Table table, String catalogName,
Map<String, String> jobProperties,
+ Map<String, String> jobSecrets, Configuration conf) {
+
+ List<StorageCredential> credentials =
+ withConfigurationOverrides(catalogName, extractCredentials(table),
conf);
+
+ if (credentials.isEmpty()) {
+ return;
+ }
+
+ if (jobSecrets != null) {
+ jobSecrets.put(
+ InputFormatConfig.vendedCredentialsKey(table.name()),
+ serializeToSingleLineBase64(Lists.newArrayList(credentials)));
+ }
+
+ for (StorageCredential credential : credentials) {
+ addCredentialEntries(catalogName, credential, jobProperties, jobSecrets,
conf);
+ }
+ }
+
+ /**
+ * Writes each key in one vended {@link StorageCredential} into job
configuration.
+ *
+ * <p>Derives the S3 bucket from {@link StorageCredential#prefix()} and
delegates to
+ * {@link #addCredentialEntry} for every entry in {@link
StorageCredential#config()}, which maps
+ * Iceberg keys to catalog-level and per-bucket S3A job properties or
secrets.
+ */
+ private static void addCredentialEntries(String catalogName,
StorageCredential credential,
+ Map<String, String> jobProperties, Map<String, String> jobSecrets,
Configuration conf) {
+
+ String bucket = bucketFromPrefix(credential.prefix());
+ for (Map.Entry<String, String> entry : credential.config().entrySet()) {
+ addCredentialEntry(
+ catalogName, bucket, entry.getKey(), entry.getValue(),
jobProperties, jobSecrets, conf);
+ }
+ }
+
+ /**
+ * Routes one Iceberg credential config entry into job properties or secrets.
+ *
+ * <p>Skips blank values, applies session catalog overrides via {@link
#resolveCredentialValue},
+ * then sends non-secret keys (endpoint, path-style access, etc.) to {@code
jobProperties} and
+ * secret keys (access key, secret key, session token) to {@code
jobSecrets}. Either map may be
+ * {@code null} when {@link #propagateToJob} is called for only properties
or only secrets.
+ */
+ private static void addCredentialEntry(String catalogName, String bucket,
String icebergKey, String value,
+ Map<String, String> jobProperties, Map<String, String> jobSecrets,
Configuration conf) {
+
+ if (StringUtils.isBlank(value)) {
+ return;
+ }
+ String resolvedValue = resolveCredentialValue(catalogName, icebergKey,
value, conf);
+
+ if (jobProperties != null && !isSecretKey(icebergKey, conf)) {
+ addNonSecretCredentialEntry(catalogName, bucket, icebergKey,
resolvedValue, jobProperties);
+ }
+
+ if (jobSecrets != null && isSecretKey(icebergKey, conf)) {
+ addSecretCredentialEntry(bucket, icebergKey, resolvedValue, jobSecrets);
+ }
+ }
+
+ /**
+ * Adds one non-secret vended value to {@code jobProperties} for Iceberg and
Hadoop S3A.
+ *
+ * <p>When {@code catalogName} is set, writes {@code
iceberg.catalog.<catalog>.<key>}.
+ * When {@code bucket} is set, also writes the matching {@code
fs.s3a.bucket.<bucket>.*}
+ * key if {@link #toS3aBucketProperty} maps the Iceberg key.
+ */
+ private static void addNonSecretCredentialEntry(String catalogName, String
bucket, String icebergKey, String value,
+ Map<String, String> jobProperties) {
+
+ if (catalogName != null) {
+ String catalogConfigKey =
+ IcebergCatalogProperties.catalogPropertyConfigKey(catalogName,
icebergKey);
+ jobProperties.putIfAbsent(catalogConfigKey, value);
+ }
+
+ if (bucket != null) {
+ String s3aKey = toS3aBucketProperty(bucket, icebergKey);
+ if (s3aKey != null) {
+ jobProperties.putIfAbsent(s3aKey, value);
+ }
+ }
+ }
+
+ /** Writes Hadoop S3A per-bucket keys only; Iceberg secrets are carried in
the serialized blob.
+ * First table wins on a shared bucket, matching the non-secret entries, so
an endpoint and its
+ * key always come from the same credential. */
+ private static void addSecretCredentialEntry(String bucket, String
icebergKey, String value,
+ Map<String, String> jobSecrets) {
+ if (bucket != null) {
+ String s3aSecretKey = toS3aBucketProperty(bucket, icebergKey);
+ if (s3aSecretKey != null) {
+ jobSecrets.putIfAbsent(s3aSecretKey, value);
+ }
+ }
+ }
+
+ /**
+ * Applies vended credentials to the table FileIO, merging session/catalog
conf overrides (e.g. S3 endpoint).
+ * Used on executors after deserialization and on HS2 commit when the table
is taken from query state.
+ */
+ public static void applyFromJobConf(Table table, Configuration conf) {
+ applyFromJobConf(table, conf != null ?
conf.get(InputFormatConfig.CATALOG_NAME) : null, conf);
+ }
+
+ /** Variant for callers that resolve the catalog name per table (HS2 commit
paths, where the
+ * job-level {@link InputFormatConfig#CATALOG_NAME} is not set). */
+ public static void applyFromJobConf(Table table, String catalogName,
Configuration conf) {
+ if (table == null || conf == null) {
+ return;
+ }
+
+ if (shouldSkipApplyFromJobConf(table.name(), catalogName, conf)) {
+ return;
+ }
+
+ FileIO io = table.io();
+ if (!(io instanceof SupportsStorageCredentials credentialIo)) {
+ return;
+ }
+
+ List<StorageCredential> credentials = resolveCredentialsForApply(table,
credentialIo, conf);
+ if (!credentials.isEmpty()) {
+ credentialIo.setCredentials(withConfigurationOverrides(catalogName,
credentials, conf));
+ }
+ }
+
+ /**
+ * Returns true when the job carries no serialized vended credentials and
the catalog is not
+ * configured for credential vending (no {@code vended-credentials} REST
delegation header).
+ * Otherwise apply may restore credentials from the job conf or from the
table FileIO.
+ */
+ private static boolean shouldSkipApplyFromJobConf(String tableName, String
catalogName, Configuration conf) {
+ return
StringUtils.isBlank(conf.get(InputFormatConfig.vendedCredentialsKey(tableName)))
&&
+ !IcebergCatalogProperties.requestsVendedCredentials(catalogName, conf);
+ }
+
+ /**
+ * Chooses which vended credentials {@link #applyFromJobConf} should install
on the FileIO.
+ *
+ * <p>Uses the first non-empty source: credentials already on the FileIO
(typical on HS2 after
+ * table load), else the base64 list from {@link
InputFormatConfig#VENDED_STORAGE_CREDENTIALS} on
+ * the task {@code conf} (restored from the HIVE-20651 Credentials channel
into table properties
+ * by {@code Utilities#copyJobSecretToTableProperties} and copied to the
task-local conf), else
+ * {@link #extractCredentials(Table)} from the table (including FileIO
property fallbacks).
+ */
+ private static List<StorageCredential> resolveCredentialsForApply(
+ Table table, SupportsStorageCredentials credentialIo, Configuration
conf) {
+
+ List<StorageCredential> credentials = credentialIo.credentials();
+ if (credentials != null && !credentials.isEmpty()) {
+ return credentials;
+ }
+ String serialized =
conf.get(InputFormatConfig.vendedCredentialsKey(table.name()));
+ if (StringUtils.isNotBlank(serialized)) {
+ return SerializationUtil.deserializeFromBase64(serialized);
+ }
+ return extractCredentials(table);
+ }
+
+ /**
+ * Returns true when the table catalog is configured for REST vended storage
credentials.
+ */
+ static boolean requestsVendedCredentials(Properties properties,
Configuration configuration) {
+ if (properties == null) {
+ return false;
+ }
+ return IcebergCatalogProperties.requestsVendedCredentials(
+ properties.getProperty(InputFormatConfig.CATALOG_NAME), configuration);
+ }
+
+ /**
+ * Loads a table and, if needed, bypasses the query-level cache so REST
vended credentials are present on the FileIO.
+ * When vended credentials are not requested for the catalog, returns the
cached table without an extra load.
+ */
+ static Table getTableWithVendedCredentials(Properties properties,
Configuration configuration) {
+ Table table = IcebergTableUtil.getTable(configuration, properties);
+ if (requestsVendedCredentials(properties, configuration) &&
extractCredentials(table).isEmpty()) {
+ table = IcebergTableUtil.getTable(configuration, properties, true);
+ }
+ return table;
+ }
+
+ /**
+ * Reloads vended credentials at job launch when compile-time propagation
missed them.
+ * Non-secret config is written to {@code jobConf}; secrets are merged into
{@code TableDesc#getJobSecrets()}
+ * for {@link org.apache.hadoop.hive.ql.plan.PlanUtils#configureJobConf}
(HIVE-20651).
+ */
+ static void refreshVendedCredentialsIfMissing(TableDesc tableDesc, JobConf
jobConf, Configuration configuration) {
+ if (tableDesc == null || tableDesc.getProperties() == null ||
+ hasSerializedCredentials(tableDesc.getJobSecrets())) {
+ return;
+ }
+
+ Properties props = tableDesc.getProperties();
+ if (!requestsVendedCredentials(props, configuration)) {
+ return;
+ }
+
+ String catalogName = props.getProperty(InputFormatConfig.CATALOG_NAME);
+ if (catalogName == null) {
+ return;
+ }
+
+ try {
+ Table table = getTableWithVendedCredentials(props, configuration);
+ Map<String, String> jobProps = new LinkedHashMap<>();
+ Map<String, String> secrets = new LinkedHashMap<>();
+ propagateToJob(table, catalogName, jobProps, secrets, configuration);
+ jobProps.forEach(jobConf::set);
+ mergeJobSecrets(tableDesc, secrets);
+ } catch (NoSuchTableException ex) {
+ // Table may not exist yet for CTAS; credentials will not be available.
+ }
+ }
+
+ /**
+ * Returns a copy of the table over a fresh secret-free {@link FileIO}, safe
to serialize into
+ * job properties. {@code SerializableTable.copyOf} embeds {@code
table.io()} — and with it the
+ * vended credentials — into the serialized bytes, so the copy is rebuilt
from the same table
+ * metadata over a FileIO that keeps only the allowlisted non-secret
properties: unknown keys
+ * never ship, whatever the storage provider. Executors restore the
credentials from the
+ * HIVE-20651 Credentials channel via {@link #applyFromJobConf}
+ * ({@code SupportsStorageCredentials#setCredentials}).
+ */
+ static Table secretFreeCopy(Table table, Configuration conf) {
+ if (table instanceof BaseMetadataTable metadataTable) {
+ Table base = secretFreeCopy(metadataTable.table(), conf);
+ return MetadataTableUtils.createMetadataTableInstance(base,
metadataTableType(metadataTable));
+ }
+ Preconditions.checkState(table instanceof HasTableOperations,
+ "Cannot build a secret-free copy of %s (%s)", table.name(),
table.getClass().getName());
+ TableMetadata metadata = ((HasTableOperations)
table).operations().current();
+
+ FileIO io = table.io();
+ Map<String, String> ioProps = new LinkedHashMap<>();
+ io.properties().forEach((k, v) -> {
+ if (!isSecretKey(k, conf)) {
+ ioProps.put(k, v);
+ }
+ });
+ FileIO cleanIo = CatalogUtil.loadFileIO(io.getClass().getName(), ioProps,
conf);
+
+ return new BaseTable(
+ new StaticTableOperations(metadata, cleanIo,
table.locationProvider()), table.name());
+ }
+
+ /** {@code metadataTableType()} is package-private; the name suffix is the
type by construction. */
+ private static MetadataTableType metadataTableType(BaseMetadataTable
metadataTable) {
+ String name = metadataTable.name();
+ MetadataTableType type =
MetadataTableType.from(name.substring(name.lastIndexOf('.') + 1));
+ Preconditions.checkState(type != null, "Cannot resolve metadata table type
from %s", name);
+ return type;
+ }
+
+ /**
+ * Single-line base64, unlike {@link SerializationUtil#serializeToBase64}
which MIME-wraps with
+ * CR/LF. The blob is restored into table properties and copied to the task
conf through
+ * {@code Utilities#copyTablePropertiesToConf}, whose {@code escapeJava}
turns CR/LF into literal
+ * {@code \r\n} — and {@code r}/{@code n} are valid base64 alphabet,
corrupting the decoded
+ * stream. {@link SerializationUtil#deserializeFromBase64} uses the MIME
decoder, which accepts
+ * unwrapped input.
+ */
+ private static String serializeToSingleLineBase64(Object obj) {
+ return
Base64.getEncoder().encodeToString(SerializationUtil.serializeToBytes(obj));
+ }
+
+ private static boolean hasSerializedCredentials(Map<String, String>
jobSecrets) {
+ return jobSecrets != null && jobSecrets.keySet().stream()
+ .anyMatch(key ->
key.startsWith(InputFormatConfig.VENDED_STORAGE_CREDENTIALS));
+ }
+
+ private static void mergeJobSecrets(TableDesc tableDesc, Map<String, String>
secrets) {
+ if (secrets.isEmpty()) {
+ return;
+ }
+ Map<String, String> existing = tableDesc.getJobSecrets();
+ if (existing == null) {
+ tableDesc.setJobSecrets(new LinkedHashMap<>(secrets));
+ } else {
+ secrets.forEach(existing::putIfAbsent);
+ }
+ }
+
+ /**
+ * A vended config key is secret — routed to the Credentials channel, never
job properties, and
+ * stripped from the serialized table's FileIO — exactly when {@code
hive.conf.hidden.list} covers
+ * it. That is Hive's own registry of sensitive configuration (the same one
behind
+ * {@code HiveConf#isHiddenConfig}) that masks values in EXPLAIN, the Tez
UI, and ATS.
+ */
+ private static boolean isSecretKey(String key, Configuration conf) {
+ return HiveConfUtil.getHiddenSet(conf).stream().anyMatch(key::startsWith);
+ }
+
+ static List<StorageCredential> extractCredentials(Table table) {
+ if (table == null) {
+ return List.of();
+ }
+ FileIO io = table.io();
+ if (io instanceof SupportsStorageCredentials credentialIo) {
+ List<StorageCredential> credentials = credentialIo.credentials();
+ if (credentials != null && !credentials.isEmpty()) {
+ return credentials;
+ }
+ }
+ return credentialsFromFileIoProperties(table, io);
+ }
+
+ private static List<StorageCredential> credentialsFromFileIoProperties(Table
table, FileIO io) {
+ Map<String, String> props = io.properties();
+ if (props == null ||
StringUtils.isBlank(props.get(S3FileIOProperties.ACCESS_KEY_ID)) ||
+ StringUtils.isBlank(props.get(S3FileIOProperties.SECRET_ACCESS_KEY))) {
+ return List.of();
+ }
+ Map<String, String> config = new LinkedHashMap<>();
+ putIfPresent(config, props, S3FileIOProperties.ACCESS_KEY_ID);
+ putIfPresent(config, props, S3FileIOProperties.SECRET_ACCESS_KEY);
+ putIfPresent(config, props, S3FileIOProperties.SESSION_TOKEN);
+ putIfPresent(config, props, S3FileIOProperties.ENDPOINT);
+ putIfPresent(config, props, S3FileIOProperties.PATH_STYLE_ACCESS);
+ putIfPresent(config, props, AwsClientProperties.CLIENT_REGION);
+ return List.of(StorageCredential.create(credentialPrefix(table), config));
+ }
+
+ private static void putIfPresent(Map<String, String> target, Map<String,
String> source, String key) {
+ if (source.containsKey(key) && StringUtils.isNotBlank(source.get(key))) {
+ target.put(key, source.get(key));
+ }
+ }
+
+ private static String credentialPrefix(Table table) {
+ String location = table.location();
+ if (StringUtils.isBlank(location)) {
+ return "";
+ }
+ String bucket = bucketFromPrefix(location);
+ if (bucket != null) {
+ return "s3://" + bucket + "/";
+ }
+ return location.endsWith("/") ? location : location + "/";
+ }
+
+ /**
+ * REST catalogs vend credentials together with S3 connectivity settings
such as endpoint
+ * and path-style access. These settings reflect the catalog's network view
and may reference
+ * hosts that are not reachable from Hive (for example, an internal {@code
s3.ozone:9878}
+ * hostname).
+ *
+ * Catalog S3 properties configured in the Hive session (for example,
+ * {@code iceberg.catalog.ice01.s3.endpoint}) override the corresponding
vended connectivity
+ * settings so the driver and executors use reachable endpoints. Vended
credentials are
+ * preserved; only non-secret connectivity properties are overridden.
+ */
+ private static List<StorageCredential> withConfigurationOverrides(
+ String catalogName, List<StorageCredential> credentials, Configuration
conf) {
+
+ if (credentials.isEmpty() || conf == null || catalogName == null) {
+ return credentials;
+ }
+
+ List<StorageCredential> updated =
Lists.newArrayListWithCapacity(credentials.size());
+ for (StorageCredential credential : credentials) {
+ Map<String, String> credsConfig = new
LinkedHashMap<>(credential.config());
+ applyCatalogConfigOverrides(catalogName, credsConfig, conf);
+ updated.add(StorageCredential.create(credential.prefix(), credsConfig));
+ }
+
+ return updated;
+ }
+
+ /**
+ * Applies session-level catalog overrides to every entry of the given
credential configuration,
+ * through the same {@link #resolveCredentialValue} used for the
job-property entries — one
+ * resolver, one scope, so both channels always carry the same values.
+ */
+ private static void applyCatalogConfigOverrides(
+ String catalogName, Map<String, String> config, Configuration conf) {
+ config.replaceAll((icebergKey, value) ->
resolveCredentialValue(catalogName, icebergKey, value, conf));
+ }
+
+ private static String resolveCredentialValue(
+ String catalogName, String icebergKey, String vendedValue, Configuration
conf) {
+ if (conf == null || catalogName == null) {
+ return vendedValue;
+ }
+ String override =
+
conf.get(IcebergCatalogProperties.catalogPropertyConfigKey(catalogName,
icebergKey));
+ return StringUtils.isNotBlank(override) ? override : vendedValue;
+ }
+
+ /** Authority (bucket) of a storage prefix such as {@code s3://bucket/path},
any scheme.
+ * Plain string parse: {@code URI.getHost()} rejects legal bucket names with
underscores. */
+ @SuppressWarnings("java:S1075") // storage URI path separator, not a
filesystem path
+ private static String bucketFromPrefix(String prefix) {
+ int schemeEnd = prefix == null ? -1 : prefix.indexOf("://");
+ if (schemeEnd < 0) {
+ return null;
+ }
+ String withoutScheme = prefix.substring(schemeEnd + 3);
+ int slash = withoutScheme.indexOf('/');
+ String bucket = slash >= 0 ? withoutScheme.substring(0, slash) :
withoutScheme;
+ return StringUtils.defaultIfBlank(bucket, null);
+ }
+
+ private static String toS3aBucketProperty(String bucket, String icebergKey) {
+ String bucketPrefix = "fs.s3a.bucket." + bucket + ".";
+ return switch (icebergKey) {
+ case S3FileIOProperties.ACCESS_KEY_ID -> bucketPrefix + "access.key";
+ case S3FileIOProperties.SECRET_ACCESS_KEY -> bucketPrefix + "secret.key";
+ case S3FileIOProperties.SESSION_TOKEN -> bucketPrefix + "session.token";
+ case S3FileIOProperties.ENDPOINT -> bucketPrefix + "endpoint";
+ case S3FileIOProperties.PATH_STYLE_ACCESS -> bucketPrefix +
"path.style.access";
+ case AwsClientProperties.CLIENT_REGION -> bucketPrefix +
"endpoint.region";
+ default -> null;
+ };
+ }
+}
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergWriter.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergWriter.java
index 0a3cce06611..d7dfc8b9539 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergWriter.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergWriter.java
@@ -24,6 +24,7 @@
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapred.Reporter;
import org.apache.iceberg.data.Record;
+import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.mr.hive.FilesForCommit;
import org.apache.iceberg.mr.mapred.Container;
@@ -32,6 +33,8 @@ public interface HiveIcebergWriter extends
FileSinkOperator.RecordWriter,
FilesForCommit files();
+ FileIO io();
+
default void close(Reporter reporter) throws IOException {
close(false);
}
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergWriterBase.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergWriterBase.java
index ad026a83040..6267a3b9988 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergWriterBase.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergWriterBase.java
@@ -69,6 +69,11 @@ abstract class HiveIcebergWriterBase implements
HiveIcebergWriter {
this.writer = writer;
}
+ @Override
+ public FileIO io() {
+ return io;
+ }
+
@Override
public void close(boolean abort) throws IOException {
writer.close();
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
index fa66b25b349..89f888255b8 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
@@ -162,6 +162,8 @@ public List<InputSplit> getSplits(JobContext context) {
.orElseGet(() -> {
Table tbl = Catalogs.loadTable(conf);
conf.set(InputFormatConfig.TABLE_IDENTIFIER, tbl.name());
+ // planning-local conf only (never shipped): for credential-vending
catalogs the loaded
+ // table's FileIO carries secrets, which must not reach a serialized
job configuration
conf.set(InputFormatConfig.SERIALIZED_TABLE_PREFIX + tbl.name(),
SerializationUtil.serializeToBase64(tbl));
return tbl;
});
diff --git
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestIcebergVendedCredentialUtil.java
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestIcebergVendedCredentialUtil.java
new file mode 100644
index 00000000000..1d80244e70d
--- /dev/null
+++
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestIcebergVendedCredentialUtil.java
@@ -0,0 +1,885 @@
+/*
+ * 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.iceberg.mr.hive;
+
+import java.nio.charset.StandardCharsets;
+import java.security.PrivilegedExceptionAction;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.plan.PlanUtils;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SerializableTable;
+import org.apache.iceberg.StaticTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableMetadataParser;
+import org.apache.iceberg.aws.s3.S3FileIOProperties;
+import org.apache.iceberg.hadoop.HadoopFileIO;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.StorageCredential;
+import org.apache.iceberg.io.SupportsStorageCredentials;
+import org.apache.iceberg.mr.InputFormatConfig;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.SerializationUtil;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link IcebergVendedCredentialUtil}: copying REST-vended storage
credentials into Hive
+ * job configuration at plan time and re-applying them on executors.
+ */
+public class TestIcebergVendedCredentialUtil {
+
+ /**
+ * Tests {@link
IcebergVendedCredentialUtil#requestsVendedCredentials(Properties,
Configuration)}.
+ *
+ * <p><b>Setup:</b> table properties name catalog {@code ice01}; session
{@code conf} has no
+ * delegation header, then gains {@code vended-credentials}.
+ *
+ * <p><b>Expects:</b> {@code false} without the header, {@code true} with it
— propagation and
+ * apply logic must not run for ordinary REST catalogs.
+ *
+ * <p><b>Why:</b> Iceberg REST only vends storage credentials when the
client sends
+ * {@code X-Iceberg-Access-Delegation: vended-credentials}; Hive mirrors
that opt-in via catalog
+ * config so static catalog keys are not replaced unnecessarily.
+ */
+ @Test
+ public void requestsVendedCredentialsRequiresDelegationHeader() {
+ Configuration conf = new Configuration();
+ Properties props = new java.util.Properties();
+ props.setProperty(InputFormatConfig.CATALOG_NAME, "ice01");
+
+ assertThat(IcebergVendedCredentialUtil.requestsVendedCredentials(props,
conf)).isFalse();
+
+ conf.set(
+ "iceberg.catalog.ice01.header.X-Iceberg-Access-Delegation",
+ "vended-credentials");
+ assertThat(IcebergVendedCredentialUtil.requestsVendedCredentials(props,
conf)).isTrue();
+ }
+
+ /**
+ * Tests {@link IcebergVendedCredentialUtil#propagateToJob} when both {@code
jobProperties} and
+ * {@code jobSecrets} are non-null (full compile-time path after both
storage-handler hooks).
+ *
+ * <p><b>Setup:</b> vended credential with internal endpoint {@code
http://minio:9000}; session
+ * {@code conf} sets {@code iceberg.catalog.ice01.s3.endpoint} to {@code
http://host:9000}.
+ *
+ * <p><b>Expects:</b> endpoint and path-style in {@code jobProps} (Iceberg
catalog keys and S3A
+ * per-bucket keys) with host endpoint; access/secret and serialized blob in
{@code jobSecrets}
+ * only; blob endpoint also host; no secret material in {@code jobProps}.
+ *
+ * <p><b>Why:</b> HIVE-20651 keeps secrets in {@code TableDesc#jobSecrets}
(later Hadoop
+ * {@code Credentials}) and non-secrets in job conf. Catalogs often vend an
internal S3 endpoint;
+ * HS2 overrides with a reachable host endpoint at propagation time so Tez
tasks and Iceberg FileIO
+ * agree on connectivity while still using vended keys.
+ */
+ @Test
+ public void propagateToJobMapsIcebergAndS3aProperties() {
+ Configuration conf = new Configuration();
+ conf.set("iceberg.catalog.ice01." + S3FileIOProperties.ENDPOINT,
"http://host:9000";);
+
+ CredentialFileIO fileIO = new CredentialFileIO();
+ fileIO.setCredentials(
+ List.of(
+ StorageCredential.create(
+ "s3://my-bucket/",
+ Map.of(
+ S3FileIOProperties.ACCESS_KEY_ID, "access",
+ S3FileIOProperties.SECRET_ACCESS_KEY, "secret",
+ S3FileIOProperties.ENDPOINT, "http://minio:9000";,
+ S3FileIOProperties.PATH_STYLE_ACCESS, "true"))));
+
+ Table table =
+ new BaseTable(new StaticTableOperations("s3://my-bucket/t", fileIO),
"db.t");
+ Map<String, String> jobProps = Maps.newHashMap();
+ Map<String, String> jobSecrets = Maps.newHashMap();
+
+ IcebergVendedCredentialUtil.propagateToJob(table, "ice01", jobProps,
jobSecrets, conf);
+
+ assertThat(jobProps)
+ .containsEntry(
+ "iceberg.catalog.ice01." + S3FileIOProperties.ENDPOINT,
+ "http://host:9000";)
+ .containsEntry(
+ "fs.s3a.bucket.my-bucket.endpoint",
+ "http://host:9000";)
+ .containsEntry(
+ "fs.s3a.bucket.my-bucket.path.style.access",
+ "true")
+ .doesNotContainKey(InputFormatConfig.vendedCredentialsKey("db.t"))
+ .doesNotContainKey("fs.s3a.bucket.my-bucket.access.key")
+ .doesNotContainKey(
+ "iceberg.catalog.ice01." + S3FileIOProperties.ACCESS_KEY_ID);
+
+ assertThat(jobSecrets)
+ .containsEntry("fs.s3a.bucket.my-bucket.access.key", "access")
+ .containsEntry("fs.s3a.bucket.my-bucket.secret.key", "secret")
+ .doesNotContainKey(
+ "iceberg.catalog.ice01." + S3FileIOProperties.ACCESS_KEY_ID)
+ .doesNotContainKey(
+ "iceberg.catalog.ice01." + S3FileIOProperties.SECRET_ACCESS_KEY)
+ .satisfies(map ->
+ assertThat(map.get(InputFormatConfig.vendedCredentialsKey("db.t")))
+ .isNotBlank());
+
+ List<StorageCredential> serialized =
+ SerializationUtil.deserializeFromBase64(
+ jobSecrets.get(InputFormatConfig.vendedCredentialsKey("db.t")));
+
+ assertThat(serialized.getFirst().prefix()).isEqualTo("s3://my-bucket/");
+ assertThat(serialized.getFirst().config())
+ .containsEntry(S3FileIOProperties.ENDPOINT, "http://host:9000";)
+ .containsEntry(S3FileIOProperties.ACCESS_KEY_ID, "access")
+ .containsEntry(S3FileIOProperties.SECRET_ACCESS_KEY, "secret")
+ .containsEntry(S3FileIOProperties.PATH_STYLE_ACCESS, "true");
+ }
+
+ /**
+ * Tests {@link IcebergVendedCredentialUtil#propagateToJob} with {@code
jobSecrets == null}.
+ *
+ * <p><b>Setup:</b> same mixed secret/non-secret credential as the full
test; only
+ * {@code jobProperties} map is passed (models {@code
configureInputJobProperties} /
+ * {@code overlayTableProperties}).
+ *
+ * <p><b>Expects:</b> overridden endpoint in {@code jobProps}; no serialized
blob and no Iceberg
+ * access-key property in {@code jobProps}.
+ *
+ * <p><b>Why:</b> Hive calls the job-properties hook without a secrets map.
This hook must add
+ * connectivity settings for S3A and Iceberg but must not place keys or the
credential blob in
+ * plain job properties — those are filled later by {@code
configureInputJobCredentials}.
+ */
+ @Test
+ public void propagateNonSecretsOnlyWhenJobSecretsNull() {
+ Configuration conf = new Configuration();
+ conf.set("iceberg.catalog.ice01." + S3FileIOProperties.ENDPOINT,
"http://host:9000";);
+
+ CredentialFileIO fileIO = new CredentialFileIO();
+ fileIO.setCredentials(
+ List.of(
+ StorageCredential.create(
+ "s3://my-bucket/",
+ Map.of(
+ S3FileIOProperties.ACCESS_KEY_ID, "access",
+ S3FileIOProperties.SECRET_ACCESS_KEY, "secret",
+ S3FileIOProperties.ENDPOINT, "http://minio:9000";,
+ S3FileIOProperties.PATH_STYLE_ACCESS, "true"))));
+
+ Table table =
+ new BaseTable(new StaticTableOperations("s3://my-bucket/t", fileIO),
"db.t");
+ Map<String, String> jobProps = Maps.newHashMap();
+
+ IcebergVendedCredentialUtil.propagateToJob(table, "ice01", jobProps, null,
conf);
+
+ assertThat(jobProps)
+ .containsEntry(
+ "iceberg.catalog.ice01." + S3FileIOProperties.ENDPOINT,
+ "http://host:9000";)
+ .containsEntry("fs.s3a.bucket.my-bucket.endpoint", "http://host:9000";)
+ .containsEntry("fs.s3a.bucket.my-bucket.path.style.access", "true")
+ .doesNotContainKey(InputFormatConfig.vendedCredentialsKey("db.t"))
+ .doesNotContainKey(
+ "iceberg.catalog.ice01." + S3FileIOProperties.ACCESS_KEY_ID);
+ }
+
+ /**
+ * Tests {@link IcebergVendedCredentialUtil#extractCredentials(Table)} when
+ * {@link SupportsStorageCredentials#credentials()} is empty.
+ *
+ * <p><b>Setup:</b> FileIO initialized with {@code s3.access-key-id} /
{@code s3.secret-access-key}
+ * in {@link FileIO#properties()} but an empty credential list.
+ *
+ * <p><b>Expects:</b> one synthetic {@link StorageCredential} with the
access key from properties.
+ *
+ * <p><b>Why:</b> after {@code loadTable}, some REST clients populate {@code
FileIO} properties
+ * before the credential list; propagation must still find vended keys or
plan/launch would skip
+ * credential vending even though the table load succeeded.
+ */
+ @Test
+ public void extractCredentialsFromFileIoPropertiesWhenCredentialListEmpty() {
+ CredentialFileIO fileIO = new CredentialFileIO();
+ fileIO.initialize(
+ Map.of(
+ S3FileIOProperties.ACCESS_KEY_ID, "access",
+ S3FileIOProperties.SECRET_ACCESS_KEY, "secret",
+ S3FileIOProperties.ENDPOINT, "http://minio:9000";));
+ Schema schema = new Schema(Types.NestedField.required(1, "x",
Types.IntegerType.get()));
+ TableMetadata metadata =
+ TableMetadata.newTableMetadata(schema, PartitionSpec.unpartitioned(),
"s3://my-bucket/warehouse/t", Map.of());
+ Table table = new BaseTable(new StaticTableOperations(metadata, fileIO),
"db.t");
+
+ StorageCredential extracted =
IcebergVendedCredentialUtil.extractCredentials(table).getFirst();
+ assertThat(extracted.prefix()).isEqualTo("s3://my-bucket/");
+ assertThat(extracted.config())
+ .containsEntry(S3FileIOProperties.ACCESS_KEY_ID, "access")
+ .containsEntry(S3FileIOProperties.SECRET_ACCESS_KEY, "secret")
+ .containsEntry(S3FileIOProperties.ENDPOINT, "http://minio:9000";);
+ }
+
+ /**
+ * Tests {@link IcebergVendedCredentialUtil#propagateToJob} with {@code
jobProperties == null}.
+ *
+ * <p><b>Setup:</b> credential with access key and secret only; only {@code
jobSecrets} map is
+ * passed (models {@code configureInputJobCredentials} for input and output
tables).
+ *
+ * <p><b>Expects:</b> serialized blob and {@code
fs.s3a.bucket.*.access.key/secret.key} in
+ * {@code jobSecrets}; no catalog-prefixed secret keys, no endpoint in
{@code jobSecrets}.
+ *
+ * <p><b>Why:</b> the credentials hook receives only the secrets map.
Propagation must still run
+ * without NPE, write material for {@code PlanUtils#configureJobConf} to
move into
+ * {@code Credentials}, and avoid duplicating non-secrets or catalog-level
secret keys that belong
+ * in job properties or only in the serialized blob.
+ */
+ @Test
+ public void propagateSecretsOnlyWhenJobPropertiesNull() {
+ CredentialFileIO fileIO = new CredentialFileIO();
+ fileIO.setCredentials(
+ List.of(
+ StorageCredential.create(
+ "s3://my-bucket/",
+ Map.of(S3FileIOProperties.ACCESS_KEY_ID, "access",
+ S3FileIOProperties.SECRET_ACCESS_KEY, "secret"))));
+ Table table =
+ new BaseTable(new StaticTableOperations("s3://my-bucket/t", fileIO),
"db.t");
+ Map<String, String> jobSecrets = Maps.newHashMap();
+
+ IcebergVendedCredentialUtil.propagateToJob(table, "ice01", null,
jobSecrets, new Configuration());
+
+ assertThat(jobSecrets)
+ .containsEntry("fs.s3a.bucket.my-bucket.access.key", "access")
+ .containsEntry("fs.s3a.bucket.my-bucket.secret.key", "secret")
+ .doesNotContainKey(
+ "iceberg.catalog.ice01." + S3FileIOProperties.ACCESS_KEY_ID)
+ .doesNotContainKey(
+ "iceberg.catalog.ice01." + S3FileIOProperties.SECRET_ACCESS_KEY)
+ .doesNotContainKey("fs.s3a.bucket.my-bucket.endpoint")
+ .satisfies(map ->
+ assertThat(map.get(InputFormatConfig.vendedCredentialsKey("db.t")))
+ .isNotBlank());
+
+ List<StorageCredential> serialized =
+ SerializationUtil.deserializeFromBase64(
+ jobSecrets.get(InputFormatConfig.vendedCredentialsKey("db.t")));
+
+ assertThat(serialized.getFirst().prefix()).isEqualTo("s3://my-bucket/");
+ assertThat(serialized.getFirst().config())
+ .containsEntry(S3FileIOProperties.ACCESS_KEY_ID, "access")
+ .containsEntry(S3FileIOProperties.SECRET_ACCESS_KEY, "secret");
+ }
+
+ /**
+ * Tests {@link IcebergVendedCredentialUtil#applyFromJobConf} for a
non-vending catalog.
+ *
+ * <p><b>Setup:</b> task {@code conf} has catalog name only — no serialized
credentials blob and
+ * no {@code vended-credentials} delegation header.
+ *
+ * <p><b>Expects:</b> FileIO credential list stays empty after apply.
+ *
+ * <p><b>Why:</b> {@link
IcebergVendedCredentialUtil#shouldSkipApplyFromJobConf} must return early
+ * so tables that use static catalog S3 configuration are not cleared or
rewritten on executors.
+ */
+ @Test
+ public void applyFromJobConfSkipsWhenVendedCredentialsNotRequested() {
+ CredentialFileIO fileIO = new CredentialFileIO();
+ Table table =
+ new BaseTable(new StaticTableOperations("s3://my-bucket/t", fileIO),
"db.t");
+
+ Configuration conf = new Configuration();
+ conf.set("iceberg.catalog", "ice01");
+
+ IcebergVendedCredentialUtil.applyFromJobConf(table, conf);
+
+ assertThat(fileIO.credentials()).isEmpty();
+ }
+
+ /**
+ * Tests {@link IcebergVendedCredentialUtil#applyFromJobConf} endpoint
override on an existing FileIO.
+ *
+ * <p><b>Setup:</b> FileIO already has vended credentials with {@code
http://minio:9000}; task
+ * {@code conf} requests vending and sets host endpoint {@code
http://host:9000}.
+ *
+ * <p><b>Expects:</b> after apply, FileIO credential endpoint is {@code
http://host:9000}.
+ *
+ * <p><b>Why:</b> HS2 may retain credentials from {@code loadTable} with the
catalog-internal
+ * endpoint; apply must re-merge session catalog S3 settings so commit paths
and tasks use the
+ * same reachable endpoint as plan-time propagation.
+ */
+ @Test
+ public void applyFromJobConfOverridesEndpointOnExistingCredentials() {
+ CredentialFileIO fileIO = new CredentialFileIO();
+ fileIO.setCredentials(
+ List.of(
+ StorageCredential.create(
+ "s3://my-bucket/",
+ Map.of(
+ S3FileIOProperties.ACCESS_KEY_ID, "access",
+ S3FileIOProperties.SECRET_ACCESS_KEY, "secret",
+ S3FileIOProperties.ENDPOINT, "http://minio:9000";))));
+ Table table =
+ new BaseTable(new StaticTableOperations("s3://my-bucket/t", fileIO),
"db.t");
+
+ Configuration conf = new Configuration();
+ conf.set("iceberg.catalog", "ice01");
+ conf.set(
+ "iceberg.catalog.ice01.header.X-Iceberg-Access-Delegation",
+ "vended-credentials");
+ conf.set("iceberg.catalog.ice01." + S3FileIOProperties.ENDPOINT,
"http://host:9000";);
+
+ IcebergVendedCredentialUtil.applyFromJobConf(table, conf);
+
+ StorageCredential applied =
+ ((SupportsStorageCredentials) table.io()).credentials().getFirst();
+ assertThat(applied.prefix()).isEqualTo("s3://my-bucket/");
+ assertThat(applied.config())
+ .containsEntry(S3FileIOProperties.ENDPOINT, "http://host:9000";)
+ .containsEntry(S3FileIOProperties.ACCESS_KEY_ID, "access")
+ .containsEntry(S3FileIOProperties.SECRET_ACCESS_KEY, "secret");
+ }
+
+ /**
+ * Tests {@link IcebergVendedCredentialUtil#applyFromJobConf} on a task with
an empty FileIO.
+ *
+ * <p><b>Setup:</b> {@code propagateToJob} fills job maps on HS2; keys are
copied into a task
+ * {@code Configuration} as on Tez/LLAP; executor table is deserialized with
a fresh FileIO (no
+ * credentials).
+ *
+ * <p><b>Expects:</b> after apply, executor FileIO has one credential with
the vended access key.
+ *
+ * <p><b>Why:</b> executors do not re-run REST {@code loadTable}; they
rebuild the table from job
+ * conf. {@link IcebergVendedCredentialUtil#resolveCredentialsForApply} must
read the serialized
+ * blob (or S3A secret keys) from the task conf and call {@code
setCredentials} so Iceberg I/O
+ * can access S3 during the task.
+ */
+ @Test
+ public void applyFromJobConfRestoresCredentialsOnExecutor() {
+ CredentialFileIO fileIO = new CredentialFileIO();
+ fileIO.setCredentials(
+ List.of(
+ StorageCredential.create(
+ "s3://my-bucket/",
+ Map.of(S3FileIOProperties.ACCESS_KEY_ID, "access",
+ S3FileIOProperties.SECRET_ACCESS_KEY, "secret"))));
+ Table table =
+ new BaseTable(new StaticTableOperations("s3://my-bucket/t", fileIO),
"db.t");
+
+ Map<String, String> jobProps = Maps.newHashMap();
+ Map<String, String> jobSecrets = Maps.newHashMap();
+ IcebergVendedCredentialUtil.propagateToJob(
+ table, "ice01", jobProps, jobSecrets, new Configuration());
+
+ Configuration taskConf = new Configuration();
+ jobProps.forEach(taskConf::set);
+ jobSecrets.forEach(taskConf::set);
+
+ CredentialFileIO executorIo = new CredentialFileIO();
+ Table executorTable =
+ new BaseTable(new StaticTableOperations("s3://my-bucket/t",
executorIo), "db.t");
+ IcebergVendedCredentialUtil.applyFromJobConf(executorTable, taskConf);
+
+ assertThat(((SupportsStorageCredentials)
executorTable.io()).credentials()).hasSize(1);
+ StorageCredential applied =
+ ((SupportsStorageCredentials)
executorTable.io()).credentials().getFirst();
+ assertThat(applied.prefix()).isEqualTo("s3://my-bucket/");
+ assertThat(applied.config())
+ .containsEntry(S3FileIOProperties.ACCESS_KEY_ID, "access")
+ .containsEntry(S3FileIOProperties.SECRET_ACCESS_KEY, "secret");
+ }
+
+ /**
+ * Tests the full HIVE-20651 secret round trip: {@code propagateToJob} →
+ * {@link org.apache.hadoop.hive.ql.plan.PlanUtils#configureJobConf}
(secrets → Hadoop
+ * {@code Credentials}, cleared from the TableDesc) → task UGI →
+ * {@code Utilities#copyJobSecretToTableProperties} → {@code
Utilities#copyTablePropertiesToConf}
+ * → {@link IcebergVendedCredentialUtil#applyFromJobConf}.
+ *
+ * <p><b>Setup:</b> the exact chain Hive runs — HS2 fills {@code
TableDesc#jobSecrets}, PlanUtils
+ * moves them to the job {@code Credentials} (asserted: nothing lands in the
plain JobConf), the
+ * task restores them into table properties and copies them into the
task-local conf the same way
+ * {@code MapRecordProcessor}/{@code FileSinkOperator} do.
+ *
+ * <p><b>Expects:</b> the executor FileIO receives the vended credential.
+ *
+ * <p><b>Why:</b> this is the only sanctioned transport for secrets — they
must survive the real
+ * restore path, including {@code copyTablePropertiesToConf}'s {@code
escapeJava}, which corrupts
+ * MIME-wrapped base64 (CR/LF become literal {@code \r\n} whose {@code
r}/{@code n} bytes are
+ * valid base64 alphabet). The blob must therefore be single-line base64.
+ */
+ @Test
+ public void applyFromJobConfRestoresCredentialsViaCredentialsChannel()
throws Exception {
+ CredentialFileIO hs2Io = new CredentialFileIO();
+ hs2Io.setCredentials(
+ List.of(
+ StorageCredential.create(
+ "s3://my-bucket/",
+ Map.of(S3FileIOProperties.ACCESS_KEY_ID, "access",
+ S3FileIOProperties.SECRET_ACCESS_KEY, "secret"))));
+ Table hs2Table =
+ new BaseTable(new StaticTableOperations("s3://my-bucket/t", hs2Io),
"db.t");
+
+ TableDesc tableDesc = new TableDesc();
+ Properties tableProps = new Properties();
+ tableProps.setProperty(hive_metastoreConstants.META_TABLE_NAME, "db.t");
+ tableDesc.setProperties(tableProps);
+
+ Map<String, String> jobSecrets = Maps.newHashMap();
+ IcebergVendedCredentialUtil.propagateToJob(hs2Table, "ice01", null,
jobSecrets, new Configuration());
+ tableDesc.setJobSecrets(jobSecrets);
+
+ JobConf jobConf = new JobConf();
+ PlanUtils.configureJobConf(tableDesc, jobConf);
+
assertThat(jobConf.get(InputFormatConfig.vendedCredentialsKey("db.t"))).isNull();
+ assertThat(tableDesc.getJobSecrets()).isEmpty();
+
+ UserGroupInformation taskUgi =
UserGroupInformation.createRemoteUser("task-user");
+ taskUgi.addCredentials(jobConf.getCredentials());
+
+ JobConf taskConf = new JobConf();
+ taskConf.set(InputFormatConfig.CATALOG_NAME, "ice01");
+ taskConf.set(
+ "iceberg.catalog.ice01.header.X-Iceberg-Access-Delegation",
+ "vended-credentials");
+
+ CredentialFileIO executorIo = new CredentialFileIO();
+ Table executorTable =
+ new BaseTable(new StaticTableOperations("s3://my-bucket/t",
executorIo), "db.t");
+
+ taskUgi.doAs((PrivilegedExceptionAction<Void>) () -> {
+ Utilities.copyJobSecretToTableProperties(tableDesc);
+ Utilities.copyTablePropertiesToConf(tableDesc, taskConf);
+ IcebergVendedCredentialUtil.applyFromJobConf(executorTable, taskConf);
+ return null;
+ });
+
+ assertThat(executorIo.credentials()).hasSize(1);
+ StorageCredential applied = executorIo.credentials().getFirst();
+ assertThat(applied.prefix()).isEqualTo("s3://my-bucket/");
+ assertThat(applied.config())
+ .containsEntry(S3FileIOProperties.ACCESS_KEY_ID, "access")
+ .containsEntry(S3FileIOProperties.SECRET_ACCESS_KEY, "secret");
+ }
+
+ /**
+ * Tests {@link HiveIcebergStorageHandler#configureJobConf} secret hygiene.
+ *
+ * <p><b>Setup:</b> vending-enabled catalog; {@code TableDesc#jobSecrets}
already holds the
+ * serialized blob and a per-bucket S3A secret key (filled at compile time by
+ * {@code configureInputJobCredentials}).
+ *
+ * <p><b>Expects:</b> after {@code configureJobConf}, neither the blob nor
the S3A secret key
+ * appears in the {@code JobConf}; the jobSecrets map is left for
+ * {@code PlanUtils#configureJobConf} to move into Hadoop {@code
Credentials}.
+ *
+ * <p><b>Why:</b> {@code TezTask} strips {@code hive.conf.hidden.list}
entries <em>before</em>
+ * storage handlers run ({@code createConfiguration} precedes
+ * {@code configureJobConfAndExtractJars}), so any secret written to the
JobConf here ships in
+ * plaintext with the DAG plan. Secrets may only travel via the Credentials
channel.
+ */
+ @Test
+ public void configureJobConfKeepsSecretsOutOfJobConf() {
+ Configuration conf = new Configuration();
+ conf.set(
+ "iceberg.catalog.ice01.header.X-Iceberg-Access-Delegation",
+ "vended-credentials");
+ HiveIcebergStorageHandler handler = new HiveIcebergStorageHandler();
+ handler.setConf(conf);
+
+ Properties props = new Properties();
+ props.setProperty(InputFormatConfig.CATALOG_NAME, "ice01");
+ TableDesc tableDesc = new TableDesc();
+ tableDesc.setProperties(props);
+
+ String secretS3aKey = "fs.s3a.bucket.my-bucket.secret.key";
+ Map<String, String> jobSecrets = Maps.newHashMap();
+ jobSecrets.put(InputFormatConfig.vendedCredentialsKey("db.t"), "blob");
+ jobSecrets.put(secretS3aKey, "secret");
+ tableDesc.setJobSecrets(jobSecrets);
+
+ JobConf jobConf = new JobConf();
+ jobConf.setBoolean(HiveConf.ConfVars.HIVE_IN_TEST_IDE.varname, true);
+
+ handler.configureJobConf(tableDesc, jobConf);
+
+
assertThat(jobConf.get(InputFormatConfig.vendedCredentialsKey("db.t"))).isNull();
+ assertThat(jobConf.get(secretS3aKey)).isNull();
+ assertThat(tableDesc.getJobSecrets())
+ .containsEntry(InputFormatConfig.vendedCredentialsKey("db.t"), "blob")
+ .containsEntry(secretS3aKey, "secret");
+ }
+
+ /**
+ * Tests per-table credential resolution when two vended tables publish into
one configuration.
+ *
+ * <p><b>Setup:</b> REST catalogs mint credentials per table; two tables on
different buckets go
+ * through the full chain — {@code propagateToJob} → {@code
PlanUtils#configureJobConf} → one
+ * task UGI → per-table restore — and both publish into a <em>shared</em>
conf the way merged
+ * map-works do. {@code Utilities#copyTableJobPropertiesToConf} is
skip-if-present for table
+ * properties, so with a single global blob key the second table would read
the first table's
+ * credentials.
+ *
+ * <p><b>Expects:</b> each executor table's FileIO receives the credential
vended for it —
+ * table 2 gets the bucket-b credential, not bucket-a's.
+ *
+ * <p><b>Why:</b> the blob key must be table-scoped (like {@code
iceberg.mr.serialized.table.*})
+ * for multi-table queries; key collisions would silently apply another
table's credentials.
+ */
+ @Test
+ public void applyFromJobConfResolvesPerTableCredentialsInSharedConf() throws
Exception {
+ TableDesc tableDesc1 = tableDescNamed("db.t1");
+ Map<String, String> secrets1 = Maps.newHashMap();
+ IcebergVendedCredentialUtil.propagateToJob(
+ tableWithCredential("s3://bucket-a/", "access-a", "db.t1"), "ice01",
null, secrets1, new Configuration());
+ tableDesc1.setJobSecrets(secrets1);
+
+ TableDesc tableDesc2 = tableDescNamed("db.t2");
+ Map<String, String> secrets2 = Maps.newHashMap();
+ IcebergVendedCredentialUtil.propagateToJob(
+ tableWithCredential("s3://bucket-b/", "access-b", "db.t2"), "ice01",
null, secrets2, new Configuration());
+ tableDesc2.setJobSecrets(secrets2);
+
+ JobConf jobConf = new JobConf();
+ PlanUtils.configureJobConf(tableDesc1, jobConf);
+ PlanUtils.configureJobConf(tableDesc2, jobConf);
+
+ UserGroupInformation taskUgi =
UserGroupInformation.createRemoteUser("task-user");
+ taskUgi.addCredentials(jobConf.getCredentials());
+
+ JobConf sharedConf = new JobConf();
+ sharedConf.set(InputFormatConfig.CATALOG_NAME, "ice01");
+ sharedConf.set(
+ "iceberg.catalog.ice01.header.X-Iceberg-Access-Delegation",
+ "vended-credentials");
+
+ CredentialFileIO executorIo1 = new CredentialFileIO();
+ Table executorTable1 =
+ new BaseTable(new StaticTableOperations("s3://bucket-a/t1",
executorIo1), "db.t1");
+ CredentialFileIO executorIo2 = new CredentialFileIO();
+ Table executorTable2 =
+ new BaseTable(new StaticTableOperations("s3://bucket-b/t2",
executorIo2), "db.t2");
+
+ taskUgi.doAs((PrivilegedExceptionAction<Void>) () -> {
+ Utilities.copyJobSecretToTableProperties(tableDesc1);
+ Utilities.copyJobSecretToTableProperties(tableDesc2);
+ Utilities.copyTableJobPropertiesToConf(tableDesc1, sharedConf);
+ Utilities.copyTableJobPropertiesToConf(tableDesc2, sharedConf);
+ IcebergVendedCredentialUtil.applyFromJobConf(executorTable1, sharedConf);
+ IcebergVendedCredentialUtil.applyFromJobConf(executorTable2, sharedConf);
+ return null;
+ });
+
+ assertThat(executorIo1.credentials()).hasSize(1);
+
assertThat(executorIo1.credentials().getFirst().prefix()).isEqualTo("s3://bucket-a/");
+ assertThat(executorIo1.credentials().getFirst().config())
+ .containsEntry(S3FileIOProperties.ACCESS_KEY_ID, "access-a");
+
+ assertThat(executorIo2.credentials()).hasSize(1);
+
assertThat(executorIo2.credentials().getFirst().prefix()).isEqualTo("s3://bucket-b/");
+ assertThat(executorIo2.credentials().getFirst().config())
+ .containsEntry(S3FileIOProperties.ACCESS_KEY_ID, "access-b");
+ }
+
+ private static TableDesc tableDescNamed(String tableName) {
+ Properties tableProps = new Properties();
+ tableProps.setProperty(hive_metastoreConstants.META_TABLE_NAME, tableName);
+ TableDesc tableDesc = new TableDesc();
+ tableDesc.setProperties(tableProps);
+ return tableDesc;
+ }
+
+ private static Table tableWithCredential(String prefix, String accessKey,
String tableName) {
+ CredentialFileIO fileIO = new CredentialFileIO();
+ fileIO.setCredentials(
+ List.of(
+ StorageCredential.create(
+ prefix,
+ Map.of(S3FileIOProperties.ACCESS_KEY_ID, accessKey,
+ S3FileIOProperties.SECRET_ACCESS_KEY, accessKey +
"-secret"))));
+ return new BaseTable(new StaticTableOperations(prefix + "t", fileIO),
tableName);
+ }
+
+ /**
+ * Tests the strip → ship → setCredentials flow for vended catalogs.
+ *
+ * <p><b>Setup:</b> HS2-side table with a credentialed FileIO (secret in the
credential list and
+ * in the FileIO properties) over a real metadata file on local disk. {@code
secretFreeCopy}
+ * rebuilds the table over a fresh FileIO keeping only allowlisted
non-secret properties before
+ * {@code serializeTable}; credentials travel separately via {@code
propagateToJob} (modelling
+ * the HIVE-20651 restore); {@code HiveTableUtil#deserializeTable} restores
them on the
+ * deserialized FileIO.
+ *
+ * <p><b>Expects:</b> the serialized table carries no secret bytes; the
deserialized table has
+ * the original qualified name, schema, FileIO class with non-secret
properties only, and the
+ * vended credentials restored from the Credentials-channel key.
+ *
+ * <p><b>Why:</b> a serialized table embeds its FileIO, so the copy shipped
through job
+ * properties must carry a FileIO with no secret state; executors re-attach
the credentials
+ * through {@code SupportsStorageCredentials#setCredentials}.
+ */
+ @Test
+ public void vendedSerializedTableShipsSecretFreeAndRestoresCredentials()
throws Exception {
+ CredentialFileIO fileIO = new CredentialFileIO();
+ fileIO.initialize(
+ Map.of(
+ S3FileIOProperties.SECRET_ACCESS_KEY, "prop-secret-11bb",
+ S3FileIOProperties.ENDPOINT, "http://minio:9000";));
+ fileIO.setCredentials(
+ List.of(
+ StorageCredential.create(
+ "s3://my-bucket/",
+ Map.of(S3FileIOProperties.ACCESS_KEY_ID, "access",
+ S3FileIOProperties.SECRET_ACCESS_KEY,
"cred-secret-11bb"))));
+
+ Schema schema = new Schema(Types.NestedField.required(1, "x",
Types.IntegerType.get()));
+ TableMetadata metadata =
+ TableMetadata.newTableMetadata(schema, PartitionSpec.unpartitioned(),
"s3://my-bucket/warehouse/t", Map.of());
+ String metadataLocation = "file://" +
+
java.nio.file.Files.createTempDirectory("iceberg-strip-test").resolve("v1.metadata.json");
+ TableMetadataParser.write(metadata,
fileIO.newOutputFile(metadataLocation));
+ Table hs2Table = new BaseTable(new StaticTableOperations(metadataLocation,
fileIO), "ice01.db.t");
+
+ Configuration conf = new Configuration();
+ String serialized = HiveTableUtil.serializeTable(
+ IcebergVendedCredentialUtil.secretFreeCopy(hs2Table, conf), conf, new
Properties(), null);
+
+ String rawBytes = new String(
+ java.util.Base64.getMimeDecoder().decode(serialized),
StandardCharsets.ISO_8859_1);
+ assertThat(rawBytes)
+ .doesNotContain("prop-secret-11bb")
+ .doesNotContain("cred-secret-11bb")
+ .contains("http://minio:9000";);
+
+ Map<String, String> jobSecrets = Maps.newHashMap();
+ IcebergVendedCredentialUtil.propagateToJob(hs2Table, "ice01", null,
jobSecrets, new Configuration());
+
+ JobConf taskConf = new JobConf();
+ taskConf.set(InputFormatConfig.SERIALIZED_TABLE_PREFIX + "db.t",
serialized);
+ jobSecrets.forEach(taskConf::set);
+ taskConf.set(InputFormatConfig.CATALOG_NAME, "ice01");
+
+ Table executorTable = HiveTableUtil.deserializeTable(taskConf, "db.t");
+
+ assertThat(executorTable).isNotNull();
+ assertThat(executorTable.name()).isEqualTo("ice01.db.t");
+ assertThat(executorTable.schema().asStruct()).isEqualTo(schema.asStruct());
+ assertThat(executorTable.io()).isInstanceOf(CredentialFileIO.class);
+ assertThat(executorTable.io().properties())
+ .containsEntry(S3FileIOProperties.ENDPOINT, "http://minio:9000";)
+ .doesNotContainKey(S3FileIOProperties.SECRET_ACCESS_KEY);
+ assertThat(((SupportsStorageCredentials)
executorTable.io()).credentials()).hasSize(1);
+ }
+
+ /**
+ * Tests the strip → ship → setCredentials flow for metadata tables ({@code
db.t.files}-style
+ * queries).
+ *
+ * <p><b>Setup:</b> a {@code FILES} metadata table over a credentialed base;
the secret-free
+ * copy is serialized and the table restored through {@code
HiveTableUtil#deserializeTable}.
+ *
+ * <p><b>Expects:</b> the serialized bytes carry no secrets; the
deserialized table keeps the
+ * metadata-table name and schema over a secret-free FileIO with the
credentials restored.
+ *
+ * <p><b>Why:</b> metadata tables share the base table FileIO and would leak
the same way; the
+ * secret-free copy must recreate the metadata-table view over the clean
FileIO.
+ */
+ @Test
+ public void vendedMetadataTableSerializesSecretFreeAndRestoresCredentials()
throws Exception {
+ CredentialFileIO fileIO = new CredentialFileIO();
+ fileIO.initialize(Map.of(S3FileIOProperties.SECRET_ACCESS_KEY,
"prop-secret-3fd2"));
+ fileIO.setCredentials(
+ List.of(
+ StorageCredential.create(
+ "s3://my-bucket/",
+ Map.of(S3FileIOProperties.SECRET_ACCESS_KEY,
"cred-secret-3fd2"))));
+
+ Schema schema = new Schema(Types.NestedField.required(1, "x",
Types.IntegerType.get()));
+ TableMetadata metadata =
+ TableMetadata.newTableMetadata(schema, PartitionSpec.unpartitioned(),
"s3://my-bucket/warehouse/t", Map.of());
+ String metadataLocation = "file://" +
+
java.nio.file.Files.createTempDirectory("iceberg-strip-meta-test").resolve("v1.metadata.json");
+ TableMetadataParser.write(metadata,
fileIO.newOutputFile(metadataLocation));
+ Table baseTable = new BaseTable(new
StaticTableOperations(metadataLocation, fileIO), "ice01.db.t");
+ Table filesTable =
MetadataTableUtils.createMetadataTableInstance(baseTable,
MetadataTableType.FILES);
+
+ Configuration conf = new Configuration();
+ String serialized = HiveTableUtil.serializeTable(
+ IcebergVendedCredentialUtil.secretFreeCopy(filesTable, conf), conf,
new Properties(), null);
+
+ String rawBytes = new String(
+ java.util.Base64.getMimeDecoder().decode(serialized),
StandardCharsets.ISO_8859_1);
+
assertThat(rawBytes).doesNotContain("cred-secret-3fd2").doesNotContain("prop-secret-3fd2");
+
+ Map<String, String> jobSecrets = Maps.newHashMap();
+ IcebergVendedCredentialUtil.propagateToJob(filesTable, "ice01", null,
jobSecrets, new Configuration());
+
+ JobConf taskConf = new JobConf();
+ taskConf.set(InputFormatConfig.SERIALIZED_TABLE_PREFIX + "db.t.files",
serialized);
+ jobSecrets.forEach(taskConf::set);
+ taskConf.set(InputFormatConfig.CATALOG_NAME, "ice01");
+
+ Table executorTable = HiveTableUtil.deserializeTable(taskConf,
"db.t.files");
+
+ assertThat(executorTable).isNotNull();
+ assertThat(executorTable.name()).isEqualTo(filesTable.name());
+
assertThat(executorTable.schema().asStruct()).isEqualTo(filesTable.schema().asStruct());
+ assertThat(executorTable.io().properties())
+ .doesNotContainKey(S3FileIOProperties.SECRET_ACCESS_KEY);
+ assertThat(((SupportsStorageCredentials)
executorTable.io()).credentials()).hasSize(1);
+ }
+
+ /**
+ * Tests the HS2 commit-path variant: {@code collectOutputs} resolves the
catalog name per table
+ * (the job-level {@code CATALOG_NAME} is not set in the DAG conf) and
session endpoint overrides
+ * must still reach the live table's FileIO through {@code setCredentials}.
+ */
+ @Test
+ public void applyFromJobConfWithExplicitCatalogAppliesEndpointOverride() {
+ CredentialFileIO fileIO = new CredentialFileIO();
+ fileIO.setCredentials(
+ List.of(
+ StorageCredential.create(
+ "s3://my-bucket/",
+ Map.of(S3FileIOProperties.ACCESS_KEY_ID, "commit-access",
+ S3FileIOProperties.SECRET_ACCESS_KEY, "commit-secret-4fa1",
+ S3FileIOProperties.ENDPOINT, "http://minio:9000";))));
+ Schema schema = new Schema(Types.NestedField.required(1, "x",
Types.IntegerType.get()));
+ TableMetadata metadata =
+ TableMetadata.newTableMetadata(schema, PartitionSpec.unpartitioned(),
"s3://my-bucket/warehouse/t", Map.of());
+ Table table = new BaseTable(new StaticTableOperations(metadata, fileIO),
"ice01.db.t");
+
+ JobConf jobConf = new JobConf();
+ jobConf.set("iceberg.catalog.ice01.header.X-Iceberg-Access-Delegation",
"vended-credentials");
+ jobConf.set("iceberg.catalog.ice01." + S3FileIOProperties.ENDPOINT,
"http://localhost:9100";);
+
+ IcebergVendedCredentialUtil.applyFromJobConf(table, "ice01", jobConf);
+
+ assertThat(fileIO.credentials()).hasSize(1);
+ assertThat(fileIO.credentials().get(0).config())
+ .containsEntry(S3FileIOProperties.ENDPOINT, "http://localhost:9100";)
+ .containsEntry(S3FileIOProperties.SECRET_ACCESS_KEY,
"commit-secret-4fa1");
+ }
+
+ /**
+ * Pins the leak that makes the secret-free copy before serialization
necessary.
+ *
+ * <p><b>Setup:</b> {@code SerializableTable.copyOf} stores a
<em>reference</em> to the live
+ * table's FileIO in a non-transient field, and Java serialization walks it
— so serializing an
+ * unsanitized copy embeds the FileIO's credential list and secret
properties in the blob. The
+ * stub mirrors {@code S3FileIO}'s serialized shape: {@code properties} and
+ * {@code storageCredentials} are both non-transient there too (the real
implementation is
+ * covered by the Gravitino qtest).
+ *
+ * <p><b>Expects:</b> secrets ARE present in the raw serialized bytes when
no sanitization runs.
+ *
+ * <p><b>Why:</b> guards the premise. If Iceberg ever stops serializing
FileIO secret state,
+ * this test fails and serializing tables as-is becomes safe again.
+ */
+ @Test
+ public void serializableTableCopyLeaksIoSecretsWithoutSanitization() {
+ CredentialFileIO fileIO = new CredentialFileIO();
+ fileIO.initialize(Map.of(S3FileIOProperties.SECRET_ACCESS_KEY,
"prop-secret-77aa"));
+ fileIO.setCredentials(
+ List.of(
+ StorageCredential.create(
+ "s3://my-bucket/",
+ Map.of(S3FileIOProperties.ACCESS_KEY_ID, "cred-access-77aa",
+ S3FileIOProperties.SECRET_ACCESS_KEY,
"cred-secret-77aa"))));
+ Schema schema = new Schema(Types.NestedField.required(1, "x",
Types.IntegerType.get()));
+ TableMetadata metadata =
+ TableMetadata.newTableMetadata(schema, PartitionSpec.unpartitioned(),
"s3://my-bucket/warehouse/t", Map.of());
+ Table table = new BaseTable(new StaticTableOperations(metadata, fileIO),
"db.t");
+
+ String unsanitized =
SerializationUtil.serializeToBase64(SerializableTable.copyOf(table));
+
+ String rawBytes = new String(
+ java.util.Base64.getMimeDecoder().decode(unsanitized),
StandardCharsets.ISO_8859_1);
+ assertThat(rawBytes)
+ .contains("cred-access-77aa")
+ .contains("cred-secret-77aa")
+ .contains("prop-secret-77aa");
+ }
+
+ /** Public with a no-arg constructor: the secret-free copy instantiates it
reflectively via
+ * {@code CatalogUtil.loadFileIO}. File operations delegate to local Hadoop
IO so metadata
+ * files can be written and read in tests. */
+ public static class CredentialFileIO implements FileIO,
SupportsStorageCredentials {
+ private List<StorageCredential> credentials = List.of();
+ private Map<String, String> properties = Map.of();
+ private transient HadoopFileIO localFiles;
+
+ @Override
+ public void initialize(Map<String, String> props) {
+ this.properties = props;
+ }
+
+ private HadoopFileIO localFiles() {
+ if (localFiles == null) {
+ localFiles = new HadoopFileIO(new Configuration());
+ }
+ return localFiles;
+ }
+
+ @Override
+ public org.apache.iceberg.io.InputFile newInputFile(String path) {
+ return localFiles().newInputFile(path);
+ }
+
+ @Override
+ public org.apache.iceberg.io.OutputFile newOutputFile(String path) {
+ return localFiles().newOutputFile(path);
+ }
+
+ @Override
+ public void deleteFile(String path) {
+ localFiles().deleteFile(path);
+ }
+
+ @Override
+ public Map<String, String> properties() {
+ return properties;
+ }
+
+ @Override
+ public void close() {
+ // No-op: test stub does not hold resources.
+ }
+
+ @Override
+ public List<StorageCredential> credentials() {
+ return credentials;
+ }
+
+ @Override
+ public void setCredentials(List<StorageCredential> creds) {
+ this.credentials = creds;
+ }
+ }
+}
diff --git
a/iceberg/iceberg-handler/src/test/queries/positive/iceberg_rest_catalog_gravitino.q
b/iceberg/iceberg-handler/src/test/queries/positive/iceberg_rest_catalog_gravitino.q
index 91aa14fdc17..66d11bbd870 100644
---
a/iceberg/iceberg-handler/src/test/queries/positive/iceberg_rest_catalog_gravitino.q
+++
b/iceberg/iceberg-handler/src/test/queries/positive/iceberg_rest_catalog_gravitino.q
@@ -3,6 +3,10 @@
--! qt:replace:/(\s+neededVirtualColumns:\s)(.*)/$1#Masked#/
-- Mask random uuid
--! qt:replace:/(\s+'uuid'=')\S+('\s*)/$1#Masked#$2/
+-- Mask Iceberg metadata file name (sequence id + UUID) in show create table
+--! qt:replace:/('metadata_location'=')[^']+(')/$1#Masked#$2/
+-- Mask metadata_location path in describe formatted ('|' delimiter: regex
contains s3:// so '/' breaks QTestReplaceHandler).
+--! qt:replace:|(metadata_location)(\s+)(s3://\S+)|$1$2#Masked#|
-- Mask random uuid
--! qt:replace:/(\s+uuid\s+)\S+(\s*)/$1#Masked#$2/
-- Mask a random snapshot id
@@ -24,9 +28,10 @@ set hive.stats.autogather=false;
set metastore.client.impl=org.apache.iceberg.hive.client.HiveRESTCatalogClient;
set metastore.catalog.default=ice01;
set iceberg.catalog.ice01.type=rest;
+set
iceberg.catalog.ice01.header.X-Iceberg-Access-Delegation=vended-credentials;
---! This config is set in the driver setup (see
TestIcebergRESTCatalogLlapLocalCliDriver.java)
---! conf.set('iceberg.catalog.ice01.uri', <RESTServer URI>);
+--! REST URI, OAuth, MinIO + Gravitino S3 warehouse / credential vending, and
host S3A are set in
+--! TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.
create database ice_rest;
use ice_rest;
@@ -64,11 +69,6 @@ show create table ice_orc2;
insert into ice_orc2 partition (company_id=100)
VALUES ('fn1','ln1', 1, 10), ('fn2','ln2', 2, 20), ('fn3','ln3', 3, 30);
---! In CI, Testcontainers' .withFileSystemBind() is not able to bind the same
host path to the same container path,
---! so as a workaround, the .metadata.json files from container are manually
synced in a daemon process,
---! since the sync can take some time, need to wait for it to happen after the
insert operation.
-! sleep 20;
-
describe formatted ice_orc2;
select * from ice_orc2;
diff --git
a/iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_rest_catalog_gravitino.q.out
b/iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_rest_catalog_gravitino.q.out
index 5821007cd3c..db4e5c2129a 100644
---
a/iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_rest_catalog_gravitino.q.out
+++
b/iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_rest_catalog_gravitino.q.out
@@ -77,14 +77,14 @@ STORED BY
WITH SERDEPROPERTIES (
'serialization.format'='1')
LOCATION
-#### A masked pattern was here ####
+ 's3://iceberg-vend/warehouse/ice_rest/ice_orc2'
TBLPROPERTIES (
'bucketing_version'='2',
'current-schema'='{"type":"struct","schema-id":0,"fields":[{"id":1,"name":"first_name","required":false,"type":"string"},{"id":2,"name":"last_name","required":false,"type":"string"},{"id":3,"name":"dept_id","required":false,"type":"long"},{"id":4,"name":"team_id","required":false,"type":"long"},{"id":5,"name":"company_id","required":false,"type":"long"}]}',
'default-partition-spec'='{"spec-id":0,"fields":[{"name":"company_id","transform":"identity","source-id":5,"field-id":1000}]}',
'format-version'='2',
'iceberg.catalog'='ice01',
-#### A masked pattern was here ####
+ 'metadata_location'='#Masked#',
'name'='ice_rest.ice_orc2',
'parquet.compression'='zstd',
'serialization.format'='1',
@@ -139,7 +139,7 @@ Table Parameters:
default-partition-spec
{\"spec-id\":0,\"fields\":[{\"name\":\"company_id\",\"transform\":\"identity\",\"source-id\":5,\"field-id\":1000}]}
format-version 2
iceberg.catalog ice01
-#### A masked pattern was here ####
+ metadata_location #Masked#
name ice_rest.ice_orc2
numFiles 1
numRows 3
@@ -220,7 +220,7 @@ Table Type: VIRTUAL_VIEW
Table Parameters:
bucketing_version 2
current-schema
{\"type\":\"struct\",\"schema-id\":0,\"fields\":[{\"id\":1,\"name\":\"first_name\",\"required\":false,\"type\":\"string\"},{\"id\":2,\"name\":\"last_name\",\"required\":false,\"type\":\"string\"}]}
-#### A masked pattern was here ####
+ metadata_location #Masked#
storage_handler
org.apache.iceberg.mr.hive.HiveIcebergStorageHandler
table_type ICEBERG-VIEW
type rest
@@ -279,7 +279,7 @@ Table Type: VIRTUAL_VIEW
Table Parameters:
bucketing_version 2
current-schema
{\"type\":\"struct\",\"schema-id\":0,\"fields\":[{\"id\":1,\"name\":\"first_name\",\"required\":false,\"type\":\"string\"},{\"id\":2,\"name\":\"last_name\",\"required\":false,\"type\":\"string\"}]}
-#### A masked pattern was here ####
+ metadata_location #Masked#
storage_handler
org.apache.iceberg.mr.hive.HiveIcebergStorageHandler
table_type ICEBERG-VIEW
type rest
@@ -336,7 +336,7 @@ Table Type: VIRTUAL_VIEW
Table Parameters:
bucketing_version 2
current-schema
{\"type\":\"struct\",\"schema-id\":1,\"fields\":[{\"id\":1,\"name\":\"_c0\",\"required\":false,\"type\":\"string\"}]}
-#### A masked pattern was here ####
+ metadata_location #Masked#
storage_handler
org.apache.iceberg.mr.hive.HiveIcebergStorageHandler
table_type ICEBERG-VIEW
type rest
@@ -405,7 +405,7 @@ Table Type: VIRTUAL_VIEW
Table Parameters:
bucketing_version 2
current-schema
{\"type\":\"struct\",\"schema-id\":0,\"fields\":[{\"id\":1,\"name\":\"first_name\",\"required\":false,\"type\":\"string\"},{\"id\":2,\"name\":\"_c1\",\"required\":false,\"type\":\"string\"}]}
-#### A masked pattern was here ####
+ metadata_location #Masked#
storage_handler
org.apache.iceberg.mr.hive.HiveIcebergStorageHandler
table_type ICEBERG-VIEW
type rest
diff --git a/itests/qtest-iceberg/pom.xml b/itests/qtest-iceberg/pom.xml
index 3dc736007f4..035c79a1445 100644
--- a/itests/qtest-iceberg/pom.xml
+++ b/itests/qtest-iceberg/pom.xml
@@ -140,6 +140,11 @@
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>com.sun.jersey</groupId>
<artifactId>jersey-servlet</artifactId>
@@ -156,6 +161,11 @@
<artifactId>hadoop-archives</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-aws</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
@@ -271,6 +281,10 @@
<artifactId>hadoop-mapreduce-client-core</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-iceberg-handler</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-iceberg-handler</artifactId>
@@ -364,6 +378,12 @@
<artifactId>hbase-mapreduce</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-aws</artifactId>
+ <version>${iceberg.version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.tez</groupId>
<artifactId>tez-tests</artifactId>
@@ -496,6 +516,12 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>bundle</artifactId>
+ <version>${aws-java-sdk.version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
diff --git
a/itests/qtest-iceberg/src/test/java/org/apache/hadoop/hive/cli/TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.java
b/itests/qtest-iceberg/src/test/java/org/apache/hadoop/hive/cli/TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.java
index 89545f83a07..4c04578e244 100644
---
a/itests/qtest-iceberg/src/test/java/org/apache/hadoop/hive/cli/TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.java
+++
b/itests/qtest-iceberg/src/test/java/org/apache/hadoop/hive/cli/TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.java
@@ -18,9 +18,6 @@
package org.apache.hadoop.hive.cli;
-import com.github.dockerjava.api.command.CopyArchiveFromContainerCmd;
-import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
-import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.cli.control.CliAdapter;
@@ -28,7 +25,10 @@
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.aws.AwsClientProperties;
+import org.apache.iceberg.aws.s3.S3FileIOProperties;
import org.apache.iceberg.hive.IcebergCatalogProperties;
+import org.apache.iceberg.hive.client.RestAccessDelegationMode;
import org.apache.iceberg.hive.client.HiveRESTCatalogClient;
import org.apache.iceberg.rest.extension.OAuth2AuthorizationServer;
import org.junit.After;
@@ -42,13 +42,13 @@
import org.junit.runners.Parameterized.Parameters;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.Network;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.containers.wait.strategy.WaitAllStrategy;
import org.testcontainers.utility.DockerImageName;
import org.testcontainers.utility.MountableFile;
-import org.testcontainers.containers.GenericContainer;
import java.io.File;
import java.io.IOException;
@@ -59,22 +59,30 @@
import java.nio.file.Paths;
import java.time.Duration;
import java.util.List;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
+/**
+ * LLAP {@link CliAdapter} qtests for Hive against the Gravitino Iceberg REST
server image
+ * ({@link #GRAVITINO_IMAGE}), with OAuth2 on the catalog HTTP API and an
Iceberg warehouse on
+ * Apache Ozone S3 Gateway using Gravitino {@code s3-secret-key} credential
vending (see
+ * {@link #GRAVITINO_S3_CONF_TEMPLATE}).
+ *
+ * <p>Table metadata and data live under {@code s3://} in {@link #S3_BUCKET}.
The host temp directory
+ * {@link #warehouseDir} is used to render the Gravitino server configuration
and as the source path for
+ * {@code .withCopyFileToContainer} (config + H2 driver JAR).</p>
+ */
@RunWith(Parameterized.class)
public class TestIcebergRESTCatalogGravitinoLlapLocalCliDriver {
private static final CliAdapter CLI_ADAPTER =
new
CliConfigs.TestIcebergRESTCatalogGravitinoLlapLocalCliDriver().getCliAdapter();
-
+
private static final Logger LOG =
LoggerFactory.getLogger(TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.class);
-
+
private static final String CATALOG_NAME = "ice01";
private static final long GRAVITINO_STARTUP_TIMEOUT_MINUTES = 5L;
private static final int GRAVITINO_HTTP_PORT = 9001;
- private static final String GRAVITINO_CONF_FILE_TEMPLATE =
"gravitino-h2-test-template.conf";
+
+ private static final String GRAVITINO_S3_CONF_TEMPLATE =
"gravitino-s3-vended-oauth-template.conf";
private static final String GRAVITINO_ROOT_DIR =
"/root/gravitino-iceberg-rest-server";
private static final String GRAVITINO_STARTUP_SCRIPT = GRAVITINO_ROOT_DIR +
"/bin/start-iceberg-rest-server.sh";
private static final String GRAVITINO_H2_LIB = GRAVITINO_ROOT_DIR +
"/libs/h2-driver.jar";
@@ -82,6 +90,8 @@ public class
TestIcebergRESTCatalogGravitinoLlapLocalCliDriver {
private static final DockerImageName GRAVITINO_IMAGE =
DockerImageName.parse("apache/gravitino-iceberg-rest:1.0.0");
+ private static final String S3_BUCKET = "iceberg-vend";
+
private static final String OAUTH2_SERVER_ICEBERG_CLIENT_ID =
"iceberg-client";
private static final String OAUTH2_SERVER_ICEBERG_CLIENT_SECRET =
"iceberg-client-secret";
@@ -89,8 +99,8 @@ public class
TestIcebergRESTCatalogGravitinoLlapLocalCliDriver {
private final File qfile;
private GenericContainer<?> gravitinoContainer;
+ private OzoneS3GatewayContainers ozoneS3;
private Path warehouseDir;
- private final ScheduledExecutorService fileSyncExecutor =
Executors.newSingleThreadScheduledExecutor();
private OAuth2AuthorizationServer oAuth2AuthorizationServer;
@Parameters(name = "{0}")
@@ -110,14 +120,15 @@ public
TestIcebergRESTCatalogGravitinoLlapLocalCliDriver(String name, File qfile
}
@Before
- public void setup() throws IOException {
+ public void setup() throws Exception {
Network dockerNetwork = Network.newNetwork();
-
+
startOAuth2AuthorizationServer(dockerNetwork);
createWarehouseDir();
+ ozoneS3 = new OzoneS3GatewayContainers();
+ ozoneS3.start(dockerNetwork, S3_BUCKET);
prepareGravitinoConfig();
startGravitinoContainer(dockerNetwork);
- fileSyncExecutor.scheduleAtFixedRate(this::syncWarehouseDir, 0, 5,
TimeUnit.SECONDS);
String host = gravitinoContainer.getHost();
Integer port = gravitinoContainer.getMappedPort(GRAVITINO_HTTP_PORT);
@@ -137,63 +148,123 @@ public void setup() throws IOException {
conf.set(restCatalogPrefix + "rest.auth.type", "oauth2");
conf.set(restCatalogPrefix + "oauth2-server-uri",
oAuth2AuthorizationServer.getTokenEndpoint());
conf.set(restCatalogPrefix + "credential",
oAuth2AuthorizationServer.getClientCredential());
+ conf.set(
+ restCatalogPrefix +
IcebergCatalogProperties.REST_ACCESS_DELEGATION_HEADER_PROPERTY,
+ RestAccessDelegationMode.VENDED_CREDENTIALS.modeName());
+
+ applyHostS3FilesystemSettings(conf);
+ applyIcebergS3ClientEndpointOverride(conf, restCatalogPrefix);
}
@After
- public void teardown() throws IOException {
+ public void teardown() throws Exception {
if (gravitinoContainer != null) {
gravitinoContainer.stop();
}
-
+
+ if (ozoneS3 != null) {
+ ozoneS3.stop();
+ }
+
if (oAuth2AuthorizationServer != null) {
oAuth2AuthorizationServer.stop();
}
- fileSyncExecutor.shutdownNow();
- FileUtils.deleteDirectory(warehouseDir.toFile());
+ if (warehouseDir != null) {
+ FileUtils.deleteDirectory(warehouseDir.toFile());
+ }
+ }
+
+ /**
+ * Puts host-reachable Iceberg S3 client settings on the HS2 session.
+ *
+ * <p>Gravitino runs in Docker and vends storage credentials whose endpoint
is
+ * {@link OzoneS3GatewayContainers#S3_DOCKER_ENDPOINT} (reachable inside the
compose network only).
+ * Hive runs on the host, so this sets {@code
iceberg.catalog.<catalog>.s3.endpoint} to the
+ * published Ozone S3G host/port, plus path-style and region, matching what
operators configure in
+ * {@code hive-site.xml} in a real deployment.
+ *
+ * <p>Those keys are the <em>source</em> for {@link
org.apache.iceberg.mr.hive.IcebergVendedCredentialUtil}:
+ * at plan time and on tasks it merges them over the vended endpoint in
credentials and job conf. This
+ * method does not replace that util — it seeds session conf so the util has
a host endpoint to apply.
+ */
+ private void applyIcebergS3ClientEndpointOverride(Configuration conf, String
restCatalogPrefix) {
+ String host = ozoneS3.getHost();
+ int port = ozoneS3.getMappedPort();
+ @SuppressWarnings("HttpUrlsUsage")
+ String icebergS3Endpoint = String.format("http://%s:%d";, host, port);
+ conf.set(restCatalogPrefix + S3FileIOProperties.ENDPOINT,
icebergS3Endpoint);
+ conf.set(restCatalogPrefix + S3FileIOProperties.PATH_STYLE_ACCESS, "true");
+ conf.set(restCatalogPrefix + AwsClientProperties.CLIENT_REGION,
"us-east-1");
+ }
+
+ /**
+ * Wires Hadoop to use S3A for {@code s3://} on the host-visible Ozone S3G
endpoint.
+ *
+ * <p><b>What:</b> sets {@code fs.s3}/{@code fs.s3a} implementation classes,
per-bucket endpoint and path-style
+ * ({@code fs.s3a.bucket.<bucket>.*}), and disables SSL for this local
Ozone S3G test.
+ *
+ * <p><b>Why:</b> Hive and Tez use Hadoop {@code FileSystem} for many {@code
s3://} paths, not only Iceberg
+ * {@code S3FileIO}. {@link
org.apache.iceberg.mr.hive.IcebergVendedCredentialUtil} propagates vended keys
and
+ * overlapping S3A settings onto <em>job</em> conf at plan time; it does not
register the S3A filesystem or seed
+ * session defaults. Session wiring is still needed for HS2 and for paths
that read session conf before job
+ * properties exist.
+ *
+ * <p>Access keys are intentionally omitted here; they are vended per query
and copied into job secrets by the util.
+ */
+ private void applyHostS3FilesystemSettings(Configuration conf) {
+ String s3Host = ozoneS3.getHost();
+ int s3Port = ozoneS3.getMappedPort();
+ @SuppressWarnings("HttpUrlsUsage")
+ String endpoint = String.format("http://%s:%d";, s3Host, s3Port);
+ conf.set("fs.s3.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem");
+ conf.set("fs.AbstractFileSystem.s3.impl", "org.apache.hadoop.fs.s3a.S3A");
+ conf.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem");
+ String bucketPrefix = "fs.s3a.bucket." + S3_BUCKET + ".";
+ conf.set(bucketPrefix + "endpoint", endpoint);
+ conf.setBoolean(bucketPrefix + "path.style.access", true);
+ conf.setBoolean(bucketPrefix + "connection.ssl.enabled", false);
}
/**
- * Starts a Gravitino container with the Iceberg REST server configured for
testing.
+ * Starts a Gravitino container with the Iceberg REST server configured for
this test.
*
* <p>This method configures the container to:
* <ul>
- * <li>Expose container REST port GRAVITINO_HTTP_PORT and map it to a host
port.</li>
- * <li>Modify the container entrypoint to create the warehouse directory
before startup.</li>
- * <li>Copy a dynamically prepared Gravitino configuration file into the
container.</li>
- * <li>Copy the H2 driver JAR into the server's lib directory.</li>
- * <li>Wait for the Gravitino Iceberg REST server to finish starting
(based on logs and port checks).</li>
- * <li>Stream container logs into the test logger for easier
debugging.</li>
+ * <li>Expose {@link #GRAVITINO_HTTP_PORT} on the container and map it to
a host port.</li>
+ * <li>Adjust the entrypoint so a bootstrap directory exists before {@link
#GRAVITINO_STARTUP_SCRIPT} runs
+ * (the Iceberg <em>warehouse</em> itself is {@code s3://} on Ozone
S3G, not this path).</li>
+ * <li>Copy the rendered Gravitino configuration from {@link
#warehouseDir} into the image at
+ * {@link #GRAVITINO_CONF_FILE}.</li>
+ * <li>Copy the H2 driver JAR (JDBC catalog backend metadata) into {@link
#GRAVITINO_H2_LIB}.</li>
+ * <li>Attach the container to {@code dockerNetwork} so it reaches the
OAuth2 server and Ozone S3G
+ * ({@link OzoneS3GatewayContainers#S3_DOCKER_ALIAS}).</li>
+ * <li>Wait for the Gravitino Iceberg REST server to finish starting (log
line + listening port).</li>
+ * <li>Stream container logs into {@link #LOG}.</li>
* </ul>
*
- * <p>Note: The {@code @SuppressWarnings("resource")} annotation is applied
because
- * IntelliJ and some compilers flag {@link
org.testcontainers.containers.GenericContainer}
- * as a resource that should be managed with try-with-resources. In this
test setup,
- * the container lifecycle is managed explicitly: it is started here and
stopped in
- * {@code @After} (via {@code gravitinoContainer.stop()}). Using
try-with-resources
- * would not work in this context, since the container must remain running
across
- * multiple test methods rather than being confined to a single block
scope.</p>
+ * <p>Note: the {@code @SuppressWarnings("resource")} annotation is applied
because IntelliJ and some compilers
+ * flag {@link GenericContainer} as a resource that should be used with
try-with-resources. Here the container
+ * lifecycle is explicit: it is started in this method and stopped in {@link
#teardown()} via
+ * {@code gravitinoContainer.stop()}.</p>
*/
@SuppressWarnings("resource")
private void startGravitinoContainer(Network dockerNetwork) {
gravitinoContainer = new GenericContainer<>(GRAVITINO_IMAGE)
.withExposedPorts(GRAVITINO_HTTP_PORT)
- // Update entrypoint to create the warehouse directory before starting
the server
+ // Bootstrap dir for the server script; warehouse is s3:// on Ozone
S3G (see template)
.withCreateContainerCmdModifier(cmd -> cmd.withEntrypoint("bash", "-c",
- String.format("mkdir -p %s && exec %s", warehouseDir.toString(),
GRAVITINO_STARTUP_SCRIPT)))
- // Mount gravitino configuration file
+ "mkdir -p /tmp/gravitino-bootstrap && exec " +
GRAVITINO_STARTUP_SCRIPT))
+ // Mount Gravitino configuration file (rendered under warehouseDir on
the host)
.withCopyFileToContainer(
- MountableFile.forHostPath(Paths.get(warehouseDir.toString(),
GRAVITINO_CONF_FILE_TEMPLATE)),
- GRAVITINO_CONF_FILE
- )
+ MountableFile.forHostPath(Paths.get(warehouseDir.toString(),
GRAVITINO_S3_CONF_TEMPLATE)),
+ GRAVITINO_CONF_FILE)
// Mount the H2 driver JAR into the server's lib directory
.withCopyFileToContainer(
MountableFile.forHostPath(
- Paths.get("target", "test-dependencies",
"h2-driver.jar").toAbsolutePath()
- ),
- GRAVITINO_H2_LIB
- )
- // Use the same Docker network as the OAuth2 server so they can
communicate
+ Paths.get("target", "test-dependencies",
"h2-driver.jar").toAbsolutePath()),
+ GRAVITINO_H2_LIB)
+ // Same Docker network as OAuth2 and Ozone S3G (Gravitino uses
http://s3.ozone:9878 in config)
.withNetwork(dockerNetwork)
// Wait for the server to be fully started
.waitingFor(
@@ -201,123 +272,71 @@ private void startGravitinoContainer(Network
dockerNetwork) {
.withStrategy(Wait.forLogMessage(".*GravitinoIcebergRESTServer
is running.*\\n", 1)
.withStartupTimeout(Duration.ofMinutes(GRAVITINO_STARTUP_TIMEOUT_MINUTES)))
.withStrategy(Wait.forListeningPort()
-
.withStartupTimeout(Duration.ofMinutes(GRAVITINO_STARTUP_TIMEOUT_MINUTES)))
- )
- .withLogConsumer(new Slf4jLogConsumer(LoggerFactory
-
.getLogger(TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.class)));
+
.withStartupTimeout(Duration.ofMinutes(GRAVITINO_STARTUP_TIMEOUT_MINUTES))))
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
gravitinoContainer.start();
}
- /**
- * Starts a background daemon that continuously synchronizes the Iceberg
warehouse
- * directory from the running Gravitino container to the host file system.
- *
- * <p>In CI environments, Testcontainers' {@code .withFileSystemBind()}
cannot reliably
- * bind the same host path to the same path inside the container, especially
when
- * using remote Docker hosts or Docker-in-Docker setups. This causes the
container's
- * writes (e.g., Iceberg metadata files like {@code .metadata.json}) to be
invisible
- * on the host.</p>
- *
- * <p>This method works around that limitation by repeatedly copying new
files from
- * the container's warehouse directory to the corresponding host directory.
Existing
- * files on the host are preserved, and only files that do not yet exist are
copied.
- * The sync runs every 1 second while the container is running.</p>
- *
- * <p>Each archive copy from the container is extracted using a {@link
TarArchiveInputStream},
- * and directories are created as needed. Files that already exist on the
host are skipped
- * to avoid overwriting container data.</p>
- */
- private void syncWarehouseDir() {
- if (gravitinoContainer.isRunning()) {
- try (CopyArchiveFromContainerCmd copyArchiveFromContainerCmd =
- gravitinoContainer
- .getDockerClient()
-
.copyArchiveFromContainerCmd(gravitinoContainer.getContainerId(),
warehouseDir.toString());
- InputStream tarStream = copyArchiveFromContainerCmd.exec();
- TarArchiveInputStream tis = new TarArchiveInputStream(tarStream)) {
-
- TarArchiveEntry entry;
- while ((entry = tis.getNextEntry()) != null) {
- // Skip directories because we only want to copy metadata files from
the container.
- if (entry.isDirectory()) {
- continue;
- }
-
- /*
- * Tar entry names include a container-specific top-level folder,
e.g.:
- *
iceberg-test-1759245909247/iceberg_warehouse/ice_rest/.../metadata.json
- *
- * Strip the first part so the relative path inside the warehouse is
preserved
- * when mapping to the host warehouseDir.
- */
-
- String[] parts = entry.getName().split("/", 2);
- if (parts.length < 2) {
- continue; // defensive guard
- }
-
- Path relativePath = Paths.get(parts[1]);
- Path outputPath = warehouseDir.resolve(relativePath);
-
- // Skip if already present on host to avoid overwriting
- if (Files.exists(outputPath)) {
- continue;
- }
-
- Files.createDirectories(outputPath.getParent());
- Files.copy(tis, outputPath);
- }
-
- } catch (Exception e) {
- LOG.error("Warehouse folder sync failed: {}", e.getMessage());
- }
- }
- }
-
+ /** Keycloak-backed OAuth2 used by Gravitino REST authentication and by the
Hive REST client. */
private void startOAuth2AuthorizationServer(Network dockerNetwork) {
oAuth2AuthorizationServer = new OAuth2AuthorizationServer(dockerNetwork,
false);
oAuth2AuthorizationServer.start();
}
+ /**
+ * Host directory used to write the rendered Gravitino config (see {@link
#prepareGravitinoConfig}) and as the
+ * source path for {@code .withCopyFileToContainer} in {@link
#startGravitinoContainer}. This is not the Iceberg
+ * warehouse root; the warehouse is {@code s3://}{@link #S3_BUCKET}{@code
/...} on Ozone S3G.
+ */
private void createWarehouseDir() {
try {
warehouseDir = Paths.get("/tmp", "iceberg-test-" +
System.currentTimeMillis()).toAbsolutePath();
Files.createDirectories(warehouseDir);
} catch (Exception e) {
- throw new RuntimeException("Failed to create the Iceberg warehouse
directory", e);
+ throw new RuntimeException("Failed to create temp directory for
Gravitino config staging", e);
}
}
+ /**
+ * Reads {@link #GRAVITINO_S3_CONF_TEMPLATE} from the classpath, substitutes
bucket / Ozone S3 / OAuth placeholders,
+ * and writes the result under {@link #warehouseDir} for copying into the
Gravitino container.
+ */
private void prepareGravitinoConfig() throws IOException {
String content;
try (InputStream in =
TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.class.getClassLoader()
- .getResourceAsStream(GRAVITINO_CONF_FILE_TEMPLATE)) {
+ .getResourceAsStream(GRAVITINO_S3_CONF_TEMPLATE)) {
if (in == null) {
- throw new IOException("Resource not found: " +
GRAVITINO_CONF_FILE_TEMPLATE);
+ throw new IOException("Resource not found: " +
GRAVITINO_S3_CONF_TEMPLATE);
}
content = new String(in.readAllBytes(), StandardCharsets.UTF_8);
}
String updatedContent = content
- .replace("/WAREHOUSE_DIR", warehouseDir.toString())
+ .replace("S3_BUCKET", S3_BUCKET)
+ .replace("S3_ACCESS_KEY", OzoneS3GatewayContainers.ACCESS_KEY)
+ .replace("S3_SECRET_KEY", OzoneS3GatewayContainers.SECRET_KEY)
.replace("OAUTH2_SERVER_URI", oAuth2AuthorizationServer.getIssuer())
.replace("OAUTH2_JWKS_URI", getJwksUri())
.replace("OAUTH2_CLIENT_ID", OAUTH2_SERVER_ICEBERG_CLIENT_ID)
.replace("OAUTH2_CLIENT_SECRET", OAUTH2_SERVER_ICEBERG_CLIENT_SECRET)
.replace("HTTP_PORT", String.valueOf(GRAVITINO_HTTP_PORT));
- Path configFile = warehouseDir.resolve(GRAVITINO_CONF_FILE_TEMPLATE);
+ Path configFile = warehouseDir.resolve(GRAVITINO_S3_CONF_TEMPLATE);
Files.writeString(configFile, updatedContent);
}
+ /**
+ * JWKS URL reachable from <em>inside</em> the Gravitino container:
host/port in the issuer are rewritten to the
+ * Keycloak container hostname and its internal HTTP port.
+ */
private String getJwksUri() {
String reachableHost =
oAuth2AuthorizationServer.getKeycloackContainerDockerInternalHostName();
int internalPort = 8080; // Keycloak container's internal port
return oAuth2AuthorizationServer.getIssuer()
.replace("localhost", reachableHost)
.replace("127.0.0.1", reachableHost)
- // replace issuer's mapped port with keyclock container's internal port
+ // Replace issuer's mapped host port with Keycloak's internal port on
the Docker network
.replaceFirst(":[0-9]+", ":" + internalPort);
}
diff --git
a/itests/qtest-iceberg/src/test/resources/gravitino-s3-vended-oauth-template.conf
b/itests/qtest-iceberg/src/test/resources/gravitino-s3-vended-oauth-template.conf
new file mode 100644
index 00000000000..9f78ec628da
--- /dev/null
+++
b/itests/qtest-iceberg/src/test/resources/gravitino-s3-vended-oauth-template.conf
@@ -0,0 +1,37 @@
+gravitino.iceberg-rest.httpPort = HTTP_PORT
+
+# --- Iceberg REST Catalog Backend (JDBC for catalog metadata) ---
+gravitino.iceberg-rest.catalog-backend = jdbc
+gravitino.iceberg-rest.uri = jdbc:h2:file:/tmp/gravitino_h2_db;AUTO_SERVER=TRUE
+gravitino.iceberg-rest.jdbc-driver = org.h2.Driver
+gravitino.iceberg-rest.jdbc-user = sa
+gravitino.iceberg-rest.jdbc-password = ""
+gravitino.iceberg-rest.jdbc-initialize = true
+
+# --- Warehouse on Ozone S3G + credential vending (s3-secret-key) ---
+gravitino.iceberg-rest.warehouse = s3://S3_BUCKET/warehouse
+gravitino.iceberg-rest.io-impl = org.apache.iceberg.aws.s3.S3FileIO
+gravitino.iceberg-rest.credential-providers = s3-secret-key
+gravitino.iceberg-rest.s3-access-key-id = S3_ACCESS_KEY
+gravitino.iceberg-rest.s3-secret-access-key = S3_SECRET_KEY
+# Docker network only (Gravitino container -> Ozone S3G). Host-side
Hive/Tez/Iceberg use
+# iceberg.catalog.ice01.s3.endpoint (no static keys); credentials are vended
per table load.
+gravitino.iceberg-rest.s3-endpoint = http://s3.ozone:9878
+gravitino.iceberg-rest.s3-path-style-access = true
+gravitino.iceberg-rest.s3-region = us-east-1
+
+# --- OAuth2 Authentication (catalog HTTP API) ---
+gravitino.authenticators = oauth
+
+gravitino.authenticator.oauth.serverUri = OAUTH2_SERVER_URI
+gravitino.authenticator.oauth.tokenPath = /protocol/openid-connect/token
+gravitino.authenticator.oauth.clientId = OAUTH2_CLIENT_ID
+gravitino.authenticator.oauth.scope = openid catalog
+gravitino.authenticator.oauth.clientSecret = OAUTH2_CLIENT_SECRET
+
+gravitino.authenticator.oauth.tokenValidatorClass =
org.apache.gravitino.server.authentication.JwksTokenValidator
+gravitino.authenticator.oauth.jwksUri =
OAUTH2_JWKS_URI/protocol/openid-connect/certs
+gravitino.authenticator.oauth.provider = default
+gravitino.authenticator.oauth.principalFields = sub
+gravitino.authenticator.oauth.allowSkewSecs = 60
+gravitino.authenticator.oauth.serviceAudience = hive-metastore
diff --git a/itests/util/pom.xml b/itests/util/pom.xml
index 7c26283e502..54b5ab17756 100644
--- a/itests/util/pom.xml
+++ b/itests/util/pom.xml
@@ -197,6 +197,10 @@
<groupId>junit</groupId>
<artifactId>junit</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers</artifactId>
+ </dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>mariadb</artifactId>
diff --git
a/itests/util/src/main/java/org/apache/hadoop/hive/cli/OzoneS3GatewayContainers.java
b/itests/util/src/main/java/org/apache/hadoop/hive/cli/OzoneS3GatewayContainers.java
new file mode 100644
index 00000000000..6a6186c70f5
--- /dev/null
+++
b/itests/util/src/main/java/org/apache/hadoop/hive/cli/OzoneS3GatewayContainers.java
@@ -0,0 +1,203 @@
+/*
+ * 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.hadoop.hive.cli;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.containers.wait.strategy.WaitAllStrategy;
+import org.testcontainers.utility.DockerImageName;
+
+/**
+ * Minimal Apache Ozone cluster with S3 Gateway for q-tests that need an
S3-compatible endpoint.
+ *
+ * <p>Layout matches {@code
packaging/src/docker/storage/ozone/docker-compose.yml}: scm, om, datanode,
+ * and s3g on a shared Docker network. S3G is reachable as {@link
#S3_DOCKER_ALIAS}:{@link #S3G_PORT}
+ * from other containers (for example Gravitino) and via the published host
port from the test JVM.
+ */
+public final class OzoneS3GatewayContainers {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(OzoneS3GatewayContainers.class);
+ private static final DockerImageName OZONE_IMAGE =
DockerImageName.parse("apache/ozone:2.1.0");
+
+ private static final int SCM_PORT = 9876;
+ private static final int OM_PORT = 9874;
+ /** Ozone 2.1.0 HDDS datanode client RPC port ({@code
hdds.datanode.client.port}). */
+ private static final int DATANODE_CLIENT_PORT = 19864;
+
+ private static final String S3_BUCKET_READY_MARKER = "OZONE_S3_BUCKET_READY";
+
+ /** S3 keys used when Ozone security is disabled (same as {@code
start-hive.sh --ozone}). */
+ public static final String ACCESS_KEY = "ozone";
+ public static final String SECRET_KEY = "secret";
+
+ public static final int S3G_PORT = 9878;
+ public static final String S3_DOCKER_ALIAS = "s3.ozone";
+ public static final String S3_DOCKER_ENDPOINT =
+ String.format("http://%s:%d";, S3_DOCKER_ALIAS, S3G_PORT);
+
+ private static final Duration STARTUP_TIMEOUT = Duration.ofMinutes(5);
+
+ private GenericContainer<?> scm;
+ private GenericContainer<?> om;
+ private GenericContainer<?> datanode;
+ private GenericContainer<?> s3g;
+
+ /**
+ * Starts scm → om → datanode → s3g on {@code network} and creates {@code
bucketName} under the
+ * default S3 volume ({@code /s3v}).
+ */
+ @SuppressWarnings("resource")
+ public void start(Network network, String bucketName) {
+ Map<String, String> commonEnv = commonEnv();
+ scm = startScm(network, commonEnv);
+ om = startOm(network, commonEnv);
+ datanode = startDatanode(network, commonEnv);
+ s3g = startS3g(network, commonEnv, bucketName);
+ LOG.info("Ozone S3G ready at {} (host {}:{})", S3_DOCKER_ENDPOINT,
s3g.getHost(), s3g.getMappedPort(S3G_PORT));
+ }
+
+ public void stop() {
+ stopQuietly(s3g);
+ stopQuietly(datanode);
+ stopQuietly(om);
+ stopQuietly(scm);
+ s3g = null;
+ datanode = null;
+ om = null;
+ scm = null;
+ }
+
+ public String getHost() {
+ return s3g.getHost();
+ }
+
+ public int getMappedPort() {
+ return s3g.getMappedPort(S3G_PORT);
+ }
+
+ private static Map<String, String> commonEnv() {
+ Map<String, String> env = new HashMap<>();
+ env.put("OZONE-SITE.XML_hdds.datanode.dir", "/data/hdds");
+ env.put("OZONE-SITE.XML_ozone.metadata.dirs", "/data/metadata");
+ env.put("OZONE-SITE.XML_ozone.om.address", "om");
+ env.put("OZONE-SITE.XML_ozone.om.http-address", "om:" + OM_PORT);
+ env.put("OZONE-SITE.XML_ozone.replication", "1");
+ env.put("OZONE-SITE.XML_ozone.scm.block.client.address", "scm");
+ env.put("OZONE-SITE.XML_ozone.scm.client.address", "scm");
+ env.put("OZONE-SITE.XML_ozone.scm.datanode.id.dir", "/data/metadata");
+ env.put("OZONE-SITE.XML_ozone.scm.names", "scm");
+ env.put("OZONE-SITE.XML_hdds.scm.safemode.min.datanode", "1");
+ env.put("OZONE-SITE.XML_hdds.scm.safemode.healthy.pipeline.pct", "0");
+ env.put("OZONE-SITE.XML_hdds.scm.safemode.enabled", "false");
+ // Path-style only: host JVM and Gravitino use http://host:9878/bucket/...
. Setting
+ // ozone.s3g.domain.name would require virtual-host Host headers (e.g.
bucket.s3.ozone).
+ // Keep CI/docker-desktop friendly (see
itests/test-docker/helm/ozone/values.yaml).
+ env.put("OZONE-SITE.XML_hdds.datanode.volume.min.free.space", "256MB");
+ env.put("OZONE-SITE.XML_ozone.scm.container.size", "128MB");
+ env.put("OZONE-SITE.XML_ozone.scm.block.size", "32MB");
+ env.put("no_proxy", "om,scm,s3g,localhost,127.0.0.1");
+ return env;
+ }
+
+ @SuppressWarnings("resource")
+ private static GenericContainer<?> startScm(Network network, Map<String,
String> commonEnv) {
+ GenericContainer<?> container = new GenericContainer<>(OZONE_IMAGE)
+ .withNetwork(network)
+ .withNetworkAliases("scm")
+ .withExposedPorts(SCM_PORT)
+ .withEnv(commonEnv)
+ .withEnv("ENSURE_SCM_INITIALIZED",
"/data/metadata/scm/current/VERSION")
+ .withCommand("ozone", "scm")
+
.waitingFor(Wait.forListeningPort().withStartupTimeout(STARTUP_TIMEOUT))
+ .withLogConsumer(outputFrame -> LOG.debug("[ozone-scm] {}",
outputFrame.getUtf8String().trim()));
+ container.start();
+ return container;
+ }
+
+ @SuppressWarnings("resource")
+ private static GenericContainer<?> startOm(Network network, Map<String,
String> commonEnv) {
+ GenericContainer<?> container = new GenericContainer<>(OZONE_IMAGE)
+ .withNetwork(network)
+ .withNetworkAliases("om")
+ .withExposedPorts(OM_PORT)
+ .withEnv(commonEnv)
+ .withEnv("CORE-SITE.XML_hadoop.proxyuser.hadoop.hosts", "*")
+ .withEnv("CORE-SITE.XML_hadoop.proxyuser.hadoop.groups", "*")
+ .withEnv("ENSURE_OM_INITIALIZED", "/data/metadata/om/current/VERSION")
+ .withEnv("WAITFOR", "scm:" + SCM_PORT)
+ .withCommand("ozone", "om")
+
.waitingFor(Wait.forListeningPort().withStartupTimeout(STARTUP_TIMEOUT))
+ .withLogConsumer(outputFrame -> LOG.debug("[ozone-om] {}",
outputFrame.getUtf8String().trim()));
+ container.start();
+ return container;
+ }
+
+ @SuppressWarnings("resource")
+ private static GenericContainer<?> startDatanode(Network network,
Map<String, String> commonEnv) {
+ GenericContainer<?> container = new GenericContainer<>(OZONE_IMAGE)
+ .withNetwork(network)
+ .withExposedPorts(DATANODE_CLIENT_PORT)
+ .withEnv(commonEnv)
+ .withCommand("ozone", "datanode")
+
.waitingFor(Wait.forListeningPort().withStartupTimeout(STARTUP_TIMEOUT))
+ .withLogConsumer(outputFrame -> LOG.debug("[ozone-dn] {}",
outputFrame.getUtf8String().trim()));
+ container.start();
+ return container;
+ }
+
+ @SuppressWarnings("resource")
+ private static GenericContainer<?> startS3g(Network network, Map<String,
String> commonEnv, String bucketName) {
+ String bootstrap = ""
+ + "set -e\n"
+ + "ozone s3g &\n"
+ + "s3g_pid=$!\n"
+ + "until ozone sh volume list >/dev/null 2>&1; do echo 'waiting for
OM...' && sleep 1; done\n"
+ + "ozone sh volume create /s3v || true\n"
+ + "ozone sh bucket delete /s3v/" + bucketName + " || true\n"
+ + "ozone sh bucket create /s3v/" + bucketName + "\n"
+ + "echo " + S3_BUCKET_READY_MARKER + "\n"
+ + "wait \"$s3g_pid\"\n";
+ GenericContainer<?> container = new GenericContainer<>(OZONE_IMAGE)
+ .withNetwork(network)
+ .withNetworkAliases(S3_DOCKER_ALIAS)
+ .withExposedPorts(S3G_PORT)
+ .withEnv(commonEnv)
+ .withEnv("WAITFOR", "om:" + OM_PORT)
+ .withCommand("sh", "-c", bootstrap)
+ .waitingFor(new WaitAllStrategy()
+
.withStrategy(Wait.forListeningPort().withStartupTimeout(STARTUP_TIMEOUT))
+ .withStrategy(Wait.forLogMessage(".*" + S3_BUCKET_READY_MARKER +
".*\\n", 1)
+ .withStartupTimeout(STARTUP_TIMEOUT)))
+ .withLogConsumer(outputFrame -> LOG.info("[ozone-s3g] {}",
outputFrame.getUtf8String().trim()));
+ container.start();
+ return container;
+ }
+
+ private static void stopQuietly(GenericContainer<?> container) {
+ if (container != null) {
+ container.stop();
+ }
+ }
+}
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java
b/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java
index 5d4bfd3dffb..8b02ec22ea5 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java
@@ -375,6 +375,13 @@ private List<FetchInputFormatSplit> getNextSplits() throws
Exception {
Class<? extends InputFormat> formatter =
currDesc.getInputFileFormatClass();
Utilities.copyTableJobPropertiesToConf(currDesc.getTableDesc(), job);
+ // A fetch task runs in-process without a DAG, so storage-handler
secrets (HIVE-20651) are
+ // never restored from the job Credentials as on a task. Surface them
into this process-local
+ // conf (never serialized or shown in EXPLAIN) so in-process readers can
access storage.
+ Map<String, String> jobSecrets = currDesc.getTableDesc().getJobSecrets();
+ if (jobSecrets != null) {
+ jobSecrets.forEach(job::set);
+ }
InputFormat inputFormat = getInputFormatFromCache(formatter, job);
if(inputFormat instanceof HiveSequenceFileInputFormat) {
// input format could be cached, in which case we need to reset the
list of files to fetch
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
b/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
index 5d3c2dccf5c..a3d48f36a31 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
@@ -942,6 +942,11 @@ protected void createBucketForFileIdx(FSPaths fsp, int
filesIdx)
updateDPCounters(fsp, filesIdx);
+ try {
+ Utilities.copyJobSecretToTableProperties(conf.getTableInfo());
+ } catch (IOException e) {
+ throw new HiveException(e);
+ }
Utilities.copyTableJobPropertiesToConf(conf.getTableInfo(), jc);
// only create bucket files only if no dynamic partitions,
// buckets of dynamic partitions will be created for each newly created
partition
@@ -1669,6 +1674,11 @@ public void checkOutputSpecs(FileSystem ignored, JobConf
job) throws IOException
private void createHiveOutputFormat(JobConf job) throws HiveException {
if (hiveOutputFormat == null) {
+ try {
+ Utilities.copyJobSecretToTableProperties(conf.getTableInfo());
+ } catch (IOException e) {
+ throw new HiveException(e);
+ }
Utilities.copyTableJobPropertiesToConf(conf.getTableInfo(), job);
}
try {
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
b/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
index f58fafc8556..8a72a22b97f 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
@@ -2517,7 +2517,7 @@ public static void
copyJobSecretToTableProperties(TableDesc tbl) throws IOExcept
String[] comps = keyString.split(TableDesc.SECRET_DELIMIT);
String tblName = comps[1];
String keyName = comps[2];
- if (tbl.getTableName().equalsIgnoreCase(tblName)) {
+ if (tblName.equalsIgnoreCase(tbl.getTableName())) {
tbl.getProperties().put(keyName, new
String(credentials.getSecretKey(key)));
}
}
@@ -3899,8 +3899,10 @@ private static void createTmpDirs(Configuration conf,
if (op instanceof FileSinkOperator) {
FileSinkDesc fdesc = ((FileSinkOperator) op).getConf();
- if (fdesc.isMmTable() || fdesc.isDirectInsert()) {
- // No need to create for MM tables, or ACID insert
+ if (fdesc.isMmTable() || fdesc.isDirectInsert() ||
+ (fdesc.getTableInfo() != null &&
fdesc.getTableInfo().isNonNative())) {
+ // No need to create for MM tables or ACID insert; storage-handler
tables manage
+ // their own output locations and commit
continue;
}
Path tempDir = fdesc.getDirName();
