[Improve][Connector-fake] Optimizing Data Generation Strategies refer to #4004 (#4061)

---------

Co-authored-by: zhilinli <lzl15844876351@163.com>
Co-authored-by: zhilinli <76689593+zhilinli123@users.noreply.github.com>

Author: 3 people

seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeDataGenerator.java

@@ -17,6 +17,7 @@
 
 package org.apache.seatunnel.connectors.seatunnel.fake.source;
 
+import org.apache.seatunnel.api.source.Collector;
 import org.apache.seatunnel.api.table.type.ArrayType;
 import org.apache.seatunnel.api.table.type.BasicType;
 import org.apache.seatunnel.api.table.type.DecimalType;
@@ -77,19 +78,21 @@ private SeaTunnelRow randomRow() {
         return new SeaTunnelRow(randomRow.toArray());
     }
 
-    public List<SeaTunnelRow> generateFakedRows(int rowNum) {
-        List<SeaTunnelRow> seaTunnelRows = new ArrayList<>();
+    /**
+     * @param rowNum The number of pieces of data to be generated by the current task
+     * @param output Data collection and distribution
+     **/
+    public void collectFakedRows(int rowNum, Collector<SeaTunnelRow> output) {
+        // Use manual configuration data preferentially
         if (fakeConfig.getFakeRows() != null) {
             for (FakeConfig.RowData rowData : fakeConfig.getFakeRows()) {
-                seaTunnelRows.add(convertRow(rowData));
+                output.collect(convertRow(rowData));
+            }
+        } else {
+            for (int i = 0; i < rowNum; i++) {
+                output.collect(randomRow());
             }
-            return seaTunnelRows;
-        }
-
-        for (int i = 0; i < rowNum; i++) {
-            seaTunnelRows.add(randomRow());
         }
-        return seaTunnelRows;
     }
 
     @SuppressWarnings("magicnumber")

seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceReader.java

@@ -70,11 +70,8 @@ public void pollNext(Collector<SeaTunnelRow> output) throws InterruptedException
         synchronized (output.getCheckpointLock()) {
             FakeSourceSplit split = splits.poll();
             if (null != split) {
-                // Generate a random number of rows to emit.
-                List<SeaTunnelRow> seaTunnelRows = fakeDataGenerator.generateFakedRows(split.getRowNum());
-                for (SeaTunnelRow seaTunnelRow : seaTunnelRows) {
-                    output.collect(seaTunnelRow);
-                }
+                // Randomly generated data are sent directly to the downstream operator
+                fakeDataGenerator.collectFakedRows(split.getRowNum(), output);
                 log.info("{} rows of data have been generated in split({}). Generation time: {}", split.getRowNum(), split.splitId(), latestTimestamp);
             } else {
                 if (!noMoreSplit) {

seatunnel-connectors-v2/connector-fake/src/test/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeDataGeneratorTest.java

@@ -17,6 +17,7 @@
 
 package org.apache.seatunnel.connectors.seatunnel.fake.source;
 
+import org.apache.seatunnel.api.source.Collector;
 import org.apache.seatunnel.api.table.type.RowKind;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
@@ -35,6 +36,7 @@
 import java.net.URISyntaxException;
 import java.net.URL;
 import java.nio.file.Paths;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
@@ -49,7 +51,18 @@ public void testComplexSchemaParse(String conf) throws FileNotFoundException, UR
         SeaTunnelRowType seaTunnelRowType = seaTunnelSchema.getSeaTunnelRowType();
         FakeConfig fakeConfig = FakeConfig.buildWithConfig(testConfig);
         FakeDataGenerator fakeDataGenerator = new FakeDataGenerator(seaTunnelSchema, fakeConfig);
-        List<SeaTunnelRow> seaTunnelRows = fakeDataGenerator.generateFakedRows(fakeConfig.getRowNum());
+        List<SeaTunnelRow> seaTunnelRows = new ArrayList<>();
+        fakeDataGenerator.collectFakedRows(fakeConfig.getRowNum(), new Collector<SeaTunnelRow>() {
+            @Override
+            public void collect(SeaTunnelRow record) {
+                seaTunnelRows.add(record);
+            }
+
+            @Override
+            public Object getCheckpointLock() {
+                throw new UnsupportedOperationException();
+            }
+        });
         Assertions.assertNotNull(seaTunnelRows);
         Assertions.assertEquals(seaTunnelRows.size(), 10);
         for (SeaTunnelRow seaTunnelRow : seaTunnelRows) {
@@ -96,7 +109,18 @@ public void testRowDataParse(String conf) throws FileNotFoundException, URISynta
         SeaTunnelSchema seaTunnelSchema = SeaTunnelSchema.buildWithConfig(testConfig.getConfig(SeaTunnelSchema.SCHEMA.key()));
         FakeConfig fakeConfig = FakeConfig.buildWithConfig(testConfig);
         FakeDataGenerator fakeDataGenerator = new FakeDataGenerator(seaTunnelSchema, fakeConfig);
-        List<SeaTunnelRow> seaTunnelRows = fakeDataGenerator.generateFakedRows(fakeConfig.getRowNum());
+        List<SeaTunnelRow> seaTunnelRows = new ArrayList<>();
+        fakeDataGenerator.collectFakedRows(fakeConfig.getRowNum(), new Collector<SeaTunnelRow>() {
+            @Override
+            public void collect(SeaTunnelRow record) {
+                seaTunnelRows.add(record);
+            }
+
+            @Override
+            public Object getCheckpointLock() {
+                throw new UnsupportedOperationException();
+            }
+        });
         Assertions.assertIterableEquals(expected, seaTunnelRows);
     }
 

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchIT.java

@@ -68,6 +68,7 @@ public class ElasticsearchIT extends TestSuiteBase implements TestResource {
     public void startUp() throws Exception {
         container = new ElasticsearchContainer(DockerImageName.parse("elasticsearch:8.0.0").asCompatibleSubstituteFor("docker.elastic.co/elasticsearch/elasticsearch"))
             .withNetwork(NETWORK)
+            .withEnv("cluster.routing.allocation.disk.threshold_enabled", "false")
             .withNetworkAliases("elasticsearch")
             .withPassword("elasticsearch")
             .withLogConsumer(new Slf4jLogConsumer(DockerLoggerFactory.getLogger("elasticsearch:8.0.0")));