[Feature][Transform][python] add data on thread

Author: FuYouJ

Diff for: seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/python/EndTagList.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.seatunnel.transform.python;
+
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class EndTagList {
+    private final List<Object> list;
+    private volatile boolean end;
+
+    public EndTagList(){
+        this.list = new ArrayList<>();
+        this.end = false;
+    }
+
+    public EndTagList(List<Object> dataList){
+        this.list = dataList;
+        this.end = false;
+    }
+
+    public List<Object> getList() {
+        return list;
+    }
+
+    public boolean isEnd() {
+        return end;
+    }
+
+    public void end(){
+        this.end = true;
+    }
+
+    public void add(Object obj) {
+        this.list.add(obj);
+    }
+}

Diff for: seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/python/PythonOperationProxy.java

@@ -17,17 +17,21 @@
 
 package org.apache.seatunnel.transform.python;
 
+import lombok.extern.slf4j.Slf4j;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowAccessor;
 import py4j.GatewayServer;
 
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
+@Slf4j
 public class PythonOperationProxy implements RowOperation {
 
     private GatewayServer javaServer;
-    private final Map<Long,SeaTunnelRowAccessor> threadDataMap = new ConcurrentHashMap<>();
-    private final Map<Long,Long> threadDataOffsetMap = new ConcurrentHashMap<>();
+    private final Map<Long, SeaTunnelRowAccessor> inputDataMap = new ConcurrentHashMap<>();
+
+    private final Map<Long, EndTagList> outputDataMap = new ConcurrentHashMap<>();
     private PythonOperationProxy(){
         if (INSTANCE != null) {
             throw new RuntimeException("Please use newInstance() method for getting the single instance of this class.");
@@ -51,14 +55,33 @@ public void shutdown(){
     }
 
     public void putThreadData(long threadId, SeaTunnelRowAccessor inputRow) {
-        this.threadDataMap.put(threadId,inputRow);
-        this.threadDataOffsetMap.compute(threadId,(thread,offset) ->{
-            if (offset == null){
-                offset = 0L;
-            }else {
-                offset = offset + 1;
-            }
-            return offset;
-        });
+        this.inputDataMap.put(threadId,inputRow);
+    }
+
+    public Object[] getOutputData(long threadId) {
+        EndTagList endTagList = outputDataMap.get(threadId);
+        while (endTagList == null || !endTagList.isEnd()){
+            log.info("wait python process data");
+        }
+        return outputDataMap.get(threadId).getList().toArray(new Object[0]);
+    }
+
+    public void addData(Long threadId,Object obj){
+        EndTagList array = this.outputDataMap.getOrDefault(threadId, new EndTagList());
+        array.add(obj);
+        this.outputDataMap.put(threadId,array);
+    }
+
+    public void addDataList(Long threadId,List<Object> dataList){
+        EndTagList array = new EndTagList();
+        this.outputDataMap.put(threadId,array);
+    }
+
+
+    public void end(Long threadId){
+        EndTagList endTagList = this.outputDataMap.get(threadId);
+        if (endTagList != null){
+            endTagList.end();
+        }
     }
 }

Diff for: seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/python/PythonTransform.java

@@ -34,7 +34,7 @@ public String getPluginName() {
     protected Object[] getOutputFieldValues(SeaTunnelRowAccessor inputRow) {
         long threadId = Thread.currentThread().getId();
         pythonOperationProxy.putThreadData(threadId,inputRow);
-        return new Object[0];
+        return pythonOperationProxy.getOutputData(threadId);
     }
 
     @Override