Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Bug][Manager][Sort] Failed to use Iceberg sink #9203

Closed
1 of 2 tasks
haifxu opened this issue Nov 3, 2023 · 1 comment
Closed
1 of 2 tasks

[Bug][Manager][Sort] Failed to use Iceberg sink #9203

haifxu opened this issue Nov 3, 2023 · 1 comment
Assignees
@EMsnap
Labels
type/bug Something is wrong
Milestone
1.10.0

Comments

@haifxu
Copy link
Contributor

haifxu commented Nov 3, 2023

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=null, 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, 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=[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
@haifxu haifxu added the type/bug Something is wrong label Nov 3, 2023
@haifxu haifxu changed the title [Bug][Manager] Failed to use Iceberg sink [Bug][Manager][Sort] Failed to use Iceberg sink Nov 3, 2023
@haifxu haifxu added this to the 1.10.0 milestone Nov 9, 2023
@vernedeng
Copy link
Contributor

vernedeng commented Nov 14, 2023 •
edited
Loading

try add JVM argument -noverify.
The reason for this problem is the duplicated dependency of hive-exec in the manager lib and packed into some connectors like iceberg-connector

@haifxu haifxu closed this as completed Nov 29, 2023
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Assignees

@EMsnap EMsnap

Labels
type/bug Something is wrong
Projects
None yet
Milestone
1.10.0
Development

No branches or pull requests
3 participants
@EMsnap @vernedeng @haifxu