[SPARK-36266][SHUFFLE] Rename classes in shuffle RPC used for block push operations #33340

Victsm · 2021-07-14T06:27:47Z

What changes were proposed in this pull request?

This is a follow-up to #29855 according to the comments
In this PR, the following changes are made:

A new BlockPushingListener interface is created specifically for block push. The existing BlockFetchingListener interface is left as is, since it might be used by external shuffle solutions. These 2 interfaces are unified under BlockTransferListener to enable code reuse.
RetryingBlockFetcher, BlockFetchStarter, and RetryingBlockFetchListener are renamed to RetryingBlockTransferor, BlockTransferStarter, and RetryingBlockTransferListener respectively. This makes their names more generic to be reused across both block fetch and push.
Comments in OneForOneBlockPusher are further clarified to better explain how we handle retries for block push.

Why are the changes needed?

To make code cleaner without sacrificing backward compatibility.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Existing unit tests.

… push. Add BlockPushingListener for block push.

Victsm · 2021-07-14T06:35:43Z

This PR picks up an earlier stale version of the same change as a follow up to the original push protocol RPC patch.
cc @mridulm @otterc @Ngone51

mridulm

Took a quick pass.
Will wait for @Ngone51, @attilapiros, @dongjoon-hyun, @tgravescs, @jiangxb1987 and @otterc to also take a pass since they all reviewed the original PR.

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/BlockPushingListener.java

core/src/main/scala/org/apache/spark/network/netty/NettyBlockTransferService.scala

Victsm · 2021-07-14T21:27:49Z

Test failure is unrelated to this change.

Ngone51 · 2021-07-15T04:52:35Z

I haven't taken a deep look into code but just wondering according to the PR description. I know some code was added in 3.1 but push-based wasn't usable at that time. So, do we have to handle the compatibility issue?

I also noticed a similar issue (see #33078 (comment)) in another PR and @mridulm told me these kinds of cases don't have to be a compatibility issue.

attilapiros · 2021-07-15T07:28:37Z

I see why the adapter pattern is used here over simple renames. You also mentioned the reason: "without sacrificing backward compatibility" but could you explain the reason a little bit more in the description to have this information for the future reference.

attilapiros · 2021-07-15T07:32:46Z

@Ngone51 on compatibility issue it is meant keeping BlockFetchingListener as it is. As the BlockFetchingListener can be used by external shuffle solutions. @Victsm correct me if I am wrong.
By the way this is what I am missing from the description.

Victsm · 2021-07-15T18:02:33Z

@attilapiros you are right. The backward compatibility is meant to keep the existing BlockFetchingListener as is. Will clarify this better in the description.

otterc

Just have few comments.

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/BlockPushingListener.java

...network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockStoreClient.java

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RetryingBlockTransferor.java

...network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockStoreClient.java

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RetryingBlockTransferor.java

Ngone51 · 2021-07-16T07:43:38Z

BTW, I think it's better to use a separate JIRA id for this PR instead of a follow-up (which usually has less change).

Victsm · 2021-07-23T00:14:25Z

Created a separate ticket for this change under SPARK-36266 and addressed the additional comments.
@Ngone51 @attilapiros @otterc , please take another look.

otterc

Looks good to me. Just a nit. Thanks for working on this @Victsm

otterc · 2021-07-23T00:54:26Z

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/BlockPushingListener.java

@@ -25,6 +25,8 @@
 * code reuse for handling block push and fetch retry.
 */
 public interface BlockPushingListener extends BlockTransferListener  {


Nit: Are we keeping BlockPushingListener? My suggestion was to call it BlockPushListener.

I think BlockPushingListener is more consistent with `BlockFetchingListener.

Yeah, but BlockPushListener sounds better and is consistent with BlockTransferListener. Not a big deal though.

BlockPushingListener is supposed to be a peer to BlockFetchingListener. Their names being consistent feels more appropriate.

Ngone51 · 2021-07-23T02:28:41Z

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RetryingBlockTransferor.java

 * particular, the listener will be invoked exactly once per blockId, with a success or failure.
 */
-public class RetryingBlockFetcher {
+public class RetryingBlockTransferor {


This's also public, although I have no idea how end-users could use it directly.
Shall we mark it as private from now? cc @mridulm

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RetryingBlockTransferor.java

Ngone51 · 2021-07-23T02:36:46Z

LGTM if tests pass.

Victsm · 2021-07-23T05:00:55Z

The tests seem to have been abnormally disrupted based on what I can see from the log.
Not sure what's causing this.
cc @mridulm

Ngone51 · 2021-07-24T13:19:43Z

ok to test

SparkQA · 2021-07-24T13:27:03Z

Kubernetes integration test unable to build dist.

exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46115/

SparkQA · 2021-07-24T13:28:10Z

Test build #141598 has finished for PR 33340 at commit a590947.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

attilapiros

@Victsm I suspect the merged code which is used for running the unittests against introduces a build error:

[error] /home/jenkins/workspace/SparkPullRequestBuilder/core/src/test/scala/org/apache/spark/shuffle/ShuffleBlockPusherSuite.scala:335:76: not found: type BlockFetchingListener
[error]         val blockFetchListener = invocation.getArguments()(4).asInstanceOf[BlockFetchingListener]
[error]                                                                            ^
[error] one error found

Could you please rebase/merge this on/with top of the master and check the ShuffleBlockPusherSuite.scala after the rebase/merge?

attilapiros · 2021-07-24T15:54:04Z

Now I am sure. Line 335 in master:

spark/core/src/test/scala/org/apache/spark/shuffle/ShuffleBlockPusherSuite.scala

Line 335 in 70a1586

    
           val blockFetchListener = invocation.getArguments()(4).asInstanceOf[BlockFetchingListener]

...on/network-shuffle/src/main/java/org/apache/spark/network/shuffle/BlockFetchingListener.java

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/BlockPushingListener.java

attilapiros

Conditional LGTM: some nits (not blocking) but it must be rebased/merged onto/with master and the test should pass.

mridulm · 2021-07-25T06:31:17Z

Thanks for the reviews @Ngone51, @attilapiros, @otterc !
@Victsm Can you address @attilapiros's comments please ?

SparkQA · 2021-07-26T08:08:09Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46136/

SparkQA · 2021-07-26T08:59:50Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46136/

SparkQA · 2021-07-26T09:31:35Z

Test build #141618 has finished for PR 33340 at commit 1d40d59.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-07-26T18:33:32Z

Kubernetes integration test unable to build dist.

exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46168/

mridulm

LGTM, pending tests

SparkQA · 2021-07-26T19:48:35Z

Test build #141652 has finished for PR 33340 at commit 9c418f4.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

…ush operations ### What changes were proposed in this pull request? This is a follow-up to #29855 according to the [comments](https://github.com/apache/spark/pull/29855/files#r505536514) In this PR, the following changes are made: 1. A new `BlockPushingListener` interface is created specifically for block push. The existing `BlockFetchingListener` interface is left as is, since it might be used by external shuffle solutions. These 2 interfaces are unified under `BlockTransferListener` to enable code reuse. 2. `RetryingBlockFetcher`, `BlockFetchStarter`, and `RetryingBlockFetchListener` are renamed to `RetryingBlockTransferor`, `BlockTransferStarter`, and `RetryingBlockTransferListener` respectively. This makes their names more generic to be reused across both block fetch and push. 3. Comments in `OneForOneBlockPusher` are further clarified to better explain how we handle retries for block push. ### Why are the changes needed? To make code cleaner without sacrificing backward compatibility. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing unit tests. Closes #33340 from Victsm/SPARK-32915-followup. Lead-authored-by: Min Shen <mshen@linkedin.com> Co-authored-by: Min Shen <victor.nju@gmail.com> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com> (cherry picked from commit c4aa54e) Signed-off-by: Mridul Muralidharan <mridulatgmail.com>

mridulm · 2021-07-26T22:41:55Z

Merged to master and branch-3.2, +CC @gengliangwang
Thanks for working on this @Victsm.
Thanks for the reviews @Ngone51, @attilapiros, @otterc !

…ush operations This is a follow-up to apache#29855 according to the [comments](https://github.com/apache/spark/pull/29855/files#r505536514) In this PR, the following changes are made: 1. A new `BlockPushingListener` interface is created specifically for block push. The existing `BlockFetchingListener` interface is left as is, since it might be used by external shuffle solutions. These 2 interfaces are unified under `BlockTransferListener` to enable code reuse. 2. `RetryingBlockFetcher`, `BlockFetchStarter`, and `RetryingBlockFetchListener` are renamed to `RetryingBlockTransferor`, `BlockTransferStarter`, and `RetryingBlockTransferListener` respectively. This makes their names more generic to be reused across both block fetch and push. 3. Comments in `OneForOneBlockPusher` are further clarified to better explain how we handle retries for block push. To make code cleaner without sacrificing backward compatibility. No Existing unit tests. Closes apache#33340 from Victsm/SPARK-32915-followup. Lead-authored-by: Min Shen <mshen@linkedin.com> Co-authored-by: Min Shen <victor.nju@gmail.com> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com> (cherry picked from commit c4aa54e)

…ush operations ### What changes were proposed in this pull request? This is a follow-up to apache#29855 according to the [comments](https://github.com/apache/spark/pull/29855/files#r505536514) In this PR, the following changes are made: 1. A new `BlockPushingListener` interface is created specifically for block push. The existing `BlockFetchingListener` interface is left as is, since it might be used by external shuffle solutions. These 2 interfaces are unified under `BlockTransferListener` to enable code reuse. 2. `RetryingBlockFetcher`, `BlockFetchStarter`, and `RetryingBlockFetchListener` are renamed to `RetryingBlockTransferor`, `BlockTransferStarter`, and `RetryingBlockTransferListener` respectively. This makes their names more generic to be reused across both block fetch and push. 3. Comments in `OneForOneBlockPusher` are further clarified to better explain how we handle retries for block push. ### Why are the changes needed? To make code cleaner without sacrificing backward compatibility. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing unit tests. Closes apache#33340 from Victsm/SPARK-32915-followup. Lead-authored-by: Min Shen <mshen@linkedin.com> Co-authored-by: Min Shen <victor.nju@gmail.com> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com> (cherry picked from commit c4aa54e) Signed-off-by: Mridul Muralidharan <mridulatgmail.com>

…ush operations This is a follow-up to #29855 according to the [comments](https://github.com/apache/spark/pull/29855/files#r505536514) In this PR, the following changes are made: 1. A new `BlockPushingListener` interface is created specifically for block push. The existing `BlockFetchingListener` interface is left as is, since it might be used by external shuffle solutions. These 2 interfaces are unified under `BlockTransferListener` to enable code reuse. 2. `RetryingBlockFetcher`, `BlockFetchStarter`, and `RetryingBlockFetchListener` are renamed to `RetryingBlockTransferor`, `BlockTransferStarter`, and `RetryingBlockTransferListener` respectively. This makes their names more generic to be reused across both block fetch and push. 3. Comments in `OneForOneBlockPusher` are further clarified to better explain how we handle retries for block push. To make code cleaner without sacrificing backward compatibility. No Existing unit tests. Closes #33340 from Victsm/SPARK-32915-followup. Lead-authored-by: Min Shen <mshen@linkedin.com> Co-authored-by: Min Shen <victor.nju@gmail.com> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

Victsm added 3 commits July 13, 2021 23:09

Rename RetryingBlockFetcher to be more generic across block fetch and…

a033dbf

… push. Add BlockPushingListener for block push.

Fix unit test failures

1052790

Fix unit test failures

994f66e

github-actions bot added the CORE label Jul 14, 2021

Victsm changed the title ~~[SPARK-32915][SHUFFLE][FOLLOW-UP] Rename classes in shuffle RPC used for block push operations~~ [WIP][SPARK-32915][SHUFFLE][FOLLOW-UP] Rename classes in shuffle RPC used for block push operations Jul 14, 2021

Victsm added 2 commits July 14, 2021 00:19

Rebase with master

2c0b5c9

Fix UT failures

3dff827

Victsm changed the title ~~[WIP][SPARK-32915][SHUFFLE][FOLLOW-UP] Rename classes in shuffle RPC used for block push operations~~ [SPARK-32915][SHUFFLE][FOLLOW-UP] Rename classes in shuffle RPC used for block push operations Jul 14, 2021

mridulm reviewed Jul 14, 2021

View reviewed changes

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/BlockPushingListener.java Show resolved Hide resolved

core/src/main/scala/org/apache/spark/network/netty/NettyBlockTransferService.scala Show resolved Hide resolved

Address review comments

c4b655b

otterc reviewed Jul 16, 2021

View reviewed changes

Ngone51 reviewed Jul 16, 2021

View reviewed changes

Victsm added 2 commits July 22, 2021 16:56

Merge branch 'master' into SPARK-32915-followup

0839512

Address additional review comments.

8410b85

Victsm changed the title ~~[SPARK-32915][SHUFFLE][FOLLOW-UP] Rename classes in shuffle RPC used for block push operations~~ [SPARK-36266][SHUFFLE] Rename classes in shuffle RPC used for block push operations Jul 23, 2021

otterc reviewed Jul 23, 2021

View reviewed changes

Ngone51 reviewed Jul 23, 2021

View reviewed changes

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RetryingBlockTransferor.java Outdated Show resolved Hide resolved

Fix comment wording.

a951368

Minor change to variable names to be consistent.

efba5d1

Minor revision to log message to be consistent.

a590947

attilapiros reviewed Jul 24, 2021

View reviewed changes

...on/network-shuffle/src/main/java/org/apache/spark/network/shuffle/BlockFetchingListener.java Outdated Show resolved Hide resolved

attilapiros reviewed Jul 24, 2021

View reviewed changes

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/BlockPushingListener.java Outdated Show resolved Hide resolved

attilapiros reviewed Jul 24, 2021

View reviewed changes

Victsm and others added 2 commits July 25, 2021 23:49

Merge branch 'apache:master' into SPARK-32915-followup

f1db4dd

Rebase with master

1d40d59

Address review comment and fix UT failures.

9c418f4

mridulm approved these changes Jul 26, 2021

View reviewed changes

asfgit closed this in c4aa54e Jul 26, 2021

pan3793 mentioned this pull request Dec 8, 2021

[SPARK-37574][CORE][SHUFFLE] Simplify fetchBlocks w/o retry #34831

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-36266][SHUFFLE] Rename classes in shuffle RPC used for block push operations #33340

[SPARK-36266][SHUFFLE] Rename classes in shuffle RPC used for block push operations #33340

Victsm commented Jul 14, 2021 •

edited

Loading

Victsm commented Jul 14, 2021

mridulm left a comment

Victsm commented Jul 14, 2021

Ngone51 commented Jul 15, 2021 •

edited

Loading

attilapiros commented Jul 15, 2021

attilapiros commented Jul 15, 2021

Victsm commented Jul 15, 2021

otterc left a comment

Ngone51 commented Jul 16, 2021

Victsm commented Jul 23, 2021

otterc left a comment

otterc Jul 23, 2021

Ngone51 Jul 23, 2021

otterc Jul 23, 2021

Victsm Jul 23, 2021

Ngone51 Jul 23, 2021

Ngone51 commented Jul 23, 2021

Victsm commented Jul 23, 2021

Ngone51 commented Jul 24, 2021

SparkQA commented Jul 24, 2021

SparkQA commented Jul 24, 2021

attilapiros left a comment •

edited

Loading

attilapiros commented Jul 24, 2021

attilapiros left a comment

mridulm commented Jul 25, 2021

SparkQA commented Jul 26, 2021

SparkQA commented Jul 26, 2021

SparkQA commented Jul 26, 2021

SparkQA commented Jul 26, 2021

mridulm left a comment

SparkQA commented Jul 26, 2021

mridulm commented Jul 26, 2021

[SPARK-36266][SHUFFLE] Rename classes in shuffle RPC used for block push operations #33340

[SPARK-36266][SHUFFLE] Rename classes in shuffle RPC used for block push operations #33340

Conversation

Victsm commented Jul 14, 2021 • edited Loading

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Victsm commented Jul 14, 2021

mridulm left a comment

Choose a reason for hiding this comment

Victsm commented Jul 14, 2021

Ngone51 commented Jul 15, 2021 • edited Loading

attilapiros commented Jul 15, 2021

attilapiros commented Jul 15, 2021

Victsm commented Jul 15, 2021

otterc left a comment

Choose a reason for hiding this comment

Ngone51 commented Jul 16, 2021

Victsm commented Jul 23, 2021

otterc left a comment

Choose a reason for hiding this comment

otterc Jul 23, 2021

Choose a reason for hiding this comment

Ngone51 Jul 23, 2021

Choose a reason for hiding this comment

otterc Jul 23, 2021

Choose a reason for hiding this comment

Victsm Jul 23, 2021

Choose a reason for hiding this comment

Ngone51 Jul 23, 2021

Choose a reason for hiding this comment

Ngone51 commented Jul 23, 2021

Victsm commented Jul 23, 2021

Ngone51 commented Jul 24, 2021

SparkQA commented Jul 24, 2021

SparkQA commented Jul 24, 2021

attilapiros left a comment • edited Loading

Choose a reason for hiding this comment

attilapiros commented Jul 24, 2021

attilapiros left a comment

Choose a reason for hiding this comment

mridulm commented Jul 25, 2021

SparkQA commented Jul 26, 2021

SparkQA commented Jul 26, 2021

SparkQA commented Jul 26, 2021

SparkQA commented Jul 26, 2021

mridulm left a comment

Choose a reason for hiding this comment

SparkQA commented Jul 26, 2021

mridulm commented Jul 26, 2021

Victsm commented Jul 14, 2021 •

edited

Loading

Ngone51 commented Jul 15, 2021 •

edited

Loading

attilapiros left a comment •

edited

Loading