haifxu opened a new issue, #9203:
URL: https://github.com/apache/inlong/issues/9203
### What happened
Failed to use Iceberg sink.
```
[ ] 2023-11-02 09:01:40.763 -ERROR [inlong-plugin-3]
o.a.i.m.p.f.FlinkService :146 - submit job from info
FlinkInfo(endpoint=null, jobName=InLong-Sort-test_group_2,
inlongStreamInfoList=[InlongStreamInfo(id=4, inlongGroupId=test_group_2,
inlongStreamId=test_stream_2, name=null, description=,
mqResource=test_stream_2, 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
18:31:22 UTC 2023, modifyTime=Thu Nov 02 01:01:40 UTC 2023,
fieldList=[StreamField(id=66, inlongGroupId=test_group_2,
inlongStreamId=test_stream_2, fieldName=id, fieldType=int, fieldComment=null,
isPredefinedField=null, fieldValue=null, preExpression=null, isMetaField=0,
metaFieldName=null, fieldFormat=null, originNodeName=null, originFieldName=nul
l, extParams=null), StreamField(id=67, inlongGroupId=test_group_2,
inlongStreamId=test_stream_2, 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=3,
inlongGroupId=test_group_2, inlongStreamId=test_stream_2,
sourceType=MYSQL_BINLOG, sourceName=test_source_2, 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
18:31:48 UTC 2023, modifyTime=Tue Oct 31 18:36:05 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, ta
bleWhiteList=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=[IcebergSink(super=StreamSink(super=StreamNode(preNodes=null,
postNodes=null, fieldList=null), id=3, inlongGroupId=test_group_2,
inlongStreamId=test_stream_2, sinkType=ICEBERG, sinkName=test_sink_2,
description=null, inlongClusterName=null,
dataNodeName=501ba772-0c2e-4c98-a13c-2ffce71cb38f, sortTaskName=null,
sortConsumerGroup=null, enableCreateResource=1, operateLog=success to create
iceberg resource, status=130, previousStatus=130, creator=admin,
modifier=admin, createTime=Tue Oct 31 18:35:35 UTC 2023, modifyTime=Thu Nov 02
01:01:40 UTC 2023, sinkFieldList=[IcebergColumnInfo(length=null,
precision=null, scale=null, partitionStrategy=null, bucketNum=null, width=null,
required=false), IcebergColumnInfo(length
=null, precision=null, scale=null, partitionStrategy=null, bucketNum=null,
width=null, required=false)], properties={}, dataEncoding=UTF-8,
dataFormat=NONE, authentication=null, version=2), catalogType=HIVE,
catalogUri=thrift://9.134.86.112:9083,
warehouse=hdfs://qy-teg-2-v3/user/tdw/warehouse, dbName=test_db,
tableName=sink_table2, dataPath=null, fileFormat=Parquet, partitionType=null,
primaryKey=null)], version=7, wrapType=INLONG_MSG_V0, useExtendedFields=false,
ignoreParseError=true)],
localJarPath=/opt/inlong-sort/sort-dist-1.10.0-SNAPSHOT.jar,
connectorJarPaths=[/opt/inlong-sort/connectors/sort-connector-iceberg-1.10.0-SNAPSHOT.jar,
/opt/inlong-sort/connectors/sort-connector-mysql-cdc-1.10.0-SNAPSHOT.jar],
localConfPath=/opt/inlong-manager/lib/InLong-Sort-test_group_2,
sourceType=null, sinkType=null, jobId=null, savepointPath=null,
isException=false, exceptionMsg=null) failed:
org.apache.flink.client.program.ProgramInvocationException: The main method
caused an error: Unable to create a sink for writing table
'default_catalog.default_database.sink_table2'.
Table options are:
'catalog-database'='test_db'
'catalog-name'='HIVE'
'catalog-table'='sink_table2'
'catalog-type'='HIVE'
'connector'='iceberg-inlong'
'default-database'='test_db'
'inlong.metric.labels'='groupId=test_group_2&streamId=test_stream_2&nodeId=test_sink_2'
'metrics.audit.key'='18'
'metrics.audit.proxy.hosts'='audit:10081'
'uri'='thrift://***:9083'
'warehouse'='hdfs://*/warehouse'
at
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
~[flink-clients_2.11-1.13.5.jar:1.13.5]
at
org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
~[flink-clients_2.11-1.13.5.jar:1.13.5]
at
org.apache.flink.client.program.PackagedProgramUtils.getPipelineFromProgram(PackagedProgramUtils.java:158)
~[flink-clients_2.11-1.13.5.jar:1.13.5]
at
org.apache.flink.client.program.PackagedProgramUtils.createJobGraph(PackagedProgramUtils.java:82)
~[flink-clients_2.11-1.13.5.jar:1.13.5]
at
org.apache.flink.client.program.PackagedProgramUtils.createJobGraph(PackagedProgramUtils.java:117)
~[flink-clients_2.11-1.13.5.jar:1.13.5]
at
org.apache.inlong.manager.plugin.flink.FlinkService.submitJobBySavepoint(FlinkService.java:192)
~[manager-plugins-base-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
at
org.apache.inlong.manager.plugin.flink.FlinkService.submit(FlinkService.java:144)
~[manager-plugins-base-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
at
org.apache.inlong.manager.plugin.flink.IntegrationTaskRunner.run(IntegrationTaskRunner.java:58)
~[manager-plugins-base-1.10.0-SNAPSHOT.jar:1.10.0-SNAPSHOT]
at
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
~[?:1.8.0_342]
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
~[?:1.8.0_342]
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]
Caused by: org.apache.flink.table.api.ValidationException: Unable to create
a sink for writing table 'default_catalog.default_database.sink_table2'.
Table options are:
'catalog-database'='test_db'
'catalog-name'='HIVE'
'catalog-table'='sink_table2'
'catalog-type'='HIVE'
'connector'='iceberg-inlong'
'default-database'='test_db'
'inlong.metric.labels'='groupId=test_group_2&streamId=test_stream_2&nodeId=test_sink_2'
'metrics.audit.key'='18'
'metrics.audit.proxy.hosts'='audit:10081'
'uri'='thrift://***:9083'
'warehouse'='hdfs://*/warehouse'
at
org.apache.flink.table.factories.FactoryUtil.createTableSink(FactoryUtil.java:171)
~[flink-table-common-1.13.5.jar:1.13.5]
at
org.apache.flink.table.planner.delegation.PlannerBase.getTableSink(PlannerBase.scala:367)
~[flink-table-planner-blink_2.11-1.13.5.jar:1.13.5]
at
org.apache.flink.table.planner.delegation.PlannerBase.translateToRel(PlannerBase.scala:201)
~[flink-table-planner-blink_2.11-1.13.5.jar:1.13.5]
at
org.apache.flink.table.planner.delegation.PlannerBase$$anonfun$1.apply(PlannerBase.scala:162)
~[flink-table-planner-blink_2.11-1.13.5.jar:1.13.5]
at
org.apache.flink.table.planner.delegation.PlannerBase$$anonfun$1.apply(PlannerBase.scala:162)
~[flink-table-planner-blink_2.11-1.13.5.jar:1.13.5]
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
~[scala-library-2.11.12.jar:?]
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
~[scala-library-2.11.12.jar:?]
at scala.collection.Iterator$class.foreach(Iterator.scala:891)
~[scala-library-2.11.12.jar:?]
at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
~[scala-library-2.11.12.jar:?]
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
~[scala-library-2.11.12.jar:?]
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
~[scala-library-2.11.12.jar:?]
at
scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
~[scala-library-2.11.12.jar:?]
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
~[scala-library-2.11.12.jar:?]
at
org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:162)
~[flink-table-planner-blink_2.11-1.13.5.jar:1.13.5]
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1518)
~[flink-table-api-java-1.13.5.jar:1.13.5]
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:740)
~[flink-table-api-java-1.13.5.jar:1.13.5]
at
org.apache.flink.table.api.internal.StatementSetImpl.execute(StatementSetImpl.java:99)
~[flink-table-api-java-1.13.5.jar:1.13.5]
at
org.apache.inlong.sort.parser.result.FlinkSqlParseResult.executeLoadSqls(FlinkSqlParseResult.java:84)
~[?:?]
at
org.apache.inlong.sort.parser.result.FlinkSqlParseResult.execute(FlinkSqlParseResult.java:63)
~[?:?]
at org.apache.inlong.sort.Entrance.main(Entrance.java:76) ~[?:?]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
~[?:1.8.0_342]
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
~[?:1.8.0_342]
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
~[?:1.8.0_342]
at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_342]
at
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
~[flink-clients_2.11-1.13.5.jar:1.13.5]
... 12 more
Caused by: java.lang.VerifyError: Stack map does not match the one at
exception handler 20
Exception Details:
Location:
org/apache/iceberg/hive/HiveCatalog.alterHiveDataBase(Lorg/apache/iceberg/catalog/Namespace;Lorg/apache/hadoop/hive/metastore/api/Database;)V
@20: astore_3
Reason:
Type 'org/apache/hadoop/hive/metastore/api/NoSuchObjectException'
(current frame, stack[0]) is not assignable to 'org/apache/thrift/TException'
(stack map, stack[0])
Current Frame:
bci: @0
flags: { }
locals: { 'org/apache/iceberg/hive/HiveCatalog',
'org/apache/iceberg/catalog/Namespace',
'org/apache/hadoop/hive/metastore/api/Database' }
stack: { 'org/apache/hadoop/hive/metastore/api/NoSuchObjectException' }
Stackmap Frame:
bci: @20
flags: { }
locals: { 'org/apache/iceberg/hive/HiveCatalog',
'org/apache/iceberg/catalog/Namespace',
'org/apache/hadoop/hive/metastore/api/Database' }
stack: { 'org/apache/thrift/TException' }
Bytecode:
0x0000000: 2ab4 009d 2b2c ba02 8100 00b9 00d1 0200
0x0000010: 57a7 0066 4ebb 0119 592d 1301 1b04 bd01
0x0000020: 1d59 032b 53b7 0282 bf4e bb01 2259 bb01
0x0000030: 2459 b701 2513 0284 b601 2b2b b601 2e13
0x0000040: 0202 b601 2bb6 0132 2db7 0135 bf4e b801
0x0000050: 3bb6 013e bb01 2259 bb01 2459 b701 2513
0x0000060: 0286 b601 2b2b b601 2e13 0202 b601 2bb6
0x0000070: 0132 2db7 0135 bfb1
Exception Handler Table:
bci [0, 17] => handler: 20
bci [0, 17] => handler: 20
bci [0, 17] => handler: 41
bci [0, 17] => handler: 77
Stackmap Table:
same_locals_1_stack_item_frame(@20,Object[#167])
same_locals_1_stack_item_frame(@41,Object[#167])
same_locals_1_stack_item_frame(@77,Object[#169])
same_frame(@119)
at java.lang.Class.forName0(Native Method) ~[?:1.8.0_342]
at java.lang.Class.forName(Class.java:348) ~[?:1.8.0_342]
at
org.apache.iceberg.common.DynConstructors$Builder.impl(DynConstructors.java:149)
~[iceberg-common-1.3.1.jar:?]
at org.apache.iceberg.CatalogUtil.loadCatalog(CatalogUtil.java:219)
~[iceberg-core-1.3.1.jar:?]
at
org.apache.iceberg.flink.CatalogLoader$HiveCatalogLoader.loadCatalog(CatalogLoader.java:113)
~[?:?]
at
org.apache.inlong.sort.iceberg.FlinkCatalog.<init>(FlinkCatalog.java:113) ~[?:?]
at
org.apache.inlong.sort.iceberg.FlinkCatalogFactory.createCatalog(FlinkCatalogFactory.java:164)
~[?:?]
at
org.apache.inlong.sort.iceberg.FlinkDynamicTableFactory.createTableLoader(FlinkDynamicTableFactory.java:211)
~[?:?]
at
org.apache.inlong.sort.iceberg.FlinkDynamicTableFactory.createDynamicTableSink(FlinkDynamicTableFactory.java:296)
~[?:?]
at
org.apache.flink.table.factories.FactoryUtil.createTableSink(FactoryUtil.java:168)
~[flink-table-common-1.13.5.jar:1.13.5]
at
org.apache.flink.table.planner.delegation.PlannerBase.getTableSink(PlannerBase.scala:367)
~[flink-table-planner-blink_2.11-1.13.5.jar:1.13.5]
at
org.apache.flink.table.planner.delegation.PlannerBase.translateToRel(PlannerBase.scala:201)
~[flink-table-planner-blink_2.11-1.13.5.jar:1.13.5]
at
org.apache.flink.table.planner.delegation.PlannerBase$$anonfun$1.apply(PlannerBase.scala:162)
~[flink-table-planner-blink_2.11-1.13.5.jar:1.13.5]
at
org.apache.flink.table.planner.delegation.PlannerBase$$anonfun$1.apply(PlannerBase.scala:162)
~[flink-table-planner-blink_2.11-1.13.5.jar:1.13.5]
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
~[scala-library-2.11.12.jar:?]
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
~[scala-library-2.11.12.jar:?]
at scala.collection.Iterator$class.foreach(Iterator.scala:891)
~[scala-library-2.11.12.jar:?]
at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
~[scala-library-2.11.12.jar:?]
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
~[scala-library-2.11.12.jar:?]
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
~[scala-library-2.11.12.jar:?]
at
scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
~[scala-library-2.11.12.jar:?]
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
~[scala-library-2.11.12.jar:?]
at
org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:162)
~[flink-table-planner-blink_2.11-1.13.5.jar:1.13.5]
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1518)
~[flink-table-api-java-1.13.5.jar:1.13.5]
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:740)
~[flink-table-api-java-1.13.5.jar:1.13.5]
at
org.apache.flink.table.api.internal.StatementSetImpl.execute(StatementSetImpl.java:99)
~[flink-table-api-java-1.13.5.jar:1.13.5]
at
org.apache.inlong.sort.parser.result.FlinkSqlParseResult.executeLoadSqls(FlinkSqlParseResult.java:84)
~[?:?]
at
org.apache.inlong.sort.parser.result.FlinkSqlParseResult.execute(FlinkSqlParseResult.java:63)
~[?:?]
at org.apache.inlong.sort.Entrance.main(Entrance.java:76) ~[?:?]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
~[?:1.8.0_342]
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
~[?:1.8.0_342]
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
~[?:1.8.0_342]
at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_342]
at
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
~[flink-clients_2.11-1.13.5.jar:1.13.5]
... 12 more
```
### What you expected to happen
Successfully run the task using iceberg sink.
### How to reproduce
Use Iceberg 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
- [X] I agree to follow this project's [Code of
Conduct](https://www.apache.org/foundation/policies/conduct)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
