Greenplum [question 05] Greenplum streaming server custom client problem handling (increasing)

1. Briefly explain

Greenplum Stream Server (GPSS) It's a ETL( extract 、 transformation 、 load ) Tools .GPSS An instance of the server receives streaming data from one or more clients , Use Greenplum An external table readable by the database transforms and inserts the data into the target Greenplum In the table . Data sources and data formats are client specific . The data source and data format are specified by the client .

2. Details of the problem

2.1 Problems caused by multithreading

INTERNAL: Load task is already opened and Need to open a load task before write The detailed error reports are as follows ：

io.grpc.StatusRuntimeException: INTERNAL: Load task is already opened
	at io.grpc.stub.ClientCalls.toStatusRuntimeException(ClientCalls.java:233)
	at io.grpc.stub.ClientCalls.getUnchecked(ClientCalls.java:214)
	at io.grpc.stub.ClientCalls.blockingUnaryCall(ClientCalls.java:139)
	at api.GpssGrpc$GpssBlockingStub.open(GpssGrpc.java:640)
	at com.xinan.ods.gpss.GpssUtil.openRequestByType(GpssUtil.java:143)
	at com.xinan.ods.gpss.GpssUtil.dealData(GpssUtil.java:54)
	at com.xinan.ods.inGPDataBase.ThreadInGPDataBase.lambda$run$1(ThreadInGPDataBase.java:46)
	at java.util.HashMap.forEach(HashMap.java:1288)
	at com.xinan.ods.inGPDataBase.ThreadInGPDataBase.run(ThreadInGPDataBase.java:43)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:748)

io.grpc.StatusRuntimeException: INTERNAL: Need to open a load task before write
	at io.grpc.stub.ClientCalls.toStatusRuntimeException(ClientCalls.java:233)
	at io.grpc.stub.ClientCalls.getUnchecked(ClientCalls.java:214)
	at io.grpc.stub.ClientCalls.blockingUnaryCall(ClientCalls.java:139)
	at api.GpssGrpc$GpssBlockingStub.write(GpssGrpc.java:650)
	at com.xinan.ods.gpss.GpssUtil.dealData(GpssUtil.java:64)
	at com.xinan.ods.inGPDataBase.ThreadInGPDataBase.lambda$run$1(ThreadInGPDataBase.java:46)
	at java.util.HashMap.forEach(HashMap.java:1288)
	at com.xinan.ods.inGPDataBase.ThreadInGPDataBase.run(ThreadInGPDataBase.java:43)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:748)