Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[HUDI-3453] Fix HoodieBackedTableMetadata concurrent reading issue #5091

Merged
merged 16 commits into from
Sep 9, 2022
Merged

[HUDI-3453] Fix HoodieBackedTableMetadata concurrent reading issue #5091

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
f65da7c
Fix HoodieBackedTableMetadata current reading issue.
Mar 22, 2022
dbea31f
close readers if set reuse false
Mar 23, 2022
da80525
merge from masterw
Mar 25, 2022
4da0484
merge from master
Jul 19, 2022
05447d8
Fix HoodieBackedTableMetadata current reading issue.
Mar 22, 2022
6c272e7
close readers if set reuse false
Jul 24, 2022
d5c4d54
Remove a warn log
codope Jul 25, 2022
a753995
merge from master
Sep 6, 2022
5076ea5
merge from master
Sep 6, 2022
5694cdd
use CountDownLatch for ut
Sep 6, 2022
daefc01
code reviewed
Sep 7, 2022
13507fc
code reviewed
Sep 7, 2022
c0dc922
code reviewed
Sep 7, 2022
3e3caec
add more debug logs
Sep 8, 2022
c711e86
add more debug logs
Sep 8, 2022
0ac5e2c
code review
Sep 8, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 57 additions & 0 deletions ...client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedTableMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,17 +58,26 @@
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.ValueSource;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import static org.apache.hudi.common.model.WriteOperationType.INSERT;
import static org.apache.hudi.common.model.WriteOperationType.UPSERT;

import static java.util.Arrays.asList;
import static java.util.Collections.emptyList;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
Expand All @@ -92,6 +101,54 @@ public void testTableOperations() throws Exception {
verifyBaseMetadataTable();
}

@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testMultiReaderForHoodieBackedTableMetadata(boolean reuse) throws Exception {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@zhangyue19921010 @nsivabalan folks, we should revisit our approach and avoid writing non-deterministic tests. Test that doesn't fail deterministically, but instead probabilistically has low net positive value b/c if it doesn't fail consistently in the presence of the issue it just becomes noisy, flaking signal.

final int taskNumber = 100;
HoodieTableType tableType = HoodieTableType.COPY_ON_WRITE;
init(tableType);
testTable.doWriteOperation("000001", INSERT, emptyList(), asList("p1"), 1);
HoodieBackedTableMetadata tableMetadata = new HoodieBackedTableMetadata(context, writeConfig.getMetadataConfig(), writeConfig.getBasePath(), writeConfig.getSpillableMapBasePath(), reuse);
assertTrue(tableMetadata.enabled());
List<String> metadataPartitions = tableMetadata.getAllPartitionPaths();
String partition = metadataPartitions.get(0);
String finalPartition = basePath + "/" + partition;
ArrayList<String> duplicatedPartitions = new ArrayList<>(taskNumber);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please add a comment explaining the intent of the test so that it's apparent and doesn't require deciphering the test to understand at least on the high levle

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added!

for (int i = 0; i < taskNumber; i++) {
duplicatedPartitions.add(finalPartition);
}
ExecutorService executors = Executors.newFixedThreadPool(taskNumber);
AtomicBoolean flag = new AtomicBoolean(false);
AtomicInteger count = new AtomicInteger(0);
AtomicInteger filesNumber = new AtomicInteger(0);

for (String part : duplicatedPartitions) {
executors.submit(new Runnable() {
@Override
public void run() {
try {
count.incrementAndGet();
while (true) {
if (count.get() == taskNumber) {
break;
}
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we add a countDownLatch here so that all threads could call tableMetadata.getAllFilesInPartition() around the same time. that way the test is deterministic.

FileStatus[] files = tableMetadata.getAllFilesInPartition(new Path(part));
filesNumber.addAndGet(files.length);
LOG.warn(Arrays.toString(files) + " : " + files.length);
codope marked this conversation as resolved.
Show resolved Hide resolved
assertEquals(1, files.length);
} catch (Exception e) {
flag.set(true);
}
}
});
}
executors.shutdown();
executors.awaitTermination(24, TimeUnit.HOURS);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's set meaningful timeout here (1 min should be more than enough to complete)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Changed!

assertFalse(flag.get());
assertEquals(filesNumber.get(), taskNumber);
}

private void doWriteInsertAndUpsert(HoodieTestTable testTable) throws Exception {
doWriteInsertAndUpsert(testTable, "0000001", "0000002", false);
}
Expand Down
9 changes: 7 additions & 2 deletions hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -229,7 +229,7 @@ public List<Pair<String, Option<HoodieRecord<HoodieMetadataPayload>>>> getRecord
throw new HoodieIOException("Error merging records from metadata table for " + sortedKeys.size() + " key : ", ioe);
} finally {
if (!reuse) {
close(Pair.of(partitionFileSlicePair.getLeft(), partitionFileSlicePair.getRight().getFileId()));
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's also clean up close method (doesn't seem like we need it)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Aha, It is also used in

  private void closePartitionReaders() {
    for (Pair<String, String> partitionFileSlicePair : partitionReaders.keySet()) {
      close(partitionFileSlicePair);
    }
    partitionReaders.clear();
  }

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, we should rewrite it to use closeReader instead

private void closePartitionReaders() {
    for (Pair<...> pair : partitionReaders.values()) {
      closeReader(pair);
    }
    partitionReaders.clear();
  }

closeReader(readers);
}
}
});
Expand Down Expand Up @@ -397,7 +397,12 @@ private Map<Pair<String, FileSlice>, List<String>> getPartitionFileSliceToKeysMa
* @return File reader and the record scanner pair for the requested file slice
*/
private Pair<HoodieFileReader, HoodieMetadataMergedLogRecordReader> getOrCreateReaders(String partitionName, FileSlice slice) {
return partitionReaders.computeIfAbsent(Pair.of(partitionName, slice.getFileId()), k -> openReaders(partitionName, slice));
if (reuse) {
return partitionReaders.computeIfAbsent(Pair.of(partitionName, slice.getFileId()), k -> {
return openReaders(partitionName, slice); });
} else {
return openReaders(partitionName, slice);
}
}

private Pair<HoodieFileReader, HoodieMetadataMergedLogRecordReader> openReaders(String partitionName, FileSlice slice) {
Expand Down