Hisoka-X commented on code in PR #9961:
URL: https://github.com/apache/seatunnel/pull/9961#discussion_r2443316891
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-milvus-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/milvus/MilvusIT.java:
##########
@@ -715,4 +718,114 @@ private void verifyIndexesExist(String database, String
collection) {
log.info("Index verification passed for collection: {}.{}", database,
collection);
}
+
+ @TestTemplate
+ public void testStreamingFakeToMilvus(TestContainer container)
+ throws IOException, InterruptedException {
+ // flush by checkpoint interval
+ String jobId = "1";
+ String database = "streaming_test";
+ String collection = "streaming_simple_example";
+ String vectorField = "book_intro";
+ int checkpointInterval = 30000;
+ new Thread(
+ () -> {
+ try {
+ container.executeJob(
+ "/streaming-fake-to-milvus.conf",
+ jobId,
+ "database=" + database,
+ "collection=" + collection,
+ "batch_size=3");
+ } catch (IOException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ })
+ .start();
+
+ // count write records
+ long count;
+ waitCollectionReady(database, collection, vectorField);
+ do {
+ count = countCollectionEntities(database, collection);
+ } while (count < 9);
Review Comment:
Please use `Awaitility.await()
.atMost(60, TimeUnit.SECONDS)
.pollInterval(2, TimeUnit.SECONDS)` to verify data in
loop.
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-milvus-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/milvus/MilvusIT.java:
##########
@@ -715,4 +718,114 @@ private void verifyIndexesExist(String database, String
collection) {
log.info("Index verification passed for collection: {}.{}", database,
collection);
}
+
+ @TestTemplate
+ public void testStreamingFakeToMilvus(TestContainer container)
+ throws IOException, InterruptedException {
+ // flush by checkpoint interval
+ String jobId = "1";
+ String database = "streaming_test";
+ String collection = "streaming_simple_example";
+ String vectorField = "book_intro";
+ int checkpointInterval = 30000;
+ new Thread(
Review Comment:
How about use `CompletedFuture.runAsync`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
