yihua commented on code in PR #19062:
URL: https://github.com/apache/hudi/pull/19062#discussion_r3470117343
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java:
##########
@@ -483,71 +501,99 @@ private void
testHoodieClientBasicMultiWriterWithEarlyConflictDetection(String t
writeConfig = buildWriteConfigForEarlyConflictDetect(markerType,
properties, InProcessLockProvider.class, earlyConflictDetectionStrategy);
}
- final SparkRDDWriteClient client1 = getHoodieWriteClient(writeConfig);
+ SparkRDDWriteClient client1 = null;
+ SparkRDDWriteClient client2 = null;
+ SparkRDDWriteClient client3 = null;
+ SparkRDDWriteClient client4 = null;
+ try {
+ client1 = getHoodieWriteClient(writeConfig);
- // Create the first commit
- final String nextCommitTime1 = "001";
- createCommitWithInserts(writeConfig, client1, "000", nextCommitTime1, 200);
+ // Create the first commit
+ final String nextCommitTime1 = "001";
+ createCommitWithInserts(writeConfig, client1, "000", nextCommitTime1,
200);
- final SparkRDDWriteClient client2 = getHoodieWriteClient(writeConfig);
- final SparkRDDWriteClient client3 = getHoodieWriteClient(writeConfig);
+ client2 = getHoodieWriteClient(writeConfig);
+ client3 = getHoodieWriteClient(writeConfig);
- final String nextCommitTime2 = "002";
+ final String nextCommitTime2 = "002";
- // start to write commit 002
- final JavaRDD<WriteStatus> writeStatusList2 =
startCommitForUpdate(writeConfig, client2, nextCommitTime2, 100);
+ // start to write commit 002
+ final SparkRDDWriteClient finalClient2 = client2;
+ final JavaRDD<WriteStatus> writeStatusList2 =
startCommitForUpdate(writeConfig, client2, nextCommitTime2, 100);
- // start to write commit 003
- // this commit 003 will fail quickly because early conflict detection
before create marker.
- final String nextCommitTime3 = "003";
- assertThrows(SparkException.class, () -> {
- final JavaRDD<WriteStatus> writeStatusList3 =
- startCommitForUpdate(writeConfig, client3, nextCommitTime3, 100);
- client3.commit(nextCommitTime3, writeStatusList3);
- }, "Early conflict detected but cannot resolve conflicts for overlapping
writes");
+ // start to write commit 003
+ // this commit 003 will fail quickly because early conflict detection
before create marker.
+ final String nextCommitTime3 = "003";
+ final SparkRDDWriteClient finalClient3 = client3;
+ assertThrows(SparkException.class, () -> {
+ final JavaRDD<WriteStatus> writeStatusList3 =
+ startCommitForUpdate(writeConfig, finalClient3, nextCommitTime3,
100);
+ finalClient3.commit(nextCommitTime3, writeStatusList3);
+ }, "Early conflict detected but cannot resolve conflicts for overlapping
writes");
- // start to commit 002 and success
- assertDoesNotThrow(() -> {
- client2.commit(nextCommitTime2, writeStatusList2);
- });
-
- HoodieWriteConfig config4 =
- HoodieWriteConfig.newBuilder().withProperties(writeConfig.getProps())
- .withHeartbeatIntervalInMs(heartBeatIntervalForCommit4).build();
- final SparkRDDWriteClient client4 = getHoodieWriteClient(config4);
+ // start to commit 002 and success
+ assertDoesNotThrow(() -> {
+ finalClient2.commit(nextCommitTime2, writeStatusList2);
+ });
- StoragePath heartbeatFilePath = new StoragePath(
- HoodieTableMetaClient.getHeartbeatFolderPath(basePath) +
StoragePath.SEPARATOR + nextCommitTime3);
- storage.create(heartbeatFilePath, true);
+ HoodieWriteConfig config4 =
+ HoodieWriteConfig.newBuilder().withProperties(writeConfig.getProps())
+ .withHeartbeatIntervalInMs(heartBeatIntervalForCommit4).build();
+ client4 = getHoodieWriteClient(config4);
- // Wait for heart beat expired for failed commitTime3 "003"
- // Otherwise commit4 still can see conflict between failed write 003.
- Thread.sleep(heartBeatIntervalForCommit4 * 2);
+ StoragePath heartbeatFilePath = new StoragePath(
+ HoodieTableMetaClient.getHeartbeatFolderPath(basePath) +
StoragePath.SEPARATOR + nextCommitTime3);
+ storage.create(heartbeatFilePath, true);
- final String nextCommitTime4 = "004";
- assertDoesNotThrow(() -> {
- final JavaRDD<WriteStatus> writeStatusList4 =
- startCommitForUpdate(writeConfig, client4, nextCommitTime4, 100);
- client4.commit(nextCommitTime4, writeStatusList4);
- });
+ // Wait for heart beat expired for failed commitTime3 "003"
+ // Otherwise commit4 still can see conflict between failed write 003.
+ Thread.sleep(heartBeatIntervalForCommit4 * 2);
- List<String> completedInstant =
metaClient.reloadActiveTimeline().getCommitsTimeline()
- .filterCompletedInstants().getInstants().stream()
- .map(HoodieInstant::requestedTime).collect(Collectors.toList());
+ final String nextCommitTime4 = "004";
+ final SparkRDDWriteClient finalClient4 = client4;
+ assertDoesNotThrow(() -> {
+ final JavaRDD<WriteStatus> writeStatusList4 =
+ startCommitForUpdate(writeConfig, finalClient4, nextCommitTime4,
100);
+ finalClient4.commit(nextCommitTime4, writeStatusList4);
+ });
- assertEquals(3, completedInstant.size());
- assertTrue(completedInstant.contains(nextCommitTime1));
- assertTrue(completedInstant.contains(nextCommitTime2));
- assertTrue(completedInstant.contains(nextCommitTime4));
+ List<String> completedInstant =
metaClient.reloadActiveTimeline().getCommitsTimeline()
+ .filterCompletedInstants().getInstants().stream()
+ .map(HoodieInstant::requestedTime).collect(Collectors.toList());
+
+ assertEquals(3, completedInstant.size());
+ assertTrue(completedInstant.contains(nextCommitTime1));
+ assertTrue(completedInstant.contains(nextCommitTime2));
+ assertTrue(completedInstant.contains(nextCommitTime4));
+
+ FileIOUtils.deleteDirectory(new File(basePath));
+ } finally {
+ // Close the server and write clients on every exit path, including
+ // failed assertions; see the currentTestingServer field comment for the
+ // surefire leak concern.
+ if (server != null) {
+ try {
+ server.close();
+ } catch (Exception e) {
+ LOG.warn("Failed to close TestingServer", e);
+ }
+ }
+ closeQuietly(client1);
+ closeQuietly(client2);
+ closeQuietly(client3);
+ closeQuietly(client4);
+ }
+ }
- FileIOUtils.deleteDirectory(new File(basePath));
- if (server != null) {
- server.close();
+ private static void closeQuietly(SparkRDDWriteClient client) {
+ if (client == null) {
+ return;
+ }
+ try {
+ client.close();
+ } catch (Exception e) {
+ LOG.warn("Failed to close SparkRDDWriteClient", e);
}
Review Comment:
There is `FileIOUtils.closeQuietly`. Could we reuse it?
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java:
##########
@@ -175,6 +183,15 @@ public void setUpMORTestTable() throws IOException {
@AfterEach
public void clean() throws IOException {
+ if (currentTestingServer != null) {
+ try {
+ currentTestingServer.close();
+ } catch (Exception e) {
+ LOG.warn("Failed to close TestingServer in @AfterEach", e);
+ } finally {
+ currentTestingServer = null;
+ }
+ }
Review Comment:
Is there a necessity to add this cleanup if
`testHoodieClientBasicMultiWriterWithEarlyConflictDetection` always closes the
server?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
