Skip to content

Commit b947b8b

Browse files
CoderYellowCoderYellow
committed
feat: add native TLS support for Kafka and improve integration across BigDataTest components
- Introduced native Kafka TLS configuration in `BigDataTestKit`, with options for SSL and SASL_SSL protocols. - Updated Kafka client properties, schema registry, and UI to support TLS-aware communication. - Added example tests demonstrating Kafka TLS usage. - Refactored container factories to dynamically generate and mount TLS certificates and truststores. - Extended user guide with example configurations and usage details.
1 parent b0b869d commit b947b8b

12 files changed

Lines changed: 219 additions & 15 deletions

File tree

‎core/src/main/kotlin/org/openprojectx/bigdata/test/core/BigDataTestKitOptions.kt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,7 @@ enum class HiveMetastoreDistribution {
8888
data class KafkaOptions(
8989
val enabled: Boolean = false,
9090
val image: String = "apache/kafka:4.1.2",
91+
val tls: HttpTlsOptions = HttpTlsOptions(),
9192
val schemaRegistryEnabled: Boolean = false,
9293
val schemaRegistryImage: String = "confluentinc/cp-schema-registry:7.8.0",
9394
val schemaRegistryTls: HttpTlsOptions = HttpTlsOptions(),

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

Lines changed: 105 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -257,13 +257,16 @@ internal class BigDataContainerFactory(
257257

258258
private fun kafka(): BigDataServiceContainer {
259259
val kafka = options.kafka
260-
if (!kafka.kerberos.enabled) return plaintextKafka(kafka)
260+
if (!kafka.kerberos.enabled) {
261+
return if (kafka.tls.enabled) tlsKafka(kafka) else plaintextKafka(kafka)
262+
}
261263

262264
val kafkaHostPort = options.portBindings.hostPort(9092, options.portBindings.kafka)
265+
val externalProtocol = if (kafka.tls.enabled) "SASL_SSL" else "SASL_PLAINTEXT"
263266
val advertisedListener = if (kafkaHostPort > 0) {
264-
"SASL_PLAINTEXT://localhost:$kafkaHostPort"
267+
"$externalProtocol://localhost:$kafkaHostPort"
265268
} else {
266-
"SASL_PLAINTEXT://broker1.example.com:9092"
269+
"$externalProtocol://broker1.example.com:9092"
267270
}
268271
val container = GenericBigDataContainer(kafka.image)
269272
.withNetwork(network)
@@ -284,11 +287,14 @@ internal class BigDataContainerFactory(
284287
.waitingFor(Wait.forListeningPort().withStartupTimeout(Duration.ofMinutes(3)))
285288
mountKerberos(container)
286289
container
287-
.withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "CONTROLLER:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT,PLAINTEXT:PLAINTEXT")
290+
.withEnv(
291+
"KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",
292+
"CONTROLLER:PLAINTEXT,$externalProtocol:$externalProtocol,PLAINTEXT:PLAINTEXT",
293+
)
288294
.withEnv("KAFKA_ADVERTISED_LISTENERS", "$advertisedListener,PLAINTEXT://kafka:19092")
289-
.withEnv("KAFKA_LISTENERS", "SASL_PLAINTEXT://0.0.0.0:9092,PLAINTEXT://0.0.0.0:19092,CONTROLLER://kafka:29093")
295+
.withEnv("KAFKA_LISTENERS", "$externalProtocol://0.0.0.0:9092,PLAINTEXT://0.0.0.0:19092,CONTROLLER://kafka:29093")
290296
.withEnv("KAFKA_CONTROLLER_QUORUM_VOTERS", "1@kafka:29093")
291-
.withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "SASL_PLAINTEXT")
297+
.withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", externalProtocol)
292298
.withEnv("KAFKA_SASL_ENABLED_MECHANISMS", "GSSAPI")
293299
.withEnv("KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL", "GSSAPI")
294300
.withEnv("KAFKA_SASL_KERBEROS_SERVICE_NAME", kafka.kerberos.servicePrincipal.substringBefore("/"))
@@ -298,6 +304,7 @@ internal class BigDataContainerFactory(
298304
Transferable.of(kafkaJaas(kafka.kerberos)),
299305
"/etc/kafka/kerberos/kafka_server_jaas.conf",
300306
)
307+
val sslProperties = if (kafka.tls.enabled) configureKafkaBrokerTls(container, kafka, "SASL_SSL") else emptyMap()
301308

302309
return BigDataServiceContainer(BigDataService.KAFKA, attachLogs("kafka", container)) {
303310
val bootstrapServers = "${container.host}:${container.getMappedPort(9092)}"
@@ -308,7 +315,41 @@ internal class BigDataContainerFactory(
308315
properties = mapOf(
309316
"bootstrap.servers" to bootstrapServers,
310317
"spring.kafka.bootstrap-servers" to bootstrapServers,
311-
) + kerberosProperties("kafka", kafka.kerberos) + kafkaClientKerberosProperties(kafka.kerberos, options.kerberos),
318+
) + kerberosProperties("kafka", kafka.kerberos) +
319+
kafkaClientKerberosProperties(kafka.kerberos, options.kerberos, kafka.tls.enabled) +
320+
sslProperties,
321+
)
322+
}
323+
}
324+
325+
private fun tlsKafka(kafka: KafkaOptions): BigDataServiceContainer {
326+
val kafkaHostPort = options.portBindings.hostPort(9092, options.portBindings.kafka)
327+
val container = if (kafkaHostPort == 0) {
328+
KafkaContainer(DockerImageName.parse(kafka.image))
329+
} else {
330+
FixedPortKafkaContainer(DockerImageName.parse(kafka.image)).withServicePort(9092, kafkaHostPort)
331+
}
332+
container
333+
.withNetwork(network)
334+
.withNetworkAliases("kafka")
335+
.withStartupTimeout(Duration.ofMinutes(3))
336+
.withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:SSL,PLAINTEXT:SSL,CONTROLLER:PLAINTEXT")
337+
.withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "BROKER")
338+
.withEnv("KAFKA_SSL_CLIENT_AUTH", "none")
339+
.withEnv("KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM", "")
340+
val sslProperties = configureKafkaBrokerTls(container, kafka, "SSL")
341+
342+
return BigDataServiceContainer(BigDataService.KAFKA, attachLogs("kafka", container)) {
343+
val bootstrapServers = container.bootstrapServers
344+
BigDataEndpoint(
345+
service = BigDataService.KAFKA,
346+
host = container.host,
347+
ports = mapOf("bootstrap" to container.getMappedPort(9092)),
348+
properties = mapOf(
349+
"bootstrap.servers" to bootstrapServers,
350+
"spring.kafka.bootstrap-servers" to bootstrapServers,
351+
"bootstrap.servers.internal" to "kafka:9093",
352+
) + sslProperties,
312353
)
313354
}
314355
}
@@ -351,6 +392,16 @@ internal class BigDataContainerFactory(
351392
.withEnv("SCHEMA_REGISTRY_LISTENERS", "http://0.0.0.0:8085")
352393
.withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "PLAINTEXT://kafka:19092")
353394
.waitingFor(Wait.forHttp("/subjects").forStatusCode(200).withStartupTimeout(Duration.ofMinutes(3)))
395+
if (kafka.tls.enabled && !kafka.kerberos.enabled) {
396+
container
397+
.withFileSystemBind(tlsMaterial.trustStorePath.toString(), "/etc/schema-registry/tls/truststore.p12", BindMode.READ_ONLY)
398+
.withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "SSL://kafka:9093")
399+
.withEnv("SCHEMA_REGISTRY_KAFKASTORE_SECURITY_PROTOCOL", "SSL")
400+
.withEnv("SCHEMA_REGISTRY_KAFKASTORE_SSL_TRUSTSTORE_LOCATION", "/etc/schema-registry/tls/truststore.p12")
401+
.withEnv("SCHEMA_REGISTRY_KAFKASTORE_SSL_TRUSTSTORE_PASSWORD", tlsMaterial.trustStorePassword)
402+
.withEnv("SCHEMA_REGISTRY_KAFKASTORE_SSL_TRUSTSTORE_TYPE", "PKCS12")
403+
.withEnv("SCHEMA_REGISTRY_KAFKASTORE_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM", "")
404+
}
354405
return BigDataServiceContainer(BigDataService.SCHEMA_REGISTRY, attachLogs("schema-registry", container)) {
355406
val tlsEndpoint = httpTlsEndpoint(
356407
name = "schema-registry",
@@ -384,11 +435,22 @@ internal class BigDataContainerFactory(
384435
container
385436
.withEnv("JAVA_OPTS", "-Djava.security.krb5.conf=/kerby/client/krb5.conf")
386437
.withEnv("KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS", "broker1.example.com:9092")
387-
.withEnv("KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL", "SASL_PLAINTEXT")
438+
.withEnv("KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL", if (kafka.tls.enabled) "SASL_SSL" else "SASL_PLAINTEXT")
388439
.withEnv("KAFKA_CLUSTERS_0_PROPERTIES_SASL_MECHANISM", "GSSAPI")
389440
.withEnv("KAFKA_CLUSTERS_0_PROPERTIES_SASL_KERBEROS_SERVICE_NAME", "kafka")
390441
.withEnv("KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG", inlineJaas(kafka.kafkaUiKerberos))
391442
}
443+
if (kafka.tls.enabled) {
444+
val trustStore = tlsMaterial.trustStorePath
445+
container
446+
.withFileSystemBind(trustStore.toString(), "/etc/kafka/tls/truststore.p12", BindMode.READ_ONLY)
447+
.withEnv("KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS", if (kafka.kerberos.enabled) "broker1.example.com:9092" else "kafka:9093")
448+
.withEnv("KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL", if (kafka.kerberos.enabled) "SASL_SSL" else "SSL")
449+
.withEnv("KAFKA_CLUSTERS_0_PROPERTIES_SSL_TRUSTSTORE_LOCATION", "/etc/kafka/tls/truststore.p12")
450+
.withEnv("KAFKA_CLUSTERS_0_PROPERTIES_SSL_TRUSTSTORE_PASSWORD", tlsMaterial.trustStorePassword)
451+
.withEnv("KAFKA_CLUSTERS_0_PROPERTIES_SSL_TRUSTSTORE_TYPE", "PKCS12")
452+
.withEnv("KAFKA_CLUSTERS_0_PROPERTIES_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM", "")
453+
}
392454

393455
return BigDataServiceContainer(BigDataService.KAFKA_UI, attachLogs("kafka-ui", container)) {
394456
val tlsEndpoint = httpTlsEndpoint(
@@ -629,6 +691,35 @@ internal class BigDataContainerFactory(
629691
return configuredHostPort
630692
}
631693

694+
private fun configureKafkaBrokerTls(
695+
container: GenericContainer<*>,
696+
kafka: KafkaOptions,
697+
securityProtocol: String,
698+
): Map<String, String> {
699+
val keyStore = tlsMaterial.keyStore(
700+
name = "kafka",
701+
domain = kafka.tls.domain,
702+
sanDomains = listOf("kafka", "broker1.example.com"),
703+
)
704+
container
705+
.withFileSystemBind(keyStore.path.toString(), "/etc/kafka/tls/kafka.keystore.p12", BindMode.READ_ONLY)
706+
.withFileSystemBind(tlsMaterial.trustStorePath.toString(), "/etc/kafka/tls/kafka.truststore.p12", BindMode.READ_ONLY)
707+
.withEnv("KAFKA_SSL_KEYSTORE_LOCATION", "/etc/kafka/tls/kafka.keystore.p12")
708+
.withEnv("KAFKA_SSL_KEYSTORE_PASSWORD", keyStore.password)
709+
.withEnv("KAFKA_SSL_KEYSTORE_TYPE", keyStore.type)
710+
.withEnv("KAFKA_SSL_KEY_PASSWORD", keyStore.password)
711+
.withEnv("KAFKA_SSL_TRUSTSTORE_LOCATION", "/etc/kafka/tls/kafka.truststore.p12")
712+
.withEnv("KAFKA_SSL_TRUSTSTORE_PASSWORD", tlsMaterial.trustStorePassword)
713+
.withEnv("KAFKA_SSL_TRUSTSTORE_TYPE", "PKCS12")
714+
715+
return mapOf(
716+
"security.protocol" to securityProtocol,
717+
"ssl.truststore.location" to tlsMaterial.trustStorePath.toString(),
718+
"ssl.truststore.password" to tlsMaterial.trustStorePassword,
719+
"ssl.truststore.type" to "PKCS12",
720+
) + tlsMaterial.properties()
721+
}
722+
632723
private data class HttpTlsEndpoint(
633724
val host: String? = null,
634725
val port: Int? = null,
@@ -758,11 +849,15 @@ internal class BigDataContainerFactory(
758849
emptyMap()
759850
}
760851

761-
private fun kafkaClientKerberosProperties(service: KerberosAuthOptions, client: KerberosOptions): Map<String, String> =
852+
private fun kafkaClientKerberosProperties(
853+
service: KerberosAuthOptions,
854+
client: KerberosOptions,
855+
tlsEnabled: Boolean = false,
856+
): Map<String, String> =
762857
if (service.enabled) {
763858
val clientKeytab = localKerberosPath("/kerby/keytabs/client.keytab")
764859
mapOf(
765-
"security.protocol" to "SASL_PLAINTEXT",
860+
"security.protocol" to if (tlsEnabled) "SASL_SSL" else "SASL_PLAINTEXT",
766861
"sasl.mechanism" to "GSSAPI",
767862
"sasl.kerberos.service.name" to service.servicePrincipal.substringBefore("/"),
768863
"sasl.jaas.config" to inlineJaas(client.clientPrincipal, clientKeytab),

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

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,32 @@ internal class TlsMaterial(
7373
return pem
7474
}
7575

76+
fun keyStore(name: String, domain: String, sanDomains: List<String> = emptyList()): KeyStoreMaterial {
77+
val keyPair = rsaKeyPair()
78+
val cert = signedCertificate(
79+
subject = X500Name("CN=$domain"),
80+
subjectKeyPair = keyPair,
81+
issuer = X500Name(ca.certificate.subjectX500Principal.name),
82+
issuerKey = ca.privateKey,
83+
issuerCert = ca.certificate,
84+
sanDomains = (listOf(domain, "localhost") + sanDomains).distinct(),
85+
)
86+
val path = directory.resolve("${name.replace(Regex("[^A-Za-z0-9._-]"), "_")}.p12")
87+
val password = options.trustStorePassword
88+
val keyStore = KeyStore.getInstance("PKCS12")
89+
keyStore.load(null, password.toCharArray())
90+
keyStore.setKeyEntry(
91+
"bigdata-test-$name",
92+
keyPair.private,
93+
password.toCharArray(),
94+
arrayOf(cert, ca.certificate),
95+
)
96+
Files.newOutputStream(path).use { output ->
97+
keyStore.store(output, password.toCharArray())
98+
}
99+
return KeyStoreMaterial(path = path, password = password, type = "PKCS12")
100+
}
101+
76102
fun properties(): Map<String, String> =
77103
mapOf(
78104
"javax.net.ssl.trustStore" to trustStorePath.toString(),
@@ -196,4 +222,10 @@ internal class TlsMaterial(
196222
val certificate: X509Certificate,
197223
val privateKey: PrivateKey,
198224
)
225+
226+
data class KeyStoreMaterial(
227+
val path: Path,
228+
val password: String,
229+
val type: String,
230+
)
199231
}

‎doc/user-guide.adoc‎

Lines changed: 22 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -282,6 +282,10 @@ domain = "schema-registry.local"
282282
enabled = true
283283
domain = "kafka-ui.local"
284284
285+
[kafkaTls]
286+
enabled = true
287+
domain = "localhost"
288+
285289
[hdfsWebTls]
286290
enabled = true
287291
domain = "hdfs.local"
@@ -309,7 +313,23 @@ schemaRegistryTls = 8443
309313

310314
Available TLS port fields are `hdfsWebTls`, `schemaRegistryTls`, `kafkaUiTls`, `localStackS3Tls`, and `fakeGcsTls`.
311315

312-
Non-HTTP services should use their native TLS implementation rather than HAProxy to avoid protocol compatibility issues. Kafka TLS is intentionally not implemented as a generic HAProxy gateway because Kafka clients depend on broker metadata and advertised listeners.
316+
Non-HTTP services use their native TLS implementation rather than HAProxy to avoid protocol compatibility issues. Kafka TLS is native broker TLS because Kafka clients depend on broker metadata and advertised listeners.
317+
318+
Kafka TLS can be enabled from the annotation or TOML:
319+
320+
[source,kotlin]
321+
----
322+
@BigDataTest(kafkaTls = true)
323+
----
324+
325+
[source,toml]
326+
----
327+
[kafkaTls]
328+
enabled = true
329+
domain = "localhost"
330+
----
331+
332+
The Kafka endpoint then exposes `security.protocol=SSL` for plaintext Kafka or `security.protocol=SASL_SSL` when Kafka Kerberos is also enabled. It also exposes `ssl.truststore.location`, `ssl.truststore.password`, and `ssl.truststore.type` for Kafka clients.
313333

314334
=== Dynamic and Fixed Ports
315335

@@ -406,7 +426,7 @@ Each started service exposes a `BigDataEndpoint` with named ports and connection
406426
|`BigDataService.KAFKA`
407427
|`bootstrap`
408428
|`9092`
409-
|`bootstrap.servers`, `spring.kafka.bootstrap-servers`, `bootstrap.servers.internal`, `kafka.kerberos.service-name`, `sasl.jaas.config`
429+
|`bootstrap.servers`, `spring.kafka.bootstrap-servers`, `bootstrap.servers.internal`, `security.protocol`, `kafka.kerberos.service-name`, `sasl.jaas.config`, `ssl.truststore.location`, `ssl.truststore.password`, `ssl.truststore.type`
410430

411431
|`BigDataService.SCHEMA_REGISTRY`
412432
|`http`, `https`

‎example/junit/build.gradle.kts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ description = "JUnit 5 example for bigdata-test"
66

77
dependencies {
88
testImplementation(project(":junit5"))
9+
testImplementation(libs.kafkaClients)
910
testImplementation(libs.junitJupiterApi)
1011
testRuntimeOnly(libs.junitJupiterEngine)
1112
testRuntimeOnly(libs.junitPlatformLauncher)

‎example/junit/src/test/kotlin/org/openprojectx/bigdata/test/example/junit/KafkaTlsExampleTest.kt‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
package org.openprojectx.bigdata.test.example.junit
2+
3+
import org.apache.kafka.clients.admin.AdminClient
4+
import org.apache.kafka.clients.admin.AdminClientConfig
5+
import org.junit.jupiter.api.Test
6+
import org.openprojectx.bigdata.test.core.BigDataService
7+
import org.openprojectx.bigdata.test.core.BigDataTestKit
8+
import org.openprojectx.bigdata.test.junit5.BigDataTest
9+
10+
@BigDataTest(kafkaTls = true, schemaRegistry = true)
11+
class KafkaTlsExampleTest {
12+
@Test
13+
fun exposesSslClientProperties(kit: BigDataTestKit) {
14+
val kafka = kit.endpoint(BigDataService.KAFKA)
15+
16+
check(kafka.property("security.protocol") == "SSL")
17+
check(kafka.property("ssl.truststore.type") == "PKCS12")
18+
check(kafka.property("ssl.truststore.location").isNotBlank())
19+
20+
AdminClient.create(
21+
mapOf(
22+
AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG to kafka.property("bootstrap.servers"),
23+
"security.protocol" to kafka.property("security.protocol"),
24+
"ssl.truststore.location" to kafka.property("ssl.truststore.location"),
25+
"ssl.truststore.password" to kafka.property("ssl.truststore.password"),
26+
"ssl.truststore.type" to kafka.property("ssl.truststore.type"),
27+
),
28+
).use { admin ->
29+
admin.listTopics().names().get()
30+
}
31+
}
32+
}

‎example/spark/src/test/kotlin/org/openprojectx/bigdata/test/example/spark/SparkBigDataScenario.kt‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@ abstract class SparkBigDataScenario {
4848
securityProtocol = context.environment.kafkaSecurityProtocol,
4949
kerberosServiceName = context.environment.kafkaKerberosServiceName,
5050
jaasConfig = context.environment.kafkaJaasConfig,
51+
sslProperties = context.environment.kafkaSslProperties,
5152
)
5253
},
5354
)
@@ -145,6 +146,7 @@ abstract class SparkBigDataScenario {
145146
kafkaKerberosServiceName = extensions.optional("kerberos-material.kafka.service-name")
146147
?: kafka.properties["sasl.kerberos.service.name"],
147148
kafkaJaasConfig = kafka.properties["sasl.jaas.config"],
149+
kafkaSslProperties = kafka.properties.filterKeys { it.startsWith("ssl.") },
148150
)
149151
}
150152

@@ -243,6 +245,7 @@ abstract class SparkBigDataScenario {
243245
securityProtocol: String?,
244246
kerberosServiceName: String?,
245247
jaasConfig: String?,
248+
sslProperties: Map<String, String>,
246249
) {
247250
val reader = spark.read()
248251
.format("kafka")
@@ -251,9 +254,14 @@ abstract class SparkBigDataScenario {
251254
.option("startingOffsets", "earliest")
252255
.option("endingOffsets", "latest")
253256

254-
if (securityProtocol == "SASL_PLAINTEXT") {
257+
if (securityProtocol != null) {
258+
reader.option("kafka.security.protocol", securityProtocol)
259+
}
260+
sslProperties.forEach { (key, value) ->
261+
reader.option("kafka.$key", value)
262+
}
263+
if (securityProtocol == "SASL_PLAINTEXT" || securityProtocol == "SASL_SSL") {
255264
reader
256-
.option("kafka.security.protocol", securityProtocol)
257265
.option("kafka.sasl.mechanism", "GSSAPI")
258266
.option("kafka.sasl.kerberos.service.name", kerberosServiceName ?: "kafka")
259267
.option("kafka.sasl.jaas.config", jaasConfig ?: "")
@@ -428,4 +436,5 @@ data class SparkScenarioEnvironment(
428436
val kafkaSecurityProtocol: String?,
429437
val kafkaKerberosServiceName: String?,
430438
val kafkaJaasConfig: String?,
439+
val kafkaSslProperties: Map<String, String>,
431440
)

‎extensions/src/main/kotlin/org/openprojectx/bigdata/test/extensions/kafka/KafkaAvroSeedExtension.kt‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,9 @@ data class KafkaAvroSeedExtension(
4444
"sasl.mechanism",
4545
"sasl.kerberos.service.name",
4646
"sasl.jaas.config",
47+
"ssl.truststore.location",
48+
"ssl.truststore.password",
49+
"ssl.truststore.type",
4750
).mapNotNull { key ->
4851
properties[key]?.let { key to it }
4952
}.toMap()

‎gradle/libs.versions.toml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ testcontainers = "2.0.4"
1212
hiveDockerTestcontainers = "0.1.4"
1313
spark = "3.5.7"
1414
hadoop = "3.4.2"
15+
kafka = "4.1.2"
1516

1617
junitJupiter = "5.14.1"
1718
junitPlatform = "1.14.1"
@@ -34,6 +35,7 @@ testcontainers = { module = "org.testcontainers:testcontainers" }
3435
testcontainersPostgresql = { module = "org.testcontainers:testcontainers-postgresql" }
3536
testcontainersKafka = { module = "org.testcontainers:testcontainers-kafka" }
3637
testcontainersJunitJupiter = { module = "org.testcontainers:testcontainers-junit-jupiter" }
38+
kafkaClients = { module = "org.apache.kafka:kafka-clients", version.ref = "kafka" }
3739
hiveDockerTestcontainers = { module = "org.openprojectx.hive.docker.core:hive-docker-testcontainers", version.ref = "hiveDockerTestcontainers" }
3840
sparkSql = { module = "org.apache.spark:spark-sql_2.12", version.ref = "spark" }
3941
sparkHive = { module = "org.apache.spark:spark-hive_2.12", version.ref = "spark" }

0 commit comments

Comments
 (0)