This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new f380f6bc56 [Improve][e2e] Improve the stability of Kafka e2e testing
(#10946)
f380f6bc56 is described below
commit f380f6bc566b88d0812a1de141e390a506b0e362
Author: zhiwei.niu <[email protected]>
AuthorDate: Thu May 28 22:53:06 2026 +0800
[Improve][e2e] Improve the stability of Kafka e2e testing (#10946)
Co-authored-by: Cursor <[email protected]>
---
.../e2e/connector/kafka/KafkaFormatIT.java | 95 ++++++++++-
.../seatunnel/e2e/connector/kafka/KafkaIT.java | 61 +++++--
.../e2e/connector/kafka/KafkaKerberosIT.java | 176 ++++++++++++---------
.../src/test/resources/kerberos/kafka.properties | 7 +-
.../src/test/resources/kerberos/start.sh | 33 +++-
5 files changed, 272 insertions(+), 100 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaFormatIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaFormatIT.java
index 5cda636266..6db41a61d9 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaFormatIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaFormatIT.java
@@ -38,6 +38,10 @@ import
org.apache.seatunnel.e2e.common.container.TestContainer;
import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
import org.apache.seatunnel.e2e.common.junit.TestContainerExtension;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -76,10 +80,12 @@ import java.time.LocalTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
+import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -121,6 +127,18 @@ public class KafkaFormatIT extends TestSuiteBase
implements TestResource {
private static final String DEBEZIUM_DATA_PATH =
"/debezium/debezium_data.txt";
private static final String DEBEZIUM_KAFKA_SOURCE_TOPIC =
"dbserver1.debezium.products";
+ private static final List<String> KAFKA_TOPICS =
+ Arrays.asList(
+ CANAL_KAFKA_SOURCE_TOPIC,
+ OGG_KAFKA_SOURCE_TOPIC,
+ MAXWELL_KAFKA_SOURCE_TOPIC,
+ COMPATIBLE_KAFKA_SOURCE_TOPIC,
+ DEBEZIUM_KAFKA_SOURCE_TOPIC,
+ CANAL_KAFKA_SINK_TOPIC,
+ OGG_KAFKA_SINK_TOPIC,
+ MAXWELL_KAFKA_SINK_TOPIC,
+ DEBEZIUM_KAFKA_SINK_TOPIC);
+
private static final String PG_SINK_TABLE1 = "sink";
private static final String PG_SINK_TABLE2 = "sink2";
@@ -303,7 +321,7 @@ public class KafkaFormatIT extends TestSuiteBase implements
TestResource {
@BeforeAll
@Override
- public void startUp() throws ClassNotFoundException, InterruptedException,
IOException {
+ public void startUp() throws ClassNotFoundException, IOException {
LOG.info("The first stage: Starting Kafka containers...");
createKafkaContainer();
@@ -327,15 +345,22 @@ public class KafkaFormatIT extends TestSuiteBase
implements TestResource {
.atLeast(100, TimeUnit.MILLISECONDS)
.pollInterval(500, TimeUnit.MILLISECONDS)
.atMost(180, TimeUnit.SECONDS)
- .untilAsserted(this::initKafkaConsumer);
+ .untilAsserted(this::initializeKafkaTopics);
+
+ given().ignoreExceptions()
+ .atLeast(100, TimeUnit.MILLISECONDS)
+ .pollInterval(500, TimeUnit.MILLISECONDS)
+ .atMost(180, TimeUnit.SECONDS)
+ .untilAsserted(this::waitForKafkaTopicsReady);
- // local file local data send kafka
given().ignoreExceptions()
.atLeast(100, TimeUnit.MILLISECONDS)
.pollInterval(500, TimeUnit.MILLISECONDS)
- .atMost(3, TimeUnit.MINUTES)
- .untilAsserted(this::initLocalDataToKafka);
- Thread.sleep(20 * 1000);
+ .atMost(180, TimeUnit.SECONDS)
+ .untilAsserted(this::initKafkaConsumer);
+
+ // local file local data send kafka
+ initLocalDataToKafka();
}
@DisabledOnContainer(
@@ -887,6 +912,56 @@ public class KafkaFormatIT extends TestSuiteBase
implements TestResource {
kafkaConsumer = new KafkaConsumer<>(prop);
}
+ private void initializeKafkaTopics() {
+ try (AdminClient adminClient = createKafkaAdmin()) {
+ Set<String> existingTopics =
adminClient.listTopics().names().get(30, TimeUnit.SECONDS);
+ List<NewTopic> topicsToCreate =
+ KAFKA_TOPICS.stream()
+ .filter(topic -> !existingTopics.contains(topic))
+ .map(topic -> new NewTopic(topic, 1, (short) 1))
+ .collect(Collectors.toList());
+ if (!topicsToCreate.isEmpty()) {
+ adminClient.createTopics(topicsToCreate).all().get(60,
TimeUnit.SECONDS);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while initializing Kafka
topics", e);
+ } catch (ExecutionException | java.util.concurrent.TimeoutException e)
{
+ throw new RuntimeException("Failed to initialize Kafka topics", e);
+ }
+ }
+
+ private void waitForKafkaTopicsReady() {
+ try (AdminClient adminClient = createKafkaAdmin()) {
+ Map<String, TopicDescription> topicDescriptions =
+ adminClient
+ .describeTopics(KAFKA_TOPICS)
+ .allTopicNames()
+ .get(30, TimeUnit.SECONDS);
+ Assertions.assertEquals(
+ new HashSet<>(KAFKA_TOPICS).size(),
+ topicDescriptions.size(),
+ "Kafka topics are not ready yet");
+ KAFKA_TOPICS.forEach(
+ topic ->
+ Assertions.assertTrue(
+ topicDescriptions.containsKey(topic),
+ "Kafka topic is not ready yet: " + topic));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting for Kafka
topics", e);
+ } catch (ExecutionException | java.util.concurrent.TimeoutException e)
{
+ throw new RuntimeException("Kafka topics are not ready yet", e);
+ }
+ }
+
+ private AdminClient createKafkaAdmin() {
+ Properties props = new Properties();
+ props.put(
+ AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
KAFKA_CONTAINER.getBootstrapServers());
+ return AdminClient.create(props);
+ }
+
// Example Initialize the pg sink table
private void initializeJdbcTable() {
try (Connection connection =
@@ -981,7 +1056,11 @@ public class KafkaFormatIT extends TestSuiteBase
implements TestResource {
producer.send(record).get();
}
} catch (IOException | InterruptedException |
ExecutionException e) {
- e.printStackTrace();
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ throw new RuntimeException(
+ "Failed to initialize local Kafka data for topic "
+ kafkaTopic, e);
}
}
}
@@ -1013,7 +1092,7 @@ public class KafkaFormatIT extends TestSuiteBase
implements TestResource {
LOG.info("truncate table sink");
}
} catch (SQLException e) {
- e.printStackTrace();
+ throw new RuntimeException("Query Postgre sink table failed: " +
tableName, e);
}
return actual;
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
index 704febe57d..6bcd103956 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
@@ -111,6 +111,7 @@ import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
@@ -197,7 +198,23 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
kafkaExactlyOnceSink,
kafkaExactlyOnceBatchSource,
kafkaExactlyOnceBatchSink);
- adminClient.createTopics(topics);
+ adminClient.createTopics(topics).all().get(60, SECONDS);
+
+ List<String> topicNames =
+
topics.stream().map(NewTopic::name).collect(Collectors.toList());
+ given().ignoreExceptions()
+ .atLeast(100, TimeUnit.MILLISECONDS)
+ .pollInterval(500, TimeUnit.MILLISECONDS)
+ .atMost(60, SECONDS)
+ .untilAsserted(
+ () -> {
+ Map<String, TopicDescription> desc =
+ adminClient
+ .describeTopics(topicNames)
+ .allTopicNames()
+ .get(30, SECONDS);
+ Assertions.assertEquals(topicNames.size(),
desc.size());
+ });
}
log.info("Write 100 records to topic test_topic_source");
@@ -616,8 +633,8 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
}
});
- // Wait for job to start and process initial data
- Awaitility.await().pollDelay(5, SECONDS).atMost(1, MINUTES).until(()
-> true);
+ // Wait for job to be RUNNING before changing partitions.
+ awaitJobRunning(container, jobId);
try (AdminClient adminClient = createKafkaAdmin()) {
Map<String, NewPartitions> newPartitions = new HashMap<>();
@@ -626,7 +643,22 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
log.info("Successfully created new partition for topic: {}",
sourceTopic);
}
- Awaitility.await().pollDelay(3, SECONDS).atMost(30, SECONDS).until(()
-> true);
+ Awaitility.await()
+ .pollInterval(1, SECONDS)
+ .atMost(30, SECONDS)
+ .untilAsserted(
+ () -> {
+ try (AdminClient adminClient = createKafkaAdmin())
{
+ Map<String, TopicDescription>
topicDescriptions =
+ adminClient
+
.describeTopics(Arrays.asList(sourceTopic))
+ .allTopicNames()
+ .get();
+ Assertions.assertTrue(
+
topicDescriptions.get(sourceTopic).partitions().size()
+ >= 2);
+ }
+ });
for (int i = 0; i < 15; i++) {
String message =
@@ -702,8 +734,7 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
}
});
- // Warm up (simple delay).
- Awaitility.await().pollDelay(5, SECONDS).atMost(1, MINUTES).until(()
-> true);
+ awaitJobRunning(container, jobId);
// Produce 10 additional records after the job starts.
for (int i = 0; i < 10; i++) {
@@ -762,7 +793,6 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
// After restore, sink should advance by the 15 newly produced records
at
// minimum.
Awaitility.await()
- .pollDelay(3, SECONDS)
.pollInterval(2, SECONDS)
.atMost(5, MINUTES)
.until(() -> visibleCountOnP0(sinkTopic) ==
expectedSinkAfterFirstRun + 15);
@@ -802,7 +832,7 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
}
});
- Awaitility.await().pollDelay(5, SECONDS).atMost(1, MINUTES).until(()
-> true);
+ awaitJobRunning(container, jobId);
// Produce 10 records after job start; latest mode should consume only
these 10
// initially.
@@ -849,7 +879,6 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
});
Awaitility.await()
- .pollDelay(3, SECONDS)
.pollInterval(2, SECONDS)
.atMost(5, MINUTES)
.until(() -> visibleCountOnP0(sinkTopic) ==
expectedSinkAfterFirstRun + 15);
@@ -888,7 +917,7 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
}
});
- Awaitility.await().pollDelay(5, SECONDS).atMost(1, MINUTES).until(()
-> true);
+ awaitJobRunning(container, jobId);
// Produce 10 records after job start.
for (int i = 0; i < 10; i++) {
@@ -936,7 +965,6 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
});
Awaitility.await()
- .pollDelay(3, SECONDS)
.pollInterval(2, SECONDS)
.atMost(5, MINUTES)
.until(() -> visibleCountOnP0(sinkTopic) ==
expectedSinkAfterFirstRun + 15);
@@ -976,7 +1004,7 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
}
});
- Awaitility.await().pollDelay(5, SECONDS).atMost(1, MINUTES).until(()
-> true);
+ awaitJobRunning(container, jobId);
// Produce 10 records after job start.
for (int i = 0; i < 10; i++) {
@@ -1025,12 +1053,19 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
});
Awaitility.await()
- .pollDelay(3, SECONDS)
.pollInterval(2, SECONDS)
.atMost(5, MINUTES)
.until(() -> visibleCountOnP0(sinkTopic) ==
expectedSinkAfterFirstRun + 15 - 11);
}
+ private void awaitJobRunning(TestContainer container, String jobId) {
+ Awaitility.await()
+ .pollInterval(2, SECONDS)
+ .atMost(1, MINUTES)
+ .untilAsserted(
+ () -> Assertions.assertEquals("RUNNING",
container.getJobStatus(jobId)));
+ }
+
/**
* Get visible record count on partition-0: endOffset - beginningOffset
(exclusive upper bound).
*/
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaKerberosIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaKerberosIT.java
index 435602ffb6..9d24670b78 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaKerberosIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaKerberosIT.java
@@ -38,6 +38,7 @@ import
org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
import org.apache.seatunnel.e2e.common.util.ContainerUtil;
import org.apache.seatunnel.format.text.TextSerializationSchema;
+import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -57,6 +58,7 @@ import org.junit.jupiter.api.TestTemplate;
import org.testcontainers.containers.Container;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.lifecycle.Startables;
import org.testcontainers.shaded.org.awaitility.Awaitility;
import org.testcontainers.utility.DockerImageName;
@@ -70,6 +72,7 @@ import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.Arrays;
+import java.util.Base64;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -94,7 +97,6 @@ public class KafkaKerberosIT extends TestSuiteBase implements
TestResource {
// The hostname is uniformly set to lowercase letters to prevent errors
during Kerberos
// authentication
private static final String KAFKA_HOST = "kafkacluster";
- private static final String BOOTSTRAP_SERVERS = KAFKA_HOST + ":9092";
private KafkaProducer<byte[], byte[]> producer;
@@ -156,7 +158,6 @@ public class KafkaKerberosIT extends TestSuiteBase
implements TestResource {
.withLogConsumer(
new Slf4jLogConsumer(
DockerLoggerFactory.getLogger(KERBEROS_IMAGE_NAME)));
- kerberosContainer.setPortBindings(Arrays.asList("88/udp:88/udp",
"749:749"));
Startables.deepStart(Stream.of(kerberosContainer)).join();
log.info("Kerberos just started");
@@ -164,10 +165,12 @@ public class KafkaKerberosIT extends TestSuiteBase
implements TestResource {
"bash",
"-c",
"kadmin.local -q \"addprinc -randkey
kafka/[email protected]\"");
+ kerberosContainer.execInContainer(
+ "bash", "-c", "kadmin.local -q \"addprinc -randkey
kafka/[email protected]\"");
kerberosContainer.execInContainer(
"bash",
"-c",
- "kadmin.local -q \"xst -k /tmp/kafka.keytab
kafka/[email protected]\"");
+ "kadmin.local -q \"xst -k /tmp/kafka.keytab
kafka/[email protected] kafka/[email protected]\"");
// test.keytab verify unprivileged keytab usage
kerberosContainer.execInContainer(
@@ -202,81 +205,91 @@ public class KafkaKerberosIT extends TestSuiteBase
implements TestResource {
.withFileSystemBind(
ContainerUtil.getResourcesFile("/kerberos/krb5.conf").getPath(),
"/etc/krb5.conf")
- .withExposedPorts(9092, 2181)
- .withFileSystemBind(
-
ContainerUtil.getResourcesFile("/kerberos/kafka.properties")
- .getPath(),
- "/etc/kafka/kafka.properties")
+ .withExposedPorts(9093, 2181)
.withFileSystemBind("/tmp/kafka.keytab",
"/tmp/kafka.keytab")
.withLogConsumer(
new Slf4jLogConsumer(
DockerLoggerFactory.getLogger(KAFKA_IMAGE_NAME)))
+ .waitingFor(
+ Wait.forLogMessage(".*ZooKeeper is ready.*", 1)
+
.withStartupTimeout(Duration.ofMinutes(2)))
.withCommand(
"bash",
"-c",
FileUtils.readFileToStr(
ContainerUtil.getResourcesFile("/kerberos/start.sh")
.toPath()));
- kafkaContainer.setPortBindings(Arrays.asList("9092:9092",
"2181:2181"));
Startables.deepStart(Stream.of(kafkaContainer)).join();
- log.info("Kafka container started");
-
- // Add Hosts, local connection kerberos kafka use
- appendToHosts("127.0.0.1", "kafkacluster");
+ log.info("Kafka container started, ZooKeeper is ready");
+
+ int externalMappedPort = kafkaContainer.getMappedPort(9093);
+ String originalProps =
+ FileUtils.readFileToStr(
+
ContainerUtil.getResourcesFile("/kerberos/kafka.properties").toPath());
+ String patchedProps =
+ originalProps.replace(
+ "EXTERNAL://localhost:9093", "EXTERNAL://localhost:" +
externalMappedPort);
+ String propsBase64 =
Base64.getEncoder().encodeToString(patchedProps.getBytes("UTF-8"));
+ kafkaContainer.execInContainer(
+ "bash",
+ "-c",
+ "echo '"
+ + propsBase64
+ + "' | base64 -d > /etc/kafka/kafka.properties"
+ + " && touch /tmp/start_kafka");
+ log.info(
+ "Kafka config written with external port {}, signaling broker
start",
+ externalMappedPort);
Awaitility.given()
.ignoreExceptions()
.atLeast(100, TimeUnit.MILLISECONDS)
.pollInterval(500, TimeUnit.MILLISECONDS)
.atMost(180, TimeUnit.SECONDS)
- .untilAsserted(this::initKafkaProducer);
+ .untilAsserted(this::waitForKafkaKerberosReady);
+ initKafkaProducer();
}
private void initKafkaProducer() {
- Properties props = new Properties();
- props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
+ Properties props = createKafkaClientSecurityProperties();
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
ByteArraySerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
ByteArraySerializer.class);
- props.put("security.protocol", "SASL_PLAINTEXT");
- props.put("sasl.mechanism", "GSSAPI");
- props.put("sasl.kerberos.service.name", "kafka");
producer = new KafkaProducer<>(props);
}
private Properties kafkaConsumerConfig() {
- Properties props = new Properties();
- props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
+ Properties props = createKafkaClientSecurityProperties();
props.put(ConsumerConfig.GROUP_ID_CONFIG,
"seatunnel-kafka-sink-group");
props.put(
ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
OffsetResetStrategy.EARLIEST.toString().toLowerCase());
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class);
- props.put("security.protocol", "SASL_PLAINTEXT");
- props.put("sasl.mechanism", "GSSAPI");
- props.put("sasl.kerberos.service.name", "kafka");
return props;
}
- private static void appendToHosts(String ip, String hostname) {
- try {
- String entry = String.format("%s %s", ip, hostname);
- ProcessBuilder processBuilder =
- new ProcessBuilder("sudo", "sh", "-c", "echo '" + entry +
"' >> /etc/hosts");
- processBuilder.redirectErrorStream(true);
-
- Process process = processBuilder.start();
-
- int exitCode = process.waitFor();
- if (exitCode == 0) {
- log.info("Successfully added to /etc/hosts: {}", entry);
- } else {
- log.error("Failed to add to /etc/hosts: {}", entry);
- }
- } catch (Exception e) {
- log.error("Failed to add to /etc/hosts: {}", e.getMessage());
- throw new RuntimeException(e);
- }
+ private Properties createKafkaClientSecurityProperties() {
+ Properties props = new Properties();
+ props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
getBootstrapServers());
+ return props;
+ }
+
+ private void waitForKafkaKerberosReady() {
+ String logs = kafkaContainer.getLogs();
+ Assertions.assertTrue(
+ logs.contains("started (kafka.server.KafkaServer)"),
+ "Kafka kerberos broker has not fully started");
+ }
+
+ static boolean containsKafkaAuthenticationFailure(String logs) {
+ return logs != null
+ && logs.contains("Failed authentication with")
+ && (logs.contains("during SASL handshake")
+ || logs.contains("SaslAuthenticationException"));
+ }
+
+ private String getBootstrapServers() {
+ return kafkaContainer.getHost() + ":" +
kafkaContainer.getMappedPort(9093);
}
@TestTemplate
@@ -285,19 +298,39 @@ public class KafkaKerberosIT extends TestSuiteBase
implements TestResource {
container.copyFileToContainer("/kerberos/krb5.conf", "/etc/krb5.conf");
container.copyAbsolutePathToContainer("/tmp/test.keytab",
"/tmp/kafka.keytab");
- Container.ExecResult execResult =
-
container.executeJob("/kerberos/kafka_sink_fake_to_kafka_kerberos.conf");
- Assertions.assertEquals(1, execResult.getExitCode());
+ int logOffsetBeforeJob = container.getServerLogs().length();
+
+ String jobId = "654321";
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ container.executeJob(
+
"/kerberos/kafka_sink_fake_to_kafka_kerberos.conf", jobId);
+ } catch (IOException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ Awaitility.given()
+ .ignoreExceptions()
+ .atLeast(100, TimeUnit.MILLISECONDS)
+ .pollInterval(500, TimeUnit.MILLISECONDS)
+ .atMost(180, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> Assertions.assertEquals("FAILED",
container.getJobStatus(jobId)));
+
+ String newLogs =
container.getServerLogs().substring(logOffsetBeforeJob);
Assertions.assertTrue(
- execResult
- .getStderr()
- .contains(
- "Could not login: the client is being asked
for a password, but the Kafka client code does not currently support obtaining
a password from the user."));
+ newLogs.contains(
+ "Could not login: the client is being asked for a
password, but the Kafka client code does not currently support obtaining a
password from the user."),
+ "Expected Kerberos login failure with wrong keytab");
}
@TestTemplate
public void testNotKerberosConfig(TestContainer container)
throws IOException, InterruptedException {
+ int brokerLogOffsetBeforeJob = kafkaContainer.getLogs().length();
+
String jobId = "123456";
CompletableFuture.runAsync(
() -> {
@@ -307,22 +340,27 @@ public class KafkaKerberosIT extends TestSuiteBase
implements TestResource {
throw new RuntimeException(e);
}
});
- // step 1. Verify whether Kafka has authentication failure logs
+
Awaitility.given()
+ .ignoreExceptions()
.atLeast(100, TimeUnit.MILLISECONDS)
.pollInterval(500, TimeUnit.MILLISECONDS)
.atMost(60, TimeUnit.SECONDS)
.untilAsserted(
- () ->
- Assertions.assertTrue(
- kafkaContainer
- .execInContainer(
- "bash",
- "-c",
- "tail
/var/log/kafka/server.log")
- .getStdout()
- .matches(
- "(?s).*Failed
authentication with /.*? \\(Unexpected Kafka request of type METADATA during
SASL handshake.*")));
+ () -> Assertions.assertEquals("RUNNING",
container.getJobStatus(jobId)));
+
+ Awaitility.given()
+ .ignoreExceptions()
+ .pollInterval(1, TimeUnit.SECONDS)
+ .atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ String newBrokerLogs =
+
kafkaContainer.getLogs().substring(brokerLogOffsetBeforeJob);
+ Assertions.assertTrue(
+
containsKafkaAuthenticationFailure(newBrokerLogs),
+ "Expected broker to log authentication
failure for non-Kerberos PLAINTEXT client");
+ });
container.cancelJob(jobId);
@@ -332,19 +370,6 @@ public class KafkaKerberosIT extends TestSuiteBase
implements TestResource {
String jobStatus =
container.getJobStatus(String.valueOf(jobId));
Assertions.assertEquals("CANCELED", jobStatus);
});
-
- // step 2. Verify that the program outputs retry logs
- Awaitility.given()
- .atLeast(100, TimeUnit.MILLISECONDS)
- .pollInterval(500, TimeUnit.MILLISECONDS)
- .atMost(60, TimeUnit.SECONDS)
- .untilAsserted(
- () ->
- Assertions.assertTrue(
- container
- .getServerLogs()
- .contains(
- "Cancelled in-flight
INIT_PRODUCER_ID request with correlation id")));
}
@TestTemplate
@@ -451,5 +476,8 @@ public class KafkaKerberosIT extends TestSuiteBase
implements TestResource {
if (kafkaContainer != null) {
kafkaContainer.close();
}
+ if (kerberosContainer != null) {
+ kerberosContainer.close();
+ }
}
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kerberos/kafka.properties
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kerberos/kafka.properties
index affde507cb..75455b60f4 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kerberos/kafka.properties
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kerberos/kafka.properties
@@ -17,9 +17,10 @@
################################################################################
zookeeper.connect=kafkacluster:2181
-listeners=SASL_PLAINTEXT://0.0.0.0:9092
-advertised.listeners=SASL_PLAINTEXT://kafkacluster:9092
-security.inter.broker.protocol=SASL_PLAINTEXT
+listeners=INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:9093
+advertised.listeners=INTERNAL://kafkacluster:9092,EXTERNAL://localhost:9093
+listener.security.protocol.map=INTERNAL:SASL_PLAINTEXT,EXTERNAL:PLAINTEXT
+inter.broker.listener.name=INTERNAL
sasl.mechanism.inter.broker.protocol=GSSAPI
sasl.enabled.mechanisms=GSSAPI
sasl.kerberos.service.name=kafka
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kerberos/start.sh
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kerberos/start.sh
index 74ce841733..522ca3c100 100755
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kerberos/start.sh
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kerberos/start.sh
@@ -26,7 +26,23 @@ nohup java -Xmx512M -Xms512M -server \
-Dsun.security.krb5.debug=true
org.apache.zookeeper.server.quorum.QuorumPeerMain \
/etc/kafka/zookeeper.properties &
-sleep 5
+ZK_READY=false
+for i in $(seq 1 30); do
+ if nc -z localhost 2181 2>/dev/null; then
+ echo "ZooKeeper is ready after ${i}s"
+ ZK_READY=true
+ break
+ fi
+ sleep 1
+done
+if [ "$ZK_READY" != "true" ]; then
+ echo "ERROR: ZooKeeper failed to start within 30s" >&2
+ exit 1
+fi
+
+echo "Waiting for Kafka config..."
+while [ ! -f /tmp/start_kafka ]; do sleep 0.1; done
+echo "Kafka config received, starting broker..."
nohup java -Xmx1G -Xms1G -server \
-XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35
\
@@ -39,6 +55,19 @@ nohup java -Xmx1G -Xms1G -server \
-Dsun.security.krb5.debug=true
-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf \
-Djava.security.krb5.conf=/etc/krb5.conf kafka.Kafka
/etc/kafka/kafka.properties &
-sleep 5
+KAFKA_READY=false
+for i in $(seq 1 60); do
+ if grep -q "started (kafka.server.KafkaServer)" /var/log/kafka/server.log
2>/dev/null; then
+ echo "Kafka broker is ready after ${i}s"
+ KAFKA_READY=true
+ break
+ fi
+ sleep 1
+done
+if [ "$KAFKA_READY" != "true" ]; then
+ echo "ERROR: Kafka broker failed to start within 60s" >&2
+ cat /var/log/kafka/server.log >&2 2>/dev/null
+ exit 1
+fi
tail -f /var/log/kafka/server.log
\ No newline at end of file
