当前位置：网站首页>Greenplum【问题 05】Greenplum Streaming Server自定义客户端问题处理（不断增加ing）

Greenplum【问题 05】Greenplum Streaming Server自定义客户端问题处理（不断增加ing）

2022-06-12 05:29:00 【シ風箏】

1.简单说明

Greenplum Stream Server (GPSS)是一个ETL(提取、转换、加载)工具。GPSS服务器的一个实例从一个或多个客户机接收流数据，使用Greenplum数据库可读的外部表将数据转换并插入到目标Greenplum表中。数据源和数据格式是特定于客户机的。数据源和数据格式由客户端指定。

2.问题详情

2.1 多线程导致的问题

INTERNAL: Load task is already opened 和 Need to open a load task before write 问题详细报错如下：

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)