You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@uniffle.apache.org by "zuston (via GitHub)" <gi...@apache.org> on 2024/05/30 06:43:23 UTC
[PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
zuston opened a new pull request, #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758
### What changes were proposed in this pull request?
Add block number check on getting shuffle result
### Why are the changes needed?
Data validation, ensure data stable and correct
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing tests.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "maobaolong (via GitHub)" <gi...@apache.org>.
maobaolong commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1704806720
##########
server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java:
##########
@@ -506,14 +506,23 @@ public void reportShuffleResult(
"appId[" + appId + "], shuffleId[" + shuffleId + "], taskAttemptId[" + taskAttemptId + "]";
try {
+ int expectedBlockCount = partitionToBlockIds.values().stream().mapToInt(x -> x.length).sum();
LOG.info(
- "Report "
- + partitionToBlockIds.size()
- + " blocks as shuffle result for the task of "
- + requestInfo);
- shuffleServer
- .getShuffleTaskManager()
- .addFinishedBlockIds(appId, shuffleId, partitionToBlockIds, bitmapNum);
+ "Accepted blockIds report for {} blocks across {} partitions as shuffle result for task {}",
+ expectedBlockCount,
+ partitionToBlockIds.size(),
+ request);
+ int updatedBlockCount =
+ shuffleServer
+ .getShuffleTaskManager()
+ .addFinishedBlockIds(appId, shuffleId, partitionToBlockIds, bitmapNum);
+ if (expectedBlockCount != updatedBlockCount) {
+ LOG.warn(
+ "Existing {} duplicated blockIds on blockId report for appId: {}, shuffleId: {}",
Review Comment:
Unfortunately, we encountered this warn level log these days, it make us stressed.
Is there indicate something wrong? Can you explain what rare condition scenario can make updatedBlockCount less than expectedBlockCount?
Why there are duplicated blockIds on blockId report?
##########
server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java:
##########
@@ -506,14 +506,23 @@ public void reportShuffleResult(
"appId[" + appId + "], shuffleId[" + shuffleId + "], taskAttemptId[" + taskAttemptId + "]";
try {
+ int expectedBlockCount = partitionToBlockIds.values().stream().mapToInt(x -> x.length).sum();
LOG.info(
- "Report "
- + partitionToBlockIds.size()
- + " blocks as shuffle result for the task of "
- + requestInfo);
- shuffleServer
- .getShuffleTaskManager()
- .addFinishedBlockIds(appId, shuffleId, partitionToBlockIds, bitmapNum);
+ "Accepted blockIds report for {} blocks across {} partitions as shuffle result for task {}",
Review Comment:
This log is verbose and fill server log, is there a better way to reduce it or do you think `DEBUG` level can be better?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston merged PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "EnricoMi (via GitHub)" <gi...@apache.org>.
EnricoMi commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1620176216
##########
server/src/main/java/org/apache/uniffle/server/ShuffleTaskManager.java:
##########
@@ -413,8 +413,14 @@ public void addFinishedBlockIds(
+ " bitmaps!");
}
+ ShuffleTaskInfo taskInfo = getShuffleTaskInfo(appId);
+ if (taskInfo == null) {
+ throw new InvalidRequestException(
+ "ShuffleTaskInfo is not found that should not happen for appId: " + appId);
+ }
for (Map.Entry<Integer, long[]> entry : partitionToBlockIds.entrySet()) {
Integer partitionId = entry.getKey();
+ taskInfo.incrBlockNumber(shuffleId, partitionId, entry.getValue().length);
Review Comment:
Would we ever see a report shuffle result request being retried (as in GRPC, not a stage retry)? If so, we would get in trouble 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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1621758049
##########
server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java:
##########
@@ -506,14 +506,24 @@ public void reportShuffleResult(
"appId[" + appId + "], shuffleId[" + shuffleId + "], taskAttemptId[" + taskAttemptId + "]";
try {
+ int expected =
+ partitionToBlockIds.values().stream().map(x -> x.length).reduce(0, (a, b) -> a + b);
Review Comment:
Sounds good.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1621864810
##########
server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java:
##########
@@ -506,14 +506,23 @@ public void reportShuffleResult(
"appId[" + appId + "], shuffleId[" + shuffleId + "], taskAttemptId[" + taskAttemptId + "]";
try {
+ int expected = partitionToBlockIds.values().stream().mapToInt(x -> x.length).sum();
LOG.info(
- "Report "
- + partitionToBlockIds.size()
- + " blocks as shuffle result for the task of "
- + requestInfo);
- shuffleServer
- .getShuffleTaskManager()
- .addFinishedBlockIds(appId, shuffleId, partitionToBlockIds, bitmapNum);
+ "Accepted blockIds report for {} blocks across {} partitions as shuffle result for task {}",
+ expected,
+ partitionToBlockIds.size(),
+ request);
+ int updated =
+ shuffleServer
+ .getShuffleTaskManager()
+ .addFinishedBlockIds(appId, shuffleId, partitionToBlockIds, bitmapNum);
+ if (expected != updated) {
Review Comment:
`expected` -> `expectedBlockCount`
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1621864242
##########
server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java:
##########
@@ -506,14 +506,23 @@ public void reportShuffleResult(
"appId[" + appId + "], shuffleId[" + shuffleId + "], taskAttemptId[" + taskAttemptId + "]";
try {
+ int expected = partitionToBlockIds.values().stream().mapToInt(x -> x.length).sum();
LOG.info(
- "Report "
- + partitionToBlockIds.size()
- + " blocks as shuffle result for the task of "
- + requestInfo);
- shuffleServer
- .getShuffleTaskManager()
- .addFinishedBlockIds(appId, shuffleId, partitionToBlockIds, bitmapNum);
+ "Accepted blockIds report for {} blocks across {} partitions as shuffle result for task {}",
+ expected,
+ partitionToBlockIds.size(),
+ request);
+ int updated =
Review Comment:
`updated` -> `updatedBlockCount`
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#issuecomment-2138820813
## Test Results
2 163 files - 270 2 163 suites - 270 2h 25m 0s :stopwatch: - 2h 35m 51s
859 tests - 74 850 :white_check_mark: - 82 1 :zzz: ±0 0 :x: ±0 8 :fire: + 8
10 280 runs - 539 10 154 :white_check_mark: - 651 14 :zzz: ±0 0 :x: ±0 112 :fire: +112
For more details on these errors, see [this check](https://github.com/apache/incubator-uniffle/runs/25590246098).
Results for commit da63e437. ± Comparison against base commit a3a49f0e.
<details>
<summary>This pull request <b>removes</b> 74 tests.</summary>
```
org.apache.uniffle.test.AQERepartitionTest ‑ resultCompareTest
org.apache.uniffle.test.AQESkewedJoinTest ‑ resultCompareTest
org.apache.uniffle.test.AQESkewedJoinWithLocalOrderTest ‑ resultCompareTest
org.apache.uniffle.test.AutoAccessTest ‑ test
org.apache.uniffle.test.CombineByKeyTest ‑ combineByKeyTest
org.apache.uniffle.test.ContinuousSelectPartitionStrategyTest ‑ resultCompareTest
org.apache.uniffle.test.DynamicConfTest ‑ dynamicConfTest
org.apache.uniffle.test.DynamicFetchClientConfTest ‑ test
org.apache.uniffle.test.FailingTasksTest ‑ testFailedTasks
org.apache.uniffle.test.GetReaderTest ‑ test
…
```
</details>
[test-results]:data:application/gzip;base64,H4sIALwkWGYC/03MSQ7DIAyF4atErLuwIUy9TEUhSKgZKgKrqHevE2Vavg/zLyymvpvZs+GoxKNhc03lvkPNrqRpJDEagISey3pgpD3Wa67ebwQXfdKXCE+ILvUE10WX85TXXyS5jmsTgRvY5xFFQNmetldPuFW3fUQROZGfhiEVmiw4JbpW6MCjtCiFUsYJ3b7BCyetVgDBRWcj+/0BpVQ9vxMBAAA=
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1621758478
##########
server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java:
##########
@@ -506,14 +506,24 @@ public void reportShuffleResult(
"appId[" + appId + "], shuffleId[" + shuffleId + "], taskAttemptId[" + taskAttemptId + "]";
try {
+ int expected =
Review Comment:
Needn't. The type has indicated everything.
##########
server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java:
##########
@@ -506,14 +506,24 @@ public void reportShuffleResult(
"appId[" + appId + "], shuffleId[" + shuffleId + "], taskAttemptId[" + taskAttemptId + "]";
try {
+ int expected =
+ partitionToBlockIds.values().stream().map(x -> x.length).reduce(0, (a, b) -> a + b);
LOG.info(
- "Report "
- + partitionToBlockIds.size()
- + " blocks as shuffle result for the task of "
- + requestInfo);
- shuffleServer
- .getShuffleTaskManager()
- .addFinishedBlockIds(appId, shuffleId, partitionToBlockIds, bitmapNum);
+ "Accepted {} blockIds report of {} partitions as shuffle result for the task of {}",
Review Comment:
Sounds good.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "rickyma (via GitHub)" <gi...@apache.org>.
rickyma commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1621913972
##########
server/src/main/java/org/apache/uniffle/server/ShuffleTaskManager.java:
##########
@@ -413,15 +422,28 @@ public void addFinishedBlockIds(
+ " bitmaps!");
}
+ ShuffleTaskInfo taskInfo = getShuffleTaskInfo(appId);
+ if (taskInfo == null) {
+ throw new InvalidRequestException(
+ "ShuffleTaskInfo is not found that should not happen for appId: " + appId);
+ }
+ int totalUpdated = 0;
for (Map.Entry<Integer, long[]> entry : partitionToBlockIds.entrySet()) {
Integer partitionId = entry.getKey();
Roaring64NavigableMap bitmap = blockIds[partitionId % bitmapNum];
+ int updated = 0;
synchronized (bitmap) {
for (long blockId : entry.getValue()) {
- bitmap.addLong(blockId);
+ if (!bitmap.contains(blockId)) {
+ bitmap.addLong(blockId);
+ updated++;
Review Comment:
Refer to @jerqi 's suggestion, you may also rename this:
updatedBlockCount
totalUpdatedBlockCount
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1620482339
##########
server/src/main/java/org/apache/uniffle/server/ShuffleTaskManager.java:
##########
@@ -413,8 +413,14 @@ public void addFinishedBlockIds(
+ " bitmaps!");
}
+ ShuffleTaskInfo taskInfo = getShuffleTaskInfo(appId);
+ if (taskInfo == null) {
+ throw new InvalidRequestException(
+ "ShuffleTaskInfo is not found that should not happen for appId: " + appId);
+ }
for (Map.Entry<Integer, long[]> entry : partitionToBlockIds.entrySet()) {
Integer partitionId = entry.getKey();
+ taskInfo.incrBlockNumber(shuffleId, partitionId, entry.getValue().length);
Review Comment:
If the blockId exists in the bitmap, the blockId counter increment will be ignored. This may need to log if having any duplicate blockId occur
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#issuecomment-2144156784
Thanks everyone for your review. Merged.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#issuecomment-2139304425
> Maybe `addFinishedBlockIds` can return int. We can add some logs if we receive some duplicated blocks.
Yes. Let me optimize this part tomorrow.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "rickyma (via GitHub)" <gi...@apache.org>.
rickyma commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1620137150
##########
server/src/main/java/org/apache/uniffle/server/ShuffleTaskInfo.java:
##########
@@ -198,6 +201,25 @@ public Set<Integer> getShuffleIds() {
return partitionDataSizes.keySet();
}
+ public void incrBlockNumber(int shuffleId, int partitionId) {
+ this.partitionBlockCounters
+ .computeIfAbsent(shuffleId, x -> JavaUtils.newConcurrentMap())
+ .computeIfAbsent(partitionId, x -> new AtomicLong())
+ .incrementAndGet();
+ }
+
+ public long getBlockNumber(int shuffleId, int partitionId) {
+ Map<Integer, AtomicLong> partitionBlockCounters = this.partitionBlockCounters.get(shuffleId);
+ if (partitionBlockCounters == null) {
Review Comment:
OK
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "EnricoMi (via GitHub)" <gi...@apache.org>.
EnricoMi commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1620252743
##########
server/src/main/java/org/apache/uniffle/server/ShuffleTaskManager.java:
##########
@@ -413,8 +413,14 @@ public void addFinishedBlockIds(
+ " bitmaps!");
}
+ ShuffleTaskInfo taskInfo = getShuffleTaskInfo(appId);
+ if (taskInfo == null) {
+ throw new InvalidRequestException(
+ "ShuffleTaskInfo is not found that should not happen for appId: " + appId);
+ }
for (Map.Entry<Integer, long[]> entry : partitionToBlockIds.entrySet()) {
Integer partitionId = entry.getKey();
+ taskInfo.incrBlockNumber(shuffleId, partitionId, entry.getValue().length);
Review Comment:
how? where?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#issuecomment-2139036346
Maybe `addFinishedBlockIds` can return int. We can add some logs if we receive some duplicated blocks.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#issuecomment-2138944231
> In which scenario do you expect the block number to disalign? What are you trying to guard against here? One side we add block ids and on the other side we retrieve block ids. You suspect incorrect usage of the API by the client or a bug server side? Sounds like that should be guarded against by testing so we are confident, not in production runtime.
We suspect the race condition may exist on the process of reporting shuffle result, which may cause the blockId miss. But this is not determinzed, and so I want to introduce the extra blockId counter check for every partition. Anyway, fast fail is better than the silcent incorrect succeed.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1621596525
##########
server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java:
##########
@@ -511,9 +511,17 @@ public void reportShuffleResult(
+ partitionToBlockIds.size()
+ " blocks as shuffle result for the task of "
+ requestInfo);
- shuffleServer
- .getShuffleTaskManager()
- .addFinishedBlockIds(appId, shuffleId, partitionToBlockIds, bitmapNum);
+ int duplicatedBlockIdNum =
+ shuffleServer
+ .getShuffleTaskManager()
+ .addFinishedBlockIds(appId, shuffleId, partitionToBlockIds, bitmapNum);
+ if (duplicatedBlockIdNum > 0) {
Review Comment:
`addFinishedBlockIds` return added blocks size. We can get total blocks size by partitionToBlockIds. We can use them to calculate the duplicate blocks size.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1620105211
##########
server/src/main/java/org/apache/uniffle/server/ShuffleTaskInfo.java:
##########
@@ -198,6 +201,25 @@ public Set<Integer> getShuffleIds() {
return partitionDataSizes.keySet();
}
+ public void incrBlockNumber(int shuffleId, int partitionId) {
+ this.partitionBlockCounters
+ .computeIfAbsent(shuffleId, x -> JavaUtils.newConcurrentMap())
+ .computeIfAbsent(partitionId, x -> new AtomicLong())
+ .incrementAndGet();
+ }
+
+ public long getBlockNumber(int shuffleId, int partitionId) {
+ Map<Integer, AtomicLong> partitionBlockCounters = this.partitionBlockCounters.get(shuffleId);
+ if (partitionBlockCounters == null) {
Review Comment:
I don't introduce extra the object creating cost.
##########
server/src/main/java/org/apache/uniffle/server/ShuffleTaskInfo.java:
##########
@@ -198,6 +201,25 @@ public Set<Integer> getShuffleIds() {
return partitionDataSizes.keySet();
}
+ public void incrBlockNumber(int shuffleId, int partitionId) {
+ this.partitionBlockCounters
+ .computeIfAbsent(shuffleId, x -> JavaUtils.newConcurrentMap())
+ .computeIfAbsent(partitionId, x -> new AtomicLong())
+ .incrementAndGet();
+ }
+
+ public long getBlockNumber(int shuffleId, int partitionId) {
+ Map<Integer, AtomicLong> partitionBlockCounters = this.partitionBlockCounters.get(shuffleId);
+ if (partitionBlockCounters == null) {
Review Comment:
I don't want to introduce extra the object creating cost.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1620206847
##########
server/src/main/java/org/apache/uniffle/server/ShuffleTaskManager.java:
##########
@@ -413,8 +413,14 @@ public void addFinishedBlockIds(
+ " bitmaps!");
}
+ ShuffleTaskInfo taskInfo = getShuffleTaskInfo(appId);
+ if (taskInfo == null) {
+ throw new InvalidRequestException(
+ "ShuffleTaskInfo is not found that should not happen for appId: " + appId);
+ }
for (Map.Entry<Integer, long[]> entry : partitionToBlockIds.entrySet()) {
Integer partitionId = entry.getKey();
+ taskInfo.incrBlockNumber(shuffleId, partitionId, entry.getValue().length);
Review Comment:
Updated. The retried request will be filtered.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1621649767
##########
server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java:
##########
@@ -511,9 +511,17 @@ public void reportShuffleResult(
+ partitionToBlockIds.size()
+ " blocks as shuffle result for the task of "
+ requestInfo);
- shuffleServer
- .getShuffleTaskManager()
- .addFinishedBlockIds(appId, shuffleId, partitionToBlockIds, bitmapNum);
+ int duplicatedBlockIdNum =
+ shuffleServer
+ .getShuffleTaskManager()
+ .addFinishedBlockIds(appId, shuffleId, partitionToBlockIds, bitmapNum);
+ if (duplicatedBlockIdNum > 0) {
Review Comment:
Done
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "rickyma (via GitHub)" <gi...@apache.org>.
rickyma commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1620083970
##########
server/src/main/java/org/apache/uniffle/server/ShuffleTaskInfo.java:
##########
@@ -198,6 +201,25 @@ public Set<Integer> getShuffleIds() {
return partitionDataSizes.keySet();
}
+ public void incrBlockNumber(int shuffleId, int partitionId) {
+ this.partitionBlockCounters
+ .computeIfAbsent(shuffleId, x -> JavaUtils.newConcurrentMap())
+ .computeIfAbsent(partitionId, x -> new AtomicLong())
+ .incrementAndGet();
+ }
+
+ public long getBlockNumber(int shuffleId, int partitionId) {
+ Map<Integer, AtomicLong> partitionBlockCounters = this.partitionBlockCounters.get(shuffleId);
+ if (partitionBlockCounters == null) {
Review Comment:
I think we can use `getOrDefault` to simplify the following codes.
##########
server/src/main/java/org/apache/uniffle/server/ShuffleTaskInfo.java:
##########
@@ -198,6 +201,25 @@ public Set<Integer> getShuffleIds() {
return partitionDataSizes.keySet();
}
+ public void incrBlockNumber(int shuffleId, int partitionId) {
Review Comment:
incrBlockNumber -> incBlockNumber
This should be a better naming.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "EnricoMi (via GitHub)" <gi...@apache.org>.
EnricoMi commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1620605675
##########
server/src/main/java/org/apache/uniffle/server/ShuffleTaskManager.java:
##########
@@ -413,8 +413,14 @@ public void addFinishedBlockIds(
+ " bitmaps!");
}
+ ShuffleTaskInfo taskInfo = getShuffleTaskInfo(appId);
+ if (taskInfo == null) {
+ throw new InvalidRequestException(
+ "ShuffleTaskInfo is not found that should not happen for appId: " + appId);
+ }
for (Map.Entry<Integer, long[]> entry : partitionToBlockIds.entrySet()) {
Integer partitionId = entry.getKey();
+ taskInfo.incrBlockNumber(shuffleId, partitionId, entry.getValue().length);
Review Comment:
got 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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#1757] feat(server): Add block number check on getting shuffle result [incubator-uniffle]
Posted by "rickyma (via GitHub)" <gi...@apache.org>.
rickyma commented on code in PR #1758:
URL: https://github.com/apache/incubator-uniffle/pull/1758#discussion_r1621653829
##########
server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java:
##########
@@ -506,14 +506,24 @@ public void reportShuffleResult(
"appId[" + appId + "], shuffleId[" + shuffleId + "], taskAttemptId[" + taskAttemptId + "]";
try {
+ int expected =
+ partitionToBlockIds.values().stream().map(x -> x.length).reduce(0, (a, b) -> a + b);
Review Comment:
This is more readable:
`int expected = partitionToBlockIds.values().stream().mapToInt(x -> x.length).sum();`
##########
server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java:
##########
@@ -506,14 +506,24 @@ public void reportShuffleResult(
"appId[" + appId + "], shuffleId[" + shuffleId + "], taskAttemptId[" + taskAttemptId + "]";
try {
+ int expected =
Review Comment:
Maybe `expectedNum` and `updatedNum`?
##########
server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java:
##########
@@ -506,14 +506,24 @@ public void reportShuffleResult(
"appId[" + appId + "], shuffleId[" + shuffleId + "], taskAttemptId[" + taskAttemptId + "]";
try {
+ int expected =
+ partitionToBlockIds.values().stream().map(x -> x.length).reduce(0, (a, b) -> a + b);
LOG.info(
- "Report "
- + partitionToBlockIds.size()
- + " blocks as shuffle result for the task of "
- + requestInfo);
- shuffleServer
- .getShuffleTaskManager()
- .addFinishedBlockIds(appId, shuffleId, partitionToBlockIds, bitmapNum);
+ "Accepted {} blockIds report of {} partitions as shuffle result for the task of {}",
Review Comment:
I think the log could be more readable:
```
LOG.info(
"Accepted blockIds report for {} blocks across {} partitions as shuffle result for task {}",
expectedNum,
partitionToBlockIds.size(),
request);
```
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org