@@ -32,7 +32,6 @@
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
-import java.nio.file.Paths;
 import java.util.HashMap;
 import java.util.Map;
 import org.apache.commons.io.FileUtils;
@@ -61,12 +60,17 @@
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.testcontainers.utility.MountableFile;
 
-// A private Kerberos Hive: the test adds principals with kadmin and changes HDFS permissions,
+// A private Kerberos Hive: the test needs principals of its own and changes HDFS permissions,
 // which would break the shared Kerberos Hive for later classes.
 @Tag("gravitino-docker-test")
-@HiveFixture(auth = HiveAuth.KERBEROS, scope = Scope.CLASS)
+@HiveFixture(
+    auth = HiveAuth.KERBEROS,
+    scope = Scope.CLASS,
+    kerberosPrincipals = {
+      HadoopUserAuthenticationIT.HADOOP_SCHEMA_PRINCIPAL,
+      HadoopUserAuthenticationIT.HADOOP_FILESET_PRINCIPAL
+    })
 public class HadoopUserAuthenticationIT extends BaseIT {
   private static final Logger LOG = LoggerFactory.getLogger(HadoopUserAuthenticationIT.class);
 
@@ -82,14 +86,13 @@ public class HadoopUserAuthenticationIT extends BaseIT {
   private static final String HADOOP_CLIENT_PRINCIPAL = "cli@HADOOPKRB";
   private static final String HADOOP_CLIENT_KEYTAB = "/client.keytab";
 
-  private static final String HADOOP_SCHEMA_PRINCIPAL = "cli_schema";
-  private static final String HADOOP_FILESET_PRINCIPAL = "cli_fileset";
+  static final String HADOOP_SCHEMA_PRINCIPAL = "cli_schema";
+  static final String HADOOP_FILESET_PRINCIPAL = "cli_fileset";
 
   private static final String HADOOP_SCHEMA_KEYTAB = "/cli_schema.keytab";
   private static final String HADOOP_FILESET_KEYTAB = "/cli_fileset.keytab";
 
-  private static final String REALM = "HADOOPKRB";
-  private static final String ADMIN_PASSWORD = "Admin12!";
+  private static final String REALM = HiveContainer.KERBEROS_REALM;
 
   private static String TMP_DIR;
 
@@ -146,7 +149,7 @@ private static void prepareKerberosConfig() throws IOException {
         .getContainer()
         .copyFileFromContainer("/gravitino_client.keytab", TMP_DIR + GRAVITINO_CLIENT_KEYTAB);
 
-    createKeyTableForSchemaAndFileset();
+    copyKeytabsOfSchemaAndFileset();
 
     // Keytab of the Gravitino server
     kerberosHive
@@ -174,57 +177,17 @@ private static void prepareKerberosConfig() throws IOException {
     System.setProperty("sun.security.krb5.debug", "true");
   }
 
-  private static void createKeyTableForSchemaAndFileset() throws IOException {
-    String shellContent =
-        "echo -e \"%s\\n%s\" | kadmin.local -q \"addprinc %s@%s\""
-            + "\n"
-            + "kadmin.local -q \"xst -k /%s.keytab -norandkey %s@%s\"";
-
-    String createSchemaShellFile = String.format("/%s.sh", HADOOP_SCHEMA_PRINCIPAL);
-    String createFileSetShellFile = String.format("/%s.sh", HADOOP_FILESET_PRINCIPAL);
-
-    FileUtils.writeStringToFile(
-        Paths.get(TMP_DIR + createSchemaShellFile).toFile(),
-        String.format(
-            shellContent,
-            ADMIN_PASSWORD,
-            ADMIN_PASSWORD,
-            HADOOP_SCHEMA_PRINCIPAL,
-            REALM,
-            HADOOP_SCHEMA_PRINCIPAL,
-            HADOOP_SCHEMA_PRINCIPAL,
-            REALM),
-        StandardCharsets.UTF_8);
+  // The fixture created both principals and their keytabs when the container started.
+  private static void copyKeytabsOfSchemaAndFileset() {
     kerberosHive
         .getContainer()
-        .copyFileToContainer(
-            MountableFile.forHostPath(TMP_DIR + createSchemaShellFile), createSchemaShellFile);
-    kerberosHive.executeInContainer("bash", createSchemaShellFile);
-
-    FileUtils.writeStringToFile(
-        Paths.get(TMP_DIR + createFileSetShellFile).toFile(),
-        String.format(
-            shellContent,
-            ADMIN_PASSWORD,
-            ADMIN_PASSWORD,
-            HADOOP_FILESET_PRINCIPAL,
-            REALM,
-            HADOOP_FILESET_PRINCIPAL,
-            HADOOP_FILESET_PRINCIPAL,
-            REALM),
-        StandardCharsets.UTF_8);
+        .copyFileFromContainer(
+            HiveContainer.kerberosKeytab(HADOOP_SCHEMA_PRINCIPAL), TMP_DIR + HADOOP_SCHEMA_KEYTAB);
     kerberosHive
         .getContainer()
-        .copyFileToContainer(
-            MountableFile.forHostPath(TMP_DIR + createFileSetShellFile), createFileSetShellFile);
-    kerberosHive.executeInContainer("bash", createFileSetShellFile);
-
-    kerberosHive
-        .getContainer()
-        .copyFileFromContainer(HADOOP_SCHEMA_KEYTAB, TMP_DIR + HADOOP_SCHEMA_KEYTAB);
-    kerberosHive
-        .getContainer()
-        .copyFileFromContainer(HADOOP_FILESET_KEYTAB, TMP_DIR + HADOOP_FILESET_KEYTAB);
+        .copyFileFromContainer(
+            HiveContainer.kerberosKeytab(HADOOP_FILESET_PRINCIPAL),
+            TMP_DIR + HADOOP_FILESET_KEYTAB);
   }
 
   private void addKerberosConfig() {
 
@@ -33,7 +33,9 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.time.Duration;
+import java.util.Collection;
 import java.util.Comparator;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -50,6 +52,7 @@
 import org.testcontainers.containers.wait.strategy.WaitAllStrategy;
 import org.testcontainers.containers.wait.strategy.WaitStrategy;
 import org.testcontainers.containers.wait.strategy.WaitStrategyTarget;
+import org.testcontainers.images.builder.Transferable;
 import org.testcontainers.utility.MountableFile;
 
 public class HiveContainer extends BaseContainer {
@@ -70,6 +73,9 @@ public class HiveContainer extends BaseContainer {
   public static final int HDFS_DEFAULTFS_PORT = 9000;
   public static final int HIVE_METASTORE_PORT = 9083;
   public static final int HIVE_SERVICE_PORT = 10000;
+
+  /** The realm of the Kerberos image's KDC. */
+  public static final String KERBEROS_REALM = "HADOOPKRB";
   // The Kerberos image's second metastore, from its hive-site1.xml.
   private static final int SECOND_HIVE_METASTORE_PORT = 19083;
 
@@ -85,8 +91,14 @@ public class HiveContainer extends BaseContainer {
   // The image's core-site.xml already lets these users impersonate.
   private static final Set<String> USERS_ALLOWED_TO_IMPERSONATE =
       ImmutableSet.of("root", "hive", "gravitino");
+  // Where the container gets the script that creates a test's Kerberos principals.
+  static final String PRINCIPALS_SCRIPT = "/tmp/gravitino-kerberos-principals.sh";
+  // The Kerberos image's start script logs in as hive/<host> right after creating its own
+  // principals and before it configures and starts HDFS; the test's principals go in before it.
+  static final String PRINCIPALS_HOOK_ANCHOR = "kinit hive/";
 
   private final boolean kerberosEnabled;
+  private final List<String> kerberosPrincipals;
 
   public static Builder builder() {
     return new Builder();
@@ -100,9 +112,26 @@ protected HiveContainer(
       Map<String, String> filesToMount,
       Map<String, String> envVars,
       Optional<Network> network,
-      boolean kerberosEnabled) {
+      boolean kerberosEnabled,
+      List<String> kerberosPrincipals) {
     super(image, hostName, ports, extraHosts, filesToMount, envVars, network);
     this.kerberosEnabled = kerberosEnabled;
+    this.kerberosPrincipals = List.copyOf(kerberosPrincipals);
+    if (!kerberosEnabled && !this.kerberosPrincipals.isEmpty()) {
+      throw new IllegalArgumentException(
+          "Kerberos principals need a Kerberos Hive: " + this.kerberosPrincipals);
+    }
+  }
+
+  /**
+   * Returns where the Kerberos Hive keeps the keytab of a principal from {@link
+   * HiveFixture#kerberosPrincipals()}. Tests copy it out with {@code copyFileFromContainer}.
+   *
+   * @param principal the principal's name, without realm
+   * @return the keytab's path in the container
+   */
+  public static String kerberosKeytab(String principal) {
+    return "/" + principal + ".keytab";
   }
 
   public String getHostName() {
@@ -140,6 +169,8 @@ protected void setupContainer() {
   public void start() {
     if (!kerberosEnabled) {
       allowHadoopUserToImpersonate();
+    } else if (!kerberosPrincipals.isEmpty()) {
+      createKerberosPrincipals();
     }
     // Set here rather than in waitStrategy(), which runs before the constructor sets the auth.
     container.waitingFor(kerberosEnabled ? kerberosReadiness() : readiness(getHadoopUser()));
@@ -182,6 +213,62 @@ private void allowHadoopUserToImpersonate() {
         cmd -> cmd.withEntrypoint("/bin/bash", "-c", script + " && exec /bin/bash " + IMAGE_START));
   }
 
+  /**
+   * Creates the test's principals and their keytabs while the container starts.
+   *
+   * <p>The KDC's database only exists once the image's start script has created it, so the
+   * entrypoint hooks a script into the start script, next to the image's own principals and before
+   * HDFS starts; readiness therefore implies the keytabs exist. A missing hook point or a failed
+   * principal stops the start script, and with it the container.
+   */
+  private void createKerberosPrincipals() {
+    container
+        .withCopyToContainer(
+            Transferable.of(principalsScript(kerberosPrincipals), 0755), PRINCIPALS_SCRIPT)
+        .withCreateContainerCmdModifier(
+            cmd -> cmd.withEntrypoint("/bin/bash", "-c", principalsEntrypoint(IMAGE_START)));
+  }
+
+  /**
+   * Returns the script that creates each principal in {@link #KERBEROS_REALM} with a keytab at
+   * {@link #kerberosKeytab(String)}, and fails if a keytab does not hold its principal. {@code
+   * kadmin.local -q} exits 0 even when its query fails, so the keytab is the check.
+   */
+  static String principalsScript(List<String> principals) {
+    StringBuilder script = new StringBuilder("set -e\n");
+    for (String name : principals) {
+      String principal = name + "@" + KERBEROS_REALM;
+      String keytab = kerberosKeytab(name);
+      script
+          .append(format("kadmin.local -q 'addprinc -randkey %s'\n", principal))
+          .append(format("kadmin.local -q 'xst -norandkey -k %s %s'\n", keytab, principal))
+          .append(
+              format(
+                  "klist -k %1$s | grep -q ' %2$s$'"
+                      + " || { echo 'No keytab for %2$s at %1$s' >&2; exit 1; }\n",
+                  keytab, principal));
+    }
+    return script.toString();
+  }
+
+  /**
+   * Returns the entrypoint that runs {@link #PRINCIPALS_SCRIPT} inside {@code imageStart}, before
+   * the line with {@link #PRINCIPALS_HOOK_ANCHOR}, and then runs {@code imageStart}. It fails
+   * without starting anything if that line is missing.
+   */
+  static String principalsEntrypoint(String imageStart) {
+    String hook =
+        format(
+            "bash %s || { echo \"Cannot create the test Kerberos principals\" >&2; exit 1; }",
+            PRINCIPALS_SCRIPT);
+    return format(
+        "sed -i '/%1$s/i %2$s' %3$s"
+            + " && grep -q '^bash %4$s ' %3$s"
+            + " || { echo 'Cannot hook the Kerberos principals into %3$s' >&2; exit 1; }"
+            + " && exec /bin/bash %3$s",
+        PRINCIPALS_HOOK_ANCHOR.replace("/", "\\/"), hook, imageStart, PRINCIPALS_SCRIPT);
+  }
+
   @Override
   public void close() {
     // A private container is closed by its JUnit store and again by the run's closer at the end of
@@ -407,6 +494,9 @@ private static String httpRequest(String method, String url) {
   }
 
   public static class Builder extends BaseContainer.Builder<Builder, HiveContainer> {
+    // Package-private so that tests can read what a spec asked for.
+    List<String> kerberosPrincipals = List.of();
+
     private Builder() {
       this.image = DEFAULT_IMAGE;
       this.hostName = HOST_NAME;
@@ -432,7 +522,20 @@ public HiveContainer build() {
           filesToMount,
           envVars,
           network,
-          kerberosEnabled);
+          kerberosEnabled,
+          kerberosPrincipals);
+    }
+
+    /**
+     * Sets the extra Kerberos principals a Kerberos Hive creates when it starts.
+     *
+     * @param kerberosPrincipals the principal names, without realm; see {@link
+     *     HiveFixture#kerberosPrincipals()}
+     * @return this builder
+     */
+    public Builder withKerberosPrincipals(Collection<String> kerberosPrincipals) {
+      this.kerberosPrincipals = List.copyOf(kerberosPrincipals);
+      return this;
     }
   }
 }
@@ -64,9 +64,19 @@
    */
   HiveStorage storage() default HiveStorage.HDFS;
 
+  /**
+   * Returns the extra Kerberos principals the Kerberos Hive creates when it starts, before it is
+   * ready. Each is a name without realm or instance, created in {@link
+   * HiveContainer#KERBEROS_REALM} with a keytab at {@link HiveContainer#kerberosKeytab(String)}.
+   * Needs {@code auth = HiveAuth.KERBEROS} and {@code scope = Scope.CLASS} or {@code Scope.METHOD}.
+   *
+   * @return the principal names; none by default
+   */
+  String[] kerberosPrincipals() default {};
+
   /**
    * Returns how long the Hive container lives. Use {@link Scope#CLASS} or {@link Scope#METHOD} for
-   * tests that change Hive in ways later tests would see, such as adding Kerberos principals.
+   * tests that change Hive in ways later tests would see, such as changing HDFS permissions.
    *
    * @return the scope; the whole run by default, shared with other classes
    */