lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-2573] Fixing double locking with multi-writers (#3827)

Browse Source 
- There are two code paths, where we are taking double locking. this was added as part of adding data table locks to update metadata table. Fixing those flows to avoid taking locks if a parent transaction already acquired a lock.
This commit is contained in:
Sivabalan Narayanan
2021-10-29 12:14:39 -04:00
committed by GitHub
parent 69ee790a47
commit 29574af239
18 changed files with 281 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
View File

@@ -424,7 +424,7 @@ public class SparkRDDWriteClient<T extends HoodieRecordPayload> extends
this.txnManager.beginTransaction();
try {
// Ensure no inflight commits by setting EAGER policy and explicitly cleaning all failed commits
this.rollbackFailedWrites(getInstantsToRollback(metaClient, HoodieFailedWritesCleaningPolicy.EAGER));
this.rollbackFailedWrites(getInstantsToRollback(metaClient, HoodieFailedWritesCleaningPolicy.EAGER, Option.of(instantTime)), true);
new UpgradeDowngrade(
metaClient, config, context, SparkUpgradeDowngradeHelper.getInstance())
.run(HoodieTableVersion.current(), instantTime);
@@ -434,6 +434,7 @@ public class SparkRDDWriteClient<T extends HoodieRecordPayload> extends
} else {
upgradeDowngrade.run(HoodieTableVersion.current(), instantTime);
}
metaClient.reloadActiveTimeline();
}
metaClient.validateTableProperties(config.getProps(), operationType);
return getTableAndInitCtx(metaClient, operationType, instantTime);

10
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkCopyOnWriteTable.java
View File

@@ -256,13 +256,15 @@ public class HoodieSparkCopyOnWriteTable<T extends HoodieRecordPayload>
}
@Override
public HoodieCleanMetadata clean(HoodieEngineContext context, String cleanInstantTime) {
return new CleanActionExecutor(context, config, this, cleanInstantTime).execute();
public HoodieCleanMetadata clean(HoodieEngineContext context, String cleanInstantTime, boolean skipLocking) {
return new CleanActionExecutor(context, config, this, cleanInstantTime, skipLocking).execute();
}
@Override
public HoodieRollbackMetadata rollback(HoodieEngineContext context, String rollbackInstantTime, HoodieInstant commitInstant, boolean deleteInstants) {
return new CopyOnWriteRollbackActionExecutor((HoodieSparkEngineContext) context, config, this, rollbackInstantTime, commitInstant, deleteInstants).execute();
public HoodieRollbackMetadata rollback(HoodieEngineContext context, String rollbackInstantTime, HoodieInstant commitInstant,
boolean deleteInstants, boolean skipLocking) {
return new CopyOnWriteRollbackActionExecutor((HoodieSparkEngineContext) context, config, this, rollbackInstantTime, commitInstant,
deleteInstants, skipLocking).execute();
}
@Override

5
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkMergeOnReadTable.java
View File

@@ -159,8 +159,9 @@ public class HoodieSparkMergeOnReadTable<T extends HoodieRecordPayload> extends
public HoodieRollbackMetadata rollback(HoodieEngineContext context,
String rollbackInstantTime,
HoodieInstant commitInstant,
boolean deleteInstants) {
return new MergeOnReadRollbackActionExecutor(context, config, this, rollbackInstantTime, commitInstant, deleteInstants).execute();
boolean deleteInstants,
boolean skipLocking) {
return new MergeOnReadRollbackActionExecutor(context, config, this, rollbackInstantTime, commitInstant, deleteInstants, skipLocking).execute();
}
@Override

133
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
View File

@@ -264,7 +264,7 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
@EnumSource(HoodieTableType.class)
public void testInsertUpsertCluster(HoodieTableType tableType) throws Exception {
init(tableType);
doWriteOperation(testTable,"0000001", INSERT);
doWriteOperation(testTable, "0000001", INSERT);
doWriteOperation(testTable, "0000002");
doClusterAndValidate(testTable, "0000003");
if (tableType == MERGE_ON_READ) {
@@ -638,6 +638,51 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
validateMetadata(writeClients[0]);
}
/**
* Tests that when inline cleaning is enabled and with auto commit set to true, there is no double locking.
* bcoz, auto clean is triggered within post commit which is already happening within a lock.
*
* @throws Exception
*/
@Test
public void testMultiWriterForDoubleLocking() throws Exception {
init(HoodieTableType.COPY_ON_WRITE);
HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc);
Properties properties = new Properties();
properties.setProperty(FILESYSTEM_LOCK_PATH_PROP_KEY, basePath + "/.hoodie/.locks");
properties.setProperty(LockConfiguration.LOCK_ACQUIRE_CLIENT_NUM_RETRIES_PROP_KEY, "3");
properties.setProperty(LockConfiguration.LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY, "5000");
HoodieWriteConfig writeConfig = getWriteConfigBuilder(true, true, false)
.withCompactionConfig(HoodieCompactionConfig.newBuilder()
.withFailedWritesCleaningPolicy(HoodieFailedWritesCleaningPolicy.LAZY).withAutoClean(true).retainCommits(4).build())
.withAutoCommit(false)
.withWriteConcurrencyMode(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL)
.withLockConfig(HoodieLockConfig.newBuilder().withLockProvider(FileSystemBasedLockProviderTestClass.class).build())
.withProperties(properties)
.build();
SparkRDDWriteClient writeClient = new SparkRDDWriteClient(engineContext, writeConfig);
String partitionPath = dataGen.getPartitionPaths()[0];
for (int j = 0; j < 6; j++) {
String newCommitTime = "000000" + j;
List<HoodieRecord> records = dataGen.generateInsertsForPartition(newCommitTime, 100, partitionPath);
writeClient.startCommitWithTime(newCommitTime);
JavaRDD writeStatuses = writeClient.insert(jsc.parallelize(records, 1), newCommitTime);
writeClient.commit(newCommitTime, writeStatuses);
}
// Ensure all commits were synced to the Metadata Table
HoodieTableMetaClient metadataMetaClient = HoodieTableMetaClient.builder().setConf(hadoopConf).setBasePath(metadataTableBasePath).build();
LOG.warn("total commits in metadata table " + metadataMetaClient.getActiveTimeline().getCommitsTimeline().countInstants());
// 6 commits and 2 cleaner commits.
assertEquals(metadataMetaClient.getActiveTimeline().getDeltaCommitTimeline().filterCompletedInstants().countInstants(), 8);
assertTrue(metadataMetaClient.getActiveTimeline().getCommitTimeline().filterCompletedInstants().countInstants() <= 1);
// Validation
validateMetadata(writeClient);
}
/**
* Lets say clustering commit succeeded in metadata table, but failed before committing to datatable.
* Next time, when clustering kicks in, hudi will rollback pending clustering and re-attempt the clustering with same instant time.
@@ -924,6 +969,92 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
assertFalse(fs.exists(new Path(metadataTableBasePath)), "Metadata table should not exist");
}
/**
* When table needs to be upgraded and when multi writer is enabled, hudi rollsback partial commits. Upgrade itself is happening
* within a lock and hence rollback should not lock again.
*
* @throws IOException
* @throws InterruptedException
*/
@Test
public void testRollbackDuringUpgradeForDoubleLocking() throws IOException, InterruptedException {
init(HoodieTableType.COPY_ON_WRITE, false);
HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc);
// Perform a commit. This should bootstrap the metadata table with latest version.
List<HoodieRecord> records;
JavaRDD<WriteStatus> writeStatuses;
String commitTimestamp = HoodieActiveTimeline.createNewInstantTime();
Properties properties = new Properties();
properties.setProperty(FILESYSTEM_LOCK_PATH_PROP_KEY, basePath + "/.hoodie/.locks");
properties.setProperty(LockConfiguration.LOCK_ACQUIRE_CLIENT_NUM_RETRIES_PROP_KEY, "3");
properties.setProperty(LockConfiguration.LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY, "5000");
HoodieWriteConfig writeConfig = getWriteConfigBuilder(false, true, false)
.withCompactionConfig(HoodieCompactionConfig.newBuilder()
.withFailedWritesCleaningPolicy(HoodieFailedWritesCleaningPolicy.LAZY).withAutoClean(false).build())
.withWriteConcurrencyMode(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL)
.withLockConfig(HoodieLockConfig.newBuilder().withLockProvider(FileSystemBasedLockProviderTestClass.class).build())
.withProperties(properties)
.build();
try (SparkRDDWriteClient client = new SparkRDDWriteClient(engineContext, writeConfig)) {
records = dataGen.generateInserts(commitTimestamp, 5);
client.startCommitWithTime(commitTimestamp);
writeStatuses = client.insert(jsc.parallelize(records, 1), commitTimestamp);
client.commit(commitTimestamp, writeStatuses);
}
// Metadata table should have been bootstrapped
assertTrue(fs.exists(new Path(metadataTableBasePath)), "Metadata table should exist");
FileStatus oldStatus = fs.getFileStatus(new Path(metadataTableBasePath));
// trigger partial commit
metaClient.reloadActiveTimeline();
commitTimestamp = HoodieActiveTimeline.createNewInstantTime();
try (SparkRDDWriteClient client = new SparkRDDWriteClient(engineContext, writeConfig)) {
records = dataGen.generateInserts(commitTimestamp, 5);
client.startCommitWithTime(commitTimestamp);
writeStatuses = client.insert(jsc.parallelize(records, 1), commitTimestamp);
}
// set hoodie.table.version to 2 in hoodie.properties file
changeTableVersion(HoodieTableVersion.TWO);
writeConfig = getWriteConfigBuilder(true, true, false)
.withRollbackUsingMarkers(false)
.withCompactionConfig(HoodieCompactionConfig.newBuilder()
.withFailedWritesCleaningPolicy(HoodieFailedWritesCleaningPolicy.LAZY).withAutoClean(false).build())
.withWriteConcurrencyMode(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL)
.withLockConfig(HoodieLockConfig.newBuilder().withLockProvider(FileSystemBasedLockProviderTestClass.class).build())
.withProperties(properties)
.build();
// With next commit the table should be deleted (as part of upgrade) and partial commit should be rolled back.
metaClient.reloadActiveTimeline();
commitTimestamp = HoodieActiveTimeline.createNewInstantTime();
try (SparkRDDWriteClient client = new SparkRDDWriteClient(engineContext, writeConfig)) {
records = dataGen.generateInserts(commitTimestamp, 5);
client.startCommitWithTime(commitTimestamp);
writeStatuses = client.insert(jsc.parallelize(records, 1), commitTimestamp);
assertNoWriteErrors(writeStatuses.collect());
}
assertFalse(fs.exists(new Path(metadataTableBasePath)), "Metadata table should not exist");
// With next commit the table should be re-bootstrapped (currently in the constructor. To be changed)
commitTimestamp = HoodieActiveTimeline.createNewInstantTime();
try (SparkRDDWriteClient client = new SparkRDDWriteClient(engineContext, writeConfig)) {
records = dataGen.generateInserts(commitTimestamp, 5);
client.startCommitWithTime(commitTimestamp);
writeStatuses = client.insert(jsc.parallelize(records, 1), commitTimestamp);
assertNoWriteErrors(writeStatuses.collect());
}
assertTrue(fs.exists(new Path(metadataTableBasePath)), "Metadata table should exist");
initMetaClient();
assertEquals(metaClient.getTableConfig().getTableVersion().versionCode(), HoodieTableVersion.THREE.versionCode());
assertTrue(fs.exists(new Path(metadataTableBasePath)), "Metadata table should exist");
FileStatus newStatus = fs.getFileStatus(new Path(metadataTableBasePath));
assertTrue(oldStatus.getModificationTime() < newStatus.getModificationTime());
}
/**
* Test various error scenarios.
*/

2
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestCleaner.java
View File

@@ -1308,7 +1308,7 @@ public class TestCleaner extends HoodieClientTestBase {
metaClient.reloadActiveTimeline();
HoodieInstant rollbackInstant = new HoodieInstant(State.INFLIGHT, HoodieTimeline.COMMIT_ACTION, "000");
table.scheduleRollback(context, "001", rollbackInstant, false);
table.rollback(context, "001", rollbackInstant, true);
table.rollback(context, "001", rollbackInstant, true, false);
final int numTempFilesAfter = testTable.listAllFilesInTempFolder().length;
assertEquals(0, numTempFilesAfter, "All temp files are deleted.");
}

6
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestCopyOnWriteRollbackActionExecutor.java
View File

@@ -88,7 +88,8 @@ public class TestCopyOnWriteRollbackActionExecutor extends HoodieClientRollbackT
BaseRollbackPlanActionExecutor copyOnWriteRollbackPlanActionExecutor =
new BaseRollbackPlanActionExecutor(context, table.getConfig(), table, "003", needRollBackInstant, false);
HoodieRollbackPlan rollbackPlan = (HoodieRollbackPlan) copyOnWriteRollbackPlanActionExecutor.execute().get();
CopyOnWriteRollbackActionExecutor copyOnWriteRollbackActionExecutor = new CopyOnWriteRollbackActionExecutor(context, table.getConfig(), table, "003", needRollBackInstant, true);
CopyOnWriteRollbackActionExecutor copyOnWriteRollbackActionExecutor = new CopyOnWriteRollbackActionExecutor(context, table.getConfig(), table, "003", needRollBackInstant, true,
false);
List<HoodieRollbackStat> hoodieRollbackStats = copyOnWriteRollbackActionExecutor.executeRollback(rollbackPlan);
// assert hoodieRollbackStats
@@ -169,7 +170,8 @@ public class TestCopyOnWriteRollbackActionExecutor extends HoodieClientRollbackT
BaseRollbackPlanActionExecutor copyOnWriteRollbackPlanActionExecutor =
new BaseRollbackPlanActionExecutor(context, table.getConfig(), table, "003", commitInstant, false);
HoodieRollbackPlan hoodieRollbackPlan = (HoodieRollbackPlan) copyOnWriteRollbackPlanActionExecutor.execute().get();
CopyOnWriteRollbackActionExecutor copyOnWriteRollbackActionExecutor = new CopyOnWriteRollbackActionExecutor(context, cfg, table, "003", commitInstant, false);
CopyOnWriteRollbackActionExecutor copyOnWriteRollbackActionExecutor = new CopyOnWriteRollbackActionExecutor(context, cfg, table, "003", commitInstant, false,
false);
Map<String, HoodieRollbackPartitionMetadata> rollbackMetadata = copyOnWriteRollbackActionExecutor.execute().getPartitionMetadata();
//3. assert the rollback stat

6
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMergeOnReadRollbackActionExecutor.java
View File

@@ -99,7 +99,8 @@ public class TestMergeOnReadRollbackActionExecutor extends HoodieClientRollbackT
table,
"003",
rollBackInstant,
true);
true,
false);
//3. assert the rollback stat
Map<String, HoodieRollbackPartitionMetadata> rollbackMetadata = mergeOnReadRollbackActionExecutor.execute().getPartitionMetadata();
assertEquals(2, rollbackMetadata.size());
@@ -148,7 +149,8 @@ public class TestMergeOnReadRollbackActionExecutor extends HoodieClientRollbackT
rollBackInstant,
true,
true,
true).execute();
true,
false).execute();
});
}