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/04/01 07:06:07 UTC
Re: [PR] [#1373][FOLLOWUP] fix(spark): shuffle manager rpc service invalid when partition data reassign is enabled [incubator-uniffle]
zuston commented on code in PR #1583:
URL: https://github.com/apache/incubator-uniffle/pull/1583#discussion_r1546031525
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java:
##########
@@ -137,6 +137,8 @@ public class RssShuffleManager extends RssShuffleManagerBase {
private Map<Integer, ShuffleHandleInfo> shuffleIdToShuffleHandleInfo;
/** Whether to enable the dynamic shuffleServer function rewrite and reread functions */
private boolean rssResubmitStage;
+
+ private boolean taskFailureRetryEnabled;
Review Comment:
`taskFailureRetryEnabled` -> `taskBlockSendFailureRetry`
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/writer/RssShuffleWriter.java:
##########
@@ -112,7 +112,7 @@ public class RssShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
private final Set<Long> blockIds = Sets.newConcurrentHashSet();
private TaskContext taskContext;
private SparkConf sparkConf;
- private boolean taskFailRetry;
+ private boolean taskFailureRetryEnabled;
Review Comment:
`taskFailureRetryEnabled` -> `blockSendFailureRetryEnabled`
##########
client-spark/common/src/main/java/org/apache/uniffle/shuffle/manager/RssShuffleManagerBase.java:
##########
@@ -488,4 +490,20 @@ protected static RemoteStorageInfo getDefaultRemoteStorageInfo(SparkConf sparkCo
return new RemoteStorageInfo(
sparkConf.get(RssSparkConfig.RSS_REMOTE_STORAGE_PATH.key(), ""), confItems);
}
+
+ protected static ShuffleHandleInfo getShuffleHandleInfo(
Review Comment:
Please revert this
##########
common/src/main/java/org/apache/uniffle/common/config/RssClientConf.java:
##########
@@ -203,8 +203,8 @@ public class RssClientConf {
"This option is only valid when the remote storage path is specified. If ture, "
+ "the remote storage conf will use the client side hadoop configuration loaded from the classpath.");
- public static final ConfigOption<Boolean> RSS_TASK_FAILED_RETRY_ENABLED =
- ConfigOptions.key("rss.task.failed.retry.enabled")
+ public static final ConfigOption<Boolean> RSS_TASK_FAILURE_RETRY_ENABLED =
+ ConfigOptions.key("rss.task.failure.retry.enabled")
Review Comment:
`rss.task.failure.retry.enabled` -> `rss.client.blockSendFailureRetry.enabled`
##########
client-spark/spark2/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java:
##########
@@ -128,6 +128,8 @@ public class RssShuffleManager extends RssShuffleManagerBase {
JavaUtils.newConcurrentMap();
/** Whether to enable the dynamic shuffleServer function rewrite and reread functions */
private boolean rssResubmitStage;
+
+ private boolean taskFailureRetryEnabled;
Review Comment:
`taskFailureRetryEnabled` -> `taskBlockFailureRetryEnabled`
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java:
##########
@@ -471,7 +478,7 @@ public <K, V, C> ShuffleHandle registerShuffle(
shuffleIdToPartitionNum.putIfAbsent(shuffleId, dependency.partitioner().numPartitions());
shuffleIdToNumMapTasks.putIfAbsent(shuffleId, dependency.rdd().partitions().length);
- if (rssResubmitStage) {
+ if (rssResubmitStage || taskFailureRetryEnabled) {
Review Comment:
ditto
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java:
##########
@@ -508,15 +515,9 @@ public <K, V> ShuffleWriter<K, V> getWriter(
} else {
writeMetrics = context.taskMetrics().shuffleWriteMetrics();
}
- ShuffleHandleInfo shuffleHandleInfo;
- if (rssResubmitStage) {
- // Get the ShuffleServer list from the Driver based on the shuffleId
- shuffleHandleInfo = getRemoteShuffleHandleInfo(shuffleId);
- } else {
- shuffleHandleInfo =
- new ShuffleHandleInfo(
- shuffleId, rssHandle.getPartitionToServers(), rssHandle.getRemoteStorage());
- }
+ ShuffleHandleInfo shuffleHandleInfo =
Review Comment:
revert this part
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java:
##########
@@ -239,10 +241,15 @@ public RssShuffleManager(SparkConf conf, boolean isDriver) {
this.rssResubmitStage =
rssConf.getBoolean(RssClientConfig.RSS_RESUBMIT_STAGE, false)
&& RssSparkShuffleUtils.isStageResubmitSupported();
+ this.taskFailureRetryEnabled =
+ sparkConf.getBoolean(
+ RssSparkConfig.SPARK_RSS_CONFIG_PREFIX
+ + RssClientConf.RSS_TASK_FAILURE_RETRY_ENABLED.key(),
+ RssClientConf.RSS_TASK_FAILURE_RETRY_ENABLED.defaultValue());
if (isDriver) {
heartBeatScheduledExecutorService =
ThreadUtils.getDaemonSingleThreadScheduledExecutor("rss-heartbeat");
- if (rssResubmitStage) {
+ if (rssResubmitStage || taskFailureRetryEnabled) {
Review Comment:
replace with `if (shuffleManagerRpcServiceEnabled)`
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java:
##########
@@ -239,10 +241,15 @@ public RssShuffleManager(SparkConf conf, boolean isDriver) {
this.rssResubmitStage =
rssConf.getBoolean(RssClientConfig.RSS_RESUBMIT_STAGE, false)
&& RssSparkShuffleUtils.isStageResubmitSupported();
+ this.taskFailureRetryEnabled =
+ sparkConf.getBoolean(
+ RssSparkConfig.SPARK_RSS_CONFIG_PREFIX
+ + RssClientConf.RSS_TASK_FAILURE_RETRY_ENABLED.key(),
+ RssClientConf.RSS_TASK_FAILURE_RETRY_ENABLED.defaultValue());
Review Comment:
And here you can introduce the new var to indicate whether we need to start shuffle manager rpc service.Like this:
`this.shuffleManagerRpcServiceEnabled = taskBlockSendFailureRetry || rssResubmitStage
`
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java:
##########
@@ -239,10 +241,15 @@ public RssShuffleManager(SparkConf conf, boolean isDriver) {
this.rssResubmitStage =
rssConf.getBoolean(RssClientConfig.RSS_RESUBMIT_STAGE, false)
&& RssSparkShuffleUtils.isStageResubmitSupported();
+ this.taskFailureRetryEnabled =
Review Comment:
this.taskBlockSendFailureRetry = rssConf.getBoolean(RssClientConf.RSS_TASK_FAILURE_RETRY_ENABLED);
--
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