Riddle4045 opened a new issue, #9495:
URL: https://github.com/apache/hudi/issues/9495
Using Flink version : 1.16
Trying to write a hudi table with metadata sync enabled to HMS, sample
DStream API job
```
final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
// generate a source of data.
DataStream<RowData> fares = env.addSource(new
TaxiFareGenerator()).map(
event -> GenericRowData.of(
event.getRideId(),
event.getDriverId(),
event.getTaxiId(),
event.getStartTime(),
event.getTip(),
event.getTolls(),
event.getTotalFare()//,
// event.getPaymentType()
));
String targetTable = "TaxiFare";
String outputPath = String.join("/",basePath, "hudi_mor_2023082101");
Map<String, String> options = new HashMap<>();
options.put(FlinkOptions.PATH.key(), outputPath);
options.put(FlinkOptions.TABLE_TYPE.key(),
HoodieTableType.MERGE_ON_READ.name());
options.put(FlinkOptions.METADATA_ENABLED.key(), "true");
options.put(FlinkOptions.HIVE_SYNC_ENABLED.key(), "true");
options.put(FlinkOptions.HIVE_SYNC_MODE.key(),
HiveSyncMode.HMS.name());
options.put(FlinkOptions.HIVE_SYNC_METASTORE_URIS.key(),
"thrift://hive-metastore:9083 ");
HoodiePipeline.Builder builder = HoodiePipeline.builder(targetTable)
.column("rideId BIGINT")
.column("driverId BIGINT")
.column("taxiId BIGINT")
.column("startTime BIGINT")
.column("tip FLOAT")
.column("tolls FLOAT")
.column("totalFare FLOAT")
.pk("driverId")
.options(options);
builder.sink(fares, false);
env.execute("Hudi Table with HMS");
```
The job fails with the following exceptions
```
2023-08-21 14:30:42
org.apache.flink.util.FlinkException: Global failure triggered by
OperatorCoordinator for 'stream_write: TaxiFare' (operator
07008715da12a894ffb19d48ffca33da).
at
org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder$LazyInitializedCoordinatorContext.failJob(OperatorCoordinatorHolder.java:617)
at
org.apache.hudi.sink.StreamWriteOperatorCoordinator.lambda$start$0(StreamWriteOperatorCoordinator.java:190)
at
org.apache.hudi.sink.utils.NonThrownExecutor.handleException(NonThrownExecutor.java:142)
at
org.apache.hudi.sink.utils.NonThrownExecutor.lambda$wrapAction$0(NonThrownExecutor.java:133)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
Source)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown
Source)
at java.base/java.lang.Thread.run(Unknown Source)
Caused by: org.apache.hudi.exception.HoodieException: Executor executes
action [initialize instant 20230821213020770] error
... 6 more
Caused by: org.apache.hudi.exception.HoodieException: Failed to update
metadata
at
org.apache.hudi.client.HoodieFlinkTableServiceClient.writeTableMetadata(HoodieFlinkTableServiceClient.java:181)
at
org.apache.hudi.client.HoodieFlinkWriteClient.writeTableMetadata(HoodieFlinkWriteClient.java:279)
at
org.apache.hudi.client.BaseHoodieWriteClient.commit(BaseHoodieWriteClient.java:282)
at
org.apache.hudi.client.BaseHoodieWriteClient.commitStats(BaseHoodieWriteClient.java:233)
at
org.apache.hudi.client.HoodieFlinkWriteClient.commit(HoodieFlinkWriteClient.java:111)
at
org.apache.hudi.client.HoodieFlinkWriteClient.commit(HoodieFlinkWriteClient.java:74)
at
org.apache.hudi.client.BaseHoodieWriteClient.commit(BaseHoodieWriteClient.java:199)
at
org.apache.hudi.sink.StreamWriteOperatorCoordinator.doCommit(StreamWriteOperatorCoordinator.java:537)
at
org.apache.hudi.sink.StreamWriteOperatorCoordinator.commitInstant(StreamWriteOperatorCoordinator.java:513)
at
org.apache.hudi.sink.StreamWriteOperatorCoordinator.commitInstant(StreamWriteOperatorCoordinator.java:484)
at
org.apache.hudi.sink.StreamWriteOperatorCoordinator.lambda$initInstant$6(StreamWriteOperatorCoordinator.java:402)
at
org.apache.hudi.sink.utils.NonThrownExecutor.lambda$wrapAction$0(NonThrownExecutor.java:130)
... 3 more
Caused by: org.apache.hudi.exception.HoodieUpsertException: Error upsetting
bucketType UPDATE for partition :files
at
org.apache.hudi.table.action.commit.BaseFlinkCommitActionExecutor.handleUpsertPartition(BaseFlinkCommitActionExecutor.java:203)
at
org.apache.hudi.table.action.commit.BaseFlinkCommitActionExecutor.execute(BaseFlinkCommitActionExecutor.java:107)
at
org.apache.hudi.table.action.commit.delta.FlinkUpsertPreppedDeltaCommitActionExecutor.execute(FlinkUpsertPreppedDeltaCommitActionExecutor.java:52)
at
org.apache.hudi.table.HoodieFlinkMergeOnReadTable.upsertPrepped(HoodieFlinkMergeOnReadTable.java:81)
at
org.apache.hudi.client.HoodieFlinkWriteClient.lambda$upsertPreppedRecords$4(HoodieFlinkWriteClient.java:167)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(Unknown
Source)
at
java.base/java.util.HashMap$ValueSpliterator.forEachRemaining(Unknown Source)
at java.base/java.util.stream.AbstractPipeline.copyInto(Unknown Source)
at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(Unknown
Source)
at java.base/java.util.stream.ReduceOps$ReduceTask.doLeaf(Unknown
Source)
at java.base/java.util.stream.ReduceOps$ReduceTask.doLeaf(Unknown
Source)
at java.base/java.util.stream.AbstractTask.compute(Unknown Source)
at java.base/java.util.concurrent.CountedCompleter.exec(Unknown Source)
at java.base/java.util.concurrent.ForkJoinTask.doExec(Unknown Source)
at java.base/java.util.concurrent.ForkJoinTask.doInvoke(Unknown Source)
at java.base/java.util.concurrent.ForkJoinTask.invoke(Unknown Source)
at
java.base/java.util.stream.ReduceOps$ReduceOp.evaluateParallel(Unknown Source)
at java.base/java.util.stream.AbstractPipeline.evaluate(Unknown Source)
at java.base/java.util.stream.ReferencePipeline.collect(Unknown Source)
at
org.apache.hudi.client.HoodieFlinkWriteClient.upsertPreppedRecords(HoodieFlinkWriteClient.java:170)
at
org.apache.hudi.metadata.FlinkHoodieBackedTableMetadataWriter.commit(FlinkHoodieBackedTableMetadataWriter.java:166)
at
org.apache.hudi.metadata.HoodieBackedTableMetadataWriter.processAndCommit(HoodieBackedTableMetadataWriter.java:830)
at
org.apache.hudi.metadata.HoodieBackedTableMetadataWriter.update(HoodieBackedTableMetadataWriter.java:897)
at
org.apache.hudi.client.HoodieFlinkTableServiceClient.writeTableMetadata(HoodieFlinkTableServiceClient.java:179)
... 14 more
Caused by: java.lang.NoClassDefFoundError: Could not initialize class
org.apache.hudi.org.apache.hadoop.hbase.io.hfile.HFile
at
org.apache.hudi.common.table.log.block.HoodieHFileDataBlock.serializeRecords(HoodieHFileDataBlock.java:142)
at
org.apache.hudi.common.table.log.block.HoodieDataBlock.getContentBytes(HoodieDataBlock.java:111)
at
org.apache.hudi.common.table.log.HoodieLogFormatWriter.appendBlocks(HoodieLogFormatWriter.java:158)
at
org.apache.hudi.io.HoodieAppendHandle.appendDataAndDeleteBlocks(HoodieAppendHandle.java:464)
at
org.apache.hudi.io.HoodieAppendHandle.doAppend(HoodieAppendHandle.java:437)
at
org.apache.hudi.table.action.commit.delta.BaseFlinkDeltaCommitActionExecutor.handleUpdate(BaseFlinkDeltaCommitActionExecutor.java:54)
at
org.apache.hudi.table.action.commit.BaseFlinkCommitActionExecutor.handleUpsertPartition(BaseFlinkCommitActionExecutor.java:195)
... 37 more
```
Based on the recommendation here, looks like I have to rebuilt hudi flink
bundle and then use that, am I correct?
https://hudi.apache.org/docs/faq/#how-can-i-resolve-the-nosuchmethoderror-from-hbase-when-using-hudi-with-metadata-table-on-hdfs
--
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: commits-unsubscr...@hudi.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
