Skip to content

Commit 81de0a6

Browse files
hawk9821hawk9821
authored
[Improve][CI]skip ui module, improve module dependent (apache#8225)
1 parent e15757b commit 81de0a6

File tree

21 files changed

+326
-237
lines changed

21 files changed

+326
-237
lines changed

‎.github/workflows/backend.yml

+39-42
Large diffs are not rendered by default.

‎seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/catalog/HudiCatalog.java

+3-1
Original file line numberDiff line numberDiff line change
@@ -100,7 +100,9 @@ public void open() throws CatalogException {
100100
@Override
101101
public void close() throws CatalogException {
102102
try {
103-
fs.close();
103+
if (fs != null) {
104+
fs.close();
105+
}
104106
} catch (Exception e) {
105107
log.info("Hudi catalog close error.", e);
106108
}

‎seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCWithSchemaChangeIT.java

+11-2
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
import org.apache.seatunnel.e2e.common.container.TestContainer;
3838
import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
3939
import org.apache.seatunnel.e2e.common.junit.TestContainerExtension;
40+
import org.apache.seatunnel.e2e.common.util.JobIdGenerator;
4041

4142
import org.junit.jupiter.api.AfterAll;
4243
import org.junit.jupiter.api.Assertions;
@@ -65,6 +66,7 @@
6566
import static org.awaitility.Awaitility.await;
6667
import static org.awaitility.Awaitility.with;
6768
import static org.awaitility.Durations.TWO_SECONDS;
69+
import static org.testcontainers.shaded.org.awaitility.Awaitility.given;
6870

6971
@Slf4j
7072
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
@@ -192,17 +194,24 @@ public void testOracleCdc2MysqlWithSchemaEvolutionCase(TestContainer container)
192194
dropTable(ORACLE_CONTAINER.getJdbcUrl(), SCEHMA_NAME, SOURCE_TABLE1);
193195
dropTable(ORACLE_CONTAINER.getJdbcUrl(), SCEHMA_NAME, SOURCE_TABLE1 + "_SINK");
194196
createAndInitialize("full_types", ADMIN_USER, ADMIN_PWD);
197+
String jobId = String.valueOf(JobIdGenerator.newJobId());
195198
CompletableFuture.runAsync(
196199
() -> {
197200
try {
198-
container.executeJob("/oraclecdc_to_mysql_with_schema_change.conf");
201+
container.executeJob("/oraclecdc_to_mysql_with_schema_change.conf", jobId);
199202
} catch (Exception e) {
200203
log.error("Commit task exception :" + e.getMessage());
201204
throw new RuntimeException(e);
202205
}
203206
});
204207

205-
Thread.sleep(10000L);
208+
given().pollDelay(10, TimeUnit.SECONDS)
209+
.await()
210+
.pollDelay(5000L, TimeUnit.MILLISECONDS)
211+
.untilAsserted(
212+
() -> {
213+
Assertions.assertEquals("RUNNING", container.getJobStatus(jobId));
214+
});
206215

207216
assertSchemaEvolution(
208217
ORACLE_CONTAINER.getJdbcUrl(),

‎seatunnel-e2e/seatunnel-connector-v2-e2e/connector-hudi-e2e/pom.xml

+8-3
Original file line numberDiff line numberDiff line change
@@ -59,11 +59,16 @@
5959

6060
<dependency>
6161
<groupId>org.apache.seatunnel</groupId>
62-
<artifactId>connector-seatunnel-e2e-base</artifactId>
62+
<artifactId>seatunnel-hadoop3-3.1.4-uber</artifactId>
6363
<version>${project.version}</version>
64-
<classifier>tests</classifier>
65-
<type>test-jar</type>
64+
<classifier>optional</classifier>
6665
<scope>test</scope>
66+
<exclusions>
67+
<exclusion>
68+
<groupId>org.apache.avro</groupId>
69+
<artifactId>avro</artifactId>
70+
</exclusion>
71+
</exclusions>
6772
</dependency>
6873

6974
<dependency>

‎seatunnel-e2e/seatunnel-connector-v2-e2e/connector-hudi-e2e/src/test/java/org/apache/seatunnel/e2e/connector/hudi/HudiSeatunnelS3MultiTableIT.java

+69-79
Original file line numberDiff line numberDiff line change
@@ -18,8 +18,8 @@
1818
package org.apache.seatunnel.e2e.connector.hudi;
1919

2020
import org.apache.seatunnel.common.utils.FileUtils;
21+
import org.apache.seatunnel.e2e.common.container.seatunnel.SeaTunnelContainer;
2122
import org.apache.seatunnel.e2e.common.util.ContainerUtil;
22-
import org.apache.seatunnel.engine.e2e.SeaTunnelContainer;
2323

2424
import org.apache.hadoop.conf.Configuration;
2525
import org.apache.hadoop.fs.LocalFileSystem;
@@ -32,16 +32,13 @@
3232
import org.junit.jupiter.api.Assertions;
3333
import org.junit.jupiter.api.BeforeAll;
3434
import org.junit.jupiter.api.Test;
35+
import org.junit.jupiter.api.TestInstance;
3536
import org.testcontainers.containers.Container;
3637
import org.testcontainers.containers.MinIOContainer;
3738

3839
import io.minio.BucketExistsArgs;
39-
import io.minio.DownloadObjectArgs;
40-
import io.minio.ListObjectsArgs;
4140
import io.minio.MakeBucketArgs;
4241
import io.minio.MinioClient;
43-
import io.minio.Result;
44-
import io.minio.messages.Item;
4542
import lombok.extern.slf4j.Slf4j;
4643

4744
import java.io.IOException;
@@ -51,6 +48,7 @@
5148
import static org.awaitility.Awaitility.given;
5249

5350
@Slf4j
51+
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
5452
public class HudiSeatunnelS3MultiTableIT extends SeaTunnelContainer {
5553

5654
private static final String MINIO_DOCKER_IMAGE = "minio/minio:RELEASE.2024-06-13T22-53-53Z";
@@ -88,7 +86,6 @@ public void startUp() throws Exception {
8886

8987
String s3URL = container.getS3URL();
9088

91-
// configuringClient
9289
minioClient =
9390
MinioClient.builder()
9491
.endpoint(s3URL)
@@ -111,11 +108,13 @@ protected String[] buildStartCommand() {
111108
"wget -P "
112109
+ SEATUNNEL_HOME
113110
+ "lib "
111+
+ " --timeout=180 "
114112
+ AWS_SDK_DOWNLOAD
115113
+ " &&"
116114
+ "wget -P "
117115
+ SEATUNNEL_HOME
118116
+ "lib "
117+
+ " --timeout=180 "
119118
+ HADOOP_AWS_DOWNLOAD
120119
+ " &&"
121120
+ ContainerUtil.adaptPathForWin(
@@ -132,90 +131,81 @@ public void tearDown() throws Exception {
132131
}
133132
}
134133

134+
@Override
135+
protected boolean isIssueWeAlreadyKnow(String threadName) {
136+
return super.isIssueWeAlreadyKnow(threadName)
137+
// hudi with s3
138+
|| threadName.startsWith("s3a-transfer");
139+
}
140+
135141
@Test
136142
public void testS3MultiWrite() throws IOException, InterruptedException {
137143
copyFileToContainer("/hudi/core-site.xml", "/tmp/seatunnel/config/core-site.xml");
138-
Container.ExecResult textWriteResult = executeSeaTunnelJob("/hudi/s3_fake_to_hudi.conf");
144+
Container.ExecResult textWriteResult = executeJob("/hudi/s3_fake_to_hudi.conf");
139145
Assertions.assertEquals(0, textWriteResult.getExitCode());
140146
Configuration configuration = new Configuration();
141147
configuration.set("fs.defaultFS", LocalFileSystem.DEFAULT_FS);
142-
given().ignoreExceptions()
148+
given().pollDelay(10, TimeUnit.SECONDS)
149+
.pollInterval(1, TimeUnit.SECONDS)
143150
.await()
144-
.atMost(60000, TimeUnit.MILLISECONDS)
151+
.atMost(300, TimeUnit.SECONDS)
145152
.untilAsserted(
146153
() -> {
147154
// copy hudi to local
148-
Path inputPath1 =
149-
downloadNewestCommitFile(DATABASE_1 + "/" + TABLE_NAME_1 + "/");
150-
Path inputPath2 =
151-
downloadNewestCommitFile(DATABASE_2 + "/" + TABLE_NAME_2 + "/");
152-
ParquetReader<Group> reader1 =
153-
ParquetReader.builder(new GroupReadSupport(), inputPath1)
154-
.withConf(configuration)
155-
.build();
156-
ParquetReader<Group> reader2 =
157-
ParquetReader.builder(new GroupReadSupport(), inputPath2)
158-
.withConf(configuration)
159-
.build();
160-
161-
long rowCount1 = 0;
162-
long rowCount2 = 0;
163-
// Read data and count rows
164-
while (reader1.read() != null) {
165-
rowCount1++;
166-
}
167-
// Read data and count rows
168-
while (reader2.read() != null) {
169-
rowCount2++;
155+
Path inputPath1 = null;
156+
Path inputPath2 = null;
157+
try {
158+
inputPath1 =
159+
new Path(
160+
MinIoUtils.downloadNewestCommitFile(
161+
minioClient,
162+
BUCKET,
163+
String.format(
164+
"%s/%s/", DATABASE_1, TABLE_NAME_1),
165+
DOWNLOAD_PATH));
166+
log.info(
167+
"download from s3 success, the parquet file is at: {}",
168+
inputPath1);
169+
inputPath2 =
170+
new Path(
171+
MinIoUtils.downloadNewestCommitFile(
172+
minioClient,
173+
BUCKET,
174+
String.format(
175+
"%s/%s/", DATABASE_2, TABLE_NAME_2),
176+
DOWNLOAD_PATH));
177+
log.info(
178+
"download from s3 success, the parquet file is at: {}",
179+
inputPath2);
180+
ParquetReader<Group> reader1 =
181+
ParquetReader.builder(new GroupReadSupport(), inputPath1)
182+
.withConf(configuration)
183+
.build();
184+
ParquetReader<Group> reader2 =
185+
ParquetReader.builder(new GroupReadSupport(), inputPath2)
186+
.withConf(configuration)
187+
.build();
188+
189+
long rowCount1 = 0;
190+
long rowCount2 = 0;
191+
// Read data and count rows
192+
while (reader1.read() != null) {
193+
rowCount1++;
194+
}
195+
// Read data and count rows
196+
while (reader2.read() != null) {
197+
rowCount2++;
198+
}
199+
Assertions.assertEquals(100, rowCount1);
200+
Assertions.assertEquals(240, rowCount2);
201+
} finally {
202+
if (inputPath1 != null) {
203+
FileUtils.deleteFile(inputPath1.toUri().getPath());
204+
}
205+
if (inputPath2 != null) {
206+
FileUtils.deleteFile(inputPath2.toUri().getPath());
207+
}
170208
}
171-
FileUtils.deleteFile(inputPath1.toUri().getPath());
172-
FileUtils.deleteFile(inputPath2.toUri().getPath());
173-
Assertions.assertEquals(100, rowCount1);
174-
Assertions.assertEquals(240, rowCount2);
175209
});
176210
}
177-
178-
public Path downloadNewestCommitFile(String pathPrefix) throws IOException {
179-
Iterable<Result<Item>> listObjects =
180-
minioClient.listObjects(
181-
ListObjectsArgs.builder().bucket(BUCKET).prefix(pathPrefix).build());
182-
String newestCommitFileabsolutePath = "";
183-
String newestCommitFileName = "";
184-
long newestCommitTime = 0L;
185-
for (Result<Item> listObject : listObjects) {
186-
Item item;
187-
try {
188-
item = listObject.get();
189-
} catch (Exception e) {
190-
throw new IOException("List minio file error.", e);
191-
}
192-
if (item.isDir() || !item.objectName().endsWith(".parquet")) {
193-
continue;
194-
}
195-
long fileCommitTime =
196-
Long.parseLong(
197-
item.objectName()
198-
.substring(
199-
item.objectName().lastIndexOf("_") + 1,
200-
item.objectName().lastIndexOf(".parquet")));
201-
if (fileCommitTime > newestCommitTime) {
202-
newestCommitFileabsolutePath = item.objectName();
203-
newestCommitFileName =
204-
newestCommitFileabsolutePath.substring(
205-
item.objectName().lastIndexOf("/") + 1);
206-
newestCommitTime = fileCommitTime;
207-
}
208-
}
209-
try {
210-
minioClient.downloadObject(
211-
DownloadObjectArgs.builder()
212-
.bucket(BUCKET)
213-
.object(newestCommitFileabsolutePath)
214-
.filename(DOWNLOAD_PATH + newestCommitFileName)
215-
.build());
216-
} catch (Exception e) {
217-
log.error("Download file from minio error.");
218-
}
219-
return new Path(DOWNLOAD_PATH + newestCommitFileName);
220-
}
221211
}

0 commit comments

Comments
 (0)