You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by si...@apache.org on 2023/09/08 20:24:56 UTC
[hudi] branch master updated: [HUDI-6820] Close write clients in tests (#9642)
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new e62d736b0c4 [HUDI-6820] Close write clients in tests (#9642)
e62d736b0c4 is described below
commit e62d736b0c43c41a83f3f411d6acb82777b752b7
Author: Tim Brown <ti...@onehouse.ai>
AuthorDate: Fri Sep 8 15:24:48 2023 -0500
[HUDI-6820] Close write clients in tests (#9642)
- Closing write clients properly in tests
---------
Co-authored-by: sivabalan <n....@gmail.com>
---
.../hudi/cli/commands/TestRestoresCommand.java | 24 +-
.../hudi/cli/integ/ITTestClusteringCommand.java | 8 +-
.../hudi/cli/integ/ITTestCompactionCommand.java | 9 +-
.../table/upgrade/SixToFiveDowngradeHandler.java | 9 +-
.../hudi/client/TestJavaHoodieBackedMetadata.java | 61 +--
.../TestHoodieJavaClientOnCopyOnWriteStorage.java | 2 +
.../callback/TestHoodieClientInitCallback.java | 61 +--
...DataValidationCheckForLogCompactionActions.java | 3 +-
.../functional/TestHoodieBackedMetadata.java | 63 +--
.../TestHoodieClientOnCopyOnWriteStorage.java | 3 +
.../TestHoodieClientOnMergeOnReadStorage.java | 169 ++++----
.../index/hbase/TestSparkHoodieHBaseIndex.java | 300 +++++++-------
.../TestHoodieSparkMergeOnReadTableCompaction.java | 8 +
...dieSparkMergeOnReadTableInsertUpdateDelete.java | 15 +-
.../TestHoodieSparkMergeOnReadTableRollback.java | 57 +--
.../ITTestFlinkConsistentHashingClustering.java | 30 +-
.../sink/cluster/ITTestHoodieFlinkClustering.java | 431 +++++++++++----------
.../sink/compact/ITTestHoodieFlinkCompactor.java | 193 ++++-----
.../apache/hudi/table/format/TestInputFormat.java | 1 +
.../org/apache/hudi/utils/TestClusteringUtil.java | 9 +
.../org/apache/hudi/utils/TestCompactionUtil.java | 23 +-
.../hudi/utils/TestViewStorageProperties.java | 8 +-
.../org/apache/hudi/TestHoodieFileIndex.scala | 1 +
.../functional/TestColumnStatsIndexWithSQL.scala | 1 +
...TestStreamSourceReadByStateTransitionTime.scala | 1 +
.../hudi/functional/TestStructuredStreaming.scala | 6 +-
.../apache/hudi/utilities/TestHoodieIndexer.java | 45 ++-
.../functional/TestHoodieSnapshotExporter.java | 14 +-
.../offlinejob/HoodieOfflineJobTestBase.java | 8 +
.../sources/TestGcsEventsHoodieIncrSource.java | 27 +-
.../utilities/sources/TestHoodieIncrSource.java | 347 +++++++++--------
.../sources/TestS3EventsHoodieIncrSource.java | 21 +-
32 files changed, 1019 insertions(+), 939 deletions(-)
diff --git a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRestoresCommand.java b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRestoresCommand.java
index 44b10b5c057..97da24bf7d0 100644
--- a/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRestoresCommand.java
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRestoresCommand.java
@@ -114,18 +114,18 @@ public class TestRestoresCommand extends CLIFunctionalTestHarness {
hoodieTestTable.addCommit("103").withBaseFilesInPartitions(partitionAndFileId);
- BaseHoodieWriteClient client = new SparkRDDWriteClient(context(), config);
- client.rollback("103");
- client.restoreToSavepoint("102");
-
- hoodieTestTable.addCommit("105").withBaseFilesInPartitions(partitionAndFileId);
- HoodieSavepointMetadata savepointMetadata = hoodieTestTable.doSavepoint("105");
- hoodieTestTable.addSavepoint("105", savepointMetadata);
-
- hoodieTestTable.addCommit("106").withBaseFilesInPartitions(partitionAndFileId);
- client.rollback("106");
- client.restoreToSavepoint("105");
- client.close();
+ try (BaseHoodieWriteClient client = new SparkRDDWriteClient(context(), config)) {
+ client.rollback("103");
+ client.restoreToSavepoint("102");
+
+ hoodieTestTable.addCommit("105").withBaseFilesInPartitions(partitionAndFileId);
+ HoodieSavepointMetadata savepointMetadata = hoodieTestTable.doSavepoint("105");
+ hoodieTestTable.addSavepoint("105", savepointMetadata);
+
+ hoodieTestTable.addCommit("106").withBaseFilesInPartitions(partitionAndFileId);
+ client.rollback("106");
+ client.restoreToSavepoint("105");
+ }
}
@Test
diff --git a/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestClusteringCommand.java b/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestClusteringCommand.java
index 22dbbe1b34b..2c6b17493d2 100644
--- a/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestClusteringCommand.java
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestClusteringCommand.java
@@ -176,10 +176,10 @@ public class ITTestClusteringCommand extends HoodieCLIIntegrationTestBase {
.withDeleteParallelism(2).forTable(tableName)
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.BLOOM).build()).build();
- SparkRDDWriteClient<HoodieAvroPayload> client = new SparkRDDWriteClient<>(new HoodieSparkEngineContext(jsc), cfg);
-
- insert(jsc, client, dataGen, "001");
- insert(jsc, client, dataGen, "002");
+ try (SparkRDDWriteClient<HoodieAvroPayload> client = new SparkRDDWriteClient<>(new HoodieSparkEngineContext(jsc), cfg)) {
+ insert(jsc, client, dataGen, "001");
+ insert(jsc, client, dataGen, "002");
+ }
}
private List<HoodieRecord> insert(JavaSparkContext jsc, SparkRDDWriteClient<HoodieAvroPayload> client,
diff --git a/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestCompactionCommand.java b/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestCompactionCommand.java
index 4e03efe4aae..6fc2d789b64 100644
--- a/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestCompactionCommand.java
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestCompactionCommand.java
@@ -305,11 +305,12 @@ public class ITTestCompactionCommand extends HoodieCLIIntegrationTestBase {
.withDeleteParallelism(2).forTable(tableName)
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.BLOOM).build()).build();
- SparkRDDWriteClient<HoodieAvroPayload> client = new SparkRDDWriteClient<>(new HoodieSparkEngineContext(jsc), cfg);
+ try (SparkRDDWriteClient<HoodieAvroPayload> client = new SparkRDDWriteClient<>(new HoodieSparkEngineContext(jsc), cfg)) {
- List<HoodieRecord> records = insert(jsc, client, dataGen);
- upsert(jsc, client, dataGen, records);
- delete(jsc, client, records);
+ List<HoodieRecord> records = insert(jsc, client, dataGen);
+ upsert(jsc, client, dataGen, records);
+ delete(jsc, client, records);
+ }
}
private List<HoodieRecord> insert(JavaSparkContext jsc, SparkRDDWriteClient<HoodieAvroPayload> client,
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/SixToFiveDowngradeHandler.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/SixToFiveDowngradeHandler.java
index 4793f368f81..dc2b7498aef 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/SixToFiveDowngradeHandler.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/SixToFiveDowngradeHandler.java
@@ -94,10 +94,11 @@ public class SixToFiveDowngradeHandler implements DowngradeHandler {
compactionConfig.setValue(HoodieCompactionConfig.INLINE_COMPACT_TRIGGER_STRATEGY.key(), CompactionTriggerStrategy.NUM_COMMITS.name());
compactionConfig.setValue(HoodieCompactionConfig.COMPACTION_STRATEGY.key(), UnBoundedCompactionStrategy.class.getName());
compactionConfig.setValue(HoodieMetadataConfig.ENABLE.key(), "false");
- BaseHoodieWriteClient writeClient = upgradeDowngradeHelper.getWriteClient(compactionConfig, context);
- Option<String> compactionInstantOpt = writeClient.scheduleCompaction(Option.empty());
- if (compactionInstantOpt.isPresent()) {
- writeClient.compact(compactionInstantOpt.get());
+ try (BaseHoodieWriteClient writeClient = upgradeDowngradeHelper.getWriteClient(compactionConfig, context)) {
+ Option<String> compactionInstantOpt = writeClient.scheduleCompaction(Option.empty());
+ if (compactionInstantOpt.isPresent()) {
+ writeClient.compact(compactionInstantOpt.get());
+ }
}
}
} catch (Exception e) {
diff --git a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java
index b22fa76788d..740b50cf9e1 100644
--- a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java
+++ b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java
@@ -1497,6 +1497,7 @@ public class TestJavaHoodieBackedMetadata extends TestHoodieMetadataBase {
metaClient.getFs().delete(toDelete);
// Write 3 (updates)
+ client.close();
client = new HoodieJavaWriteClient(engineContext, writeConfig);
String commit3 = HoodieActiveTimeline.createNewInstantTime();
client.startCommitWithTime(commit3);
@@ -1518,6 +1519,7 @@ public class TestJavaHoodieBackedMetadata extends TestHoodieMetadataBase {
// ensure commit3's delta commit in MDT has last mod time > the actual rollback for previous failed commit i.e. commit2.
// if rollback wasn't eager, rollback's last mod time will be lower than the commit3'd delta commit last mod time.
assertTrue(commit3Files.get(0).getModificationTime() > rollbackFiles.get(0).getModificationTime());
+ client.close();
}
/**
@@ -1680,6 +1682,7 @@ public class TestJavaHoodieBackedMetadata extends TestHoodieMetadataBase {
// Validation
validateMetadata(writeClients[0]);
+ Arrays.stream(writeClients).forEach(HoodieJavaWriteClient::close);
}
/**
@@ -1706,25 +1709,26 @@ public class TestJavaHoodieBackedMetadata extends TestHoodieMetadataBase {
.withProperties(properties)
.build();
- HoodieJavaWriteClient writeClient = new HoodieJavaWriteClient(engineContext, writeConfig);
- String partitionPath = dataGen.getPartitionPaths()[0];
- for (int j = 0; j < 6; j++) {
- String newCommitTime = HoodieActiveTimeline.createNewInstantTime();
- List<HoodieRecord> records = dataGen.generateInsertsForPartition(newCommitTime, 100, partitionPath);
- writeClient.startCommitWithTime(newCommitTime);
- List<WriteStatus> writeStatuses = writeClient.insert(records, newCommitTime);
- writeClient.commit(newCommitTime, writeStatuses);
- }
+ try (HoodieJavaWriteClient writeClient = new HoodieJavaWriteClient(engineContext, writeConfig)) {
+ String partitionPath = dataGen.getPartitionPaths()[0];
+ for (int j = 0; j < 6; j++) {
+ String newCommitTime = HoodieActiveTimeline.createNewInstantTime();
+ List<HoodieRecord> records = dataGen.generateInsertsForPartition(newCommitTime, 100, partitionPath);
+ writeClient.startCommitWithTime(newCommitTime);
+ List<WriteStatus> writeStatuses = writeClient.insert(records, 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());
+ // 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);
+ // 6 commits and 2 cleaner commits.
+ assertEquals(metadataMetaClient.getActiveTimeline().getDeltaCommitTimeline().filterCompletedInstants().countInstants(), 8);
+ assertTrue(metadataMetaClient.getActiveTimeline().getCommitTimeline().filterCompletedInstants().countInstants() <= 1);
+ // Validation
+ validateMetadata(writeClient);
+ }
}
/**
@@ -2584,20 +2588,21 @@ public class TestJavaHoodieBackedMetadata extends TestHoodieMetadataBase {
metadataProps.setProperty(INLINE_COMPACT_NUM_DELTA_COMMITS.key(), "3");
HoodieWriteConfig metadataWriteConfig = HoodieWriteConfig.newBuilder()
.withProperties(metadataProps).build();
- HoodieJavaWriteClient metadataWriteClient = new HoodieJavaWriteClient(context, metadataWriteConfig);
- final String compactionInstantTime = HoodieTableMetadataUtil.createCompactionTimestamp(commitTime);
- assertTrue(metadataWriteClient.scheduleCompactionAtInstant(compactionInstantTime, Option.empty()));
- metadataWriteClient.compact(compactionInstantTime);
+ try (HoodieJavaWriteClient metadataWriteClient = new HoodieJavaWriteClient(context, metadataWriteConfig)) {
+ final String compactionInstantTime = HoodieTableMetadataUtil.createCompactionTimestamp(commitTime);
+ assertTrue(metadataWriteClient.scheduleCompactionAtInstant(compactionInstantTime, Option.empty()));
+ metadataWriteClient.compact(compactionInstantTime);
- // verify metadata table
- validateMetadata(client);
+ // verify metadata table
+ validateMetadata(client);
- // Execute pending clustering operation
- clusteringClient = getHoodieWriteClient(clusterWriteCfg);
- clusteringClient.cluster("0000003", true);
+ // Execute pending clustering operation
+ clusteringClient = getHoodieWriteClient(clusterWriteCfg);
+ clusteringClient.cluster("0000003", true);
- // verify metadata table
- validateMetadata(client);
+ // verify metadata table
+ validateMetadata(client);
+ }
}
private void validateMetadata(HoodieJavaWriteClient testClient) throws IOException {
diff --git a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/functional/TestHoodieJavaClientOnCopyOnWriteStorage.java b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/functional/TestHoodieJavaClientOnCopyOnWriteStorage.java
index 211dc0129e6..ee4c1fca352 100644
--- a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/functional/TestHoodieJavaClientOnCopyOnWriteStorage.java
+++ b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/functional/TestHoodieJavaClientOnCopyOnWriteStorage.java
@@ -1335,6 +1335,7 @@ public class TestHoodieJavaClientOnCopyOnWriteStorage extends HoodieJavaClientTe
100, dataGen::generateInserts, HoodieJavaWriteClient::bulkInsert, false, 100, 300,
0, true);
client.clean();
+ client.close();
HoodieActiveTimeline timeline = metaClient.getActiveTimeline().reload();
if (cleaningPolicy.isLazy()) {
assertTrue(
@@ -1474,6 +1475,7 @@ public class TestHoodieJavaClientOnCopyOnWriteStorage extends HoodieJavaClientTe
Future<HoodieCleanMetadata> clean1 = service.submit(() -> new HoodieJavaWriteClient(context, getParallelWritingWriteConfig(cleaningPolicy, true)).clean());
commit4.get();
clean1.get();
+ client.close();
HoodieActiveTimeline timeline = metaClient.getActiveTimeline().reload();
assertTrue(timeline.getTimelineOfActions(
CollectionUtils.createSet(ROLLBACK_ACTION)).countInstants() == 2);
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/callback/TestHoodieClientInitCallback.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/callback/TestHoodieClientInitCallback.java
index 1ede02413fb..691214a71c5 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/callback/TestHoodieClientInitCallback.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/callback/TestHoodieClientInitCallback.java
@@ -81,10 +81,11 @@ public class TestHoodieClientInitCallback {
.build(false);
assertFalse(config.contains(CUSTOM_CONFIG_KEY1));
- SparkRDDWriteClient<Object> writeClient = new SparkRDDWriteClient<>(engineContext, config);
+ try (SparkRDDWriteClient<Object> writeClient = new SparkRDDWriteClient<>(engineContext, config)) {
- assertFalse(writeClient.getConfig().contains(CUSTOM_CONFIG_KEY1));
- assertFalse(writeClient.getTableServiceClient().getConfig().contains(CUSTOM_CONFIG_KEY1));
+ assertFalse(writeClient.getConfig().contains(CUSTOM_CONFIG_KEY1));
+ assertFalse(writeClient.getTableServiceClient().getConfig().contains(CUSTOM_CONFIG_KEY1));
+ }
}
@Test
@@ -100,19 +101,20 @@ public class TestHoodieClientInitCallback {
assertFalse(new Schema.Parser().parse(config.getWriteSchema())
.getObjectProps().containsKey(CUSTOM_CONFIG_KEY2));
- SparkRDDWriteClient<Object> writeClient = new SparkRDDWriteClient<>(engineContext, config);
+ try (SparkRDDWriteClient<Object> writeClient = new SparkRDDWriteClient<>(engineContext, config)) {
- HoodieWriteConfig updatedConfig = writeClient.getConfig();
- assertFalse(updatedConfig.contains(CUSTOM_CONFIG_KEY1));
- Schema actualSchema = new Schema.Parser().parse(updatedConfig.getWriteSchema());
- assertTrue(actualSchema.getObjectProps().containsKey(CUSTOM_CONFIG_KEY2));
- assertEquals(CUSTOM_CONFIG_VALUE2, actualSchema.getObjectProps().get(CUSTOM_CONFIG_KEY2));
+ HoodieWriteConfig updatedConfig = writeClient.getConfig();
+ assertFalse(updatedConfig.contains(CUSTOM_CONFIG_KEY1));
+ Schema actualSchema = new Schema.Parser().parse(updatedConfig.getWriteSchema());
+ assertTrue(actualSchema.getObjectProps().containsKey(CUSTOM_CONFIG_KEY2));
+ assertEquals(CUSTOM_CONFIG_VALUE2, actualSchema.getObjectProps().get(CUSTOM_CONFIG_KEY2));
- updatedConfig = writeClient.getTableServiceClient().getConfig();
- assertFalse(updatedConfig.contains(CUSTOM_CONFIG_KEY1));
- actualSchema = new Schema.Parser().parse(updatedConfig.getWriteSchema());
- assertTrue(actualSchema.getObjectProps().containsKey(CUSTOM_CONFIG_KEY2));
- assertEquals(CUSTOM_CONFIG_VALUE2, actualSchema.getObjectProps().get(CUSTOM_CONFIG_KEY2));
+ updatedConfig = writeClient.getTableServiceClient().getConfig();
+ assertFalse(updatedConfig.contains(CUSTOM_CONFIG_KEY1));
+ actualSchema = new Schema.Parser().parse(updatedConfig.getWriteSchema());
+ assertTrue(actualSchema.getObjectProps().containsKey(CUSTOM_CONFIG_KEY2));
+ assertEquals(CUSTOM_CONFIG_VALUE2, actualSchema.getObjectProps().get(CUSTOM_CONFIG_KEY2));
+ }
}
@Test
@@ -130,21 +132,22 @@ public class TestHoodieClientInitCallback {
assertFalse(new Schema.Parser().parse(config.getWriteSchema())
.getObjectProps().containsKey(CUSTOM_CONFIG_KEY2));
- SparkRDDWriteClient<Object> writeClient = new SparkRDDWriteClient<>(engineContext, config);
-
- HoodieWriteConfig updatedConfig = writeClient.getConfig();
- assertTrue(updatedConfig.contains(CUSTOM_CONFIG_KEY1));
- assertEquals(CUSTOM_CONFIG_VALUE1, updatedConfig.getString(CUSTOM_CONFIG_KEY1));
- Schema actualSchema = new Schema.Parser().parse(updatedConfig.getWriteSchema());
- assertTrue(actualSchema.getObjectProps().containsKey(CUSTOM_CONFIG_KEY2));
- assertEquals(CUSTOM_CONFIG_VALUE2, actualSchema.getObjectProps().get(CUSTOM_CONFIG_KEY2));
-
- updatedConfig = writeClient.getTableServiceClient().getConfig();
- assertTrue(updatedConfig.contains(CUSTOM_CONFIG_KEY1));
- assertEquals(CUSTOM_CONFIG_VALUE1, updatedConfig.getString(CUSTOM_CONFIG_KEY1));
- actualSchema = new Schema.Parser().parse(updatedConfig.getWriteSchema());
- assertTrue(actualSchema.getObjectProps().containsKey(CUSTOM_CONFIG_KEY2));
- assertEquals(CUSTOM_CONFIG_VALUE2, actualSchema.getObjectProps().get(CUSTOM_CONFIG_KEY2));
+ try (SparkRDDWriteClient<Object> writeClient = new SparkRDDWriteClient<>(engineContext, config)) {
+
+ HoodieWriteConfig updatedConfig = writeClient.getConfig();
+ assertTrue(updatedConfig.contains(CUSTOM_CONFIG_KEY1));
+ assertEquals(CUSTOM_CONFIG_VALUE1, updatedConfig.getString(CUSTOM_CONFIG_KEY1));
+ Schema actualSchema = new Schema.Parser().parse(updatedConfig.getWriteSchema());
+ assertTrue(actualSchema.getObjectProps().containsKey(CUSTOM_CONFIG_KEY2));
+ assertEquals(CUSTOM_CONFIG_VALUE2, actualSchema.getObjectProps().get(CUSTOM_CONFIG_KEY2));
+
+ updatedConfig = writeClient.getTableServiceClient().getConfig();
+ assertTrue(updatedConfig.contains(CUSTOM_CONFIG_KEY1));
+ assertEquals(CUSTOM_CONFIG_VALUE1, updatedConfig.getString(CUSTOM_CONFIG_KEY1));
+ actualSchema = new Schema.Parser().parse(updatedConfig.getWriteSchema());
+ assertTrue(actualSchema.getObjectProps().containsKey(CUSTOM_CONFIG_KEY2));
+ assertEquals(CUSTOM_CONFIG_VALUE2, actualSchema.getObjectProps().get(CUSTOM_CONFIG_KEY2));
+ }
}
@Test
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestDataValidationCheckForLogCompactionActions.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestDataValidationCheckForLogCompactionActions.java
index a04182e3379..635f1c651ac 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestDataValidationCheckForLogCompactionActions.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestDataValidationCheckForLogCompactionActions.java
@@ -166,7 +166,8 @@ public class TestDataValidationCheckForLogCompactionActions extends HoodieClient
}
curr++;
}
-
+ mainTable.client.close();
+ experimentTable.client.close();
}
private void verifyRecords(TestTableContents mainTable, TestTableContents experimentTable) {
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
index 6f6c4b65b11..05c67c02686 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
@@ -1910,6 +1910,7 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
metaClient.getFs().delete(toDelete);
// Write 3 (updates)
+ client.close();
client = new SparkRDDWriteClient(engineContext, writeConfig);
String commit3 = HoodieActiveTimeline.createNewInstantTime();
client.startCommitWithTime(commit3);
@@ -1931,6 +1932,7 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
// ensure commit3's delta commit in MDT has last mod time > the actual rollback for previous failed commit i.e. commit2.
// if rollback wasn't eager, rollback's last mod time will be lower than the commit3'd delta commit last mod time.
assertTrue(commit3Files.get(0).getModificationTime() > rollbackFiles.get(0).getModificationTime());
+ client.close();
}
/**
@@ -2080,6 +2082,7 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
for (Future future : futures) {
future.get();
}
+ executors.shutdown();
// Ensure all commits were synced to the Metadata Table
HoodieTableMetaClient metadataMetaClient = HoodieTableMetaClient.builder().setConf(hadoopConf).setBasePath(metadataTableBasePath).build();
@@ -2093,6 +2096,7 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
// Validation
validateMetadata(writeClients[0]);
+ Arrays.stream(writeClients).forEach(SparkRDDWriteClient::close);
}
/**
@@ -2119,25 +2123,27 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
.withProperties(properties)
.build();
- SparkRDDWriteClient writeClient = new SparkRDDWriteClient(engineContext, writeConfig);
- String partitionPath = dataGen.getPartitionPaths()[0];
- for (int j = 0; j < 6; j++) {
- String newCommitTime = HoodieActiveTimeline.createNewInstantTime();
- List<HoodieRecord> records = dataGen.generateInsertsForPartition(newCommitTime, 100, partitionPath);
- writeClient.startCommitWithTime(newCommitTime);
- JavaRDD writeStatuses = writeClient.insert(jsc.parallelize(records, 1), newCommitTime);
- writeClient.commit(newCommitTime, writeStatuses);
- }
+ try (SparkRDDWriteClient writeClient = new SparkRDDWriteClient(engineContext, writeConfig)) {
+ String partitionPath = dataGen.getPartitionPaths()[0];
+ for (int j = 0; j < 6; j++) {
+ String newCommitTime = HoodieActiveTimeline.createNewInstantTime();
+ 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);
+ // 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);
+ }
}
/**
@@ -3200,20 +3206,21 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
metadataProps.setProperty(INLINE_COMPACT_NUM_DELTA_COMMITS.key(), "3");
HoodieWriteConfig metadataWriteConfig = HoodieWriteConfig.newBuilder()
.withProperties(metadataProps).build();
- SparkRDDWriteClient metadataWriteClient = new SparkRDDWriteClient(context, metadataWriteConfig, true);
- final String compactionInstantTime = HoodieTableMetadataUtil.createCompactionTimestamp(commitTime);
- assertTrue(metadataWriteClient.scheduleCompactionAtInstant(compactionInstantTime, Option.empty()));
- metadataWriteClient.compact(compactionInstantTime);
+ try (SparkRDDWriteClient metadataWriteClient = new SparkRDDWriteClient(context, metadataWriteConfig, true)) {
+ final String compactionInstantTime = HoodieTableMetadataUtil.createCompactionTimestamp(commitTime);
+ assertTrue(metadataWriteClient.scheduleCompactionAtInstant(compactionInstantTime, Option.empty()));
+ metadataWriteClient.compact(compactionInstantTime);
- // verify metadata table
- validateMetadata(client);
+ // verify metadata table
+ validateMetadata(client);
- // Execute pending clustering operation
- clusteringClient = getHoodieWriteClient(clusterWriteCfg);
- clusteringClient.cluster("0000003", true);
+ // Execute pending clustering operation
+ clusteringClient = getHoodieWriteClient(clusterWriteCfg);
+ clusteringClient.cluster("0000003", true);
- // verify metadata table
- validateMetadata(client);
+ // verify metadata table
+ validateMetadata(client);
+ }
}
@Test
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
index 4802d09a2b9..72690ed8409 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
@@ -2438,6 +2438,7 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
100, dataGen::generateInserts, SparkRDDWriteClient::bulkInsert, false, 100, 300,
0, true);
client.clean();
+ client.close();
HoodieActiveTimeline timeline = metaClient.getActiveTimeline().reload();
if (cleaningPolicy.isLazy()) {
assertTrue(
@@ -2523,6 +2524,7 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
cleaningPolicy = EAGER;
client = new SparkRDDWriteClient(context, getParallelWritingWriteConfig(cleaningPolicy, populateMetaFields));
client.startCommit();
+ client.close();
timeline = metaClient.getActiveTimeline().reload();
// since OCC is enabled, hudi auto flips the cleaningPolicy to Lazy.
assertTrue(timeline.getTimelineOfActions(
@@ -2584,6 +2586,7 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
assertTrue(timeline.getTimelineOfActions(
CollectionUtils.createSet(CLEAN_ACTION)).countInstants() == 0);
assertTrue(timeline.getCommitsTimeline().filterCompletedInstants().countInstants() == 3);
+ client.close();
}
private Pair<Path, JavaRDD<WriteStatus>> testConsistencyCheck(HoodieTableMetaClient metaClient, String instantTime, boolean enableOptimisticConsistencyGuard)
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnMergeOnReadStorage.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnMergeOnReadStorage.java
index 001e56d97bc..90dbcd5ee7e 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnMergeOnReadStorage.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnMergeOnReadStorage.java
@@ -375,57 +375,58 @@ public class TestHoodieClientOnMergeOnReadStorage extends HoodieClientTestBase {
.build();
HoodieWriteConfig lcConfig = getConfigBuilder(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA, HoodieIndex.IndexType.INMEMORY)
.withAutoCommit(false).withCompactionConfig(compactionConfig).build();
- SparkRDDWriteClient lcClient = new SparkRDDWriteClient(context, lcConfig);
HoodieWriteConfig config = getConfigBuilder(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA, HoodieIndex.IndexType.INMEMORY)
.withAutoCommit(true).build();
- SparkRDDWriteClient client = new SparkRDDWriteClient(context, config);
+ try (SparkRDDWriteClient lcClient = new SparkRDDWriteClient(context, lcConfig);
+ SparkRDDWriteClient client = new SparkRDDWriteClient(context, config)) {
- // First insert
- String newCommitTime = HoodieActiveTimeline.createNewInstantTime();
- insertBatch(config, client, newCommitTime, "000", 100,
- SparkRDDWriteClient::insert, false, false, 100, 100,
- 1, Option.empty());
- String prevCommitTime = newCommitTime;
+ // First insert
+ String newCommitTime = HoodieActiveTimeline.createNewInstantTime();
+ insertBatch(config, client, newCommitTime, "000", 100,
+ SparkRDDWriteClient::insert, false, false, 100, 100,
+ 1, Option.empty());
+ String prevCommitTime = newCommitTime;
- // Upsert
- newCommitTime = HoodieActiveTimeline.createNewInstantTime();
- updateBatch(config, client, newCommitTime, prevCommitTime,
- Option.of(Arrays.asList(prevCommitTime)), "000", 10, SparkRDDWriteClient::upsert,
- false, false, 10, 100, 4, config.populateMetaFields());
- prevCommitTime = newCommitTime;
+ // Upsert
+ newCommitTime = HoodieActiveTimeline.createNewInstantTime();
+ updateBatch(config, client, newCommitTime, prevCommitTime,
+ Option.of(Arrays.asList(prevCommitTime)), "000", 10, SparkRDDWriteClient::upsert,
+ false, false, 10, 100, 4, config.populateMetaFields());
+ prevCommitTime = newCommitTime;
- // Schedule and execute log-compaction but do not commit.
- Option<String> logCompactionTimeStamp = lcClient.scheduleLogCompaction(Option.empty());
- assertTrue(logCompactionTimeStamp.isPresent());
- lcClient.logCompact(logCompactionTimeStamp.get());
+ // Schedule and execute log-compaction but do not commit.
+ Option<String> logCompactionTimeStamp = lcClient.scheduleLogCompaction(Option.empty());
+ assertTrue(logCompactionTimeStamp.isPresent());
+ lcClient.logCompact(logCompactionTimeStamp.get());
- // Rollback the log compaction commit.
- HoodieInstant instant = new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.LOG_COMPACTION_ACTION, logCompactionTimeStamp.get());
- getHoodieTable(metaClient, config).rollbackInflightLogCompaction(instant);
+ // Rollback the log compaction commit.
+ HoodieInstant instant = new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.LOG_COMPACTION_ACTION, logCompactionTimeStamp.get());
+ getHoodieTable(metaClient, config).rollbackInflightLogCompaction(instant);
- // Validate timeline.
- HoodieTimeline activeTimeline = metaClient.reloadActiveTimeline();
- HoodieInstant rollbackInstant = activeTimeline.lastInstant().get();
- assertEquals(3, activeTimeline.countInstants());
- assertEquals(HoodieTimeline.ROLLBACK_ACTION, rollbackInstant.getAction());
+ // Validate timeline.
+ HoodieTimeline activeTimeline = metaClient.reloadActiveTimeline();
+ HoodieInstant rollbackInstant = activeTimeline.lastInstant().get();
+ assertEquals(3, activeTimeline.countInstants());
+ assertEquals(HoodieTimeline.ROLLBACK_ACTION, rollbackInstant.getAction());
- // Validate block instant times.
- validateBlockInstantsBeforeAndAfterRollback(config, prevCommitTime, rollbackInstant.getTimestamp());
- prevCommitTime = rollbackInstant.getTimestamp();
+ // Validate block instant times.
+ validateBlockInstantsBeforeAndAfterRollback(config, prevCommitTime, rollbackInstant.getTimestamp());
+ prevCommitTime = rollbackInstant.getTimestamp();
- // Do one more upsert
- newCommitTime = HoodieActiveTimeline.createNewInstantTime();
- updateBatch(config, client, newCommitTime, prevCommitTime,
- Option.of(Arrays.asList(prevCommitTime)), "000", 10, SparkRDDWriteClient::upsert,
- false, false, 10, 100, 4, config.populateMetaFields());
- prevCommitTime = newCommitTime;
+ // Do one more upsert
+ newCommitTime = HoodieActiveTimeline.createNewInstantTime();
+ updateBatch(config, client, newCommitTime, prevCommitTime,
+ Option.of(Arrays.asList(prevCommitTime)), "000", 10, SparkRDDWriteClient::upsert,
+ false, false, 10, 100, 4, config.populateMetaFields());
+ prevCommitTime = newCommitTime;
- // Complete log-compaction now.
- logCompactionTimeStamp = lcClient.scheduleLogCompaction(Option.empty());
- assertTrue(logCompactionTimeStamp.isPresent());
- HoodieWriteMetadata metadata = lcClient.logCompact(logCompactionTimeStamp.get());
- lcClient.commitLogCompaction(logCompactionTimeStamp.get(), (HoodieCommitMetadata) metadata.getCommitMetadata().get(), Option.empty());
- assertDataInMORTable(config, prevCommitTime, logCompactionTimeStamp.get(), hadoopConf, Arrays.asList(dataGen.getPartitionPaths()));
+ // Complete log-compaction now.
+ logCompactionTimeStamp = lcClient.scheduleLogCompaction(Option.empty());
+ assertTrue(logCompactionTimeStamp.isPresent());
+ HoodieWriteMetadata metadata = lcClient.logCompact(logCompactionTimeStamp.get());
+ lcClient.commitLogCompaction(logCompactionTimeStamp.get(), (HoodieCommitMetadata) metadata.getCommitMetadata().get(), Option.empty());
+ assertDataInMORTable(config, prevCommitTime, logCompactionTimeStamp.get(), hadoopConf, Arrays.asList(dataGen.getPartitionPaths()));
+ }
}
private void validateBlockInstantsBeforeAndAfterRollback(HoodieWriteConfig config, String instant, String currentInstant) {
@@ -473,7 +474,6 @@ public class TestHoodieClientOnMergeOnReadStorage extends HoodieClientTestBase {
.build();
HoodieWriteConfig lcWriteConfig = getConfigBuilder(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA,
HoodieIndex.IndexType.INMEMORY).withAutoCommit(true).withCompactionConfig(logCompactionConfig).build();
- SparkRDDWriteClient lcWriteClient = new SparkRDDWriteClient(context, lcWriteConfig);
HoodieCompactionConfig compactionConfig = HoodieCompactionConfig.newBuilder()
.withMaxNumDeltaCommitsBeforeCompaction(1)
@@ -484,51 +484,54 @@ public class TestHoodieClientOnMergeOnReadStorage extends HoodieClientTestBase {
.withArchivalConfig(HoodieArchivalConfig.newBuilder().archiveCommitsWith(4, 5).build())
.withMetadataConfig(HoodieMetadataConfig.newBuilder().withMaxNumDeltaCommitsBeforeCompaction(2).build())
.build();
- SparkRDDWriteClient client = new SparkRDDWriteClient(context, config);
- // First insert
- String newCommitTime = HoodieActiveTimeline.createNewInstantTime();
- insertBatch(config, client, newCommitTime, "000", 100,
- SparkRDDWriteClient::insert, false, false, 10, 100,
- 1, Option.empty());
- String prevCommitTime = newCommitTime;
- List<String> logCompactionInstantTimes = new ArrayList<>();
-
- for (int i = 0; i < 6; i++) {
- if (i % 4 == 0) {
- // Schedule compaction.
- Option<String> compactionTimeStamp = client.scheduleCompaction(Option.empty());
- assertTrue(compactionTimeStamp.isPresent());
- client.compact(compactionTimeStamp.get());
- prevCommitTime = compactionTimeStamp.get();
+ try (SparkRDDWriteClient lcWriteClient = new SparkRDDWriteClient(context, lcWriteConfig);
+ SparkRDDWriteClient client = new SparkRDDWriteClient(context, config)) {
+
+ // First insert
+ String newCommitTime = HoodieActiveTimeline.createNewInstantTime();
+ insertBatch(config, client, newCommitTime, "000", 100,
+ SparkRDDWriteClient::insert, false, false, 10, 100,
+ 1, Option.empty());
+ String prevCommitTime = newCommitTime;
+ List<String> logCompactionInstantTimes = new ArrayList<>();
+
+ for (int i = 0; i < 6; i++) {
+ if (i % 4 == 0) {
+ // Schedule compaction.
+ Option<String> compactionTimeStamp = client.scheduleCompaction(Option.empty());
+ assertTrue(compactionTimeStamp.isPresent());
+ client.compact(compactionTimeStamp.get());
+ prevCommitTime = compactionTimeStamp.get();
+ }
+
+ // Upsert
+ newCommitTime = HoodieActiveTimeline.createNewInstantTime();
+ updateBatch(config, client, newCommitTime, prevCommitTime,
+ Option.of(Arrays.asList(prevCommitTime)), "000", 50, SparkRDDWriteClient::upsert,
+ false, false, 50, 10, 0, config.populateMetaFields());
+ // Schedule log compaction.
+ Option<String> logCompactionTimeStamp = lcWriteClient.scheduleLogCompaction(Option.empty());
+ if (logCompactionTimeStamp.isPresent()) {
+ logCompactionInstantTimes.add(logCompactionTimeStamp.get());
+ lcWriteClient.logCompact(logCompactionTimeStamp.get());
+ prevCommitTime = logCompactionTimeStamp.get();
+ }
}
-
- // Upsert
- newCommitTime = HoodieActiveTimeline.createNewInstantTime();
- updateBatch(config, client, newCommitTime, prevCommitTime,
- Option.of(Arrays.asList(prevCommitTime)), "000", 50, SparkRDDWriteClient::upsert,
- false, false, 50, 10, 0, config.populateMetaFields());
- // Schedule log compaction.
- Option<String> logCompactionTimeStamp = lcWriteClient.scheduleLogCompaction(Option.empty());
- if (logCompactionTimeStamp.isPresent()) {
- logCompactionInstantTimes.add(logCompactionTimeStamp.get());
- lcWriteClient.logCompact(logCompactionTimeStamp.get());
- prevCommitTime = logCompactionTimeStamp.get();
- }
- }
- boolean logCompactionInstantArchived = false;
- Map<String, List<HoodieInstant>> instantsMap = metaClient.getArchivedTimeline().getInstantsAsStream()
- .collect(Collectors.groupingBy(HoodieInstant::getTimestamp));
- for (String logCompactionTimeStamp: logCompactionInstantTimes) {
- List<HoodieInstant> instants = instantsMap.get(logCompactionTimeStamp);
- if (instants == null) {
- continue;
+ boolean logCompactionInstantArchived = false;
+ Map<String, List<HoodieInstant>> instantsMap = metaClient.getArchivedTimeline().getInstantsAsStream()
+ .collect(Collectors.groupingBy(HoodieInstant::getTimestamp));
+ for (String logCompactionTimeStamp : logCompactionInstantTimes) {
+ List<HoodieInstant> instants = instantsMap.get(logCompactionTimeStamp);
+ if (instants == null) {
+ continue;
+ }
+ assertEquals(1, instants.size());
+ assertEquals(HoodieTimeline.DELTA_COMMIT_ACTION, instants.get(0).getAction());
+ logCompactionInstantArchived = true;
}
- assertEquals(1, instants.size());
- assertEquals(HoodieTimeline.DELTA_COMMIT_ACTION, instants.get(0).getAction());
- logCompactionInstantArchived = true;
+ assertTrue(logCompactionInstantArchived);
}
- assertTrue(logCompactionInstantArchived);
}
@Override
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/hbase/TestSparkHoodieHBaseIndex.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/hbase/TestSparkHoodieHBaseIndex.java
index be663d05bfe..6767e38a543 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/hbase/TestSparkHoodieHBaseIndex.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/hbase/TestSparkHoodieHBaseIndex.java
@@ -284,33 +284,34 @@ public class TestSparkHoodieHBaseIndex extends SparkClientFunctionalTestHarness
// Load to memory
HoodieWriteConfig config = getConfig();
SparkHoodieHBaseIndex index = new SparkHoodieHBaseIndex(config);
- SparkRDDWriteClient writeClient = getHoodieWriteClient(config);
- writeClient.startCommitWithTime(newCommitTime);
- metaClient = HoodieTableMetaClient.reload(metaClient);
- HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(config)) {
+ writeClient.startCommitWithTime(newCommitTime);
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
- JavaRDD<WriteStatus> writeStatues = writeClient.upsert(writeRecords, newCommitTime);
- tagLocation(index, writeRecords, hoodieTable);
+ JavaRDD<WriteStatus> writeStatues = writeClient.upsert(writeRecords, newCommitTime);
+ tagLocation(index, writeRecords, hoodieTable);
- // Duplicate upsert and ensure correctness is maintained
- // We are trying to approximately imitate the case when the RDD is recomputed. For RDD creating, driver code is not
- // recomputed. This includes the state transitions. We need to delete the inflight instance so that subsequent
- // upsert will not run into conflicts.
- metaClient.getFs().delete(new Path(metaClient.getMetaPath(), "001.inflight"));
+ // Duplicate upsert and ensure correctness is maintained
+ // We are trying to approximately imitate the case when the RDD is recomputed. For RDD creating, driver code is not
+ // recomputed. This includes the state transitions. We need to delete the inflight instance so that subsequent
+ // upsert will not run into conflicts.
+ metaClient.getFs().delete(new Path(metaClient.getMetaPath(), "001.inflight"));
- writeClient.upsert(writeRecords, newCommitTime);
- assertNoWriteErrors(writeStatues.collect());
+ writeClient.upsert(writeRecords, newCommitTime);
+ assertNoWriteErrors(writeStatues.collect());
- // Now commit this & update location of records inserted and validate no errors
- writeClient.commit(newCommitTime, writeStatues);
- // Now tagLocation for these records, hbaseIndex should tag them correctly
- metaClient = HoodieTableMetaClient.reload(metaClient);
- hoodieTable = HoodieSparkTable.create(config, context, metaClient);
- List<HoodieRecord> taggedRecords = tagLocation(index, writeRecords, hoodieTable).collect();
- assertEquals(numRecords, taggedRecords.stream().filter(HoodieRecord::isCurrentLocationKnown).count());
- assertEquals(numRecords, taggedRecords.stream().map(record -> record.getKey().getRecordKey()).distinct().count());
- assertEquals(numRecords, taggedRecords.stream().filter(record -> (record.getCurrentLocation() != null
- && record.getCurrentLocation().getInstantTime().equals(newCommitTime))).distinct().count());
+ // Now commit this & update location of records inserted and validate no errors
+ writeClient.commit(newCommitTime, writeStatues);
+ // Now tagLocation for these records, hbaseIndex should tag them correctly
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ hoodieTable = HoodieSparkTable.create(config, context, metaClient);
+ List<HoodieRecord> taggedRecords = tagLocation(index, writeRecords, hoodieTable).collect();
+ assertEquals(numRecords, taggedRecords.stream().filter(HoodieRecord::isCurrentLocationKnown).count());
+ assertEquals(numRecords, taggedRecords.stream().map(record -> record.getKey().getRecordKey()).distinct().count());
+ assertEquals(numRecords, taggedRecords.stream().filter(record -> (record.getCurrentLocation() != null
+ && record.getCurrentLocation().getInstantTime().equals(newCommitTime))).distinct().count());
+ }
}
@Disabled("HUDI-6460")
@@ -379,41 +380,42 @@ public class TestSparkHoodieHBaseIndex extends SparkClientFunctionalTestHarness
HoodieWriteConfig config = getConfigBuilder(100, false, false)
.withRollbackUsingMarkers(false).build();
SparkHoodieHBaseIndex index = new SparkHoodieHBaseIndex(config);
- SparkRDDWriteClient writeClient = getHoodieWriteClient(config);
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(config)) {
- final String newCommitTime = writeClient.startCommit();
- final int numRecords = 10;
- List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, numRecords);
- JavaRDD<HoodieRecord> writeRecords = jsc().parallelize(records, 1);
- metaClient = HoodieTableMetaClient.reload(metaClient);
+ final String newCommitTime = writeClient.startCommit();
+ final int numRecords = 10;
+ List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, numRecords);
+ JavaRDD<HoodieRecord> writeRecords = jsc().parallelize(records, 1);
+ metaClient = HoodieTableMetaClient.reload(metaClient);
- // Insert 200 records
- JavaRDD<WriteStatus> writeStatues = writeClient.upsert(writeRecords, newCommitTime);
- assertNoWriteErrors(writeStatues.collect());
+ // Insert 200 records
+ JavaRDD<WriteStatus> writeStatues = writeClient.upsert(writeRecords, newCommitTime);
+ assertNoWriteErrors(writeStatues.collect());
- // commit this upsert
- writeClient.commit(newCommitTime, writeStatues);
- HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
- // Now tagLocation for these records, hbaseIndex should tag them
- List<HoodieRecord> records2 = tagLocation(index, writeRecords, hoodieTable).collect();
- assertEquals(numRecords, records2.stream().filter(HoodieRecord::isCurrentLocationKnown).count());
-
- // check tagged records are tagged with correct fileIds
- List<String> fileIds = writeStatues.map(WriteStatus::getFileId).collect();
- assertEquals(0, records2.stream().filter(record -> record.getCurrentLocation().getFileId() == null).count());
- List<String> taggedFileIds = records2.stream().map(record -> record.getCurrentLocation().getFileId()).distinct().collect(Collectors.toList());
-
- // both lists should match
- assertTrue(taggedFileIds.containsAll(fileIds) && fileIds.containsAll(taggedFileIds));
- // Rollback the last commit
- writeClient.rollback(newCommitTime);
-
- hoodieTable = HoodieSparkTable.create(config, context, metaClient);
- // Now tagLocation for these records, hbaseIndex should not tag them since it was a rolled
- // back commit
- List<HoodieRecord> records3 = tagLocation(index, writeRecords, hoodieTable).collect();
- assertEquals(0, records3.stream().filter(HoodieRecord::isCurrentLocationKnown).count());
- assertEquals(0, records3.stream().filter(record -> record.getCurrentLocation() != null).count());
+ // commit this upsert
+ writeClient.commit(newCommitTime, writeStatues);
+ HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
+ // Now tagLocation for these records, hbaseIndex should tag them
+ List<HoodieRecord> records2 = tagLocation(index, writeRecords, hoodieTable).collect();
+ assertEquals(numRecords, records2.stream().filter(HoodieRecord::isCurrentLocationKnown).count());
+
+ // check tagged records are tagged with correct fileIds
+ List<String> fileIds = writeStatues.map(WriteStatus::getFileId).collect();
+ assertEquals(0, records2.stream().filter(record -> record.getCurrentLocation().getFileId() == null).count());
+ List<String> taggedFileIds = records2.stream().map(record -> record.getCurrentLocation().getFileId()).distinct().collect(Collectors.toList());
+
+ // both lists should match
+ assertTrue(taggedFileIds.containsAll(fileIds) && fileIds.containsAll(taggedFileIds));
+ // Rollback the last commit
+ writeClient.rollback(newCommitTime);
+
+ hoodieTable = HoodieSparkTable.create(config, context, metaClient);
+ // Now tagLocation for these records, hbaseIndex should not tag them since it was a rolled
+ // back commit
+ List<HoodieRecord> records3 = tagLocation(index, writeRecords, hoodieTable).collect();
+ assertEquals(0, records3.stream().filter(HoodieRecord::isCurrentLocationKnown).count());
+ assertEquals(0, records3.stream().filter(record -> record.getCurrentLocation() != null).count());
+ }
}
/*
@@ -425,36 +427,37 @@ public class TestSparkHoodieHBaseIndex extends SparkClientFunctionalTestHarness
// Load to memory
HoodieWriteConfig config = getConfigBuilder(100, false, false).withRollbackUsingMarkers(false).build();
SparkHoodieHBaseIndex index = new SparkHoodieHBaseIndex(config);
- SparkRDDWriteClient writeClient = getHoodieWriteClient(config);
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(config)) {
- String newCommitTime = writeClient.startCommit();
- // make a commit with 199 records
- JavaRDD<HoodieRecord> writeRecords = generateAndCommitRecords(writeClient, 199, newCommitTime);
+ String newCommitTime = writeClient.startCommit();
+ // make a commit with 199 records
+ JavaRDD<HoodieRecord> writeRecords = generateAndCommitRecords(writeClient, 199, newCommitTime);
- // make a second commit with a single record
- String invalidCommit = writeClient.startCommit();
- JavaRDD<HoodieRecord> invalidWriteRecords = generateAndCommitRecords(writeClient, 1, invalidCommit);
+ // make a second commit with a single record
+ String invalidCommit = writeClient.startCommit();
+ JavaRDD<HoodieRecord> invalidWriteRecords = generateAndCommitRecords(writeClient, 1, invalidCommit);
- // verify location is tagged.
- HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
- JavaRDD<HoodieRecord> javaRDD0 = tagLocation(index, invalidWriteRecords, hoodieTable);
- assert (javaRDD0.collect().size() == 1); // one record present
- assert (javaRDD0.filter(HoodieRecord::isCurrentLocationKnown).collect().size() == 1); // it is tagged
- assert (javaRDD0.collect().get(0).getCurrentLocation().getInstantTime().equals(invalidCommit));
+ // verify location is tagged.
+ HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
+ JavaRDD<HoodieRecord> javaRDD0 = tagLocation(index, invalidWriteRecords, hoodieTable);
+ assert (javaRDD0.collect().size() == 1); // one record present
+ assert (javaRDD0.filter(HoodieRecord::isCurrentLocationKnown).collect().size() == 1); // it is tagged
+ assert (javaRDD0.collect().get(0).getCurrentLocation().getInstantTime().equals(invalidCommit));
- // rollback the invalid commit, so that hbase will be left with a stale entry.
- writeClient.rollback(invalidCommit);
+ // rollback the invalid commit, so that hbase will be left with a stale entry.
+ writeClient.rollback(invalidCommit);
- // Now tagLocation for the valid records, hbaseIndex should tag them
- metaClient = HoodieTableMetaClient.reload(metaClient);
- hoodieTable = HoodieSparkTable.create(config, context, metaClient);
- JavaRDD<HoodieRecord> javaRDD1 = tagLocation(index, writeRecords, hoodieTable);
- assert (javaRDD1.filter(HoodieRecord::isCurrentLocationKnown).collect().size() == 199);
-
- // tagLocation for the invalid record - commit is not present in timeline due to rollback.
- JavaRDD<HoodieRecord> javaRDD2 = tagLocation(index, invalidWriteRecords, hoodieTable);
- assert (javaRDD2.collect().size() == 1); // one record present
- assert (javaRDD2.filter(HoodieRecord::isCurrentLocationKnown).collect().size() == 0); // it is not tagged
+ // Now tagLocation for the valid records, hbaseIndex should tag them
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ hoodieTable = HoodieSparkTable.create(config, context, metaClient);
+ JavaRDD<HoodieRecord> javaRDD1 = tagLocation(index, writeRecords, hoodieTable);
+ assert (javaRDD1.filter(HoodieRecord::isCurrentLocationKnown).collect().size() == 199);
+
+ // tagLocation for the invalid record - commit is not present in timeline due to rollback.
+ JavaRDD<HoodieRecord> javaRDD2 = tagLocation(index, invalidWriteRecords, hoodieTable);
+ assert (javaRDD2.collect().size() == 1); // one record present
+ assert (javaRDD2.filter(HoodieRecord::isCurrentLocationKnown).collect().size() == 0); // it is not tagged
+ }
}
/*
@@ -467,23 +470,24 @@ public class TestSparkHoodieHBaseIndex extends SparkClientFunctionalTestHarness
HoodieWriteConfig config = getConfigBuilder(100, false, false)
.withRollbackUsingMarkers(false).build();
SparkHoodieHBaseIndex index = new SparkHoodieHBaseIndex(config);
- SparkRDDWriteClient writeClient = getHoodieWriteClient(config);
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(config)) {
- String commitTime1 = writeClient.startCommit();
- JavaRDD<HoodieRecord> writeRecords1 = generateAndCommitRecords(writeClient, 20, commitTime1);
+ String commitTime1 = writeClient.startCommit();
+ JavaRDD<HoodieRecord> writeRecords1 = generateAndCommitRecords(writeClient, 20, commitTime1);
- // rollback the commit - leaves a clean file in timeline.
- writeClient.rollback(commitTime1);
+ // rollback the commit - leaves a clean file in timeline.
+ writeClient.rollback(commitTime1);
- // create a second commit with 20 records
- metaClient = HoodieTableMetaClient.reload(metaClient);
- generateAndCommitRecords(writeClient, 20);
+ // create a second commit with 20 records
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ generateAndCommitRecords(writeClient, 20);
- // Now tagLocation for the first set of rolledback records, hbaseIndex should tag them
- metaClient = HoodieTableMetaClient.reload(metaClient);
- HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
- JavaRDD<HoodieRecord> javaRDD1 = tagLocation(index, writeRecords1, hoodieTable);
- assert (javaRDD1.filter(HoodieRecord::isCurrentLocationKnown).collect().size() == 20);
+ // Now tagLocation for the first set of rolledback records, hbaseIndex should tag them
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
+ JavaRDD<HoodieRecord> javaRDD1 = tagLocation(index, writeRecords1, hoodieTable);
+ assert (javaRDD1.filter(HoodieRecord::isCurrentLocationKnown).collect().size() == 20);
+ }
}
private JavaRDD<HoodieRecord> generateAndCommitRecords(SparkRDDWriteClient writeClient, int numRecs) throws Exception {
@@ -520,24 +524,25 @@ public class TestSparkHoodieHBaseIndex extends SparkClientFunctionalTestHarness
HoodieWriteConfig config = getConfigBuilder(100, false, false).withProps(params).build();
SparkHoodieHBaseIndex index = new SparkHoodieHBaseIndex(config);
- SparkRDDWriteClient writeClient = getHoodieWriteClient(config);
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(config)) {
- // make first commit with 20 records
- JavaRDD<HoodieRecord> writeRecords1 = generateAndCommitRecords(writeClient, 20);
- metaClient = HoodieTableMetaClient.reload(metaClient);
- String commit1 = metaClient.getActiveTimeline().firstInstant().get().getTimestamp();
+ // make first commit with 20 records
+ JavaRDD<HoodieRecord> writeRecords1 = generateAndCommitRecords(writeClient, 20);
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ String commit1 = metaClient.getActiveTimeline().firstInstant().get().getTimestamp();
- // Make 6 additional commits, so that first commit is archived
- for (int nCommit = 0; nCommit < 6; nCommit++) {
- generateAndCommitRecords(writeClient, 20);
- }
+ // Make 6 additional commits, so that first commit is archived
+ for (int nCommit = 0; nCommit < 6; nCommit++) {
+ generateAndCommitRecords(writeClient, 20);
+ }
- // tagLocation for the first set of records (for the archived commit), hbaseIndex should tag them as valid
- metaClient = HoodieTableMetaClient.reload(metaClient);
- assertTrue(metaClient.getArchivedTimeline().containsInstant(commit1));
- HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
- JavaRDD<HoodieRecord> javaRDD1 = tagLocation(index, writeRecords1, hoodieTable);
- assertEquals(20, javaRDD1.filter(HoodieRecord::isCurrentLocationKnown).collect().size());
+ // tagLocation for the first set of records (for the archived commit), hbaseIndex should tag them as valid
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ assertTrue(metaClient.getArchivedTimeline().containsInstant(commit1));
+ HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
+ JavaRDD<HoodieRecord> javaRDD1 = tagLocation(index, writeRecords1, hoodieTable);
+ assertEquals(20, javaRDD1.filter(HoodieRecord::isCurrentLocationKnown).collect().size());
+ }
}
@Test
@@ -554,62 +559,63 @@ public class TestSparkHoodieHBaseIndex extends SparkClientFunctionalTestHarness
// only for test, set the hbaseConnection to mocked object
index.setHbaseConnection(hbaseConnection);
- SparkRDDWriteClient writeClient = getHoodieWriteClient(config);
-
- // start a commit and generate test data
- String newCommitTime = writeClient.startCommit();
- List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 250);
- JavaRDD<HoodieRecord> writeRecords = jsc().parallelize(records, 1);
- metaClient = HoodieTableMetaClient.reload(metaClient);
- HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(config)) {
- // Insert 250 records
- JavaRDD<WriteStatus> writeStatues = writeClient.upsert(writeRecords, newCommitTime);
- assertNoWriteErrors(writeStatues.collect());
+ // start a commit and generate test data
+ String newCommitTime = writeClient.startCommit();
+ List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 250);
+ JavaRDD<HoodieRecord> writeRecords = jsc().parallelize(records, 1);
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
- // Now tagLocation for these records, hbaseIndex should tag them
- tagLocation(index, writeRecords, hoodieTable);
+ // Insert 250 records
+ JavaRDD<WriteStatus> writeStatues = writeClient.upsert(writeRecords, newCommitTime);
+ assertNoWriteErrors(writeStatues.collect());
- // 3 batches should be executed given batchSize = 100 and parallelism = 1
- verify(table, times(3)).get((List<Get>) any());
+ // Now tagLocation for these records, hbaseIndex should tag them
+ tagLocation(index, writeRecords, hoodieTable);
+ // 3 batches should be executed given batchSize = 100 and parallelism = 1
+ verify(table, times(3)).get((List<Get>) any());
+ }
}
@Test
public void testTotalPutsBatching() throws Exception {
HoodieWriteConfig config = getConfig();
SparkHoodieHBaseIndex index = new SparkHoodieHBaseIndex(config);
- SparkRDDWriteClient writeClient = getHoodieWriteClient(config);
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(config)) {
- // start a commit and generate test data
- String newCommitTime = writeClient.startCommit();
- List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 250);
- JavaRDD<HoodieRecord> writeRecords = jsc().parallelize(records, 1);
- metaClient = HoodieTableMetaClient.reload(metaClient);
- HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
+ // start a commit and generate test data
+ String newCommitTime = writeClient.startCommit();
+ List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 250);
+ JavaRDD<HoodieRecord> writeRecords = jsc().parallelize(records, 1);
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
- // Insert 200 records
- JavaRDD<WriteStatus> writeStatues = writeClient.upsert(writeRecords, newCommitTime);
+ // Insert 200 records
+ JavaRDD<WriteStatus> writeStatues = writeClient.upsert(writeRecords, newCommitTime);
- // commit this upsert
- writeClient.commit(newCommitTime, writeStatues);
+ // commit this upsert
+ writeClient.commit(newCommitTime, writeStatues);
- // Mock hbaseConnection and related entities
- Connection hbaseConnection = mock(Connection.class);
- HTable table = mock(HTable.class);
- when(hbaseConnection.getTable(TableName.valueOf(TABLE_NAME))).thenReturn(table);
- when(table.get((List<Get>) any())).thenReturn(new Result[0]);
+ // Mock hbaseConnection and related entities
+ Connection hbaseConnection = mock(Connection.class);
+ HTable table = mock(HTable.class);
+ when(hbaseConnection.getTable(TableName.valueOf(TABLE_NAME))).thenReturn(table);
+ when(table.get((List<Get>) any())).thenReturn(new Result[0]);
- // only for test, set the hbaseConnection to mocked object
- index.setHbaseConnection(hbaseConnection);
+ // only for test, set the hbaseConnection to mocked object
+ index.setHbaseConnection(hbaseConnection);
- // Get all the files generated
- int numberOfDataFileIds = (int) writeStatues.map(status -> status.getFileId()).distinct().count();
+ // Get all the files generated
+ int numberOfDataFileIds = (int) writeStatues.map(status -> status.getFileId()).distinct().count();
- updateLocation(index, writeStatues, hoodieTable);
- // 3 batches should be executed given batchSize = 100 and <=numberOfDataFileIds getting updated,
- // so each fileId ideally gets updates
- verify(table, atMost(numberOfDataFileIds)).put((List<Put>) any());
+ updateLocation(index, writeStatues, hoodieTable);
+ // 3 batches should be executed given batchSize = 100 and <=numberOfDataFileIds getting updated,
+ // so each fileId ideally gets updates
+ verify(table, atMost(numberOfDataFileIds)).put((List<Put>) any());
+ }
}
@Test
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableCompaction.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableCompaction.java
index 4676abbbe8c..d145958a057 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableCompaction.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableCompaction.java
@@ -44,6 +44,7 @@ import org.apache.hudi.testutils.HoodieMergeOnReadTestUtils;
import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
import org.apache.spark.api.java.JavaRDD;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
@@ -90,6 +91,13 @@ public class TestHoodieSparkMergeOnReadTableCompaction extends SparkClientFuncti
dataGen = new HoodieTestDataGenerator();
}
+ @AfterEach
+ public void teardown() throws IOException {
+ if (client != null) {
+ client.close();
+ }
+ }
+
@ParameterizedTest
@MethodSource("writePayloadTest")
public void testWriteDuringCompaction(String payloadClass) throws IOException {
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableInsertUpdateDelete.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableInsertUpdateDelete.java
index 46e5e9eb24b..73d551b0ae0 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableInsertUpdateDelete.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableInsertUpdateDelete.java
@@ -226,13 +226,14 @@ public class TestHoodieSparkMergeOnReadTableInsertUpdateDelete extends SparkClie
FileCreateUtils.deleteRollbackCommit(metaClient.getBasePath().substring(metaClient.getBasePath().indexOf(":") + 1),
rollbackInstant.getTimestamp());
metaClient.reloadActiveTimeline();
- SparkRDDWriteClient client1 = getHoodieWriteClient(cfg);
- // trigger compaction again.
- client1.compact(compactionInstant.get());
- metaClient.reloadActiveTimeline();
- // verify that there is no new rollback instant generated
- HoodieInstant newRollbackInstant = metaClient.getActiveTimeline().getRollbackTimeline().lastInstant().get();
- assertEquals(rollbackInstant.getTimestamp(), newRollbackInstant.getTimestamp());
+ try (SparkRDDWriteClient client1 = getHoodieWriteClient(cfg)) {
+ // trigger compaction again.
+ client1.compact(compactionInstant.get());
+ metaClient.reloadActiveTimeline();
+ // verify that there is no new rollback instant generated
+ HoodieInstant newRollbackInstant = metaClient.getActiveTimeline().getRollbackTimeline().lastInstant().get();
+ assertEquals(rollbackInstant.getTimestamp(), newRollbackInstant.getTimestamp());
+ }
}
}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java
index de8c218c1a8..e492682fef3 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java
@@ -560,9 +560,9 @@ public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunction
.withMarkersType(MarkerType.DIRECT.name());
addConfigsForPopulateMetaFields(cfgBuilder, populateMetaFields);
HoodieWriteConfig cfg1 = cfgBuilder.build();
- final SparkRDDWriteClient client1 = getHoodieWriteClient(cfg1);
- client1.clean();
- client1.close();
+ try (final SparkRDDWriteClient client1 = getHoodieWriteClient(cfg1)) {
+ client1.clean();
+ }
metaClient = HoodieTableMetaClient.reload(metaClient);
upsertRecords(client, "011", records, dataGen);
@@ -882,31 +882,32 @@ public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunction
HoodieWriteConfig autoCommitFalseCfg = getWriteConfig(false, rollbackUsingMarkers);
HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator();
- SparkRDDWriteClient client = getHoodieWriteClient(cfg);
- // commit 1
- List<HoodieRecord> records = insertRecords(client, dataGen, "001");
- // commit 2 to create log files
- List<HoodieRecord> updates1 = updateRecords(client, dataGen, "002", records, metaClient, cfg, true);
-
- // trigger a inflight commit 3 which will be later be rolled back explicitly.
- SparkRDDWriteClient autoCommitFalseClient = getHoodieWriteClient(autoCommitFalseCfg);
- List<HoodieRecord> updates2 = updateRecords(autoCommitFalseClient, dataGen, "003", records, metaClient, autoCommitFalseCfg, false);
-
- // commit 4 successful (mimic multi-writer scenario)
- List<HoodieRecord> updates3 = updateRecords(client, dataGen, "004", records, metaClient, cfg, false);
-
- // trigger compaction
- long numLogFiles = getNumLogFilesInLatestFileSlice(metaClient, cfg, dataGen);
- doCompaction(autoCommitFalseClient, metaClient, cfg, numLogFiles);
- long numLogFilesAfterCompaction = getNumLogFilesInLatestFileSlice(metaClient, cfg, dataGen);
- assertNotEquals(numLogFiles, numLogFilesAfterCompaction);
-
- // rollback 3rd commit.
- client.rollback("003");
- long numLogFilesAfterRollback = getNumLogFilesInLatestFileSlice(metaClient, cfg, dataGen);
- // lazy rollback should have added the rollback block to previous file slice and not the latest. And so the latest slice's log file count should
- // remain the same.
- assertEquals(numLogFilesAfterRollback, numLogFilesAfterCompaction);
+ try (SparkRDDWriteClient client = getHoodieWriteClient(cfg);
+ SparkRDDWriteClient autoCommitFalseClient = getHoodieWriteClient(autoCommitFalseCfg)) {
+ // commit 1
+ List<HoodieRecord> records = insertRecords(client, dataGen, "001");
+ // commit 2 to create log files
+ List<HoodieRecord> updates1 = updateRecords(client, dataGen, "002", records, metaClient, cfg, true);
+
+ // trigger a inflight commit 3 which will be later be rolled back explicitly.
+ List<HoodieRecord> updates2 = updateRecords(autoCommitFalseClient, dataGen, "003", records, metaClient, autoCommitFalseCfg, false);
+
+ // commit 4 successful (mimic multi-writer scenario)
+ List<HoodieRecord> updates3 = updateRecords(client, dataGen, "004", records, metaClient, cfg, false);
+
+ // trigger compaction
+ long numLogFiles = getNumLogFilesInLatestFileSlice(metaClient, cfg, dataGen);
+ doCompaction(autoCommitFalseClient, metaClient, cfg, numLogFiles);
+ long numLogFilesAfterCompaction = getNumLogFilesInLatestFileSlice(metaClient, cfg, dataGen);
+ assertNotEquals(numLogFiles, numLogFilesAfterCompaction);
+
+ // rollback 3rd commit.
+ client.rollback("003");
+ long numLogFilesAfterRollback = getNumLogFilesInLatestFileSlice(metaClient, cfg, dataGen);
+ // lazy rollback should have added the rollback block to previous file slice and not the latest. And so the latest slice's log file count should
+ // remain the same.
+ assertEquals(numLogFilesAfterRollback, numLogFilesAfterCompaction);
+ }
}
private List<HoodieRecord> insertRecords(SparkRDDWriteClient client, HoodieTestDataGenerator dataGen, String commitTime) {
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestFlinkConsistentHashingClustering.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestFlinkConsistentHashingClustering.java
index e52fe8b976a..f2684d69809 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestFlinkConsistentHashingClustering.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestFlinkConsistentHashingClustering.java
@@ -84,17 +84,18 @@ public class ITTestFlinkConsistentHashingClustering {
// Manually set the split threshold to trigger split in the clustering
conf.set(FlinkOptions.WRITE_PARQUET_MAX_FILE_SIZE, 1);
conf.setString(HoodieIndexConfig.BUCKET_SPLIT_THRESHOLD.key(), String.valueOf(1 / 1024.0 / 1024.0));
- HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf);
- Option<String> clusteringInstantOption = writeClient.scheduleClustering(Option.empty());
- Assertions.assertTrue(clusteringInstantOption.isPresent());
-
- // Validate clustering plan
- HoodieClusteringPlan clusteringPlan = getLatestClusteringPlan(writeClient);
- Assertions.assertEquals(4, clusteringPlan.getInputGroups().size());
- Assertions.assertEquals(1, clusteringPlan.getInputGroups().get(0).getSlices().size());
- Assertions.assertEquals(1, clusteringPlan.getInputGroups().get(1).getSlices().size());
- Assertions.assertEquals(1, clusteringPlan.getInputGroups().get(2).getSlices().size());
- Assertions.assertEquals(1, clusteringPlan.getInputGroups().get(3).getSlices().size());
+ try (HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf)) {
+ Option<String> clusteringInstantOption = writeClient.scheduleClustering(Option.empty());
+ Assertions.assertTrue(clusteringInstantOption.isPresent());
+
+ // Validate clustering plan
+ HoodieClusteringPlan clusteringPlan = getLatestClusteringPlan(writeClient);
+ Assertions.assertEquals(4, clusteringPlan.getInputGroups().size());
+ Assertions.assertEquals(1, clusteringPlan.getInputGroups().get(0).getSlices().size());
+ Assertions.assertEquals(1, clusteringPlan.getInputGroups().get(1).getSlices().size());
+ Assertions.assertEquals(1, clusteringPlan.getInputGroups().get(2).getSlices().size());
+ Assertions.assertEquals(1, clusteringPlan.getInputGroups().get(3).getSlices().size());
+ }
}
@Test
@@ -103,9 +104,10 @@ public class ITTestFlinkConsistentHashingClustering {
prepareData(tableEnv);
Configuration conf = getDefaultConfiguration();
- HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf);
- Option<String> clusteringInstantOption = writeClient.scheduleClustering(Option.empty());
- Assertions.assertFalse(clusteringInstantOption.isPresent());
+ try (HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf)) {
+ Option<String> clusteringInstantOption = writeClient.scheduleClustering(Option.empty());
+ Assertions.assertFalse(clusteringInstantOption.isPresent());
+ }
}
private HoodieClusteringPlan getLatestClusteringPlan(HoodieFlinkWriteClient writeClient) {
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestHoodieFlinkClustering.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestHoodieFlinkClustering.java
index 4c817a7927a..ec2211f02cf 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestHoodieFlinkClustering.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestHoodieFlinkClustering.java
@@ -157,53 +157,54 @@ public class ITTestHoodieFlinkClustering {
// To compute the clustering instant time and do clustering.
String clusteringInstantTime = HoodieActiveTimeline.createNewInstantTime();
- HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf);
- HoodieFlinkTable<?> table = writeClient.getHoodieTable();
-
- boolean scheduled = writeClient.scheduleClusteringAtInstant(clusteringInstantTime, Option.empty());
-
- assertTrue(scheduled, "The clustering plan should be scheduled");
-
- // fetch the instant based on the configured execution sequence
- table.getMetaClient().reloadActiveTimeline();
- HoodieTimeline timeline = table.getActiveTimeline().filterPendingReplaceTimeline()
- .filter(instant -> instant.getState() == HoodieInstant.State.REQUESTED);
-
- // generate clustering plan
- // should support configurable commit metadata
- Option<Pair<HoodieInstant, HoodieClusteringPlan>> clusteringPlanOption = ClusteringUtils.getClusteringPlan(
- table.getMetaClient(), timeline.lastInstant().get());
-
- HoodieClusteringPlan clusteringPlan = clusteringPlanOption.get().getRight();
-
- // Mark instant as clustering inflight
- HoodieInstant instant = HoodieTimeline.getReplaceCommitRequestedInstant(clusteringInstantTime);
- table.getActiveTimeline().transitionReplaceRequestedToInflight(instant, Option.empty());
-
- final Schema tableAvroSchema = StreamerUtil.getTableAvroSchema(table.getMetaClient(), false);
- final DataType rowDataType = AvroSchemaConverter.convertToDataType(tableAvroSchema);
- final RowType rowType = (RowType) rowDataType.getLogicalType();
-
- DataStream<ClusteringCommitEvent> dataStream = env.addSource(new ClusteringPlanSourceFunction(clusteringInstantTime, clusteringPlan, conf))
- .name("clustering_source")
- .uid("uid_clustering_source")
- .rebalance()
- .transform("clustering_task",
- TypeInformation.of(ClusteringCommitEvent.class),
- new ClusteringOperator(conf, rowType))
- .setParallelism(clusteringPlan.getInputGroups().size());
-
- ExecNodeUtil.setManagedMemoryWeight(dataStream.getTransformation(),
- conf.getInteger(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L);
-
- dataStream
- .addSink(new ClusteringCommitSink(conf))
- .name("clustering_commit")
- .uid("uid_clustering_commit")
- .setParallelism(1);
-
- env.execute("flink_hudi_clustering");
- TestData.checkWrittenData(tempFile, EXPECTED, 4);
+ try (HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf)) {
+ HoodieFlinkTable<?> table = writeClient.getHoodieTable();
+
+ boolean scheduled = writeClient.scheduleClusteringAtInstant(clusteringInstantTime, Option.empty());
+
+ assertTrue(scheduled, "The clustering plan should be scheduled");
+
+ // fetch the instant based on the configured execution sequence
+ table.getMetaClient().reloadActiveTimeline();
+ HoodieTimeline timeline = table.getActiveTimeline().filterPendingReplaceTimeline()
+ .filter(instant -> instant.getState() == HoodieInstant.State.REQUESTED);
+
+ // generate clustering plan
+ // should support configurable commit metadata
+ Option<Pair<HoodieInstant, HoodieClusteringPlan>> clusteringPlanOption = ClusteringUtils.getClusteringPlan(
+ table.getMetaClient(), timeline.lastInstant().get());
+
+ HoodieClusteringPlan clusteringPlan = clusteringPlanOption.get().getRight();
+
+ // Mark instant as clustering inflight
+ HoodieInstant instant = HoodieTimeline.getReplaceCommitRequestedInstant(clusteringInstantTime);
+ table.getActiveTimeline().transitionReplaceRequestedToInflight(instant, Option.empty());
+
+ final Schema tableAvroSchema = StreamerUtil.getTableAvroSchema(table.getMetaClient(), false);
+ final DataType rowDataType = AvroSchemaConverter.convertToDataType(tableAvroSchema);
+ final RowType rowType = (RowType) rowDataType.getLogicalType();
+
+ DataStream<ClusteringCommitEvent> dataStream = env.addSource(new ClusteringPlanSourceFunction(clusteringInstantTime, clusteringPlan, conf))
+ .name("clustering_source")
+ .uid("uid_clustering_source")
+ .rebalance()
+ .transform("clustering_task",
+ TypeInformation.of(ClusteringCommitEvent.class),
+ new ClusteringOperator(conf, rowType))
+ .setParallelism(clusteringPlan.getInputGroups().size());
+
+ ExecNodeUtil.setManagedMemoryWeight(dataStream.getTransformation(),
+ conf.getInteger(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L);
+
+ dataStream
+ .addSink(new ClusteringCommitSink(conf))
+ .name("clustering_commit")
+ .uid("uid_clustering_commit")
+ .setParallelism(1);
+
+ env.execute("flink_hudi_clustering");
+ TestData.checkWrittenData(tempFile, EXPECTED, 4);
+ }
}
@Test
@@ -292,21 +293,22 @@ public class ITTestHoodieFlinkClustering {
// To compute the clustering instant time.
String clusteringInstantTime = HoodieActiveTimeline.createNewInstantTime();
- HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf);
+ try (HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf)) {
- boolean scheduled = writeClient.scheduleClusteringAtInstant(clusteringInstantTime, Option.empty());
+ boolean scheduled = writeClient.scheduleClusteringAtInstant(clusteringInstantTime, Option.empty());
- assertFalse(scheduled, "1 delta commit, the clustering plan should not be scheduled");
+ assertFalse(scheduled, "1 delta commit, the clustering plan should not be scheduled");
- tableEnv.executeSql(TestSQL.INSERT_T1).await();
- // wait for the asynchronous commit to finish
- TimeUnit.SECONDS.sleep(3);
+ tableEnv.executeSql(TestSQL.INSERT_T1).await();
+ // wait for the asynchronous commit to finish
+ TimeUnit.SECONDS.sleep(3);
- clusteringInstantTime = HoodieActiveTimeline.createNewInstantTime();
+ clusteringInstantTime = HoodieActiveTimeline.createNewInstantTime();
- scheduled = writeClient.scheduleClusteringAtInstant(clusteringInstantTime, Option.empty());
+ scheduled = writeClient.scheduleClusteringAtInstant(clusteringInstantTime, Option.empty());
- assertTrue(scheduled, "2 delta commits, the clustering plan should be scheduled");
+ assertTrue(scheduled, "2 delta commits, the clustering plan should be scheduled");
+ }
}
@Test
@@ -365,77 +367,78 @@ public class ITTestHoodieFlinkClustering {
// To compute the clustering instant time and do clustering.
String firstClusteringInstant = HoodieActiveTimeline.createNewInstantTime();
- HoodieFlinkWriteClient<?> writeClient = FlinkWriteClients.createWriteClient(conf);
- HoodieFlinkTable<?> table = writeClient.getHoodieTable();
-
- boolean scheduled = writeClient.scheduleClusteringAtInstant(firstClusteringInstant, Option.empty());
-
- assertTrue(scheduled, "The clustering plan should be scheduled");
-
- // fetch the instant based on the configured execution sequence
- table.getMetaClient().reloadActiveTimeline();
- HoodieTimeline timeline = table.getActiveTimeline().filterPendingReplaceTimeline()
- .filter(i -> i.getState() == HoodieInstant.State.REQUESTED);
-
- // generate clustering plan
- // should support configurable commit metadata
- Option<Pair<HoodieInstant, HoodieClusteringPlan>> clusteringPlanOption = ClusteringUtils.getClusteringPlan(
- table.getMetaClient(), timeline.lastInstant().get());
-
- HoodieClusteringPlan clusteringPlan = clusteringPlanOption.get().getRight();
-
- // Mark instant as clustering inflight
- HoodieInstant instant = HoodieTimeline.getReplaceCommitRequestedInstant(firstClusteringInstant);
- table.getActiveTimeline().transitionReplaceRequestedToInflight(instant, Option.empty());
-
- final Schema tableAvroSchema = StreamerUtil.getTableAvroSchema(table.getMetaClient(), false);
- final DataType rowDataType = AvroSchemaConverter.convertToDataType(tableAvroSchema);
- final RowType rowType = (RowType) rowDataType.getLogicalType();
-
- DataStream<ClusteringCommitEvent> dataStream =
- env.addSource(new ClusteringPlanSourceFunction(firstClusteringInstant, clusteringPlan, conf))
- .name("clustering_source")
- .uid("uid_clustering_source")
- .rebalance()
- .transform(
- "clustering_task",
- TypeInformation.of(ClusteringCommitEvent.class),
- new ClusteringOperator(conf, rowType))
- .setParallelism(clusteringPlan.getInputGroups().size());
-
- ExecNodeUtil.setManagedMemoryWeight(
- dataStream.getTransformation(),
- conf.getInteger(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L);
-
- // keep pending clustering, not committing clustering
- dataStream
- .addSink(new DiscardingSink<>())
- .name("discarding-sink")
- .uid("uid_discarding-sink")
- .setParallelism(1);
-
- env.execute("flink_hudi_clustering");
-
- tableEnv.executeSql(TestSQL.INSERT_T1).await();
- // wait for the asynchronous commit to finish
- TimeUnit.SECONDS.sleep(3);
-
- // archive the first commit, retain the second commit before the inflight replacecommit
- writeClient.archive();
-
- scheduled = writeClient.scheduleClusteringAtInstant(HoodieActiveTimeline.createNewInstantTime(), Option.empty());
-
- assertTrue(scheduled, "The clustering plan should be scheduled");
- table.getMetaClient().reloadActiveTimeline();
- timeline = table.getActiveTimeline().filterPendingReplaceTimeline()
- .filter(i -> i.getState() == HoodieInstant.State.REQUESTED);
-
- HoodieInstant secondClusteringInstant = timeline.lastInstant().get();
- List<HoodieClusteringGroup> inputFileGroups = ClusteringUtils.getClusteringPlan(table.getMetaClient(), secondClusteringInstant).get().getRight().getInputGroups();
- // clustering plan has no previous file slice generated by previous pending clustering
- assertFalse(inputFileGroups
- .stream().anyMatch(fg -> fg.getSlices()
- .stream().anyMatch(s -> s.getDataFilePath().contains(firstClusteringInstant))));
+ try (HoodieFlinkWriteClient<?> writeClient = FlinkWriteClients.createWriteClient(conf)) {
+ HoodieFlinkTable<?> table = writeClient.getHoodieTable();
+
+ boolean scheduled = writeClient.scheduleClusteringAtInstant(firstClusteringInstant, Option.empty());
+
+ assertTrue(scheduled, "The clustering plan should be scheduled");
+
+ // fetch the instant based on the configured execution sequence
+ table.getMetaClient().reloadActiveTimeline();
+ HoodieTimeline timeline = table.getActiveTimeline().filterPendingReplaceTimeline()
+ .filter(i -> i.getState() == HoodieInstant.State.REQUESTED);
+
+ // generate clustering plan
+ // should support configurable commit metadata
+ Option<Pair<HoodieInstant, HoodieClusteringPlan>> clusteringPlanOption = ClusteringUtils.getClusteringPlan(
+ table.getMetaClient(), timeline.lastInstant().get());
+
+ HoodieClusteringPlan clusteringPlan = clusteringPlanOption.get().getRight();
+
+ // Mark instant as clustering inflight
+ HoodieInstant instant = HoodieTimeline.getReplaceCommitRequestedInstant(firstClusteringInstant);
+ table.getActiveTimeline().transitionReplaceRequestedToInflight(instant, Option.empty());
+
+ final Schema tableAvroSchema = StreamerUtil.getTableAvroSchema(table.getMetaClient(), false);
+ final DataType rowDataType = AvroSchemaConverter.convertToDataType(tableAvroSchema);
+ final RowType rowType = (RowType) rowDataType.getLogicalType();
+
+ DataStream<ClusteringCommitEvent> dataStream =
+ env.addSource(new ClusteringPlanSourceFunction(firstClusteringInstant, clusteringPlan, conf))
+ .name("clustering_source")
+ .uid("uid_clustering_source")
+ .rebalance()
+ .transform(
+ "clustering_task",
+ TypeInformation.of(ClusteringCommitEvent.class),
+ new ClusteringOperator(conf, rowType))
+ .setParallelism(clusteringPlan.getInputGroups().size());
+
+ ExecNodeUtil.setManagedMemoryWeight(
+ dataStream.getTransformation(),
+ conf.getInteger(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L);
+
+ // keep pending clustering, not committing clustering
+ dataStream
+ .addSink(new DiscardingSink<>())
+ .name("discarding-sink")
+ .uid("uid_discarding-sink")
+ .setParallelism(1);
+
+ env.execute("flink_hudi_clustering");
+
+ tableEnv.executeSql(TestSQL.INSERT_T1).await();
+ // wait for the asynchronous commit to finish
+ TimeUnit.SECONDS.sleep(3);
+
+ // archive the first commit, retain the second commit before the inflight replacecommit
+ writeClient.archive();
+
+ scheduled = writeClient.scheduleClusteringAtInstant(HoodieActiveTimeline.createNewInstantTime(), Option.empty());
+
+ assertTrue(scheduled, "The clustering plan should be scheduled");
+ table.getMetaClient().reloadActiveTimeline();
+ timeline = table.getActiveTimeline().filterPendingReplaceTimeline()
+ .filter(i -> i.getState() == HoodieInstant.State.REQUESTED);
+
+ HoodieInstant secondClusteringInstant = timeline.lastInstant().get();
+ List<HoodieClusteringGroup> inputFileGroups = ClusteringUtils.getClusteringPlan(table.getMetaClient(), secondClusteringInstant).get().getRight().getInputGroups();
+ // clustering plan has no previous file slice generated by previous pending clustering
+ assertFalse(inputFileGroups
+ .stream().anyMatch(fg -> fg.getSlices()
+ .stream().anyMatch(s -> s.getDataFilePath().contains(firstClusteringInstant))));
+ }
}
/**
@@ -561,56 +564,57 @@ public class ITTestHoodieFlinkClustering {
// To compute the clustering instant time and do clustering.
String clusteringInstantTime = HoodieActiveTimeline.createNewInstantTime();
- HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf);
- HoodieFlinkTable<?> table = writeClient.getHoodieTable();
-
- boolean scheduled = writeClient.scheduleClusteringAtInstant(clusteringInstantTime, Option.empty());
-
- assertTrue(scheduled, "The clustering plan should be scheduled");
-
- // fetch the instant based on the configured execution sequence
- table.getMetaClient().reloadActiveTimeline();
- HoodieTimeline timeline = table.getActiveTimeline().filterPendingReplaceTimeline()
- .filter(instant -> instant.getState() == HoodieInstant.State.REQUESTED);
-
- // generate clustering plan
- // should support configurable commit metadata
- Option<Pair<HoodieInstant, HoodieClusteringPlan>> clusteringPlanOption = ClusteringUtils.getClusteringPlan(
- table.getMetaClient(), timeline.lastInstant().get());
-
- HoodieClusteringPlan clusteringPlan = clusteringPlanOption.get().getRight();
-
- // Mark instant as clustering inflight
- HoodieInstant instant = HoodieTimeline.getReplaceCommitRequestedInstant(clusteringInstantTime);
- table.getActiveTimeline().transitionReplaceRequestedToInflight(instant, Option.empty());
-
- DataStream<ClusteringCommitEvent> dataStream = env.addSource(new ClusteringPlanSourceFunction(clusteringInstantTime, clusteringPlan, conf))
- .name("clustering_source")
- .uid("uid_clustering_source")
- .rebalance()
- .transform("clustering_task",
- TypeInformation.of(ClusteringCommitEvent.class),
- new ClusteringOperator(conf, rowType))
- .setParallelism(clusteringPlan.getInputGroups().size());
-
- ExecNodeUtil.setManagedMemoryWeight(dataStream.getTransformation(),
- conf.getInteger(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L);
-
- dataStream
- .addSink(new ClusteringCommitSink(conf))
- .name("clustering_commit")
- .uid("uid_clustering_commit")
- .setParallelism(1);
-
- env.execute("flink_hudi_clustering");
-
- // test output
- final Map<String, String> expected = new HashMap<>();
- expected.put("par1", "[id1,par1,id1,Danny,23,1100001,par1, id2,par1,id2,Stephen,33,2100001,par1]");
- expected.put("par2", "[id3,par2,id3,Julian,53,3100001,par2, id4,par2,id4,Fabian,31,4100001,par2]");
- expected.put("par3", "[id5,par3,id5,Sophia,18,5100001,par3, id6,par3,id6,Emma,20,6100001,par3]");
- expected.put("par4", "[id7,par4,id7,Bob,44,7100001,par4, id8,par4,id8,Han,56,8100001,par4]");
- TestData.checkWrittenData(tempFile, expected, 4);
+ try (HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf)) {
+ HoodieFlinkTable<?> table = writeClient.getHoodieTable();
+
+ boolean scheduled = writeClient.scheduleClusteringAtInstant(clusteringInstantTime, Option.empty());
+
+ assertTrue(scheduled, "The clustering plan should be scheduled");
+
+ // fetch the instant based on the configured execution sequence
+ table.getMetaClient().reloadActiveTimeline();
+ HoodieTimeline timeline = table.getActiveTimeline().filterPendingReplaceTimeline()
+ .filter(instant -> instant.getState() == HoodieInstant.State.REQUESTED);
+
+ // generate clustering plan
+ // should support configurable commit metadata
+ Option<Pair<HoodieInstant, HoodieClusteringPlan>> clusteringPlanOption = ClusteringUtils.getClusteringPlan(
+ table.getMetaClient(), timeline.lastInstant().get());
+
+ HoodieClusteringPlan clusteringPlan = clusteringPlanOption.get().getRight();
+
+ // Mark instant as clustering inflight
+ HoodieInstant instant = HoodieTimeline.getReplaceCommitRequestedInstant(clusteringInstantTime);
+ table.getActiveTimeline().transitionReplaceRequestedToInflight(instant, Option.empty());
+
+ DataStream<ClusteringCommitEvent> dataStream = env.addSource(new ClusteringPlanSourceFunction(clusteringInstantTime, clusteringPlan, conf))
+ .name("clustering_source")
+ .uid("uid_clustering_source")
+ .rebalance()
+ .transform("clustering_task",
+ TypeInformation.of(ClusteringCommitEvent.class),
+ new ClusteringOperator(conf, rowType))
+ .setParallelism(clusteringPlan.getInputGroups().size());
+
+ ExecNodeUtil.setManagedMemoryWeight(dataStream.getTransformation(),
+ conf.getInteger(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L);
+
+ dataStream
+ .addSink(new ClusteringCommitSink(conf))
+ .name("clustering_commit")
+ .uid("uid_clustering_commit")
+ .setParallelism(1);
+
+ env.execute("flink_hudi_clustering");
+
+ // test output
+ final Map<String, String> expected = new HashMap<>();
+ expected.put("par1", "[id1,par1,id1,Danny,23,1100001,par1, id2,par1,id2,Stephen,33,2100001,par1]");
+ expected.put("par2", "[id3,par2,id3,Julian,53,3100001,par2, id4,par2,id4,Fabian,31,4100001,par2]");
+ expected.put("par3", "[id5,par3,id5,Sophia,18,5100001,par3, id6,par3,id6,Emma,20,6100001,par3]");
+ expected.put("par4", "[id7,par4,id7,Bob,44,7100001,par4, id8,par4,id8,Han,56,8100001,par4]");
+ TestData.checkWrittenData(tempFile, expected, 4);
+ }
}
@Test
@@ -679,53 +683,54 @@ public class ITTestHoodieFlinkClustering {
// To compute the clustering instant time and do clustering.
String clusteringInstantTime = HoodieActiveTimeline.createNewInstantTime();
- HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf);
- HoodieFlinkTable<?> table = writeClient.getHoodieTable();
+ try (HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf)) {
+ HoodieFlinkTable<?> table = writeClient.getHoodieTable();
- boolean scheduled = writeClient.scheduleClusteringAtInstant(clusteringInstantTime, Option.empty());
+ boolean scheduled = writeClient.scheduleClusteringAtInstant(clusteringInstantTime, Option.empty());
- assertTrue(scheduled, "The clustering plan should be scheduled");
+ assertTrue(scheduled, "The clustering plan should be scheduled");
- tableEnv.executeSql(TestSQL.INSERT_T1);
+ tableEnv.executeSql(TestSQL.INSERT_T1);
- // fetch the instant based on the configured execution sequence
- table.getMetaClient().reloadActiveTimeline();
- HoodieTimeline timeline = table.getActiveTimeline().filterPendingReplaceTimeline()
- .filter(instant -> instant.getState() == HoodieInstant.State.REQUESTED);
+ // fetch the instant based on the configured execution sequence
+ table.getMetaClient().reloadActiveTimeline();
+ HoodieTimeline timeline = table.getActiveTimeline().filterPendingReplaceTimeline()
+ .filter(instant -> instant.getState() == HoodieInstant.State.REQUESTED);
- // generate clustering plan
- // should support configurable commit metadata
- Option<Pair<HoodieInstant, HoodieClusteringPlan>> clusteringPlanOption = ClusteringUtils.getClusteringPlan(
- table.getMetaClient(), timeline.lastInstant().get());
+ // generate clustering plan
+ // should support configurable commit metadata
+ Option<Pair<HoodieInstant, HoodieClusteringPlan>> clusteringPlanOption = ClusteringUtils.getClusteringPlan(
+ table.getMetaClient(), timeline.lastInstant().get());
- HoodieClusteringPlan clusteringPlan = clusteringPlanOption.get().getRight();
+ HoodieClusteringPlan clusteringPlan = clusteringPlanOption.get().getRight();
- // Mark instant as clustering inflight
- HoodieInstant instant = HoodieTimeline.getReplaceCommitRequestedInstant(clusteringInstantTime);
- table.getActiveTimeline().transitionReplaceRequestedToInflight(instant, Option.empty());
+ // Mark instant as clustering inflight
+ HoodieInstant instant = HoodieTimeline.getReplaceCommitRequestedInstant(clusteringInstantTime);
+ table.getActiveTimeline().transitionReplaceRequestedToInflight(instant, Option.empty());
- final Schema tableAvroSchema = StreamerUtil.getTableAvroSchema(table.getMetaClient(), false);
- final DataType rowDataType = AvroSchemaConverter.convertToDataType(tableAvroSchema);
- final RowType rowType = (RowType) rowDataType.getLogicalType();
+ final Schema tableAvroSchema = StreamerUtil.getTableAvroSchema(table.getMetaClient(), false);
+ final DataType rowDataType = AvroSchemaConverter.convertToDataType(tableAvroSchema);
+ final RowType rowType = (RowType) rowDataType.getLogicalType();
- DataStream<ClusteringCommitEvent> dataStream = env.addSource(new ClusteringPlanSourceFunction(clusteringInstantTime, clusteringPlan, conf))
- .name("clustering_source")
- .uid("uid_clustering_source")
- .rebalance()
- .transform("clustering_task",
- TypeInformation.of(ClusteringCommitEvent.class),
- new ClusteringOperator(conf, rowType))
- .setParallelism(clusteringPlan.getInputGroups().size());
+ DataStream<ClusteringCommitEvent> dataStream = env.addSource(new ClusteringPlanSourceFunction(clusteringInstantTime, clusteringPlan, conf))
+ .name("clustering_source")
+ .uid("uid_clustering_source")
+ .rebalance()
+ .transform("clustering_task",
+ TypeInformation.of(ClusteringCommitEvent.class),
+ new ClusteringOperator(conf, rowType))
+ .setParallelism(clusteringPlan.getInputGroups().size());
- ExecNodeUtil.setManagedMemoryWeight(dataStream.getTransformation(),
- conf.getInteger(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L);
+ ExecNodeUtil.setManagedMemoryWeight(dataStream.getTransformation(),
+ conf.getInteger(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L);
- dataStream
- .addSink(new ClusteringCommitTestSink(conf))
- .name("clustering_commit")
- .uid("uid_clustering_commit")
- .setParallelism(1);
+ dataStream
+ .addSink(new ClusteringCommitTestSink(conf))
+ .name("clustering_commit")
+ .uid("uid_clustering_commit")
+ .setParallelism(1);
- env.execute("flink_hudi_clustering");
+ env.execute("flink_hudi_clustering");
+ }
}
}
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/ITTestHoodieFlinkCompactor.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/ITTestHoodieFlinkCompactor.java
index ac2d93a7305..7b07f306982 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/ITTestHoodieFlinkCompactor.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/ITTestHoodieFlinkCompactor.java
@@ -152,36 +152,36 @@ public class ITTestHoodieFlinkCompactor {
// infer changelog mode
CompactionUtil.inferChangelogMode(conf, metaClient);
- HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf);
-
- String compactionInstantTime = scheduleCompactionPlan(metaClient, writeClient);
-
- HoodieFlinkTable<?> table = writeClient.getHoodieTable();
- // generate compaction plan
- // should support configurable commit metadata
- HoodieCompactionPlan compactionPlan = CompactionUtils.getCompactionPlan(
- table.getMetaClient(), compactionInstantTime);
-
- HoodieInstant instant = HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
- // Mark instant as compaction inflight
- table.getActiveTimeline().transitionCompactionRequestedToInflight(instant);
-
- env.addSource(new CompactionPlanSourceFunction(Collections.singletonList(Pair.of(compactionInstantTime, compactionPlan)), conf))
- .name("compaction_source")
- .uid("uid_compaction_source")
- .rebalance()
- .transform("compact_task",
- TypeInformation.of(CompactionCommitEvent.class),
- new CompactOperator(conf))
- .setParallelism(FlinkMiniCluster.DEFAULT_PARALLELISM)
- .addSink(new CompactionCommitSink(conf))
- .name("compaction_commit")
- .uid("uid_compaction_commit")
- .setParallelism(1);
-
- env.execute("flink_hudi_compaction");
- writeClient.close();
- TestData.checkWrittenDataCOW(tempFile, EXPECTED1);
+ try (HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf)) {
+
+ String compactionInstantTime = scheduleCompactionPlan(metaClient, writeClient);
+
+ HoodieFlinkTable<?> table = writeClient.getHoodieTable();
+ // generate compaction plan
+ // should support configurable commit metadata
+ HoodieCompactionPlan compactionPlan = CompactionUtils.getCompactionPlan(
+ table.getMetaClient(), compactionInstantTime);
+
+ HoodieInstant instant = HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
+ // Mark instant as compaction inflight
+ table.getActiveTimeline().transitionCompactionRequestedToInflight(instant);
+
+ env.addSource(new CompactionPlanSourceFunction(Collections.singletonList(Pair.of(compactionInstantTime, compactionPlan)), conf))
+ .name("compaction_source")
+ .uid("uid_compaction_source")
+ .rebalance()
+ .transform("compact_task",
+ TypeInformation.of(CompactionCommitEvent.class),
+ new CompactOperator(conf))
+ .setParallelism(FlinkMiniCluster.DEFAULT_PARALLELISM)
+ .addSink(new CompactionCommitSink(conf))
+ .name("compaction_commit")
+ .uid("uid_compaction_commit")
+ .setParallelism(1);
+
+ env.execute("flink_hudi_compaction");
+ TestData.checkWrittenDataCOW(tempFile, EXPECTED1);
+ }
}
@ParameterizedTest
@@ -223,46 +223,46 @@ public class ITTestHoodieFlinkCompactor {
// infer changelog mode
CompactionUtil.inferChangelogMode(conf, metaClient);
- HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf);
-
- String compactionInstantTime = scheduleCompactionPlan(metaClient, writeClient);
-
- HoodieFlinkTable<?> table = writeClient.getHoodieTable();
-
- // try to upgrade or downgrade
- if (upgrade) {
- metaClient.getTableConfig().setTableVersion(HoodieTableVersion.FIVE);
- new UpgradeDowngrade(metaClient, writeClient.getConfig(), writeClient.getEngineContext(), FlinkUpgradeDowngradeHelper.getInstance()).run(HoodieTableVersion.SIX, "none");
- } else {
- metaClient.getTableConfig().setTableVersion(HoodieTableVersion.SIX);
- new UpgradeDowngrade(metaClient, writeClient.getConfig(), writeClient.getEngineContext(), FlinkUpgradeDowngradeHelper.getInstance()).run(HoodieTableVersion.FIVE, "none");
+ try (HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf)) {
+
+ String compactionInstantTime = scheduleCompactionPlan(metaClient, writeClient);
+
+ HoodieFlinkTable<?> table = writeClient.getHoodieTable();
+
+ // try to upgrade or downgrade
+ if (upgrade) {
+ metaClient.getTableConfig().setTableVersion(HoodieTableVersion.FIVE);
+ new UpgradeDowngrade(metaClient, writeClient.getConfig(), writeClient.getEngineContext(), FlinkUpgradeDowngradeHelper.getInstance()).run(HoodieTableVersion.SIX, "none");
+ } else {
+ metaClient.getTableConfig().setTableVersion(HoodieTableVersion.SIX);
+ new UpgradeDowngrade(metaClient, writeClient.getConfig(), writeClient.getEngineContext(), FlinkUpgradeDowngradeHelper.getInstance()).run(HoodieTableVersion.FIVE, "none");
+ }
+
+ // generate compaction plan
+ // should support configurable commit metadata
+ HoodieCompactionPlan compactionPlan = CompactionUtils.getCompactionPlan(
+ table.getMetaClient(), compactionInstantTime);
+
+ HoodieInstant instant = HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
+ // Mark instant as compaction inflight
+ table.getActiveTimeline().transitionCompactionRequestedToInflight(instant);
+
+ env.addSource(new CompactionPlanSourceFunction(Collections.singletonList(Pair.of(compactionInstantTime, compactionPlan)), conf))
+ .name("compaction_source")
+ .uid("uid_compaction_source")
+ .rebalance()
+ .transform("compact_task",
+ TypeInformation.of(CompactionCommitEvent.class),
+ new CompactOperator(conf))
+ .setParallelism(FlinkMiniCluster.DEFAULT_PARALLELISM)
+ .addSink(new CompactionCommitSink(conf))
+ .name("compaction_commit")
+ .uid("uid_compaction_commit")
+ .setParallelism(1);
+
+ env.execute("flink_hudi_compaction");
+ TestData.checkWrittenDataCOW(tempFile, EXPECTED1);
}
-
- // generate compaction plan
- // should support configurable commit metadata
- HoodieCompactionPlan compactionPlan = CompactionUtils.getCompactionPlan(
- table.getMetaClient(), compactionInstantTime);
-
- HoodieInstant instant = HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
- // Mark instant as compaction inflight
- table.getActiveTimeline().transitionCompactionRequestedToInflight(instant);
-
- env.addSource(new CompactionPlanSourceFunction(Collections.singletonList(Pair.of(compactionInstantTime, compactionPlan)), conf))
- .name("compaction_source")
- .uid("uid_compaction_source")
- .rebalance()
- .transform("compact_task",
- TypeInformation.of(CompactionCommitEvent.class),
- new CompactOperator(conf))
- .setParallelism(FlinkMiniCluster.DEFAULT_PARALLELISM)
- .addSink(new CompactionCommitSink(conf))
- .name("compaction_commit")
- .uid("uid_compaction_commit")
- .setParallelism(1);
-
- env.execute("flink_hudi_compaction");
- writeClient.close();
- TestData.checkWrittenDataCOW(tempFile, EXPECTED1);
}
@ParameterizedTest
@@ -474,40 +474,41 @@ public class ITTestHoodieFlinkCompactor {
// infer changelog mode
CompactionUtil.inferChangelogMode(conf, metaClient);
- HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf);
+ try (HoodieFlinkWriteClient writeClient = FlinkWriteClients.createWriteClient(conf)) {
- String compactionInstantTime = scheduleCompactionPlan(metaClient, writeClient);
+ String compactionInstantTime = scheduleCompactionPlan(metaClient, writeClient);
- HoodieFlinkTable<?> table = writeClient.getHoodieTable();
- // generate compaction plan
- // should support configurable commit metadata
- HoodieCompactionPlan compactionPlan = CompactionUtils.getCompactionPlan(
- table.getMetaClient(), compactionInstantTime);
+ HoodieFlinkTable<?> table = writeClient.getHoodieTable();
+ // generate compaction plan
+ // should support configurable commit metadata
+ HoodieCompactionPlan compactionPlan = CompactionUtils.getCompactionPlan(
+ table.getMetaClient(), compactionInstantTime);
- HoodieInstant instant = HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
- // Mark instant as compaction inflight
- table.getActiveTimeline().transitionCompactionRequestedToInflight(instant);
+ HoodieInstant instant = HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
+ // Mark instant as compaction inflight
+ table.getActiveTimeline().transitionCompactionRequestedToInflight(instant);
- tableEnv.executeSql(TestSQL.INSERT_T1);
+ tableEnv.executeSql(TestSQL.INSERT_T1);
- // Make configuration and setAvroSchema.
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, Time.milliseconds(1)));
+ // Make configuration and setAvroSchema.
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, Time.milliseconds(1)));
- env.addSource(new CompactionPlanSourceFunction(Collections.singletonList(Pair.of(compactionInstantTime, compactionPlan)), conf))
- .name("compaction_source")
- .uid("uid_compaction_source")
- .rebalance()
- .transform("compact_task",
- TypeInformation.of(CompactionCommitEvent.class),
- new CompactOperator(conf))
- .setParallelism(1)
- .addSink(new CompactionCommitTestSink(conf))
- .name("compaction_commit")
- .uid("uid_compaction_commit")
- .setParallelism(1);
+ env.addSource(new CompactionPlanSourceFunction(Collections.singletonList(Pair.of(compactionInstantTime, compactionPlan)), conf))
+ .name("compaction_source")
+ .uid("uid_compaction_source")
+ .rebalance()
+ .transform("compact_task",
+ TypeInformation.of(CompactionCommitEvent.class),
+ new CompactOperator(conf))
+ .setParallelism(1)
+ .addSink(new CompactionCommitTestSink(conf))
+ .name("compaction_commit")
+ .uid("uid_compaction_commit")
+ .setParallelism(1);
- env.execute("flink_hudi_compaction");
+ env.execute("flink_hudi_compaction");
+ }
}
private String scheduleCompactionPlan(HoodieTableMetaClient metaClient, HoodieFlinkWriteClient<?> writeClient) {
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java
index f4ecb3e67d0..f69477c3df0 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java
@@ -1064,6 +1064,7 @@ public class TestInputFormat {
List<RowData> actual4 = readData(inputFormat4);
// final List<RowData> expected4 = TestData.dataSetInsert(3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14);
TestData.assertRowDataEquals(actual4, Collections.emptyList());
+ writeClient.close();
}
@ParameterizedTest
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestClusteringUtil.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestClusteringUtil.java
index 11a5b874325..9a3c17c45c5 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestClusteringUtil.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestClusteringUtil.java
@@ -38,6 +38,8 @@ import org.apache.hudi.util.ClusteringUtil;
import org.apache.hudi.util.FlinkTables;
import org.apache.hudi.util.FlinkWriteClients;
import org.apache.hudi.util.StreamerUtil;
+
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -72,6 +74,13 @@ public class TestClusteringUtil {
beforeEach(Collections.emptyMap());
}
+ @AfterEach
+ void afterEach() {
+ if (this.writeClient != null) {
+ this.writeClient.close();
+ }
+ }
+
void beforeEach(Map<String, String> options) throws IOException {
this.conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
conf.setString(FlinkOptions.OPERATION, WriteOperationType.INSERT.value());
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestCompactionUtil.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestCompactionUtil.java
index 9be28d02c27..aa35eb72397 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestCompactionUtil.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestCompactionUtil.java
@@ -140,20 +140,21 @@ public class TestCompactionUtil {
// write a commit with data first
TestData.writeDataAsBatch(TestData.DATA_SET_SINGLE_INSERT, conf);
- HoodieFlinkWriteClient<?> writeClient = FlinkWriteClients.createWriteClient(conf);
- CompactionUtil.scheduleCompaction(metaClient, writeClient, true, true);
+ try (HoodieFlinkWriteClient<?> writeClient = FlinkWriteClients.createWriteClient(conf)) {
+ CompactionUtil.scheduleCompaction(metaClient, writeClient, true, true);
- Option<HoodieInstant> pendingCompactionInstant = metaClient.reloadActiveTimeline().filterPendingCompactionTimeline().lastInstant();
- assertTrue(pendingCompactionInstant.isPresent(), "A compaction plan expects to be scheduled");
+ Option<HoodieInstant> pendingCompactionInstant = metaClient.reloadActiveTimeline().filterPendingCompactionTimeline().lastInstant();
+ assertTrue(pendingCompactionInstant.isPresent(), "A compaction plan expects to be scheduled");
- // write another commit with data and start a new instant
- TestData.writeDataAsBatch(TestData.DATA_SET_INSERT, conf);
- TimeUnit.SECONDS.sleep(3); // in case the instant time interval is too close
- writeClient.startCommit();
+ // write another commit with data and start a new instant
+ TestData.writeDataAsBatch(TestData.DATA_SET_INSERT, conf);
+ TimeUnit.SECONDS.sleep(3); // in case the instant time interval is too close
+ writeClient.startCommit();
- CompactionUtil.scheduleCompaction(metaClient, writeClient, true, false);
- int numCompactionCommits = metaClient.reloadActiveTimeline().filterPendingCompactionTimeline().countInstants();
- assertThat("Two compaction plan expects to be scheduled", numCompactionCommits, is(2));
+ CompactionUtil.scheduleCompaction(metaClient, writeClient, true, false);
+ int numCompactionCommits = metaClient.reloadActiveTimeline().filterPendingCompactionTimeline().countInstants();
+ assertThat("Two compaction plan expects to be scheduled", numCompactionCommits, is(2));
+ }
}
@ParameterizedTest
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestViewStorageProperties.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestViewStorageProperties.java
index a8b06c111cd..e3e449bbd41 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestViewStorageProperties.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestViewStorageProperties.java
@@ -18,6 +18,7 @@
package org.apache.hudi.utils;
+import org.apache.hudi.client.HoodieFlinkWriteClient;
import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
import org.apache.hudi.common.table.view.FileSystemViewStorageType;
import org.apache.hudi.configuration.FlinkOptions;
@@ -66,8 +67,9 @@ public class TestViewStorageProperties {
@Test
void testDumpRemoteViewStorageConfig() throws IOException {
Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
- FlinkWriteClients.createWriteClient(conf);
- FileSystemViewStorageConfig storageConfig = ViewStorageProperties.loadFromProperties(conf.getString(FlinkOptions.PATH), new Configuration());
- assertThat(storageConfig.getStorageType(), is(FileSystemViewStorageType.REMOTE_FIRST));
+ try (HoodieFlinkWriteClient<?> writeClient = FlinkWriteClients.createWriteClient(conf)) {
+ FileSystemViewStorageConfig storageConfig = ViewStorageProperties.loadFromProperties(conf.getString(FlinkOptions.PATH), new Configuration());
+ assertThat(storageConfig.getStorageType(), is(FileSystemViewStorageType.REMOTE_FIRST));
+ }
}
}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieFileIndex.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieFileIndex.scala
index cd9dbc8df79..1ccb4081fb8 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieFileIndex.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieFileIndex.scala
@@ -194,6 +194,7 @@ class TestHoodieFileIndex extends HoodieSparkClientTestBase with ScalaAssertionS
val fileIndex = HoodieFileIndex(spark, metaClient, None, queryOpts)
assertEquals("partition_path", fileIndex.partitionSchema.fields.map(_.name).mkString(","))
+ writeClient.close()
}
@ParameterizedTest
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndexWithSQL.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndexWithSQL.scala
index bb0c0065a91..3fae2964549 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndexWithSQL.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndexWithSQL.scala
@@ -194,6 +194,7 @@ class TestColumnStatsIndexWithSQL extends ColumnStatIndexTestBase {
val writeClient = new SparkRDDWriteClient(new HoodieSparkEngineContext(jsc), getWriteConfig(commonOpts))
writeClient.scheduleCompaction(org.apache.hudi.common.util.Option.empty())
+ writeClient.close()
doWriteAndValidateColumnStats(testCase, metadataOpts, commonOpts,
dataSourcePath = "index/colstats/update-input-table-json",
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamSourceReadByStateTransitionTime.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamSourceReadByStateTransitionTime.scala
index 4cdfe45eb15..ef83b280956 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamSourceReadByStateTransitionTime.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamSourceReadByStateTransitionTime.scala
@@ -91,6 +91,7 @@ class TestStreamSourceReadByStateTransitionTime extends TestStreamingSource {
assertCountMatched(10, true),
StopStream
)
+ writeClient.close()
}
}
}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStructuredStreaming.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStructuredStreaming.scala
index 1fcd3b1975f..1bbcf1833dd 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStructuredStreaming.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStructuredStreaming.scala
@@ -329,6 +329,7 @@ class TestStructuredStreaming extends HoodieSparkClientTestBase {
.start(destPath)
query3.processAllAvailable()
+ query3.stop()
metaClient = HoodieTableMetaClient.builder
.setConf(fs.getConf).setBasePath(destPath).setLoadActiveTimelineOnLoad(true).build
@@ -473,7 +474,7 @@ class TestStructuredStreaming extends HoodieSparkClientTestBase {
}
private def streamingWrite(schema: StructType, sourcePath: String, destPath: String, hudiOptions: Map[String, String], checkpoint: String): Unit = {
- spark.readStream
+ val query = spark.readStream
.schema(schema)
.json(sourcePath)
.writeStream
@@ -483,7 +484,8 @@ class TestStructuredStreaming extends HoodieSparkClientTestBase {
.option("checkpointLocation", basePath + "/checkpoint" + checkpoint)
.outputMode(OutputMode.Append)
.start(destPath)
- .processAllAvailable()
+ query.processAllAvailable()
+ query.stop()
}
@Test
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieIndexer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieIndexer.java
index e1acde4fcd6..e6fd7f20833 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieIndexer.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieIndexer.java
@@ -373,13 +373,14 @@ public class TestHoodieIndexer extends SparkClientFunctionalTestHarness implemen
HoodieWriteConfig.Builder writeConfigBuilder = getWriteConfigBuilder(basePath(), tableName);
HoodieWriteConfig writeConfig = writeConfigBuilder.withMetadataConfig(metadataConfig).build();
// do one upsert with synchronous metadata update
- SparkRDDWriteClient writeClient = new SparkRDDWriteClient(context(), writeConfig);
- String instant = HoodieActiveTimeline.createNewInstantTime();
- writeClient.startCommitWithTime(instant);
- List<HoodieRecord> records = DATA_GENERATOR.generateInserts(instant, 100);
- JavaRDD<WriteStatus> result = writeClient.upsert(jsc().parallelize(records, 1), instant);
- List<WriteStatus> statuses = result.collect();
- assertNoWriteErrors(statuses);
+ try (SparkRDDWriteClient writeClient = new SparkRDDWriteClient(context(), writeConfig)) {
+ String instant = HoodieActiveTimeline.createNewInstantTime();
+ writeClient.startCommitWithTime(instant);
+ List<HoodieRecord> records = DATA_GENERATOR.generateInserts(instant, 100);
+ JavaRDD<WriteStatus> result = writeClient.upsert(jsc().parallelize(records, 1), instant);
+ List<WriteStatus> statuses = result.collect();
+ assertNoWriteErrors(statuses);
+ }
}
private void scheduleAndExecuteIndexing(MetadataPartitionType partitionTypeToIndex, String tableName) {
@@ -424,13 +425,14 @@ public class TestHoodieIndexer extends SparkClientFunctionalTestHarness implemen
HoodieMetadataConfig.Builder metadataConfigBuilder = getMetadataConfigBuilder(true, false).withMetadataIndexBloomFilter(true);
HoodieWriteConfig writeConfig = writeConfigBuilder.withMetadataConfig(metadataConfigBuilder.build()).build();
// do one upsert with synchronous metadata update
- SparkRDDWriteClient writeClient = new SparkRDDWriteClient(context(), writeConfig);
- String instant = HoodieActiveTimeline.createNewInstantTime();
- writeClient.startCommitWithTime(instant);
- List<HoodieRecord> records = DATA_GENERATOR.generateInserts(instant, 100);
- JavaRDD<WriteStatus> result = writeClient.upsert(jsc().parallelize(records, 1), instant);
- List<WriteStatus> statuses = result.collect();
- assertNoWriteErrors(statuses);
+ try (SparkRDDWriteClient writeClient = new SparkRDDWriteClient(context(), writeConfig)) {
+ String instant = HoodieActiveTimeline.createNewInstantTime();
+ writeClient.startCommitWithTime(instant);
+ List<HoodieRecord> records = DATA_GENERATOR.generateInserts(instant, 100);
+ JavaRDD<WriteStatus> result = writeClient.upsert(jsc().parallelize(records, 1), instant);
+ List<WriteStatus> statuses = result.collect();
+ assertNoWriteErrors(statuses);
+ }
// validate partitions built successfully
assertTrue(reload(metaClient).getTableConfig().getMetadataPartitions().contains(FILES.getPartitionPath()));
@@ -477,13 +479,14 @@ public class TestHoodieIndexer extends SparkClientFunctionalTestHarness implemen
HoodieMetadataConfig.Builder metadataConfigBuilder = getMetadataConfigBuilder(true, false);
HoodieWriteConfig writeConfig = writeConfigBuilder.withMetadataConfig(metadataConfigBuilder.build()).build();
// do one upsert with synchronous metadata update
- SparkRDDWriteClient writeClient = new SparkRDDWriteClient(context(), writeConfig);
- String instant = HoodieActiveTimeline.createNewInstantTime();
- writeClient.startCommitWithTime(instant);
- List<HoodieRecord> records = DATA_GENERATOR.generateInserts(instant, 100);
- JavaRDD<WriteStatus> result = writeClient.upsert(jsc().parallelize(records, 1), instant);
- List<WriteStatus> statuses = result.collect();
- assertNoWriteErrors(statuses);
+ try (SparkRDDWriteClient writeClient = new SparkRDDWriteClient(context(), writeConfig)) {
+ String instant = HoodieActiveTimeline.createNewInstantTime();
+ writeClient.startCommitWithTime(instant);
+ List<HoodieRecord> records = DATA_GENERATOR.generateInserts(instant, 100);
+ JavaRDD<WriteStatus> result = writeClient.upsert(jsc().parallelize(records, 1), instant);
+ List<WriteStatus> statuses = result.collect();
+ assertNoWriteErrors(statuses);
+ }
// validate files partition built successfully
assertTrue(reload(metaClient).getTableConfig().getMetadataPartitions().contains(FILES.getPartitionPath()));
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestHoodieSnapshotExporter.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestHoodieSnapshotExporter.java
index 5a0279fdf4a..b6187e989d9 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestHoodieSnapshotExporter.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestHoodieSnapshotExporter.java
@@ -93,13 +93,13 @@ public class TestHoodieSnapshotExporter extends SparkClientFunctionalTestHarness
// Prepare data as source Hudi dataset
HoodieWriteConfig cfg = getHoodieWriteConfig(sourcePath);
- SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg);
- writeClient.startCommitWithTime(COMMIT_TIME);
- HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator(new String[] {PARTITION_PATH});
- List<HoodieRecord> records = dataGen.generateInserts(COMMIT_TIME, NUM_RECORDS);
- JavaRDD<HoodieRecord> recordsRDD = jsc().parallelize(records, 1);
- writeClient.bulkInsert(recordsRDD, COMMIT_TIME);
- writeClient.close();
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) {
+ writeClient.startCommitWithTime(COMMIT_TIME);
+ HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator(new String[] {PARTITION_PATH});
+ List<HoodieRecord> records = dataGen.generateInserts(COMMIT_TIME, NUM_RECORDS);
+ JavaRDD<HoodieRecord> recordsRDD = jsc().parallelize(records, 1);
+ writeClient.bulkInsert(recordsRDD, COMMIT_TIME);
+ }
RemoteIterator<LocatedFileStatus> itr = lfs.listFiles(new Path(sourcePath), true);
while (itr.hasNext()) {
LOG.info(">>> Prepared test file: " + itr.next().getPath());
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/offlinejob/HoodieOfflineJobTestBase.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/offlinejob/HoodieOfflineJobTestBase.java
index f613945db48..33615cdddee 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/offlinejob/HoodieOfflineJobTestBase.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/offlinejob/HoodieOfflineJobTestBase.java
@@ -31,6 +31,7 @@ import org.apache.hudi.utilities.testutils.UtilitiesTestBase;
import org.apache.hadoop.fs.FileSystem;
import org.apache.spark.api.java.JavaRDD;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
@@ -59,6 +60,13 @@ public class HoodieOfflineJobTestBase extends UtilitiesTestBase {
dataGen = new HoodieTestDataGenerator();
}
+ @AfterEach
+ public void teardown() {
+ if (client != null) {
+ client.close();
+ }
+ }
+
// -------------------------------------------------------------------------
// Utilities
// -------------------------------------------------------------------------
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestGcsEventsHoodieIncrSource.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestGcsEventsHoodieIncrSource.java
index 2d76c1b3d2e..cc80123a19c 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestGcsEventsHoodieIncrSource.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestGcsEventsHoodieIncrSource.java
@@ -350,21 +350,22 @@ public class TestGcsEventsHoodieIncrSource extends SparkClientFunctionalTestHarn
private Pair<String, List<HoodieRecord>> writeGcsMetadataRecords(String commitTime) throws IOException {
HoodieWriteConfig writeConfig = getWriteConfig();
- SparkRDDWriteClient writeClient = getHoodieWriteClient(writeConfig);
-
- writeClient.startCommitWithTime(commitTime);
- List<HoodieRecord> gcsMetadataRecords = Arrays.asList(
- getGcsMetadataRecord(commitTime, "data-file-1.json", "bucket-1", "1"),
- getGcsMetadataRecord(commitTime, "data-file-2.json", "bucket-1", "1"),
- getGcsMetadataRecord(commitTime, "data-file-3.json", "bucket-1", "1"),
- getGcsMetadataRecord(commitTime, "data-file-4.json", "bucket-1", "1")
- );
- JavaRDD<WriteStatus> result = writeClient.upsert(jsc().parallelize(gcsMetadataRecords, 1), commitTime);
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(writeConfig)) {
+
+ writeClient.startCommitWithTime(commitTime);
+ List<HoodieRecord> gcsMetadataRecords = Arrays.asList(
+ getGcsMetadataRecord(commitTime, "data-file-1.json", "bucket-1", "1"),
+ getGcsMetadataRecord(commitTime, "data-file-2.json", "bucket-1", "1"),
+ getGcsMetadataRecord(commitTime, "data-file-3.json", "bucket-1", "1"),
+ getGcsMetadataRecord(commitTime, "data-file-4.json", "bucket-1", "1")
+ );
+ JavaRDD<WriteStatus> result = writeClient.upsert(jsc().parallelize(gcsMetadataRecords, 1), commitTime);
- List<WriteStatus> statuses = result.collect();
- assertNoWriteErrors(statuses);
+ List<WriteStatus> statuses = result.collect();
+ assertNoWriteErrors(statuses);
- return Pair.of(commitTime, gcsMetadataRecords);
+ return Pair.of(commitTime, gcsMetadataRecords);
+ }
}
private TypedProperties setProps(IncrSourceHelper.MissingCheckpointStrategy missingCheckpointStrategy) {
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestHoodieIncrSource.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestHoodieIncrSource.java
index 301b6472de1..d4b0d6defa2 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestHoodieIncrSource.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestHoodieIncrSource.java
@@ -41,10 +41,10 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.sources.helpers.IncrSourceHelper;
+import org.apache.hudi.utilities.sources.helpers.TestSnapshotQuerySplitterImpl;
import org.apache.avro.Schema;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hudi.utilities.sources.helpers.TestSnapshotQuerySplitterImpl;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
@@ -104,33 +104,33 @@ public class TestHoodieIncrSource extends SparkClientFunctionalTestHarness {
.enable(false).build())
.build();
- SparkRDDWriteClient writeClient = getHoodieWriteClient(writeConfig);
- Pair<String, List<HoodieRecord>> inserts = writeRecords(writeClient, INSERT, null, "100");
- Pair<String, List<HoodieRecord>> inserts2 = writeRecords(writeClient, INSERT, null, "200");
- Pair<String, List<HoodieRecord>> inserts3 = writeRecords(writeClient, INSERT, null, "300");
- Pair<String, List<HoodieRecord>> inserts4 = writeRecords(writeClient, INSERT, null, "400");
- Pair<String, List<HoodieRecord>> inserts5 = writeRecords(writeClient, INSERT, null, "500");
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(writeConfig)) {
+ Pair<String, List<HoodieRecord>> inserts = writeRecords(writeClient, INSERT, null, "100");
+ Pair<String, List<HoodieRecord>> inserts2 = writeRecords(writeClient, INSERT, null, "200");
+ Pair<String, List<HoodieRecord>> inserts3 = writeRecords(writeClient, INSERT, null, "300");
+ Pair<String, List<HoodieRecord>> inserts4 = writeRecords(writeClient, INSERT, null, "400");
+ Pair<String, List<HoodieRecord>> inserts5 = writeRecords(writeClient, INSERT, null, "500");
- // read everything upto latest
- readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT, Option.empty(), 500, inserts5.getKey());
+ // read everything upto latest
+ readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT, Option.empty(), 500, inserts5.getKey());
- // even if the begin timestamp is archived (100), full table scan should kick in, but should filter for records having commit time > 100
- readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT, Option.of("100"), 400, inserts5.getKey());
+ // even if the begin timestamp is archived (100), full table scan should kick in, but should filter for records having commit time > 100
+ readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT, Option.of("100"), 400, inserts5.getKey());
- // even if the read upto latest is set, if begin timestamp is in active timeline, only incremental should kick in.
- readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT, Option.of("400"), 100, inserts5.getKey());
+ // even if the read upto latest is set, if begin timestamp is in active timeline, only incremental should kick in.
+ readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT, Option.of("400"), 100, inserts5.getKey());
- // read just the latest
- readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST, Option.empty(), 100, inserts5.getKey());
+ // read just the latest
+ readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST, Option.empty(), 100, inserts5.getKey());
- // ensure checkpoint does not move
- readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST, Option.of(inserts5.getKey()), 0, inserts5.getKey());
+ // ensure checkpoint does not move
+ readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST, Option.of(inserts5.getKey()), 0, inserts5.getKey());
- Pair<String, List<HoodieRecord>> inserts6 = writeRecords(writeClient, INSERT, null, "600");
+ Pair<String, List<HoodieRecord>> inserts6 = writeRecords(writeClient, INSERT, null, "600");
- // insert new batch and ensure the checkpoint moves
- readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST, Option.of(inserts5.getKey()), 100, inserts6.getKey());
- writeClient.close();
+ // insert new batch and ensure the checkpoint moves
+ readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST, Option.of(inserts5.getKey()), 100, inserts6.getKey());
+ }
}
@ParameterizedTest
@@ -149,69 +149,68 @@ public class TestHoodieIncrSource extends SparkClientFunctionalTestHarness {
.withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(false).build())
.build();
- SparkRDDWriteClient writeClient = getHoodieWriteClient(writeConfig);
- List<Pair<String, List<HoodieRecord>>> inserts = new ArrayList<>();
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(writeConfig)) {
+ List<Pair<String, List<HoodieRecord>>> inserts = new ArrayList<>();
- for (int i = 0; i < 6; i++) {
- inserts.add(writeRecords(writeClient, INSERT, null, HoodieActiveTimeline.createNewInstantTime()));
- }
+ for (int i = 0; i < 6; i++) {
+ inserts.add(writeRecords(writeClient, INSERT, null, HoodieActiveTimeline.createNewInstantTime()));
+ }
- // Emulates a scenario where an inflight commit is before a completed commit
- // The checkpoint should not go past this commit
- HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
- HoodieInstant instant4 = activeTimeline
- .filter(instant -> instant.getTimestamp().equals(inserts.get(4).getKey())).firstInstant().get();
- Option<byte[]> instant4CommitData = activeTimeline.getInstantDetails(instant4);
- activeTimeline.revertToInflight(instant4);
- metaClient.reloadActiveTimeline();
-
- // Reads everything up to latest
- readAndAssert(
- IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT,
- Option.empty(),
- 400,
- inserts.get(3).getKey());
-
- // Even if the beginning timestamp is archived, full table scan should kick in, but should filter for records having commit time > first instant time
- readAndAssert(
- IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT,
- Option.of(inserts.get(0).getKey()),
- 300,
- inserts.get(3).getKey());
-
- // Even if the read upto latest is set, if begin timestamp is in active timeline, only incremental should kick in.
- readAndAssert(
- IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT,
- Option.of(inserts.get(2).getKey()),
- 100,
- inserts.get(3).getKey());
-
- // Reads just the latest
- readAndAssert(
- IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST,
- Option.empty(),
- 100,
- inserts.get(3).getKey());
-
- // Ensures checkpoint does not move
- readAndAssert(
- IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST,
- Option.of(inserts.get(3).getKey()),
- 0,
- inserts.get(3).getKey());
-
- activeTimeline.reload().saveAsComplete(
- new HoodieInstant(HoodieInstant.State.INFLIGHT, instant4.getAction(), inserts.get(4).getKey()),
- instant4CommitData);
-
- // After the inflight commit completes, the checkpoint should move on after incremental pull
- readAndAssert(
- IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST,
- Option.of(inserts.get(3).getKey()),
- 200,
- inserts.get(5).getKey());
-
- writeClient.close();
+ // Emulates a scenario where an inflight commit is before a completed commit
+ // The checkpoint should not go past this commit
+ HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
+ HoodieInstant instant4 = activeTimeline
+ .filter(instant -> instant.getTimestamp().equals(inserts.get(4).getKey())).firstInstant().get();
+ Option<byte[]> instant4CommitData = activeTimeline.getInstantDetails(instant4);
+ activeTimeline.revertToInflight(instant4);
+ metaClient.reloadActiveTimeline();
+
+ // Reads everything up to latest
+ readAndAssert(
+ IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT,
+ Option.empty(),
+ 400,
+ inserts.get(3).getKey());
+
+ // Even if the beginning timestamp is archived, full table scan should kick in, but should filter for records having commit time > first instant time
+ readAndAssert(
+ IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT,
+ Option.of(inserts.get(0).getKey()),
+ 300,
+ inserts.get(3).getKey());
+
+ // Even if the read upto latest is set, if begin timestamp is in active timeline, only incremental should kick in.
+ readAndAssert(
+ IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT,
+ Option.of(inserts.get(2).getKey()),
+ 100,
+ inserts.get(3).getKey());
+
+ // Reads just the latest
+ readAndAssert(
+ IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST,
+ Option.empty(),
+ 100,
+ inserts.get(3).getKey());
+
+ // Ensures checkpoint does not move
+ readAndAssert(
+ IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST,
+ Option.of(inserts.get(3).getKey()),
+ 0,
+ inserts.get(3).getKey());
+
+ activeTimeline.reload().saveAsComplete(
+ new HoodieInstant(HoodieInstant.State.INFLIGHT, instant4.getAction(), inserts.get(4).getKey()),
+ instant4CommitData);
+
+ // After the inflight commit completes, the checkpoint should move on after incremental pull
+ readAndAssert(
+ IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST,
+ Option.of(inserts.get(3).getKey()),
+ 200,
+ inserts.get(5).getKey());
+ }
}
@ParameterizedTest
@@ -230,101 +229,101 @@ public class TestHoodieIncrSource extends SparkClientFunctionalTestHarness {
.withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(false).build())
.build();
- SparkRDDWriteClient writeClient = getHoodieWriteClient(writeConfig);
- List<Pair<String, List<HoodieRecord>>> dataBatches = new ArrayList<>();
-
- // For COW:
- // 0: bulk_insert of 100 records
- // 1: bulk_insert of 100 records
- // 2: bulk_insert of 100 records
- // schedule clustering
- // 3: bulk_insert of 100 records
- // 4: upsert of 100 records (updates only based on round 3)
- // 5: upsert of 100 records (updates only based on round 3)
- // 6: bulk_insert of 100 records
- // For MOR:
- // 0: bulk_insert of 100 records
- // 1: bulk_insert of 100 records
- // 2: bulk_insert of 100 records
- // 3: bulk_insert of 100 records
- // 4: upsert of 100 records (updates only based on round 3)
- // schedule compaction
- // 5: upsert of 100 records (updates only based on round 3)
- // schedule clustering
- // 6: bulk_insert of 100 records
- for (int i = 0; i < 6; i++) {
- WriteOperationType opType = i < 4 ? BULK_INSERT : UPSERT;
- List<HoodieRecord> recordsForUpdate = i < 4 ? null : dataBatches.get(3).getRight();
- dataBatches.add(writeRecords(writeClient, opType, recordsForUpdate, HoodieActiveTimeline.createNewInstantTime()));
- if (tableType == COPY_ON_WRITE) {
- if (i == 2) {
- writeClient.scheduleClustering(Option.empty());
- }
- } else if (tableType == MERGE_ON_READ) {
- if (i == 4) {
- writeClient.scheduleCompaction(Option.empty());
- }
- if (i == 5) {
- writeClient.scheduleClustering(Option.empty());
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(writeConfig)) {
+ List<Pair<String, List<HoodieRecord>>> dataBatches = new ArrayList<>();
+
+ // For COW:
+ // 0: bulk_insert of 100 records
+ // 1: bulk_insert of 100 records
+ // 2: bulk_insert of 100 records
+ // schedule clustering
+ // 3: bulk_insert of 100 records
+ // 4: upsert of 100 records (updates only based on round 3)
+ // 5: upsert of 100 records (updates only based on round 3)
+ // 6: bulk_insert of 100 records
+ // For MOR:
+ // 0: bulk_insert of 100 records
+ // 1: bulk_insert of 100 records
+ // 2: bulk_insert of 100 records
+ // 3: bulk_insert of 100 records
+ // 4: upsert of 100 records (updates only based on round 3)
+ // schedule compaction
+ // 5: upsert of 100 records (updates only based on round 3)
+ // schedule clustering
+ // 6: bulk_insert of 100 records
+ for (int i = 0; i < 6; i++) {
+ WriteOperationType opType = i < 4 ? BULK_INSERT : UPSERT;
+ List<HoodieRecord> recordsForUpdate = i < 4 ? null : dataBatches.get(3).getRight();
+ dataBatches.add(writeRecords(writeClient, opType, recordsForUpdate, HoodieActiveTimeline.createNewInstantTime()));
+ if (tableType == COPY_ON_WRITE) {
+ if (i == 2) {
+ writeClient.scheduleClustering(Option.empty());
+ }
+ } else if (tableType == MERGE_ON_READ) {
+ if (i == 4) {
+ writeClient.scheduleCompaction(Option.empty());
+ }
+ if (i == 5) {
+ writeClient.scheduleClustering(Option.empty());
+ }
}
}
- }
- dataBatches.add(writeRecords(writeClient, BULK_INSERT, null, HoodieActiveTimeline.createNewInstantTime()));
-
- String latestCommitTimestamp = dataBatches.get(dataBatches.size() - 1).getKey();
- // Pending clustering exists
- Option<HoodieInstant> clusteringInstant =
- metaClient.getActiveTimeline().filterPendingReplaceTimeline()
- .filter(instant -> ClusteringUtils.getClusteringPlan(metaClient, instant).isPresent())
- .firstInstant();
- assertTrue(clusteringInstant.isPresent());
- assertTrue(clusteringInstant.get().getTimestamp().compareTo(latestCommitTimestamp) < 0);
-
- if (tableType == MERGE_ON_READ) {
- // Pending compaction exists
- Option<HoodieInstant> compactionInstant =
- metaClient.getActiveTimeline().filterPendingCompactionTimeline().firstInstant();
- assertTrue(compactionInstant.isPresent());
- assertTrue(compactionInstant.get().getTimestamp().compareTo(latestCommitTimestamp) < 0);
- }
+ dataBatches.add(writeRecords(writeClient, BULK_INSERT, null, HoodieActiveTimeline.createNewInstantTime()));
+
+ String latestCommitTimestamp = dataBatches.get(dataBatches.size() - 1).getKey();
+ // Pending clustering exists
+ Option<HoodieInstant> clusteringInstant =
+ metaClient.getActiveTimeline().filterPendingReplaceTimeline()
+ .filter(instant -> ClusteringUtils.getClusteringPlan(metaClient, instant).isPresent())
+ .firstInstant();
+ assertTrue(clusteringInstant.isPresent());
+ assertTrue(clusteringInstant.get().getTimestamp().compareTo(latestCommitTimestamp) < 0);
+
+ if (tableType == MERGE_ON_READ) {
+ // Pending compaction exists
+ Option<HoodieInstant> compactionInstant =
+ metaClient.getActiveTimeline().filterPendingCompactionTimeline().firstInstant();
+ assertTrue(compactionInstant.isPresent());
+ assertTrue(compactionInstant.get().getTimestamp().compareTo(latestCommitTimestamp) < 0);
+ }
- // test SnapshotLoadQuerySpliiter to split snapshot query .
- // Reads only first commit
- readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT,
- Option.empty(),
- 100,
- dataBatches.get(0).getKey(),
- Option.of(TestSnapshotQuerySplitterImpl.class.getName()));
- writeClient.close();
-
- // The pending tables services should not block the incremental pulls
- // Reads everything up to latest
- readAndAssert(
- IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT,
- Option.empty(),
- 500,
- dataBatches.get(6).getKey());
-
- // Even if the read upto latest is set, if begin timestamp is in active timeline, only incremental should kick in.
- readAndAssert(
- IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT,
- Option.of(dataBatches.get(2).getKey()),
- 200,
- dataBatches.get(6).getKey());
-
- // Reads just the latest
- readAndAssert(
- IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST,
- Option.empty(),
- 100,
- dataBatches.get(6).getKey());
-
- // Ensures checkpoint does not move
- readAndAssert(
- IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST,
- Option.of(dataBatches.get(6).getKey()),
- 0,
- dataBatches.get(6).getKey());
+ // test SnapshotLoadQuerySpliiter to split snapshot query .
+ // Reads only first commit
+ readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT,
+ Option.empty(),
+ 100,
+ dataBatches.get(0).getKey(),
+ Option.of(TestSnapshotQuerySplitterImpl.class.getName()));
+
+ // The pending tables services should not block the incremental pulls
+ // Reads everything up to latest
+ readAndAssert(
+ IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT,
+ Option.empty(),
+ 500,
+ dataBatches.get(6).getKey());
+
+ // Even if the read upto latest is set, if begin timestamp is in active timeline, only incremental should kick in.
+ readAndAssert(
+ IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT,
+ Option.of(dataBatches.get(2).getKey()),
+ 200,
+ dataBatches.get(6).getKey());
+
+ // Reads just the latest
+ readAndAssert(
+ IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST,
+ Option.empty(),
+ 100,
+ dataBatches.get(6).getKey());
+
+ // Ensures checkpoint does not move
+ readAndAssert(
+ IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST,
+ Option.of(dataBatches.get(6).getKey()),
+ 0,
+ dataBatches.get(6).getKey());
+ }
}
private void readAndAssert(IncrSourceHelper.MissingCheckpointStrategy missingCheckpointStrategy, Option<String> checkpointToPull, int expectedCount,
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestS3EventsHoodieIncrSource.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestS3EventsHoodieIncrSource.java
index d40d7adce52..7d58d21d874 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestS3EventsHoodieIncrSource.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestS3EventsHoodieIncrSource.java
@@ -41,13 +41,13 @@ import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.sources.helpers.CloudDataFetcher;
import org.apache.hudi.utilities.sources.helpers.IncrSourceHelper;
import org.apache.hudi.utilities.sources.helpers.QueryRunner;
+import org.apache.hudi.utilities.sources.helpers.TestCloudObjectsSelectorCommon;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
-import org.apache.hudi.utilities.sources.helpers.TestCloudObjectsSelectorCommon;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
@@ -207,18 +207,19 @@ public class TestS3EventsHoodieIncrSource extends SparkClientFunctionalTestHarne
private Pair<String, List<HoodieRecord>> writeS3MetadataRecords(String commitTime) throws IOException {
HoodieWriteConfig writeConfig = getWriteConfig();
- SparkRDDWriteClient writeClient = getHoodieWriteClient(writeConfig);
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(writeConfig)) {
- writeClient.startCommitWithTime(commitTime);
- List<HoodieRecord> s3MetadataRecords = Arrays.asList(
- generateS3EventMetadata(commitTime, "bucket-1", "data-file-1.json", 1L)
- );
- JavaRDD<WriteStatus> result = writeClient.upsert(jsc().parallelize(s3MetadataRecords, 1), commitTime);
+ writeClient.startCommitWithTime(commitTime);
+ List<HoodieRecord> s3MetadataRecords = Arrays.asList(
+ generateS3EventMetadata(commitTime, "bucket-1", "data-file-1.json", 1L)
+ );
+ JavaRDD<WriteStatus> result = writeClient.upsert(jsc().parallelize(s3MetadataRecords, 1), commitTime);
- List<WriteStatus> statuses = result.collect();
- assertNoWriteErrors(statuses);
+ List<WriteStatus> statuses = result.collect();
+ assertNoWriteErrors(statuses);
- return Pair.of(commitTime, s3MetadataRecords);
+ return Pair.of(commitTime, s3MetadataRecords);
+ }
}
@Test