lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

Fix various errors found by long running delta-streamer tests

Browse Source 
1. Parquet Avro schema mismatch errors when ingesting are sometimes silently ignored due to race-condition in BoundedInMemoryExecutor. This was reproducible when running long-running delta-streamer with wrong schema and it caused data-loss
  2. Fix behavior of Delta-Streamer to error out by default if there are any error records
  3. Fix a bug in tracking write errors in WriteStats. Earlier the write errors were tracking sampled errors as opposed to total errors.
  4. Delta Streamer does not commit the changes done as part of inline compaction as auto-commit is force disabled. Fix this behavior to always auto-commit inline compaction as it would not otherwise commit.
This commit is contained in:
Balaji Varadarajan
2019-05-12 11:57:04 -07:00
committed by vinoth chandar
parent a0e62b7919
commit 9cce9abf4d
8 changed files with 48 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
hoodie-client/src/main/java/com/uber/hoodie/HoodieWriteClient.java
View File

@@ -1348,7 +1348,8 @@ public class HoodieWriteClient<T extends HoodieRecordPayload> implements Seriali
Optional<String> compactionInstantTimeOpt = scheduleCompaction(extraMetadata);
compactionInstantTimeOpt.ifPresent(compactionInstantTime -> {
try {
compact(compactionInstantTime);
// inline compaction should auto commit as the user is never given control
compact(compactionInstantTime, true);
} catch (IOException ioe) {
throw new HoodieIOException(ioe.getMessage(), ioe);
}

3
hoodie-client/src/main/java/com/uber/hoodie/WriteStatus.java
View File

@@ -89,7 +89,8 @@ public class WriteStatus implements Serializable {
* HoodieRecord} before deflation.
*/
public void markFailure(HoodieRecord record, Throwable t, Optional<Map<String, String>> optionalRecordMetadata) {
if (random.nextDouble() <= failureFraction) {
if (failedRecords.isEmpty() || (random.nextDouble() <= failureFraction)) {
// Guaranteed to have at-least one error
failedRecords.add(record);
errors.put(record.getKey(), t);
}

2
hoodie-client/src/main/java/com/uber/hoodie/io/HoodieAppendHandle.java
View File

@@ -251,7 +251,7 @@ public class HoodieAppendHandle<T extends HoodieRecordPayload> extends HoodieIOH
writeStatus.getStat().setNumInserts(insertRecordsWritten);
writeStatus.getStat().setNumDeletes(recordsDeleted);
writeStatus.getStat().setTotalWriteBytes(estimatedNumberOfBytesWritten);
writeStatus.getStat().setTotalWriteErrors(writeStatus.getFailedRecords().size());
writeStatus.getStat().setTotalWriteErrors(writeStatus.getTotalErrorRecords());
RuntimeStats runtimeStats = new RuntimeStats();
runtimeStats.setTotalUpsertTime(timer.endTimer());
writeStatus.getStat().setRuntimeStats(runtimeStats);

2
hoodie-client/src/main/java/com/uber/hoodie/io/HoodieCreateHandle.java
View File

@@ -166,7 +166,7 @@ public class HoodieCreateHandle<T extends HoodieRecordPayload> extends HoodieIOH
stat.setFileId(writeStatus.getFileId());
stat.setPaths(new Path(config.getBasePath()), path, tempPath);
stat.setTotalWriteBytes(FSUtils.getFileSize(fs, getStorageWriterPath()));
stat.setTotalWriteErrors(writeStatus.getFailedRecords().size());
stat.setTotalWriteErrors(writeStatus.getTotalErrorRecords());
RuntimeStats runtimeStats = new RuntimeStats();
runtimeStats.setTotalCreateTime(timer.endTimer());
stat.setRuntimeStats(runtimeStats);

2
hoodie-client/src/main/java/com/uber/hoodie/io/HoodieMergeHandle.java
View File

@@ -281,7 +281,7 @@ public class HoodieMergeHandle<T extends HoodieRecordPayload> extends HoodieIOHa
writeStatus.getStat().setNumDeletes(recordsDeleted);
writeStatus.getStat().setNumUpdateWrites(updatedRecordsWritten);
writeStatus.getStat().setNumInserts(insertRecordsWritten);
writeStatus.getStat().setTotalWriteErrors(writeStatus.getFailedRecords().size());
writeStatus.getStat().setTotalWriteErrors(writeStatus.getTotalErrorRecords());
RuntimeStats runtimeStats = new RuntimeStats();
runtimeStats.setTotalUpsertTime(timer.endTimer());
writeStatus.getStat().setRuntimeStats(runtimeStats);