codope commented on code in PR #8684:
URL: https://github.com/apache/hudi/pull/8684#discussion_r1222567953
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java:
##########
@@ -104,40 +106,19 @@ protected void initRegistry() {
}
@Override
- protected <T extends SpecificRecordBase> void initialize(HoodieEngineContext
engineContext,
- Option<T>
actionMetadata,
- Option<String>
inflightInstantTimestamp) {
- try {
- if (enabled) {
- initializeIfNeeded(dataMetaClient, actionMetadata,
inflightInstantTimestamp);
- }
- } catch (IOException e) {
- LOG.error("Failed to initialize metadata table. Disabling the writer.",
e);
- enabled = false;
- }
- }
-
- @Override
- protected void commit(String instantTime, Map<MetadataPartitionType,
HoodieData<HoodieRecord>> partitionRecordsMap,
- boolean canTriggerTableService) {
- ValidationUtils.checkState(enabled, "Metadata table cannot be committed to
as it is not enabled");
+ protected void commit(String instantTime, Map<MetadataPartitionType,
HoodieData<HoodieRecord>> partitionRecordsMap) {
ValidationUtils.checkState(metadataMetaClient != null, "Metadata table is
not fully initialized yet.");
HoodieData<HoodieRecord> preppedRecords = prepRecords(partitionRecordsMap);
List<HoodieRecord> preppedRecordList = preppedRecords.collectAsList();
- try (HoodieFlinkWriteClient writeClient = new
HoodieFlinkWriteClient(engineContext, metadataWriteConfig)) {
+ // Flink engine does not optimize initialCommit to MDT as bulk insert is
not yet supported
+
+ try (HoodieFlinkWriteClient writeClient = (HoodieFlinkWriteClient)
getWriteClient()) {
// rollback partially failed writes if any.
if (writeClient.rollbackFailedWrites()) {
metadataMetaClient = HoodieTableMetaClient.reload(metadataMetaClient);
}
- if (canTriggerTableService) {
- // trigger compaction before doing the delta commit. this is to
ensure, if this delta commit succeeds in metadata table, but failed in data
table,
- // we would have compacted metadata table and so could have included
uncommitted data which will never be ignored while reading from metadata
- // table (since reader will filter out only from delta commits)
- compactIfNecessary(writeClient, instantTime);
- }
Review Comment:
why is this removed? Don't we want compaction to run?
--
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]
