lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

Adding support for optional skipping single archiving failures

Browse Source
This commit is contained in:
Ron Barabash
2019-06-19 09:46:53 +03:00
committed by vinoth chandar
parent 66c7fa2322
commit 1b61eb45e0
2 changed files with 18 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
hoodie-client/src/main/java/com/uber/hoodie/config/HoodieWriteConfig.java
View File

@@ -70,7 +70,8 @@ public class HoodieWriteConfig extends DefaultHoodieConfig {
private static final String DEFAULT_CONSISTENCY_CHECK_ENABLED = "false"; private static final String DEFAULT_CONSISTENCY_CHECK_ENABLED = "false";
private static final String EMBEDDED_TIMELINE_SERVER_ENABLED = "hoodie.embed.timeline.server"; private static final String EMBEDDED_TIMELINE_SERVER_ENABLED = "hoodie.embed.timeline.server";
private static final String DEFAULT_EMBEDDED_TIMELINE_SERVER_ENABLED = "false"; private static final String DEFAULT_EMBEDDED_TIMELINE_SERVER_ENABLED = "false";
private static final String FAIL_ON_TIMELINE_ARCHIVING_ENABLED_PROP = "hoodie.fail.on.timeline.archiving";
private static final String DEFAULT_FAIL_ON_TIMELINE_ARCHIVING_ENABLED = "true";
// time between successive attempts to ensure written data's metadata is consistent on storage // time between successive attempts to ensure written data's metadata is consistent on storage
private static final String INITIAL_CONSISTENCY_CHECK_INTERVAL_MS_PROP = private static final String INITIAL_CONSISTENCY_CHECK_INTERVAL_MS_PROP =
"hoodie.consistency.check.initial_interval_ms"; "hoodie.consistency.check.initial_interval_ms";
@@ -169,6 +170,10 @@ public class HoodieWriteConfig extends DefaultHoodieConfig {
return Boolean.parseBoolean(props.getProperty(EMBEDDED_TIMELINE_SERVER_ENABLED)); return Boolean.parseBoolean(props.getProperty(EMBEDDED_TIMELINE_SERVER_ENABLED));
} }
public boolean isFailOnTimelineArchivingEnabled() {
return Boolean.parseBoolean(props.getProperty(FAIL_ON_TIMELINE_ARCHIVING_ENABLED_PROP));
}
public int getMaxConsistencyChecks() { public int getMaxConsistencyChecks() {
return Integer.parseInt(props.getProperty(MAX_CONSISTENCY_CHECKS_PROP)); return Integer.parseInt(props.getProperty(MAX_CONSISTENCY_CHECKS_PROP));
} }
@@ -696,6 +701,8 @@ public class HoodieWriteConfig extends DefaultHoodieConfig {
MAX_CONSISTENCY_CHECK_INTERVAL_MS_PROP, String.valueOf(DEFAULT_MAX_CONSISTENCY_CHECK_INTERVAL_MS)); MAX_CONSISTENCY_CHECK_INTERVAL_MS_PROP, String.valueOf(DEFAULT_MAX_CONSISTENCY_CHECK_INTERVAL_MS));
setDefaultOnCondition(props, !props.containsKey(MAX_CONSISTENCY_CHECKS_PROP), setDefaultOnCondition(props, !props.containsKey(MAX_CONSISTENCY_CHECKS_PROP),
MAX_CONSISTENCY_CHECKS_PROP, String.valueOf(DEFAULT_MAX_CONSISTENCY_CHECKS)); MAX_CONSISTENCY_CHECKS_PROP, String.valueOf(DEFAULT_MAX_CONSISTENCY_CHECKS));
setDefaultOnCondition(props, !props.containsKey(FAIL_ON_TIMELINE_ARCHIVING_ENABLED_PROP),
FAIL_ON_TIMELINE_ARCHIVING_ENABLED_PROP, DEFAULT_FAIL_ON_TIMELINE_ARCHIVING_ENABLED);
// Make sure the props is propagated // Make sure the props is propagated
setDefaultOnCondition(props, !isIndexConfigSet, setDefaultOnCondition(props, !isIndexConfigSet,

7
hoodie-client/src/main/java/com/uber/hoodie/io/HoodieCommitArchiveLog.java
View File

@@ -246,10 +246,17 @@ public class HoodieCommitArchiveLog {
log.info("Wrapper schema " + wrapperSchema.toString()); log.info("Wrapper schema " + wrapperSchema.toString());
List<IndexedRecord> records = new ArrayList<>(); List<IndexedRecord> records = new ArrayList<>();
for (HoodieInstant hoodieInstant : instants) { for (HoodieInstant hoodieInstant : instants) {
try {
records.add(convertToAvroRecord(commitTimeline, hoodieInstant)); records.add(convertToAvroRecord(commitTimeline, hoodieInstant));
if (records.size() >= this.config.getCommitArchivalBatchSize()) { if (records.size() >= this.config.getCommitArchivalBatchSize()) {
writeToFile(wrapperSchema, records); writeToFile(wrapperSchema, records);
} }
} catch (Exception e) {
log.error("Failed to archive commits, .commit file: " + hoodieInstant.getFileName(), e);
if (this.config.isFailOnTimelineArchivingEnabled()) {
throw e;
}
}
} }
writeToFile(wrapperSchema, records); writeToFile(wrapperSchema, records);
} catch (Exception e) { } catch (Exception e) {