You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@uniffle.apache.org by xi...@apache.org on 2023/10/31 06:30:00 UTC
(incubator-uniffle) branch branch-0.8 updated (cdb38ae78 -> bb5676832)
This is an automated email from the ASF dual-hosted git repository.
xianjingfeng pushed a change to branch branch-0.8
in repository https://gitbox.apache.org/repos/asf/incubator-uniffle.git
from cdb38ae78 [#1254] fix(test): Fix the flaky test RssShuffleTest. (#1255)
new f6eae7e94 [#1261] fix(spark): Throw out InterruptedException for sleep in requestExecutorMemory #1262
new 657919e0c [#1254][FOLLOWUP] fix(test): Fix the flaky test RssShuffleTest. (#1259)
new bb5676832 [#1252] fix(server): Incorrect storage write fail metric (#1253)
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../spark/shuffle/writer/WriteBufferManager.java | 2 +-
.../shuffle/orderedgrouped/RssShuffleTest.java | 2 +-
.../apache/uniffle/server/ShuffleFlushManager.java | 93 +++++++++++++---------
3 files changed, 59 insertions(+), 38 deletions(-)
(incubator-uniffle) 02/03: [#1254][FOLLOWUP] fix(test): Fix the flaky test RssShuffleTest. (#1259)
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xianjingfeng pushed a commit to branch branch-0.8
in repository https://gitbox.apache.org/repos/asf/incubator-uniffle.git
commit 657919e0cedea1eed00f57ca40a42b5bfa7db2fb
Author: zhengchenyu <zh...@163.com>
AuthorDate: Tue Oct 24 14:05:02 2023 +0800
[#1254][FOLLOWUP] fix(test): Fix the flaky test RssShuffleTest. (#1259)
### What changes were proposed in this pull request?
### Why are the changes needed?
Fix: #1254
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
unit test
(cherry picked from commit 3071099338ad76722745468391871d5124adef54)
---
.../runtime/library/common/shuffle/orderedgrouped/RssShuffleTest.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/client-tez/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/RssShuffleTest.java b/client-tez/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/RssShuffleTest.java
index b352f1015..39462bf3a 100644
--- a/client-tez/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/RssShuffleTest.java
+++ b/client-tez/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/RssShuffleTest.java
@@ -87,11 +87,11 @@ public class RssShuffleTest {
conf.setLong(Constants.TEZ_RUNTIME_TASK_MEMORY, 300000L);
RssShuffle shuffle = new RssShuffle(inputContext, conf, 1, 3000000L, 0, APPATTEMPT_ID);
try {
- shuffle.run();
ShuffleScheduler scheduler = shuffle.rssScheduler;
MergeManager mergeManager = shuffle.merger;
assertFalse(scheduler.isShutdown());
assertFalse(mergeManager.isShutdown());
+ shuffle.run();
String exceptionMessage = "Simulating fetch failure";
shuffle.reportException(new IOException(exceptionMessage));
(incubator-uniffle) 01/03: [#1261] fix(spark): Throw out InterruptedException for sleep in requestExecutorMemory #1262
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xianjingfeng pushed a commit to branch branch-0.8
in repository https://gitbox.apache.org/repos/asf/incubator-uniffle.git
commit f6eae7e94bc4b56fd71dfa86602fd7cbea5aa0bb
Author: Zhen Wang <64...@qq.com>
AuthorDate: Tue Oct 24 13:37:07 2023 +0800
[#1261] fix(spark): Throw out InterruptedException for sleep in requestExecutorMemory #1262
### What changes were proposed in this pull request?
Pass InterruptedException in requestExecutorMemory.
### Why are the changes needed?
Fix: #1261
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing tests.
(cherry picked from commit 9dd4c4df3617e72288ae77b58ac8907229e9e9ef)
---
.../main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java b/client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java
index 27fb40a21..027be97af 100644
--- a/client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java
+++ b/client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java
@@ -365,7 +365,7 @@ public class WriteBufferManager extends MemoryConsumer {
try {
Thread.sleep(requireMemoryInterval);
} catch (InterruptedException ie) {
- LOG.warn("Exception happened when waiting for memory.", ie);
+ throw new RssException("Interrupted when waiting for memory.", ie);
}
gotMem = acquireMemory(askExecutorMemory);
allocatedBytes.addAndGet(gotMem);
(incubator-uniffle) 03/03: [#1252] fix(server): Incorrect storage write fail metric (#1253)
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xianjingfeng pushed a commit to branch branch-0.8
in repository https://gitbox.apache.org/repos/asf/incubator-uniffle.git
commit bb56768325347940aa0e3c21e9ae78a58b4a5d46
Author: summaryzb <su...@gmail.com>
AuthorDate: Wed Oct 25 19:35:58 2023 +0800
[#1252] fix(server): Incorrect storage write fail metric (#1253)
What changes were proposed in this pull request?
As title
Why are the changes needed?
Fix: #1252
Does this PR introduce any user-facing change?
No.
How was this patch tested?
unit test
(cherry picked from commit 931d6cda0e3a691867721f51eabb567d568831e8)
---
.../apache/uniffle/server/ShuffleFlushManager.java | 93 +++++++++++++---------
1 file changed, 57 insertions(+), 36 deletions(-)
diff --git a/server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java b/server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java
index ba46dfab0..eec574179 100644
--- a/server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java
+++ b/server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java
@@ -83,37 +83,41 @@ public class ShuffleFlushManager {
eventHandler.handle(event);
}
+ private void recordFinalFail(ShuffleDataFlushEvent event, long start) {
+ LOG.error(
+ "Failed to write data for {} in {} times, shuffle data will be lost", event, retryMax);
+ if (event.getUnderStorage() != null) {
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ }
+ event.doCleanup();
+ if (shuffleServer != null) {
+ long duration = System.currentTimeMillis() - start;
+ ShuffleServerMetrics.counterTotalFailedWrittenEventNum.inc();
+ LOG.error(
+ "Flush to file for {} failed in {} ms and release {} bytes",
+ event,
+ duration,
+ event.getSize());
+ }
+ }
+
+ private void recordSuccess(ShuffleDataFlushEvent event, long start) {
+ updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), event.getShuffleBlocks());
+ ShuffleServerMetrics.incStorageSuccessCounter(event.getUnderStorage().getStorageHost());
+ event.doCleanup();
+ if (shuffleServer != null) {
+ if (LOG.isDebugEnabled()) {
+ long duration = System.currentTimeMillis() - start;
+ LOG.debug("Flush to file success in {} ms and release {} bytes", duration, event.getSize());
+ }
+ }
+ }
+
public void processEvent(ShuffleDataFlushEvent event) {
try {
ShuffleServerMetrics.gaugeWriteHandler.inc();
- long start = System.currentTimeMillis();
- boolean writeSuccess = flushToFile(event);
- if (writeSuccess || event.getRetryTimes() > retryMax) {
- if (event.getRetryTimes() > retryMax) {
- LOG.error(
- "Failed to write data for {} in {} times, shuffle data will be lost",
- event,
- retryMax);
- if (event.getUnderStorage() != null) {
- ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
- }
- }
- event.doCleanup();
- if (shuffleServer != null) {
- long duration = System.currentTimeMillis() - start;
- if (writeSuccess) {
- LOG.debug(
- "Flush to file success in {} ms and release {} bytes", duration, event.getSize());
- } else {
- ShuffleServerMetrics.counterTotalFailedWrittenEventNum.inc();
- LOG.error(
- "Flush to file for {} failed in {} ms and release {} bytes",
- event,
- duration,
- event.getSize());
- }
- }
- }
+ flushToFile(event);
+ // for thread safety we should not use or change any event info when write to file is failed
} catch (Exception e) {
LOG.error("Exception happened when flush data for " + event, e);
} finally {
@@ -122,7 +126,12 @@ public class ShuffleFlushManager {
}
}
+ private boolean reachRetryMax(ShuffleDataFlushEvent event) {
+ return event.getRetryTimes() > retryMax;
+ }
+
private boolean flushToFile(ShuffleDataFlushEvent event) {
+ long start = System.currentTimeMillis();
boolean writeSuccess = false;
try {
@@ -158,7 +167,7 @@ public class ShuffleFlushManager {
if (!storage.canWrite()) {
// todo: Could we add an interface supportPending for storageManager
// to unify following logic of multiple different storage managers
- if (event.getRetryTimes() <= retryMax) {
+ if (!reachRetryMax(event)) {
if (event.isPended()) {
LOG.error(
"Drop this event directly due to already having entered pending queue. event: {}",
@@ -166,9 +175,15 @@ public class ShuffleFlushManager {
return true;
}
event.increaseRetryTimes();
- ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
event.markPended();
- eventHandler.handle(event);
+ if (!reachRetryMax(event)) {
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ eventHandler.handle(event);
+ } else {
+ recordFinalFail(event, start);
+ }
+ } else {
+ recordFinalFail(event, start);
}
return false;
}
@@ -194,23 +209,29 @@ public class ShuffleFlushManager {
ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
writeSuccess = storageManager.write(storage, handler, event);
if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
- } else if (event.getRetryTimes() <= retryMax) {
+ recordSuccess(event, start);
+ } else if (!reachRetryMax(event)) {
if (event.isPended()) {
LOG.error(
"Drop this event directly due to already having entered pending queue. event: {}",
event);
}
event.increaseRetryTimes();
- ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
event.markPended();
- eventHandler.handle(event);
+ if (!reachRetryMax(event)) {
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ eventHandler.handle(event);
+ } else {
+ recordFinalFail(event, start);
+ }
}
} catch (Throwable throwable) {
// just log the error, don't throw the exception and stop the flush thread
LOG.error("Exception happened when process flush shuffle data for {}", event, throwable);
event.increaseRetryTimes();
+ if (reachRetryMax(event)) {
+ recordFinalFail(event, start);
+ }
}
return writeSuccess;
}