You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2021/11/19 20:51:29 UTC
[GitHub] [hudi] manojpec opened a new pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
manojpec opened a new pull request #4046:
URL: https://github.com/apache/hudi/pull/4046
## What is the purpose of the pull request
* Fixed the test not to do compaction for COW table
* Since the thread pool can pick up the multi writer operations in any order,
the test is flaky. Added a latch to make it more ordered. Still the thread
pool count and the order of operation has to be fixed in the test.
## Brief change log
*(for example:)*
- *Modify AnnotationLocation checkstyle rule in checkstyle.xml*
## Verify this pull request
*(Please pick either of the following options)*
This pull request is a trivial rework / code cleanup without any test coverage.
*(or)*
This pull request is already covered by existing tests, such as *(please describe tests)*.
(or)
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end.*
- *Added HoodieClientWriteTest to verify the change.*
- *Manually verified the change by running a job locally.*
## Committer checklist
- [ ] Has a corresponding JIRA in PR title & commit
- [ ] Commit message is descriptive of the change
- [ ] CI is green
- [ ] Necessary doc changes done or have another open PR
- [ ] For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot removed a comment on pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
hudi-bot removed a comment on pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#issuecomment-974495373
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516",
"triggerID" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7ca3b984957252fb5400f3d1b471bf75a6d6e330 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] nsivabalan commented on pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
nsivabalan commented on pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#issuecomment-991397729
synced up directly. changes look good to me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot removed a comment on pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
hudi-bot removed a comment on pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#issuecomment-990292349
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516",
"triggerID" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"triggerType" : "PUSH"
}, {
"hash" : "042f367161decec941613cd60cd4fe3c129029e2",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "042f367161decec941613cd60cd4fe3c129029e2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7ca3b984957252fb5400f3d1b471bf75a6d6e330 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516)
* 042f367161decec941613cd60cd4fe3c129029e2 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#issuecomment-990296740
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516",
"triggerID" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"triggerType" : "PUSH"
}, {
"hash" : "042f367161decec941613cd60cd4fe3c129029e2",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=4137",
"triggerID" : "042f367161decec941613cd60cd4fe3c129029e2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7ca3b984957252fb5400f3d1b471bf75a6d6e330 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516)
* 042f367161decec941613cd60cd4fe3c129029e2 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=4137)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#issuecomment-990292349
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516",
"triggerID" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"triggerType" : "PUSH"
}, {
"hash" : "042f367161decec941613cd60cd4fe3c129029e2",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "042f367161decec941613cd60cd4fe3c129029e2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7ca3b984957252fb5400f3d1b471bf75a6d6e330 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516)
* 042f367161decec941613cd60cd4fe3c129029e2 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] manojpec commented on pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
manojpec commented on pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#issuecomment-974560935
@nsivabalan
CI passed in the re-run - https://dev.azure.com/apache-hudi-ci-org/apache-hudi-ci/_build/results?buildId=3516&view=results
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#issuecomment-974467606
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516",
"triggerID" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7ca3b984957252fb5400f3d1b471bf75a6d6e330 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#issuecomment-990348621
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516",
"triggerID" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"triggerType" : "PUSH"
}, {
"hash" : "042f367161decec941613cd60cd4fe3c129029e2",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=4137",
"triggerID" : "042f367161decec941613cd60cd4fe3c129029e2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 042f367161decec941613cd60cd4fe3c129029e2 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=4137)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] nsivabalan merged pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
nsivabalan merged pull request #4046:
URL: https://github.com/apache/hudi/pull/4046
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] manojpec commented on a change in pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
manojpec commented on a change in pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#discussion_r766171591
##########
File path: hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
##########
@@ -228,78 +231,88 @@ private void testMultiWriterWithAsyncTableServicesWithConflict(HoodieTableType t
createCommitWithUpserts(cfg, client, "002", "000", "003", 100);
validInstants.add("002");
validInstants.add("003");
- ExecutorService executors = Executors.newFixedThreadPool(2);
+
+ final int threadCount = 2;
+ CountDownLatch countDownLatch = new CountDownLatch(threadCount);
+ ExecutorService executors = Executors.newFixedThreadPool(threadCount);
SparkRDDWriteClient client1 = getHoodieWriteClient(cfg);
SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+
// Create upserts, schedule cleaning, schedule compaction in parallel
Future future1 = executors.submit(() -> {
- String newCommitTime = "004";
- int numRecords = 100;
- String commitTimeBetweenPrevAndNew = "002";
- try {
- createCommitWithUpserts(cfg, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- Assertions.fail("Conflicts not handled correctly");
- }
- validInstants.add("004");
- } catch (Exception e1) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- Assertions.assertTrue(e1 instanceof HoodieWriteConflictException);
- }
+ final String newCommitTime = "004";
+ final int numRecords = 100;
+ final String commitTimeBetweenPrevAndNew = "002";
+
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ assertThrows(HoodieWriteConflictException.class, () -> {
+ createCommitWithUpserts(cfg, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
+ countDownLatch.countDown();
+ try {
+ countDownLatch.await();
Review comment:
fixed it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] nsivabalan commented on a change in pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
nsivabalan commented on a change in pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#discussion_r766182027
##########
File path: hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
##########
@@ -238,82 +255,101 @@ private void testMultiWriterWithAsyncTableServicesWithConflict(HoodieTableType t
createCommitWithUpserts(cfg, client, "002", "000", "003", 100);
validInstants.add("002");
validInstants.add("003");
- ExecutorService executors = Executors.newFixedThreadPool(2);
- // write config with clustering enabled
- HoodieWriteConfig cfg2 = writeConfigBuilder
- .withClusteringConfig(HoodieClusteringConfig.newBuilder().withInlineClustering(true).withInlineClusteringNumCommits(1).build())
+
+ // Three clients running actions in parallel
+ final int threadCount = 3;
+ final CountDownLatch scheduleCountDownLatch = new CountDownLatch(threadCount);
+ final ExecutorService executors = Executors.newFixedThreadPool(threadCount);
+
+ // Write config with clustering enabled
+ final HoodieWriteConfig cfg2 = writeConfigBuilder
+ .withClusteringConfig(HoodieClusteringConfig.newBuilder()
+ .withInlineClustering(true)
+ .withInlineClusteringNumCommits(1)
+ .build())
.build();
- SparkRDDWriteClient client1 = getHoodieWriteClient(cfg2);
- SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+ final SparkRDDWriteClient client1 = getHoodieWriteClient(cfg2);
+ final SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+ final SparkRDDWriteClient client3 = getHoodieWriteClient(cfg);
+
// Create upserts, schedule cleaning, schedule compaction in parallel
Future future1 = executors.submit(() -> {
- String newCommitTime = "004";
- int numRecords = 100;
- String commitTimeBetweenPrevAndNew = "002";
- try {
- createCommitWithUpserts(cfg2, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- fail("Conflicts not handled correctly");
- }
- validInstants.add("004");
- } catch (Exception e1) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- assertTrue(e1 instanceof HoodieWriteConflictException);
- }
+ final String newCommitTime = "004";
+ final int numRecords = 100;
+ final String commitTimeBetweenPrevAndNew = "002";
+
+ // We want the upsert to go through only after the compaction
+ // and cleaning schedule completion. So, waiting on latch here.
+ latchCountDownAndWait(scheduleCountDownLatch, 30000);
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ // Since the compaction already went in, this upsert has
+ // to fail
+ assertThrows(IllegalArgumentException.class, () -> {
+ createCommitWithUpserts(cfg, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
+ });
+ } else {
+ // We don't have the compaction for COW and so this upsert
+ // has to pass
+ assertDoesNotThrow(() -> {
+ createCommitWithUpserts(cfg, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
+ });
+ validInstants.add(newCommitTime);
}
});
+
Future future2 = executors.submit(() -> {
- try {
- client2.scheduleTableService("005", Option.empty(), TableServiceType.COMPACT);
- } catch (Exception e2) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- throw new RuntimeException(e2);
- }
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ assertDoesNotThrow(() -> {
+ client2.scheduleTableService("005", Option.empty(), TableServiceType.COMPACT);
+ });
}
+ latchCountDownAndWait(scheduleCountDownLatch, 30000);
});
+
Future future3 = executors.submit(() -> {
- try {
- client2.scheduleTableService("006", Option.empty(), TableServiceType.CLEAN);
- } catch (Exception e2) {
- throw new RuntimeException(e2);
- }
+ assertDoesNotThrow(() -> {
+ latchCountDownAndWait(scheduleCountDownLatch, 30000);
+ client3.scheduleTableService("006", Option.empty(), TableServiceType.CLEAN);
+ });
});
future1.get();
future2.get();
future3.get();
+
+ CountDownLatch runCountDownLatch = new CountDownLatch(threadCount);
// Create inserts, run cleaning, run compaction in parallel
future1 = executors.submit(() -> {
- String newCommitTime = "007";
- int numRecords = 100;
- try {
- createCommitWithInserts(cfg2, client1, "003", newCommitTime, numRecords);
+ final String newCommitTime = "007";
+ final int numRecords = 100;
+ latchCountDownAndWait(runCountDownLatch, 30000);
+ assertDoesNotThrow(() -> {
+ createCommitWithInserts(cfg, client1, "003", newCommitTime, numRecords);
validInstants.add("007");
- } catch (Exception e1) {
- throw new RuntimeException(e1);
- }
+ });
});
+
future2 = executors.submit(() -> {
- try {
- JavaRDD<WriteStatus> writeStatusJavaRDD = (JavaRDD<WriteStatus>) client2.compact("005");
- client2.commitCompaction("005", writeStatusJavaRDD, Option.empty());
- validInstants.add("005");
- } catch (Exception e2) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- throw new RuntimeException(e2);
- }
+ latchCountDownAndWait(runCountDownLatch, 30000);
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
Review comment:
in the 2nd batch, all 3 can go in parallel right. why do we need countdown latch ? can you help me understand
##########
File path: hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
##########
@@ -238,82 +255,101 @@ private void testMultiWriterWithAsyncTableServicesWithConflict(HoodieTableType t
createCommitWithUpserts(cfg, client, "002", "000", "003", 100);
validInstants.add("002");
validInstants.add("003");
- ExecutorService executors = Executors.newFixedThreadPool(2);
- // write config with clustering enabled
- HoodieWriteConfig cfg2 = writeConfigBuilder
- .withClusteringConfig(HoodieClusteringConfig.newBuilder().withInlineClustering(true).withInlineClusteringNumCommits(1).build())
+
+ // Three clients running actions in parallel
+ final int threadCount = 3;
+ final CountDownLatch scheduleCountDownLatch = new CountDownLatch(threadCount);
+ final ExecutorService executors = Executors.newFixedThreadPool(threadCount);
+
+ // Write config with clustering enabled
+ final HoodieWriteConfig cfg2 = writeConfigBuilder
+ .withClusteringConfig(HoodieClusteringConfig.newBuilder()
+ .withInlineClustering(true)
+ .withInlineClusteringNumCommits(1)
+ .build())
.build();
- SparkRDDWriteClient client1 = getHoodieWriteClient(cfg2);
- SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+ final SparkRDDWriteClient client1 = getHoodieWriteClient(cfg2);
+ final SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+ final SparkRDDWriteClient client3 = getHoodieWriteClient(cfg);
+
// Create upserts, schedule cleaning, schedule compaction in parallel
Future future1 = executors.submit(() -> {
- String newCommitTime = "004";
- int numRecords = 100;
- String commitTimeBetweenPrevAndNew = "002";
- try {
- createCommitWithUpserts(cfg2, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- fail("Conflicts not handled correctly");
- }
- validInstants.add("004");
- } catch (Exception e1) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- assertTrue(e1 instanceof HoodieWriteConflictException);
- }
+ final String newCommitTime = "004";
+ final int numRecords = 100;
+ final String commitTimeBetweenPrevAndNew = "002";
+
+ // We want the upsert to go through only after the compaction
+ // and cleaning schedule completion. So, waiting on latch here.
+ latchCountDownAndWait(scheduleCountDownLatch, 30000);
Review comment:
why do we have to instantiate the count down latch with 3. 1 would suffice right. we just need compaction scheduling to happen as first and then rest of the two operations can proceed concurrently. can you help me understand.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] manojpec commented on a change in pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
manojpec commented on a change in pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#discussion_r767074057
##########
File path: hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
##########
@@ -238,82 +255,101 @@ private void testMultiWriterWithAsyncTableServicesWithConflict(HoodieTableType t
createCommitWithUpserts(cfg, client, "002", "000", "003", 100);
validInstants.add("002");
validInstants.add("003");
- ExecutorService executors = Executors.newFixedThreadPool(2);
- // write config with clustering enabled
- HoodieWriteConfig cfg2 = writeConfigBuilder
- .withClusteringConfig(HoodieClusteringConfig.newBuilder().withInlineClustering(true).withInlineClusteringNumCommits(1).build())
+
+ // Three clients running actions in parallel
+ final int threadCount = 3;
+ final CountDownLatch scheduleCountDownLatch = new CountDownLatch(threadCount);
+ final ExecutorService executors = Executors.newFixedThreadPool(threadCount);
+
+ // Write config with clustering enabled
+ final HoodieWriteConfig cfg2 = writeConfigBuilder
+ .withClusteringConfig(HoodieClusteringConfig.newBuilder()
+ .withInlineClustering(true)
+ .withInlineClusteringNumCommits(1)
+ .build())
.build();
- SparkRDDWriteClient client1 = getHoodieWriteClient(cfg2);
- SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+ final SparkRDDWriteClient client1 = getHoodieWriteClient(cfg2);
+ final SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+ final SparkRDDWriteClient client3 = getHoodieWriteClient(cfg);
+
// Create upserts, schedule cleaning, schedule compaction in parallel
Future future1 = executors.submit(() -> {
- String newCommitTime = "004";
- int numRecords = 100;
- String commitTimeBetweenPrevAndNew = "002";
- try {
- createCommitWithUpserts(cfg2, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- fail("Conflicts not handled correctly");
- }
- validInstants.add("004");
- } catch (Exception e1) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- assertTrue(e1 instanceof HoodieWriteConflictException);
- }
+ final String newCommitTime = "004";
+ final int numRecords = 100;
+ final String commitTimeBetweenPrevAndNew = "002";
+
+ // We want the upsert to go through only after the compaction
+ // and cleaning schedule completion. So, waiting on latch here.
+ latchCountDownAndWait(scheduleCountDownLatch, 30000);
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ // Since the compaction already went in, this upsert has
+ // to fail
+ assertThrows(IllegalArgumentException.class, () -> {
+ createCommitWithUpserts(cfg, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
+ });
+ } else {
+ // We don't have the compaction for COW and so this upsert
+ // has to pass
+ assertDoesNotThrow(() -> {
+ createCommitWithUpserts(cfg, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
+ });
+ validInstants.add(newCommitTime);
}
});
+
Future future2 = executors.submit(() -> {
- try {
- client2.scheduleTableService("005", Option.empty(), TableServiceType.COMPACT);
- } catch (Exception e2) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- throw new RuntimeException(e2);
- }
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ assertDoesNotThrow(() -> {
+ client2.scheduleTableService("005", Option.empty(), TableServiceType.COMPACT);
+ });
}
+ latchCountDownAndWait(scheduleCountDownLatch, 30000);
});
+
Future future3 = executors.submit(() -> {
- try {
- client2.scheduleTableService("006", Option.empty(), TableServiceType.CLEAN);
- } catch (Exception e2) {
- throw new RuntimeException(e2);
- }
+ assertDoesNotThrow(() -> {
+ latchCountDownAndWait(scheduleCountDownLatch, 30000);
+ client3.scheduleTableService("006", Option.empty(), TableServiceType.CLEAN);
+ });
});
future1.get();
future2.get();
future3.get();
+
+ CountDownLatch runCountDownLatch = new CountDownLatch(threadCount);
// Create inserts, run cleaning, run compaction in parallel
future1 = executors.submit(() -> {
- String newCommitTime = "007";
- int numRecords = 100;
- try {
- createCommitWithInserts(cfg2, client1, "003", newCommitTime, numRecords);
+ final String newCommitTime = "007";
+ final int numRecords = 100;
+ latchCountDownAndWait(runCountDownLatch, 30000);
+ assertDoesNotThrow(() -> {
+ createCommitWithInserts(cfg, client1, "003", newCommitTime, numRecords);
validInstants.add("007");
- } catch (Exception e1) {
- throw new RuntimeException(e1);
- }
+ });
});
+
future2 = executors.submit(() -> {
- try {
- JavaRDD<WriteStatus> writeStatusJavaRDD = (JavaRDD<WriteStatus>) client2.compact("005");
- client2.commitCompaction("005", writeStatusJavaRDD, Option.empty());
- validInstants.add("005");
- } catch (Exception e2) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- throw new RuntimeException(e2);
- }
+ latchCountDownAndWait(runCountDownLatch, 30000);
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
Review comment:
Executor submit is not same as thread started running the operation. By having the count down latch inside the block we are sure all three threads are ready for the operations and are running in parallel. This is a more stricter case than without latch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot removed a comment on pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
hudi-bot removed a comment on pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#issuecomment-974467606
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516",
"triggerID" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7ca3b984957252fb5400f3d1b471bf75a6d6e330 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot removed a comment on pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
hudi-bot removed a comment on pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#issuecomment-974466096
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7ca3b984957252fb5400f3d1b471bf75a6d6e330 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#issuecomment-974495373
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516",
"triggerID" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7ca3b984957252fb5400f3d1b471bf75a6d6e330 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] nsivabalan commented on pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
nsivabalan commented on pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#issuecomment-989349638
@manojpec : once you addressed the comment, let me know. I can take a look
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot removed a comment on pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
hudi-bot removed a comment on pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#issuecomment-990296740
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516",
"triggerID" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"triggerType" : "PUSH"
}, {
"hash" : "042f367161decec941613cd60cd4fe3c129029e2",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=4137",
"triggerID" : "042f367161decec941613cd60cd4fe3c129029e2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7ca3b984957252fb5400f3d1b471bf75a6d6e330 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=3516)
* 042f367161decec941613cd60cd4fe3c129029e2 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=4137)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on a change in pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#discussion_r753816359
##########
File path: hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
##########
@@ -228,78 +231,88 @@ private void testMultiWriterWithAsyncTableServicesWithConflict(HoodieTableType t
createCommitWithUpserts(cfg, client, "002", "000", "003", 100);
validInstants.add("002");
validInstants.add("003");
- ExecutorService executors = Executors.newFixedThreadPool(2);
+
+ final int threadCount = 2;
+ CountDownLatch countDownLatch = new CountDownLatch(threadCount);
+ ExecutorService executors = Executors.newFixedThreadPool(threadCount);
SparkRDDWriteClient client1 = getHoodieWriteClient(cfg);
SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+
// Create upserts, schedule cleaning, schedule compaction in parallel
Future future1 = executors.submit(() -> {
- String newCommitTime = "004";
- int numRecords = 100;
- String commitTimeBetweenPrevAndNew = "002";
- try {
- createCommitWithUpserts(cfg, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- Assertions.fail("Conflicts not handled correctly");
- }
- validInstants.add("004");
- } catch (Exception e1) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- Assertions.assertTrue(e1 instanceof HoodieWriteConflictException);
- }
+ final String newCommitTime = "004";
+ final int numRecords = 100;
+ final String commitTimeBetweenPrevAndNew = "002";
+
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ assertThrows(HoodieWriteConflictException.class, () -> {
+ createCommitWithUpserts(cfg, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
+ countDownLatch.countDown();
+ try {
+ countDownLatch.await();
Review comment:
lets add a large timeout here, to ensure the tests don't just hang on CI?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] nsivabalan commented on a change in pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
nsivabalan commented on a change in pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#discussion_r753804804
##########
File path: hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
##########
@@ -228,78 +231,88 @@ private void testMultiWriterWithAsyncTableServicesWithConflict(HoodieTableType t
createCommitWithUpserts(cfg, client, "002", "000", "003", 100);
validInstants.add("002");
validInstants.add("003");
- ExecutorService executors = Executors.newFixedThreadPool(2);
+
+ final int threadCount = 2;
+ CountDownLatch countDownLatch = new CountDownLatch(threadCount);
+ ExecutorService executors = Executors.newFixedThreadPool(threadCount);
SparkRDDWriteClient client1 = getHoodieWriteClient(cfg);
SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+
// Create upserts, schedule cleaning, schedule compaction in parallel
Future future1 = executors.submit(() -> {
- String newCommitTime = "004";
- int numRecords = 100;
- String commitTimeBetweenPrevAndNew = "002";
- try {
- createCommitWithUpserts(cfg, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- Assertions.fail("Conflicts not handled correctly");
- }
- validInstants.add("004");
- } catch (Exception e1) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- Assertions.assertTrue(e1 instanceof HoodieWriteConflictException);
- }
+ final String newCommitTime = "004";
+ final int numRecords = 100;
+ final String commitTimeBetweenPrevAndNew = "002";
+
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ assertThrows(HoodieWriteConflictException.class, () -> {
+ createCommitWithUpserts(cfg, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
+ countDownLatch.countDown();
+ try {
+ countDownLatch.await();
Review comment:
should we add `countDownLatch.await()` at L284 just after all futures are complete instead of here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] manojpec commented on a change in pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
manojpec commented on a change in pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#discussion_r767073882
##########
File path: hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
##########
@@ -238,82 +255,101 @@ private void testMultiWriterWithAsyncTableServicesWithConflict(HoodieTableType t
createCommitWithUpserts(cfg, client, "002", "000", "003", 100);
validInstants.add("002");
validInstants.add("003");
- ExecutorService executors = Executors.newFixedThreadPool(2);
- // write config with clustering enabled
- HoodieWriteConfig cfg2 = writeConfigBuilder
- .withClusteringConfig(HoodieClusteringConfig.newBuilder().withInlineClustering(true).withInlineClusteringNumCommits(1).build())
+
+ // Three clients running actions in parallel
+ final int threadCount = 3;
+ final CountDownLatch scheduleCountDownLatch = new CountDownLatch(threadCount);
+ final ExecutorService executors = Executors.newFixedThreadPool(threadCount);
+
+ // Write config with clustering enabled
+ final HoodieWriteConfig cfg2 = writeConfigBuilder
+ .withClusteringConfig(HoodieClusteringConfig.newBuilder()
+ .withInlineClustering(true)
+ .withInlineClusteringNumCommits(1)
+ .build())
.build();
- SparkRDDWriteClient client1 = getHoodieWriteClient(cfg2);
- SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+ final SparkRDDWriteClient client1 = getHoodieWriteClient(cfg2);
+ final SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+ final SparkRDDWriteClient client3 = getHoodieWriteClient(cfg);
+
// Create upserts, schedule cleaning, schedule compaction in parallel
Future future1 = executors.submit(() -> {
- String newCommitTime = "004";
- int numRecords = 100;
- String commitTimeBetweenPrevAndNew = "002";
- try {
- createCommitWithUpserts(cfg2, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- fail("Conflicts not handled correctly");
- }
- validInstants.add("004");
- } catch (Exception e1) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- assertTrue(e1 instanceof HoodieWriteConflictException);
- }
+ final String newCommitTime = "004";
+ final int numRecords = 100;
+ final String commitTimeBetweenPrevAndNew = "002";
+
+ // We want the upsert to go through only after the compaction
+ // and cleaning schedule completion. So, waiting on latch here.
+ latchCountDownAndWait(scheduleCountDownLatch, 30000);
Review comment:
By having the count as 3, we can make sure all the clients have been successfully scheduled at ready to start to next set of actions in parallel. Your suggested approach of 1 will also work here (but without they doing countdown). Having the count at 3 gives us more controlled parallelism.
##########
File path: hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
##########
@@ -238,82 +255,101 @@ private void testMultiWriterWithAsyncTableServicesWithConflict(HoodieTableType t
createCommitWithUpserts(cfg, client, "002", "000", "003", 100);
validInstants.add("002");
validInstants.add("003");
- ExecutorService executors = Executors.newFixedThreadPool(2);
- // write config with clustering enabled
- HoodieWriteConfig cfg2 = writeConfigBuilder
- .withClusteringConfig(HoodieClusteringConfig.newBuilder().withInlineClustering(true).withInlineClusteringNumCommits(1).build())
+
+ // Three clients running actions in parallel
+ final int threadCount = 3;
+ final CountDownLatch scheduleCountDownLatch = new CountDownLatch(threadCount);
+ final ExecutorService executors = Executors.newFixedThreadPool(threadCount);
+
+ // Write config with clustering enabled
+ final HoodieWriteConfig cfg2 = writeConfigBuilder
+ .withClusteringConfig(HoodieClusteringConfig.newBuilder()
+ .withInlineClustering(true)
+ .withInlineClusteringNumCommits(1)
+ .build())
.build();
- SparkRDDWriteClient client1 = getHoodieWriteClient(cfg2);
- SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+ final SparkRDDWriteClient client1 = getHoodieWriteClient(cfg2);
+ final SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+ final SparkRDDWriteClient client3 = getHoodieWriteClient(cfg);
+
// Create upserts, schedule cleaning, schedule compaction in parallel
Future future1 = executors.submit(() -> {
- String newCommitTime = "004";
- int numRecords = 100;
- String commitTimeBetweenPrevAndNew = "002";
- try {
- createCommitWithUpserts(cfg2, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- fail("Conflicts not handled correctly");
- }
- validInstants.add("004");
- } catch (Exception e1) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- assertTrue(e1 instanceof HoodieWriteConflictException);
- }
+ final String newCommitTime = "004";
+ final int numRecords = 100;
+ final String commitTimeBetweenPrevAndNew = "002";
+
+ // We want the upsert to go through only after the compaction
+ // and cleaning schedule completion. So, waiting on latch here.
+ latchCountDownAndWait(scheduleCountDownLatch, 30000);
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ // Since the compaction already went in, this upsert has
+ // to fail
+ assertThrows(IllegalArgumentException.class, () -> {
+ createCommitWithUpserts(cfg, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
+ });
+ } else {
+ // We don't have the compaction for COW and so this upsert
+ // has to pass
+ assertDoesNotThrow(() -> {
+ createCommitWithUpserts(cfg, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
+ });
+ validInstants.add(newCommitTime);
}
});
+
Future future2 = executors.submit(() -> {
- try {
- client2.scheduleTableService("005", Option.empty(), TableServiceType.COMPACT);
- } catch (Exception e2) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- throw new RuntimeException(e2);
- }
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ assertDoesNotThrow(() -> {
+ client2.scheduleTableService("005", Option.empty(), TableServiceType.COMPACT);
+ });
}
+ latchCountDownAndWait(scheduleCountDownLatch, 30000);
});
+
Future future3 = executors.submit(() -> {
- try {
- client2.scheduleTableService("006", Option.empty(), TableServiceType.CLEAN);
- } catch (Exception e2) {
- throw new RuntimeException(e2);
- }
+ assertDoesNotThrow(() -> {
+ latchCountDownAndWait(scheduleCountDownLatch, 30000);
+ client3.scheduleTableService("006", Option.empty(), TableServiceType.CLEAN);
+ });
});
future1.get();
future2.get();
future3.get();
+
+ CountDownLatch runCountDownLatch = new CountDownLatch(threadCount);
// Create inserts, run cleaning, run compaction in parallel
future1 = executors.submit(() -> {
- String newCommitTime = "007";
- int numRecords = 100;
- try {
- createCommitWithInserts(cfg2, client1, "003", newCommitTime, numRecords);
+ final String newCommitTime = "007";
+ final int numRecords = 100;
+ latchCountDownAndWait(runCountDownLatch, 30000);
+ assertDoesNotThrow(() -> {
+ createCommitWithInserts(cfg, client1, "003", newCommitTime, numRecords);
validInstants.add("007");
- } catch (Exception e1) {
- throw new RuntimeException(e1);
- }
+ });
});
+
future2 = executors.submit(() -> {
- try {
- JavaRDD<WriteStatus> writeStatusJavaRDD = (JavaRDD<WriteStatus>) client2.compact("005");
- client2.commitCompaction("005", writeStatusJavaRDD, Option.empty());
- validInstants.add("005");
- } catch (Exception e2) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- throw new RuntimeException(e2);
- }
+ latchCountDownAndWait(runCountDownLatch, 30000);
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
Review comment:
Executor submit is not same as thread started running the operation. By having the count down latch inside the block we are sure all three threads are ready for the operations and are running in parallel. This is a more stricter case than without latch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] manojpec commented on a change in pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
manojpec commented on a change in pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#discussion_r767073882
##########
File path: hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
##########
@@ -238,82 +255,101 @@ private void testMultiWriterWithAsyncTableServicesWithConflict(HoodieTableType t
createCommitWithUpserts(cfg, client, "002", "000", "003", 100);
validInstants.add("002");
validInstants.add("003");
- ExecutorService executors = Executors.newFixedThreadPool(2);
- // write config with clustering enabled
- HoodieWriteConfig cfg2 = writeConfigBuilder
- .withClusteringConfig(HoodieClusteringConfig.newBuilder().withInlineClustering(true).withInlineClusteringNumCommits(1).build())
+
+ // Three clients running actions in parallel
+ final int threadCount = 3;
+ final CountDownLatch scheduleCountDownLatch = new CountDownLatch(threadCount);
+ final ExecutorService executors = Executors.newFixedThreadPool(threadCount);
+
+ // Write config with clustering enabled
+ final HoodieWriteConfig cfg2 = writeConfigBuilder
+ .withClusteringConfig(HoodieClusteringConfig.newBuilder()
+ .withInlineClustering(true)
+ .withInlineClusteringNumCommits(1)
+ .build())
.build();
- SparkRDDWriteClient client1 = getHoodieWriteClient(cfg2);
- SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+ final SparkRDDWriteClient client1 = getHoodieWriteClient(cfg2);
+ final SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+ final SparkRDDWriteClient client3 = getHoodieWriteClient(cfg);
+
// Create upserts, schedule cleaning, schedule compaction in parallel
Future future1 = executors.submit(() -> {
- String newCommitTime = "004";
- int numRecords = 100;
- String commitTimeBetweenPrevAndNew = "002";
- try {
- createCommitWithUpserts(cfg2, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- fail("Conflicts not handled correctly");
- }
- validInstants.add("004");
- } catch (Exception e1) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- assertTrue(e1 instanceof HoodieWriteConflictException);
- }
+ final String newCommitTime = "004";
+ final int numRecords = 100;
+ final String commitTimeBetweenPrevAndNew = "002";
+
+ // We want the upsert to go through only after the compaction
+ // and cleaning schedule completion. So, waiting on latch here.
+ latchCountDownAndWait(scheduleCountDownLatch, 30000);
Review comment:
By having the count as 3, we can make sure all the clients have been successfully scheduled at ready to start to next set of actions in parallel. Your suggested approach of 1 will also work here (but without they doing countdown). Having the count at 3 gives us more controlled parallelism.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] nsivabalan commented on pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
nsivabalan commented on pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#issuecomment-991397729
synced up directly. changes look good to me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] nsivabalan merged pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
nsivabalan merged pull request #4046:
URL: https://github.com/apache/hudi/pull/4046
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#issuecomment-974466096
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7ca3b984957252fb5400f3d1b471bf75a6d6e330",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7ca3b984957252fb5400f3d1b471bf75a6d6e330 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on a change in pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#discussion_r753816359
##########
File path: hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
##########
@@ -228,78 +231,88 @@ private void testMultiWriterWithAsyncTableServicesWithConflict(HoodieTableType t
createCommitWithUpserts(cfg, client, "002", "000", "003", 100);
validInstants.add("002");
validInstants.add("003");
- ExecutorService executors = Executors.newFixedThreadPool(2);
+
+ final int threadCount = 2;
+ CountDownLatch countDownLatch = new CountDownLatch(threadCount);
+ ExecutorService executors = Executors.newFixedThreadPool(threadCount);
SparkRDDWriteClient client1 = getHoodieWriteClient(cfg);
SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+
// Create upserts, schedule cleaning, schedule compaction in parallel
Future future1 = executors.submit(() -> {
- String newCommitTime = "004";
- int numRecords = 100;
- String commitTimeBetweenPrevAndNew = "002";
- try {
- createCommitWithUpserts(cfg, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- Assertions.fail("Conflicts not handled correctly");
- }
- validInstants.add("004");
- } catch (Exception e1) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- Assertions.assertTrue(e1 instanceof HoodieWriteConflictException);
- }
+ final String newCommitTime = "004";
+ final int numRecords = 100;
+ final String commitTimeBetweenPrevAndNew = "002";
+
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ assertThrows(HoodieWriteConflictException.class, () -> {
+ createCommitWithUpserts(cfg, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
+ countDownLatch.countDown();
+ try {
+ countDownLatch.await();
Review comment:
lets add a large timeout here, to ensure the tests don't just hang on CI?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] nsivabalan commented on a change in pull request #4046: [HUDI-2527] Multi writer test with conflicting async table services
Posted by GitBox <gi...@apache.org>.
nsivabalan commented on a change in pull request #4046:
URL: https://github.com/apache/hudi/pull/4046#discussion_r753804804
##########
File path: hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
##########
@@ -228,78 +231,88 @@ private void testMultiWriterWithAsyncTableServicesWithConflict(HoodieTableType t
createCommitWithUpserts(cfg, client, "002", "000", "003", 100);
validInstants.add("002");
validInstants.add("003");
- ExecutorService executors = Executors.newFixedThreadPool(2);
+
+ final int threadCount = 2;
+ CountDownLatch countDownLatch = new CountDownLatch(threadCount);
+ ExecutorService executors = Executors.newFixedThreadPool(threadCount);
SparkRDDWriteClient client1 = getHoodieWriteClient(cfg);
SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+
// Create upserts, schedule cleaning, schedule compaction in parallel
Future future1 = executors.submit(() -> {
- String newCommitTime = "004";
- int numRecords = 100;
- String commitTimeBetweenPrevAndNew = "002";
- try {
- createCommitWithUpserts(cfg, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- Assertions.fail("Conflicts not handled correctly");
- }
- validInstants.add("004");
- } catch (Exception e1) {
- if (tableType == HoodieTableType.MERGE_ON_READ) {
- Assertions.assertTrue(e1 instanceof HoodieWriteConflictException);
- }
+ final String newCommitTime = "004";
+ final int numRecords = 100;
+ final String commitTimeBetweenPrevAndNew = "002";
+
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ assertThrows(HoodieWriteConflictException.class, () -> {
+ createCommitWithUpserts(cfg, client1, "003", commitTimeBetweenPrevAndNew, newCommitTime, numRecords);
+ countDownLatch.countDown();
+ try {
+ countDownLatch.await();
Review comment:
should we add `countDownLatch.await()` at L284 just after all futures are complete instead of here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org