[Bug][Manager] Failed to use PostgreSQL sink

[haifxu]

What happened

[ ] 2023-11-02 09:10:06.475 -ERROR [inlong-workflow-0] a.i.m.s.l.s.SortConfigListener:128 - failed to build sort config for groupId=test_group_4, streamInfos=[InlongStreamInfo(id=6, inlongGroupId=test_group_4, inlongStreamId=test_stream_4, name=null, description=, mqResource=test_stream_4, dataType=null, dataEncoding=UTF-8, dataSeparator=124, dataEscapeChar=null, syncSend=0, dailyRecords=10, dailyStorage=10, peakRecords=1000, maxLength=10240, storagePeriod=1, extParams={"ignoreParseError":true,"useExtendedFields":false}, status=130, previousStatus=100, creator=admin, modifier=admin, createTime=Tue Oct 31 19:15:24 UTC 2023, modifyTime=Thu Nov 02 01:10:06 UTC 2023, fieldList=[StreamField(id=76, inlongGroupId=test_group_4, inlongStreamId=test_stream_4, fieldName=id, fieldType=int, fieldComment=null, isPredefinedField=null, fieldValue=null, preExpression=null, isMetaField=0, metaFieldName=null, fieldFormat=null, originNodeName=null, originFieldName=null, extParams=null), StreamField(id=77, inlongGroupId=test_group_4, inlongStreamId=test_stream_4, fieldName=name, fieldType=string, fieldComment=null, isPredefinedField=null, fieldValue=null, preExpression=null, isMetaField=0, metaFieldName=null, fieldFormat=null, originNodeName=null, originFieldName=null, extParams=null)], extList=null, sourceList=[MySQLBinlogSource(super=StreamSource(id=5, inlongGroupId=test_group_4, inlongStreamId=test_stream_4, sourceType=MYSQL_BINLOG, sourceName=test_source_4, agentIp=null, uuid=null, inlongClusterName=null, inlongClusterNodeTag=null, dataNodeName=null, serializationType=debezium_json, snapshot=null, version=1, status=101, previousStatus=110, creator=admin, modifier=admin, createTime=Tue Oct 31 19:15:50 UTC 2023, modifyTime=Wed Nov 01 23:10:43 UTC 2023, properties={}, templateId=null, subSourceList=null, ignoreParseError=false), user=root, password=****** hostname=9.134.53.69, port=3306, serverId=0, includeSchema=null, databaseWhiteList=test, tableWhiteList=test.source_table, serverTimezone=null, intervalMs=500, snapshotMode=initial, offsetFilename=null, historyFilename=null, monitoredDdl=null, timestampFormatStandard=SQL, allMigration=false, primaryKey=null, specificOffsetFile=null, specificOffsetPos=null)], sinkList=[PostgreSQLSink(super=StreamSink(super=StreamNode(preNodes=null, postNodes=null, fieldList=null), id=5, inlongGroupId=test_group_4, inlongStreamId=test_stream_4, sinkType=POSTGRESQL, sinkName=test_sink_4, description=null, inlongClusterName=null, dataNodeName=null, sortTaskName=null, sortConsumerGroup=null, enableCreateResource=1, operateLog=success to create PostgreSQL resource, status=130, previousStatus=130, creator=admin, modifier=admin, createTime=Tue Oct 31 19:17:22 UTC 2023, modifyTime=Thu Nov 02 01:10:06 UTC 2023, sinkFieldList=[SinkField(id=45, sinkType=null, inlongGroupId=null, inlongStreamId=null, fieldName=id, fieldType=INTEGER, fieldComment=id, isMetaField=0, metaFieldName=null, fieldFormat=null, originNodeName=null, originFieldName=null, sourceFieldName=id, sourceFieldType=int, extParams=null), SinkField(id=46, sinkType=null, inlongGroupId=null, inlongStreamId=null, fieldName=name, fieldType=VARCHAR(10), fieldComment=name, isMetaField=0, metaFieldName=null, fieldFormat=null, originNodeName=null, originFieldName=null, sourceFieldName=name, sourceFieldType=string, extParams=null)], properties={}, dataEncoding=UTF-8, dataFormat=NONE, authentication=null, version=7), jdbcUrl=jdbc:postgresql://9.135.74.226:5432/wedata_dev, username=postgres_admin, password=****** dbName=wedata_dev, tableName=sink_table2, primaryKey=id)], version=7, wrapType=INLONG_MSG_V0, useExtendedFields=false, ignoreParseError=true)] 
java.lang.IllegalArgumentException: Unsupported FieldType : INTEGER
	at org.apache.inlong.manager.common.enums.FieldType.forName(FieldType.java:66) ~[manager-common-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.pojo.sort.util.FieldInfoUtils.convertFieldFormat(FieldInfoUtils.java:246) ~[manager-pojo-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.pojo.sort.node.base.LoadNodeProvider.lambda$parseSinkFields$2(LoadNodeProvider.java:99) ~[manager-pojo-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193) ~[?:1.8.0_342]
	at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175) ~[?:1.8.0_342]
	at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384) ~[?:1.8.0_342]
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482) ~[?:1.8.0_342]
	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472) ~[?:1.8.0_342]
	at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708) ~[?:1.8.0_342]
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:1.8.0_342]
	at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:566) ~[?:1.8.0_342]
	at org.apache.inlong.manager.pojo.sort.node.base.LoadNodeProvider.parseSinkFields(LoadNodeProvider.java:117) ~[manager-pojo-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.pojo.sort.node.provider.PostgreSQLProvider.createLoadNode(PostgreSQLProvider.java:82) ~[manager-pojo-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.pojo.sort.node.NodeFactory.addBuiltInField(NodeFactory.java:117) ~[manager-pojo-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.service.resource.sort.DefaultSortConfigOperator.createNodes(DefaultSortConfigOperator.java:238) ~[manager-service-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.service.resource.sort.DefaultSortConfigOperator.getGroupInfo(DefaultSortConfigOperator.java:178) ~[manager-service-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.service.resource.sort.DefaultSortConfigOperator.buildConfig(DefaultSortConfigOperator.java:97) ~[manager-service-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.service.listener.sort.SortConfigListener.listen(SortConfigListener.java:125) ~[manager-service-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.event.LogableEventListener.executeListenerWithLog(LogableEventListener.java:79) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.event.LogableEventListener.listen(LogableEventListener.java:60) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.event.task.TaskEventNotifier.notify(TaskEventNotifier.java:57) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.processor.ServiceTaskProcessor.complete(ServiceTaskProcessor.java:115) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.core.impl.ProcessorExecutorImpl.executeComplete(ProcessorExecutorImpl.java:99) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.core.impl.ProcessorExecutorImpl.executeStart(ProcessorExecutorImpl.java:92) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.core.impl.ProcessorExecutorImpl.executeComplete(ProcessorExecutorImpl.java:104) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.core.impl.ProcessorExecutorImpl.executeStart(ProcessorExecutorImpl.java:92) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.core.impl.ProcessorExecutorImpl.executeComplete(ProcessorExecutorImpl.java:104) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.core.impl.ProcessorExecutorImpl.executeStart(ProcessorExecutorImpl.java:92) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.core.impl.ProcessServiceImpl.start(ProcessServiceImpl.java:75) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.service.workflow.WorkflowServiceImpl.start(WorkflowServiceImpl.java:90) ~[manager-service-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.service.workflow.WorkflowService.startAsync(WorkflowService.java:66) ~[manager-service-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.service.listener.group.apply.ApproveApplyProcessListener.lambda$listen$0(ApproveApplyProcessListener.java:79) ~[manager-service-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_342]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_342]
	at java.lang.Thread.run(Thread.java:750) ~[?:1.8.0_342]
[ ] 2023-11-02 09:10:06.520 -ERROR [inlong-workflow-0] a.i.m.w.e.LogableEventListener:88 - execute listener WorkflowEventLogEntity(id=null, processId=71, processName=CREATE_GROUP_RESOURCE, processDisplayName=Create Group, inlongGroupId=test_group_4, taskId=140, elementName=InitSort, elementDisplayName=Group-InitSort, eventType=TaskEvent, event=COMPLETE, listener=SortConfigListener, startTime=Thu Nov 02 09:10:06 UTC 2023, endTime=null, status=-1, async=0, ip=192.168.16.6, remark=null, exception=failed to build sort config for groupId=test_group_4, Unsupported FieldType : INTEGER) error:  
org.apache.inlong.manager.common.exceptions.WorkflowListenerException: failed to build sort config for groupId=test_group_4, Unsupported FieldType : INTEGER
	at org.apache.inlong.manager.service.listener.sort.SortConfigListener.listen(SortConfigListener.java:129) ~[manager-service-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.event.LogableEventListener.executeListenerWithLog(LogableEventListener.java:79) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.event.LogableEventListener.listen(LogableEventListener.java:60) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.event.task.TaskEventNotifier.notify(TaskEventNotifier.java:57) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.processor.ServiceTaskProcessor.complete(ServiceTaskProcessor.java:115) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.core.impl.ProcessorExecutorImpl.executeComplete(ProcessorExecutorImpl.java:99) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.core.impl.ProcessorExecutorImpl.executeStart(ProcessorExecutorImpl.java:92) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.core.impl.ProcessorExecutorImpl.executeComplete(ProcessorExecutorImpl.java:104) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.core.impl.ProcessorExecutorImpl.executeStart(ProcessorExecutorImpl.java:92) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.core.impl.ProcessorExecutorImpl.executeComplete(ProcessorExecutorImpl.java:104) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.core.impl.ProcessorExecutorImpl.executeStart(ProcessorExecutorImpl.java:92) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.workflow.core.impl.ProcessServiceImpl.start(ProcessServiceImpl.java:75) ~[manager-workflow-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.service.workflow.WorkflowServiceImpl.start(WorkflowServiceImpl.java:90) ~[manager-service-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.service.workflow.WorkflowService.startAsync(WorkflowService.java:66) ~[manager-service-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at org.apache.inlong.manager.service.listener.group.apply.ApproveApplyProcessListener.lambda$listen$0(ApproveApplyProcessListener.java:79) ~[manager-service-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_342]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_342]
	at java.lang.Thread.run(Thread.java:750) ~[?:1.8.0_342]

What you expected to happen

Use PostgreSQL sink

How to reproduce

Use PostgreSQL sink

Environment

No response

InLong version

master

InLong Component

InLong Manager

Are you willing to submit PR?

• Yes, I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[vernedeng]

There is not type of INTERGER in the definition of FieldType, try INT TINYINT BIGINT