You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2022/03/30 20:09:13 UTC
[GitHub] [hudi] nsivabalan commented on a change in pull request #5138: [HUDI-3713] Guarding archival for multi-writer
nsivabalan commented on a change in pull request #5138:
URL: https://github.com/apache/hudi/pull/5138#discussion_r838921139
##########
File path: hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiver.java
##########
@@ -404,6 +423,79 @@ public void testLoadArchiveTimelineWithDamagedPlanFile(boolean enableArchiveMerg
assertThrows(HoodieException.class, () -> metaClient.getArchivedTimeline().reload());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testArchivalWithMultiWriters(boolean enableMetadata) throws Exception {
+ HoodieWriteConfig writeConfig = initTestTableAndGetWriteConfig(enableMetadata, 2, 4, 5, 2,
+ HoodieTableType.COPY_ON_WRITE, false, 10, 209715200,
+ HoodieFailedWritesCleaningPolicy.LAZY, WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL);
+
+ final ExecutorService executors = Executors.newFixedThreadPool(2);
+ List<CompletableFuture<Boolean>> completableFutureList = new ArrayList<>();
+ CountDownLatch countDownLatch = new CountDownLatch(1);
+ IntStream.range(0, 2).forEach(index -> {
+ completableFutureList.add(CompletableFuture.supplyAsync(() -> {
+ HoodieTable table = HoodieSparkTable.create(writeConfig, context, metaClient);
+ try {
+ countDownLatch.await(30, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ metaClient.reloadActiveTimeline();
+ while (!metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants().lastInstant().get().getTimestamp().endsWith("29")
+ || metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants().countInstants() > 4) {
+ try {
+ //System.out.println("Archiving " + index + ", total completed instants " + metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants().countInstants());
+ //System.out.println("Last active instant " + metaClient.reloadActiveTimeline().getCommitsTimeline().filterCompletedInstants().lastInstant().get().toString());
+ HoodieTimelineArchiver archiver = new HoodieTimelineArchiver(writeConfig, table);
+ archiver.archiveIfRequired(context, true);
+ // if not for below sleep, both archiving threads acquires lock in quick succession and does not give space for main thread
+ // to complete the write operation when metadata table is enabled.
+ if (enableMetadata) {
+ Thread.sleep(2);
Review comment:
this is being invoked in a loop by 2 different threads. and archiveIfRequired is acquiring a lock as you might be aware. Whats happening is that, these two threads keeps acquiring the lock in quick successing and does not give much space for the writeClient to acquire lock to perform more writes (needs lock to write to metadata).
##########
File path: hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
##########
@@ -468,14 +469,16 @@ protected void preWrite(String instantTime, WriteOperationType writeOperationTyp
* @param metadata Commit Metadata corresponding to committed instant
* @param instantTime Instant Time
* @param extraMetadata Additional Metadata passed by user
+ * @param acquireLockForArchival true if lock has to be acquired for archival. false otherwise.
*/
- protected void postCommit(HoodieTable table, HoodieCommitMetadata metadata, String instantTime, Option<Map<String, String>> extraMetadata) {
+ protected void postCommit(HoodieTable table, HoodieCommitMetadata metadata, String instantTime, Option<Map<String, String>> extraMetadata,
+ boolean acquireLockForArchival) {
Review comment:
there are two code paths from WriteClient stand point. in one flow, the caller calls archival from within a lock already. where as, for another flow, the caller does not have a lock. So, to avoid double locking, we have to do this. We already have a tracking ticket to clean up this w/ a nice abstraction.
https://issues.apache.org/jira/browse/HUDI-2635
##########
File path: hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
##########
@@ -468,14 +469,16 @@ protected void preWrite(String instantTime, WriteOperationType writeOperationTyp
* @param metadata Commit Metadata corresponding to committed instant
* @param instantTime Instant Time
* @param extraMetadata Additional Metadata passed by user
+ * @param acquireLockForArchival true if lock has to be acquired for archival. false otherwise.
*/
- protected void postCommit(HoodieTable table, HoodieCommitMetadata metadata, String instantTime, Option<Map<String, String>> extraMetadata) {
+ protected void postCommit(HoodieTable table, HoodieCommitMetadata metadata, String instantTime, Option<Map<String, String>> extraMetadata,
+ boolean acquireLockForArchival) {
Review comment:
So, this is not about OCC or not.
##########
File path: hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieTimelineArchiver.java
##########
@@ -167,6 +178,9 @@ public boolean archiveIfRequired(HoodieEngineContext context) throws IOException
return success;
} finally {
close();
+ if (acquireLock) {
+ txnManager.close();
Review comment:
my bad. it has to be end transaction.
##########
File path: hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieTimelineArchiver.java
##########
@@ -143,11 +147,18 @@ private void close() {
}
}
+ public boolean archiveIfRequired(HoodieEngineContext context) throws IOException {
+ return archiveIfRequired(context, false);
+ }
+
/**
* Check if commits need to be archived. If yes, archive commits.
*/
- public boolean archiveIfRequired(HoodieEngineContext context) throws IOException {
+ public boolean archiveIfRequired(HoodieEngineContext context, boolean acquireLock) throws IOException {
try {
+ if (acquireLock) {
+ txnManager.beginTransaction();
Review comment:
again, this is for double locking purpose. not for OCC vs single writer.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org