feat: add compatibility for custom Kafka and PostgreSQL images

- Updated `BigDataContainerFactory` to support Testcontainers image compatibility checks for Kafka and PostgreSQL.
- Introduced helper methods to mark overridden Kafka and PostgreSQL images as compatible substitutes.
- Enhanced Kerberos configurations to handle escaped characters in JAAS keytab paths and principal values.
- Updated user guide with details on Kafka and PostgreSQL image compatibility logic.

Author: CoderYellow

core/src/main/kotlin/org/openprojectx/bigdata/test/core/container/BigDataContainerFactory.kt

@@ -207,7 +207,7 @@ internal class BigDataContainerFactory(
         if (hive.distribution == HiveMetastoreDistribution.CLOUDERA) {
             return clouderaHms()
         }
-        val postgres = PostgreSQLContainer(DockerImageName.parse(hive.databaseImage))
+        val postgres = PostgreSQLContainer(compatiblePostgresImage(hive.databaseImage))
             .withNetwork(network)
             .withNetworkAliases("hive-metastore-postgres")
             .withDatabaseName(hive.databaseName)
@@ -257,6 +257,9 @@ internal class BigDataContainerFactory(
         }
     }
 
+    private fun compatiblePostgresImage(image: String): DockerImageName =
+        DockerImageName.parse(image).asCompatibleSubstituteFor("postgres")
+
     private fun clouderaHms(): BigDataServiceContainer {
         val hive = options.hiveMetastore
         val container = GenericBigDataContainer(hive.image)
@@ -370,9 +373,9 @@ internal class BigDataContainerFactory(
     private fun tlsKafka(kafka: KafkaOptions): BigDataServiceContainer {
         val kafkaHostPort = options.portBindings.hostPort(9092, options.portBindings.kafka)
         val container = if (kafkaHostPort == 0) {
-            KafkaContainer(DockerImageName.parse(kafka.image))
+            KafkaContainer(compatibleKafkaImage(kafka.image))
         } else {
-            FixedPortKafkaContainer(DockerImageName.parse(kafka.image)).withServicePort(9092, kafkaHostPort)
+            FixedPortKafkaContainer(compatibleKafkaImage(kafka.image)).withServicePort(9092, kafkaHostPort)
         }
         container
             .withNetwork(network)
@@ -402,9 +405,9 @@ internal class BigDataContainerFactory(
     private fun plaintextKafka(kafka: KafkaOptions): BigDataServiceContainer {
         val kafkaHostPort = options.portBindings.hostPort(9092, options.portBindings.kafka)
         val container = if (kafkaHostPort == 0) {
-            KafkaContainer(DockerImageName.parse(kafka.image))
+            KafkaContainer(compatibleKafkaImage(kafka.image))
         } else {
-            FixedPortKafkaContainer(DockerImageName.parse(kafka.image)).withServicePort(9092, kafkaHostPort)
+            FixedPortKafkaContainer(compatibleKafkaImage(kafka.image)).withServicePort(9092, kafkaHostPort)
         }
         container
             .withNetwork(network)
@@ -427,6 +430,9 @@ internal class BigDataContainerFactory(
         }
     }
 
+    private fun compatibleKafkaImage(image: String): DockerImageName =
+        DockerImageName.parse(image).asCompatibleSubstituteFor("apache/kafka")
+
     private fun schemaRegistry(): BigDataServiceContainer {
         val kafka = options.kafka
         val container = GenericBigDataContainer(kafka.schemaRegistryImage)
@@ -1206,16 +1212,22 @@ internal class BigDataContainerFactory(
           com.sun.security.auth.module.Krb5LoginModule required
           useKeyTab=true
           storeKey=true
-          keyTab="${options.keytabPath}"
-          principal="${options.servicePrincipal}";
+          keyTab="${jaasKeytab(options.keytabPath)}"
+          principal="${jaasValue(options.servicePrincipal)}";
         };
         """.trimIndent()
 
     private fun inlineJaas(options: KerberosAuthOptions): String =
-        """com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="${options.keytabPath}" principal="${options.servicePrincipal}";"""
+        """com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="${jaasKeytab(options.keytabPath)}" principal="${jaasValue(options.servicePrincipal)}";"""
 
     private fun inlineJaas(principal: String, keytabPath: String): String =
-        """com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="$keytabPath" principal="$principal";"""
+        """com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="${jaasKeytab(keytabPath)}" principal="${jaasValue(principal)}";"""
+
+    private fun jaasKeytab(path: String): String =
+        jaasValue(path.replace('\\', '/'))
+
+    private fun jaasValue(value: String): String =
+        value.replace("\\", "\\\\").replace("\"", "\\\"")
 
     private fun localKerberosPath(containerPath: String): String =
         when {

doc/user-guide.adoc

@@ -233,6 +233,8 @@ localStackS3 = "localstack/localstack:4.14.0"
 fakeGcs = "fsouza/fake-gcs-server:1.54"
 ----
 
+Kafka and the open-source HMS PostgreSQL sidecar use typed Testcontainers classes instead of plain generic containers. When you override those images, `bigdata-test` marks the configured Kafka image as a compatible substitute for `apache/kafka` and the configured PostgreSQL image as a compatible substitute for `postgres`, so private mirrors or rebuilt images can still pass Testcontainers image checks.
+
 You can place `[images]` in the same TOML files used by `@BigDataExtensions`, because `@BigDataTest` reads those locations before startup when the extensions annotation is present:
 
 [source,kotlin]