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 07:36:11 UTC
[GitHub] [hudi] codope commented on a change in pull request #5138: [HUDI-3713] Guarding archival for multi-writer
codope commented on a change in pull request #5138:
URL: https://github.com/apache/hudi/pull/5138#discussion_r838204510
##########
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());
Review comment:
let's remove these
##########
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:
is there a need to pass this boolean argument? can't we deduce base on config? typically, for multi-writer scenario users would have already enabled OCC mode. so why not take a lock whenever this mode is on?
##########
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:
should we have finally block and 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:
we just beginTransaction() in any case.. if there is no lock then beginTransaction() is a no-op anyway right?
##########
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:
why is this delay required only when metadata is enabled?
--
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