This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 4859b981e31 [FLINK-40048][e2e] Replace Kafka with filesystem connector
in SqlClientITCase (#28606)
4859b981e31 is described below
commit 4859b981e31cc49f38af66adb392545c411dd096
Author: Martijn Visser <[email protected]>
AuthorDate: Fri Jul 3 07:58:09 2026 +0200
[FLINK-40048][e2e] Replace Kafka with filesystem connector in
SqlClientITCase (#28606)
* [FLINK-40048][e2e] Replace Kafka with filesystem connector in
SqlClientITCase
testMatchRecognize has been @Disabled since the Kafka connector was
externalized (FLINK-30859), leaving kafka-clients, the 1.17-era
flink-sql-connector-kafka and a KafkaContainer behind for no active
test while still requiring dependency upgrades.
Read the test records from a file staged into the JobManager and
TaskManager containers instead, and re-enable the test. Verify results
by waiting for the bounded job to reach FINISHED rather than polling
the sink file in an unbounded loop that swallowed job failures.
Also drop the RegReplace function registration: its invocation moved to
the Kafka-specific script in FLINK-10624 and CREATE FUNCTION is covered
by the flink-end-to-end-tests-sql module, which still consumes the
SqlToolbox.jar built here.
Generated-by: Claude Code (Fable 5)
---
.../flink-sql-client-test/pom.xml | 44 -----
.../src/test/java/SqlClientITCase.java | 179 ++++++++-------------
2 files changed, 65 insertions(+), 158 deletions(-)
diff --git a/flink-end-to-end-tests/flink-sql-client-test/pom.xml
b/flink-end-to-end-tests/flink-sql-client-test/pom.xml
index 849ef672608..78df21239ee 100644
--- a/flink-end-to-end-tests/flink-sql-client-test/pom.xml
+++ b/flink-end-to-end-tests/flink-sql-client-test/pom.xml
@@ -51,32 +51,6 @@ under the License.
<version>${project.version}</version>
</dependency>
- <!-- The following dependencies are for connector/format
sql-jars that
- we copy using the maven-dependency-plugin. When
extending the test
- to cover more connectors/formats, add a dependency here
and an entry
- to the dependency-plugin configuration below.
- This ensures that all modules we actually need (as
defined by the
- dependency-plugin configuration) are built before this
module. -->
- <dependency>
- <!-- Used by maven-dependency-plugin -->
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-sql-connector-kafka</artifactId>
- <version>3.0.0-1.17</version>
- <scope>provided</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka-clients</artifactId>
- <version>3.9.2</version>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>testcontainers-kafka</artifactId>
- <scope>test</scope>
- </dependency>
-
<dependency>
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
@@ -85,18 +59,6 @@ under the License.
</dependencies>
- <dependencyManagement>
- <dependencies>
- <dependency>
- <!-- Pick an arbitrary version here to satisfy
the enforcer-plugin,
- as we neither access nor package the
kafka dependencies -->
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka-clients</artifactId>
- <version>2.2.2</version>
- </dependency>
- </dependencies>
- </dependencyManagement>
-
<build>
<plugins>
<!-- Build toolbox jar. -->
@@ -135,12 +97,6 @@ under the License.
When extending
this list please also add a dependency
for the
respective module. -->
<artifactItems>
- <artifactItem>
-
<groupId>org.apache.flink</groupId>
-
<artifactId>flink-sql-connector-kafka</artifactId>
-
<version>3.0.0-1.17</version>
-
<type>jar</type>
- </artifactItem>
<artifactItem>
<groupId>org.apache.flink</groupId>
<artifactId>flink-test-utils</artifactId>
diff --git
a/flink-end-to-end-tests/flink-sql-client-test/src/test/java/SqlClientITCase.java
b/flink-end-to-end-tests/flink-sql-client-test/src/test/java/SqlClientITCase.java
index 0414854c7ba..fdd2295d827 100644
---
a/flink-end-to-end-tests/flink-sql-client-test/src/test/java/SqlClientITCase.java
+++
b/flink-end-to-end-tests/flink-sql-client-test/src/test/java/SqlClientITCase.java
@@ -16,75 +16,45 @@
* limitations under the License.
*/
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.client.deployment.StandaloneClusterId;
+import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.connector.testframe.container.FlinkContainers;
import org.apache.flink.connector.testframe.container.FlinkContainersSettings;
import org.apache.flink.connector.testframe.container.TestcontainersSettings;
import org.apache.flink.connector.upserttest.sink.UpsertTestFileUtil;
+import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.test.resources.ResourceTestUtils;
import org.apache.flink.test.util.SQLJobSubmission;
-import org.apache.flink.util.DockerImageVersions;
-import org.apache.kafka.clients.admin.AdminClient;
-import org.apache.kafka.clients.admin.NewTopic;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.serialization.BytesSerializer;
-import org.apache.kafka.common.serialization.StringSerializer;
-import org.apache.kafka.common.utils.Bytes;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.Network;
-import org.testcontainers.containers.output.Slf4jLogConsumer;
-import org.testcontainers.junit.jupiter.Container;
-import org.testcontainers.junit.jupiter.Testcontainers;
-import org.testcontainers.kafka.ConfluentKafkaContainer;
-import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.MountableFile;
import java.io.File;
-import java.io.IOException;
+import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
-import java.util.Arrays;
-import java.util.Collections;
+import java.util.Collection;
import java.util.List;
-import java.util.Properties;
import static org.assertj.core.api.Assertions.assertThat;
/** E2E Test for SqlClient. */
-@Testcontainers
class SqlClientITCase {
private static final Logger LOG =
LoggerFactory.getLogger(SqlClientITCase.class);
- private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
-
- private static final Slf4jLogConsumer LOG_CONSUMER = new
Slf4jLogConsumer(LOG);
- private static final Path sqlToolBoxJar =
ResourceTestUtils.getResource(".*/SqlToolbox\\.jar");
-
- private final Path sqlConnectorKafkaJar =
ResourceTestUtils.getResource(".*kafka.*\\.jar");
-
private final Path sqlConnectorUpsertTestJar =
ResourceTestUtils.getResource(".*flink-test-utils.*\\.jar");
- private static final Network NETWORK = Network.newNetwork();
-
- @Container
- private static final ConfluentKafkaContainer KAFKA =
- new
ConfluentKafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA))
- .withNetwork(NETWORK)
- .withListener(INTER_CONTAINER_KAFKA_ALIAS + ":19092")
- .withLogConsumer(LOG_CONSUMER);
-
private final FlinkContainers flink =
FlinkContainers.builder()
.withFlinkContainersSettings(
@@ -96,11 +66,7 @@ class SqlClientITCase {
Duration.ofMillis(500))
.build())
.withTestcontainersSettings(
- TestcontainersSettings.builder()
- .network(NETWORK)
- .logger(LOG)
- .dependsOn(KAFKA)
- .build())
+
TestcontainersSettings.builder().logger(LOG).build())
.build();
@TempDir private File tempDir;
@@ -120,7 +86,7 @@ class SqlClientITCase {
String outputFilepath = "/flink/records-upsert.out";
List<String> sqlLines =
- Arrays.asList(
+ List.of(
"SET 'execution.runtime-mode' = 'batch';",
"",
"CREATE TABLE UpsertSinkTable (",
@@ -158,7 +124,7 @@ class SqlClientITCase {
String outputFilepath = "/flink/records-append.out";
List<String> sqlLines =
- Arrays.asList(
+ List.of(
"SET 'execution.runtime-mode' = 'batch';",
"",
"CREATE TABLE AppendSinkTable (",
@@ -189,26 +155,33 @@ class SqlClientITCase {
}
@Test
- @Disabled("Disable due to Kafka connector need to release a new version
2.0.")
void testMatchRecognize() throws Exception {
String outputFilepath = "/flink/records-matchrecognize.out";
-
- String[] messages =
- new String[] {
- "{\"timestamp\": \"2018-03-12T08:00:00Z\", \"user\":
\"Alice\", \"event\": { \"type\": \"WARNING\", \"message\": \"This is a
warning.\"}}",
- "{\"timestamp\": \"2018-03-12T08:10:00Z\", \"user\":
\"Alice\", \"event\": { \"type\": \"WARNING\", \"message\": \"This is a
warning.\"}}",
- "{\"timestamp\": \"2018-03-12T09:00:00Z\", \"user\":
\"Bob\", \"event\": { \"type\": \"WARNING\", \"message\": \"This is another
warning.\"}}",
- "{\"timestamp\": \"2018-03-12T09:10:00Z\", \"user\":
\"Alice\", \"event\": { \"type\": \"INFO\", \"message\": \"This is a info.\"}}",
- "{\"timestamp\": \"2018-03-12T09:20:00Z\", \"user\":
\"Steve\", \"event\": { \"type\": \"INFO\", \"message\": \"This is another
info.\"}}",
- "{\"timestamp\": \"2018-03-12T09:30:00Z\", \"user\":
\"Steve\", \"event\": { \"type\": \"INFO\", \"message\": \"This is another
info.\"}}",
- "{\"timestamp\": \"2018-03-12T09:30:00Z\", \"user\": null,
\"event\": { \"type\": \"WARNING\", \"message\": \"This is a bad message
because the user is missing.\"}}",
- "{\"timestamp\": \"2018-03-12T10:40:00Z\", \"user\":
\"Bob\", \"event\": { \"type\": \"ERROR\", \"message\": \"This is an error.\"}}"
- };
- sendMessages("test-json", messages);
+ String inputFilepath = "/tmp/test-json.jsonl";
+
+ List<String> messages =
+ List.of(
+ "{\"timestamp\": \"2018-03-12T08:00:00Z\", \"user\":
\"Alice\", \"event\": { \"type\": \"WARNING\", \"message\": \"This is a
warning.\"}}",
+ "{\"timestamp\": \"2018-03-12T08:10:00Z\", \"user\":
\"Alice\", \"event\": { \"type\": \"WARNING\", \"message\": \"This is a
warning.\"}}",
+ "{\"timestamp\": \"2018-03-12T09:00:00Z\", \"user\":
\"Bob\", \"event\": { \"type\": \"WARNING\", \"message\": \"This is another
warning.\"}}",
+ "{\"timestamp\": \"2018-03-12T09:10:00Z\", \"user\":
\"Alice\", \"event\": { \"type\": \"INFO\", \"message\": \"This is a info.\"}}",
+ "{\"timestamp\": \"2018-03-12T09:20:00Z\", \"user\":
\"Steve\", \"event\": { \"type\": \"INFO\", \"message\": \"This is another
info.\"}}",
+ "{\"timestamp\": \"2018-03-12T09:30:00Z\", \"user\":
\"Steve\", \"event\": { \"type\": \"INFO\", \"message\": \"This is another
info.\"}}",
+ "{\"timestamp\": \"2018-03-12T09:30:00Z\", \"user\":
null, \"event\": { \"type\": \"WARNING\", \"message\": \"This is a bad message
because the user is missing.\"}}",
+ "{\"timestamp\": \"2018-03-12T10:40:00Z\", \"user\":
\"Bob\", \"event\": { \"type\": \"ERROR\", \"message\": \"This is an
error.\"}}");
+ File inputFile = new File(tempDir, "test-json.jsonl");
+ Files.write(inputFile.toPath(), messages);
+ MountableFile mountableFile =
MountableFile.forHostPath(inputFile.toPath());
+ flink.getJobManager().copyFileToContainer(mountableFile,
inputFilepath);
+ for (GenericContainer<?> taskManager : flink.getTaskManagers()) {
+ taskManager.copyFileToContainer(mountableFile, inputFilepath);
+ }
List<String> sqlLines =
- Arrays.asList(
- "CREATE FUNCTION RegReplace AS
'org.apache.flink.table.toolbox.StringRegexReplaceFunction';",
+ List.of(
+ // MATCH_RECOGNIZE requires streaming mode; the
bounded filesystem source
+ // emits MAX_WATERMARK at end of input which flushes
the pending match
+ "SET 'execution.runtime-mode' = 'streaming';",
"",
"CREATE TABLE JsonSourceTable (",
" `timestamp` TIMESTAMP_LTZ(3),",
@@ -217,12 +190,8 @@ class SqlClientITCase {
" `rowtime` AS `timestamp`,",
" WATERMARK FOR `rowtime` AS `rowtime` - INTERVAL
'2' SECOND",
") WITH (",
- " 'connector' = 'kafka',",
- " 'topic' = 'test-json',",
- " 'properties.bootstrap.servers' = '"
- + INTER_CONTAINER_KAFKA_ALIAS
- + ":19092',",
- " 'scan.startup.mode' = 'earliest-offset',",
+ " 'connector' = 'filesystem',",
+ " 'path' = 'file://" + inputFilepath + "',",
" 'format' = 'json',",
" 'json.timestamp-format.standard' = 'ISO-8601'",
");",
@@ -257,60 +226,42 @@ class SqlClientITCase {
verifyNumberOfResultRecords(outputFilepath, 1);
}
- // duplicated from KafkaContainerClient@flink-end-to-end-tests-common
- private void sendMessages(String topic, String... messages) {
- Properties props = new Properties();
- props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
KAFKA.getBootstrapServers());
- props.put(ProducerConfig.ACKS_CONFIG, "all");
-
- final int numPartitions = 1;
- final short replicationFactor = 1;
- try (AdminClient admin = AdminClient.create(props)) {
- admin.createTopics(
- Collections.singletonList(
- new NewTopic(topic, numPartitions,
replicationFactor)))
- .all()
- .get();
- } catch (Exception e) {
- throw new IllegalStateException(
- String.format(
- "Fail to create topic [%s partitions: %d replicas:
%d].",
- topic, numPartitions, replicationFactor),
- e);
- }
-
- try (Producer<Bytes, String> producer =
- new KafkaProducer<>(props, new BytesSerializer(), new
StringSerializer())) {
- for (String message : messages) {
- producer.send(new ProducerRecord<>(topic, message));
- }
- }
- }
-
private void verifyNumberOfResultRecords(String resultFilePath, int
expectedNumberOfRecords)
- throws IOException, InterruptedException {
+ throws Exception {
+ RestClusterClient<StandaloneClusterId> clusterClient =
flink.getRestClusterClient();
+ // the sink flushes all records at the latest when the bounded job
finishes, so the
+ // result file is complete once the job reaches a terminal state
+ CommonTestUtils.waitUntilIgnoringExceptions(
+ () -> {
+ try {
+ Collection<JobStatusMessage> jobs =
clusterClient.listJobs().get();
+ return !jobs.isEmpty()
+ && jobs.stream()
+ .allMatch(
+ job ->
job.getJobState().isGloballyTerminalState());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ },
+ Duration.ofMinutes(2),
+ Duration.ofMillis(100),
+ "Job did not terminate within 2 minutes");
+ Collection<JobStatusMessage> jobs = clusterClient.listJobs().get();
+ assertThat(jobs).hasSize(1);
+
assertThat(jobs.iterator().next().getJobState()).isEqualTo(JobStatus.FINISHED);
+
File tempOutputFile = new File(tempDir, "records.out");
- String tempOutputFilepath = tempOutputFile.toString();
- GenericContainer<?> taskManager = flink.getTaskManagers().get(0);
- int numberOfResultRecords;
- while (true) {
- Thread.sleep(50); // prevent NotFoundException: Status 404
- try {
- taskManager.copyFileFromContainer(resultFilePath,
tempOutputFilepath);
- numberOfResultRecords =
UpsertTestFileUtil.getNumberOfRecords(tempOutputFile);
- if (numberOfResultRecords == expectedNumberOfRecords) {
- break;
- }
- } catch (Exception ignored) {
- }
- }
- assertThat(numberOfResultRecords).isEqualTo(expectedNumberOfRecords);
+ flink.getTaskManagers()
+ .get(0)
+ .copyFileFromContainer(resultFilePath,
tempOutputFile.toString());
+ assertThat(UpsertTestFileUtil.getNumberOfRecords(tempOutputFile))
+ .isEqualTo(expectedNumberOfRecords);
}
private void executeSql(List<String> sqlLines) throws Exception {
flink.submitSQLJob(
new SQLJobSubmission.SQLJobSubmissionBuilder(sqlLines)
- .addJars(sqlConnectorUpsertTestJar,
sqlConnectorKafkaJar, sqlToolBoxJar)
+ .addJars(sqlConnectorUpsertTestJar)
.build());
}
}
