lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

HoodieSortedMergeHandle#close write data disorder (#4841)

Browse Source 
Co-authored-by: 854194341@qq.com <loukey_7821>
This commit is contained in:
luokey
2022-02-18 17:31:38 +08:00
committed by GitHub
parent ed106f671e
commit de8161ae96
1 changed files with 3 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
View File

@@ -108,8 +108,9 @@ public class HoodieSortedMergeHandle<T extends HoodieRecordPayload, I, K, O> ext
@Override @Override
public List<WriteStatus> close() { public List<WriteStatus> close() {
// write out any pending records (this can happen when inserts are turned into updates) // write out any pending records (this can happen when inserts are turned into updates)
newRecordKeysSorted.stream().forEach(key -> { while (!newRecordKeysSorted.isEmpty()) {
try { try {
String key = newRecordKeysSorted.poll();
HoodieRecord<T> hoodieRecord = keyToNewRecords.get(key); HoodieRecord<T> hoodieRecord = keyToNewRecords.get(key);
if (!writtenRecordKeys.contains(hoodieRecord.getRecordKey())) { if (!writtenRecordKeys.contains(hoodieRecord.getRecordKey())) {
if (useWriterSchema) { if (useWriterSchema) {
@@ -122,7 +123,7 @@ public class HoodieSortedMergeHandle<T extends HoodieRecordPayload, I, K, O> ext
} catch (IOException e) { } catch (IOException e) {
throw new HoodieUpsertException("Failed to close UpdateHandle", e); throw new HoodieUpsertException("Failed to close UpdateHandle", e);
} }
}); }
newRecordKeysSorted.clear(); newRecordKeysSorted.clear();
keyToNewRecords.clear(); keyToNewRecords.clear();