Skip to content

Commit 752ef0d

Browse files
CoderYellowCoderYellow
committed
feat: add plaintext Kafka support and enhance Spark integration
- Added plaintext Kafka setup alongside Kerberos-enabled Kafka. - Improved Spark example with Avro-Kafka integration and Iceberg tables on S3/GCS. - Introduced new dependencies for Kafka, Avro, Iceberg, and GCS connectors. - Enhanced configuration management with CI-based repository resolution.
1 parent 2a7a86d commit 752ef0d

6 files changed

Lines changed: 328 additions & 49 deletions

File tree

‎core/build.gradle.kts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,4 +6,5 @@ plugins {
66
dependencies {
77
api(libs.testcontainers)
88
api(libs.testcontainersPostgresql)
9+
api(libs.testcontainersKafka)
910
}

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

Lines changed: 43 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -5,11 +5,13 @@ import org.openprojectx.bigdata.test.core.BigDataService
55
import org.openprojectx.bigdata.test.core.BigDataTestKitOptions
66
import org.openprojectx.bigdata.test.core.ContainerLogMode
77
import org.openprojectx.bigdata.test.core.KerberosAuthOptions
8+
import org.openprojectx.bigdata.test.core.KafkaOptions
89
import org.testcontainers.containers.BindMode
910
import org.testcontainers.containers.GenericContainer
1011
import org.testcontainers.containers.Network
1112
import org.testcontainers.containers.PostgreSQLContainer
1213
import org.testcontainers.containers.wait.strategy.Wait
14+
import org.testcontainers.kafka.KafkaContainer
1315
import org.testcontainers.containers.output.OutputFrame
1416
import org.testcontainers.images.builder.Transferable
1517
import org.testcontainers.lifecycle.Startable
@@ -197,6 +199,8 @@ internal class BigDataContainerFactory(
197199

198200
private fun kafka(): BigDataServiceContainer {
199201
val kafka = options.kafka
202+
if (!kafka.kerberos.enabled) return plaintextKafka(kafka)
203+
200204
val container = GenericBigDataContainer(kafka.image)
201205
.withNetwork(network)
202206
.withNetworkAliases("kafka", "broker1.example.com")
@@ -214,24 +218,22 @@ internal class BigDataContainerFactory(
214218
.withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
215219
.withEnv("CLUSTER_ID", kafka.clusterId)
216220
.waitingFor(Wait.forListeningPort().withStartupTimeout(Duration.ofMinutes(3)))
217-
if (kafka.kerberos.enabled) {
218-
mountKerberos(container)
219-
container
220-
.withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "CONTROLLER:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT")
221-
.withEnv("KAFKA_ADVERTISED_LISTENERS", "SASL_PLAINTEXT://broker1.example.com:9092")
222-
.withEnv("KAFKA_LISTENERS", "SASL_PLAINTEXT://broker1.example.com:9092,CONTROLLER://broker1.example.com:29093")
223-
.withEnv("KAFKA_CONTROLLER_QUORUM_VOTERS", "1@broker1.example.com:29093")
224-
.withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "SASL_PLAINTEXT")
225-
.withEnv("KAFKA_SASL_ENABLED_MECHANISMS", "GSSAPI")
226-
.withEnv("KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL", "GSSAPI")
227-
.withEnv("KAFKA_SASL_KERBEROS_SERVICE_NAME", kafka.kerberos.servicePrincipal.substringBefore("/"))
228-
.withEnv("KAFKA_OPTS", "-Djava.security.krb5.conf=/kerby/client/krb5.conf -Djava.security.auth.login.config=/etc/kafka/kerberos/kafka_server_jaas.conf")
229-
.withEnv("KRB5_CONFIG", "/kerby/client/krb5.conf")
230-
.withCopyToContainer(
231-
Transferable.of(kafkaJaas(kafka.kerberos)),
232-
"/etc/kafka/kerberos/kafka_server_jaas.conf",
233-
)
234-
}
221+
mountKerberos(container)
222+
container
223+
.withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "CONTROLLER:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT")
224+
.withEnv("KAFKA_ADVERTISED_LISTENERS", "SASL_PLAINTEXT://broker1.example.com:9092")
225+
.withEnv("KAFKA_LISTENERS", "SASL_PLAINTEXT://broker1.example.com:9092,CONTROLLER://broker1.example.com:29093")
226+
.withEnv("KAFKA_CONTROLLER_QUORUM_VOTERS", "1@broker1.example.com:29093")
227+
.withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "SASL_PLAINTEXT")
228+
.withEnv("KAFKA_SASL_ENABLED_MECHANISMS", "GSSAPI")
229+
.withEnv("KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL", "GSSAPI")
230+
.withEnv("KAFKA_SASL_KERBEROS_SERVICE_NAME", kafka.kerberos.servicePrincipal.substringBefore("/"))
231+
.withEnv("KAFKA_OPTS", "-Djava.security.krb5.conf=/kerby/client/krb5.conf -Djava.security.auth.login.config=/etc/kafka/kerberos/kafka_server_jaas.conf")
232+
.withEnv("KRB5_CONFIG", "/kerby/client/krb5.conf")
233+
.withCopyToContainer(
234+
Transferable.of(kafkaJaas(kafka.kerberos)),
235+
"/etc/kafka/kerberos/kafka_server_jaas.conf",
236+
)
235237

236238
return BigDataServiceContainer(BigDataService.KAFKA, attachLogs("kafka", container)) {
237239
val bootstrapServers = "${container.host}:${container.getMappedPort(9092)}"
@@ -247,6 +249,28 @@ internal class BigDataContainerFactory(
247249
}
248250
}
249251

252+
private fun plaintextKafka(kafka: KafkaOptions): BigDataServiceContainer {
253+
val container = KafkaContainer(DockerImageName.parse(kafka.image))
254+
.withNetwork(network)
255+
.withNetworkAliases("kafka")
256+
.withListener("kafka:19092")
257+
.withStartupTimeout(Duration.ofMinutes(3))
258+
259+
return BigDataServiceContainer(BigDataService.KAFKA, attachLogs("kafka", container)) {
260+
val bootstrapServers = container.bootstrapServers
261+
BigDataEndpoint(
262+
service = BigDataService.KAFKA,
263+
host = container.host,
264+
ports = mapOf("bootstrap" to container.getMappedPort(9092)),
265+
properties = mapOf(
266+
"bootstrap.servers" to bootstrapServers,
267+
"spring.kafka.bootstrap-servers" to bootstrapServers,
268+
"bootstrap.servers.internal" to "kafka:19092",
269+
),
270+
)
271+
}
272+
}
273+
250274
private fun schemaRegistry(): BigDataServiceContainer {
251275
val kafka = options.kafka
252276
val container = GenericBigDataContainer(kafka.schemaRegistryImage)
@@ -255,7 +279,7 @@ internal class BigDataContainerFactory(
255279
.withServicePort(8085, options.portBindings.schemaRegistry)
256280
.withEnv("SCHEMA_REGISTRY_HOST_NAME", "schema-registry")
257281
.withEnv("SCHEMA_REGISTRY_LISTENERS", "http://0.0.0.0:8085")
258-
.withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "PLAINTEXT://kafka:9092")
282+
.withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "PLAINTEXT://kafka:19092")
259283
.waitingFor(Wait.forHttp("/subjects").forStatusCode(200).withStartupTimeout(Duration.ofMinutes(3)))
260284
if (kafka.schemaRegistryKerberos.enabled || kafka.kerberos.enabled) {
261285
mountKerberos(container)

‎example/spark/build.gradle.kts‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,40 @@
1+
import org.gradle.api.artifacts.component.ModuleComponentIdentifier
2+
13
plugins {
24
id("buildsrc.convention.kotlin-jvm")
35
}
46

57
description = "Spark JUnit 5 example for bigdata-test"
68

9+
configurations.all {
10+
resolutionStrategy.force("org.apache.kafka:kafka-clients:3.4.1")
11+
resolutionStrategy.capabilitiesResolution {
12+
withCapability("org.lz4:lz4-java") {
13+
select(candidates.first {
14+
val id = it.id
15+
id is ModuleComponentIdentifier && id.group == "at.yawk.lz4"
16+
})
17+
because("at.yawk.lz4:lz4-java 1.10.1 is the maintained fork with newer fixes")
18+
}
19+
}
20+
}
21+
722
dependencies {
823
testImplementation(project(":junit5"))
924
testImplementation(libs.sparkSql)
1025
testImplementation(libs.sparkHive)
1126
testImplementation(libs.sparkSqlKafka)
27+
testImplementation(libs.sparkAvro)
1228
testImplementation(libs.hadoopAws)
29+
testImplementation(libs.hadoopClientApi)
30+
testImplementation(libs.icebergSparkRuntime)
31+
testImplementation(libs.icebergAwsBundle)
32+
testImplementation(libs.gcsConnector)
33+
testImplementation(libs.gcsio)
34+
testImplementation(libs.gcsUtilHadoop)
35+
testImplementation(libs.kafkaAvroSerializer)
36+
testImplementation(libs.kafkaSchemaRegistryClient)
37+
testImplementation(libs.avro)
1338
testImplementation(libs.junitJupiterApi)
1439
testRuntimeOnly(libs.junitJupiterEngine)
1540
testRuntimeOnly(libs.junitPlatformLauncher)
@@ -19,5 +44,6 @@ dependencies {
1944
tasks.withType<Test>().configureEach {
2045
jvmArgs(
2146
"--add-exports=java.base/sun.nio.ch=ALL-UNNAMED",
47+
"--add-opens=java.base/java.nio=ALL-UNNAMED",
2248
)
2349
}

0 commit comments

Comments
 (0)